Logstash 详解：复杂日志处理的语法与方法（实践记录）

Logstash 作为一种强大的开源数据处理管道工具，能够高效地解析、过滤和传输日志数据，广泛应用于 ELK（Elasticsearch、Logstash、Kibana）栈中。

本文将详细介绍 Logstash 的核心语法与方法，帮助你在复杂日志处理场景中更好地发挥其强大功能。

一. Logstash 配置基础

Logstash 的配置文件由三个主要部分组成：input、filter 和 output。input 部分定义了日志的来源，filter 部分用于解析和处理日志数据，而 output 部分则决定了数据的去向。

一个简单的 Logstash 配置文件示例如下：

input {
  file {
path => "/var/log/syslog"
    start_position => "beginning"
  }
}

filter {
  grok {
match => { "message" => "%{SYSLOGLINE}" }
  }
}

output {
  elasticsearch {
hosts => ["localhost:9200"]
    index => "syslog-%{+YYYY.MM.dd}"
  }
}

在这个示例中，Logstash 从 /var/log/syslog 文件读取日志，使用 grok 过滤器解析 syslog 格式的日志数据，并将结果输出到 Elasticsearch。

二. Filters：复杂日志处理的核心

Logstash 提供了丰富的 filter 插件，用于处理不同格式的日志数据，并对其进行复杂的转换和增强。

1. Grok：提取结构化数据

grok 是 Logstash 中最常用的过滤器之一，专用于从非结构化数据中提取结构化字段。grok 使用类似正则表达式的模式匹配语言，允许你从日志消息中提取有意义的信息。

filter {
  grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
}

在这个示例中，grok 使用内置的 COMBINEDAPACHELOG 模式来解析 Apache 的访问日志。Logstash 提供了大量预定义的 grok 模式，能够处理常见的日志格式。

2. Mutate：数据转换与增强

mutate 过滤器用于对日志事件的字段进行修改，例如重命名字段、移除不必要的数据、类型转换等。

filter {
  mutate {
    rename => { "clientip" => "client_ip" }
    convert => { "response_bytes" => "integer" }
    remove_field => [ "unnecessary_field" ]
  }
}

通过 mutate 过滤器，你可以对数据进行必要的调整，使其符合后续处理和分析的需求。

3. Date：时间戳处理

日志数据中的时间戳通常使用多种格式表示，而 Logstash 的 date 过滤器能够将这些不同格式的时间戳解析为标准的 @timestamp 字段。

filter {
  date {
match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
    target => "@timestamp"
  }
}

这个配置将日志中的 timestamp 字段解析为标准时间格式，并覆盖默认的 @timestamp 字段，从而在分析时使用统一的时间基准。

4. Ruby：自定义逻辑

如果内置的过滤器无法满足需求，Logstash 允许在配置文件中嵌入 Ruby 代码，提供更高的灵活性。

filter {
  ruby {
code => "event.set('combined_field', event.get('field1') + ' ' + event.get('field2'))"
  }
}

这个示例中，使用 Ruby 代码将两个字段的内容组合为一个新的字段。通过 ruby 过滤器，你可以实现几乎无限的自定义逻辑，处理复杂的场景。

5. Aggregate：多事件聚合

对于需要跨多个事件处理的数据，可以使用 aggregate 过滤器。例如，在处理多行日志时，可以使用 aggregate 将它们组合成一个事件。

filter {
  aggregate {
    task_id => "%{some_id}"
    code => "map['count'] ||= 0; map['count'] += 1"
    push_map_as_event_on_timeout => true
    timeout => 120
  }
}

这个配置将在 120 秒内聚合同一个 task_id 的日志行，并生成一个包含计数的事件。这在处理复杂日志数据时非常有用。

三. 条件语句与多事件处理

Logstash 支持使用条件语句根据事件内容选择性地应用不同的过滤器或输出插件，这使得配置文件可以更加灵活。

filter {
if [source] == "apache_logs" {
    grok {
match => { "message" => "%{COMMONAPACHELOG}" }
    }
  } else if [source] == "syslog" {
    grok {
match => { "message" => "%{SYSLOGLINE}" }
    }
  }
}