4. Logstash管道深入

Logstash 的核心是一条可编排的事件处理管道。每个事件（Event）从 input 进入，被 filter 逐层加工，再由 output 写出。理解这条管道的三个阶段，以及 grok、mutate、date、geoip 等过滤器的组合方式，是让日志从“能收进来”变成“能查、能算、能告警”的关键。

三阶段管道的职责边界

input 负责接入与解码，决定数据从哪里来、以什么格式进入管道；filter 负责解析、改写、补全与裁剪，决定事件最终长什么样；output 负责投递，决定事件写到哪里。三者之间传递的是同一个 Event 对象，filter 阶段对字段的增删改会直接影响 output 的写入结果，因此字段命名与类型必须在 filter 阶段就统一。

一个典型的接入配置如下，input 使用 beats 接收 Filebeat 采集的 Nginx 访问日志，codec 为 json，表示上游已按 JSON 编码：

input {
  beats {
    port => 5044
    codec => json
  }
}

如果上游发送的是纯文本行，则应使用 plain codec，并在 filter 中用 grok 解析。codec 选错会导致事件结构错位，这是排障时首先要确认的点。

grok：把非结构化文本切成字段

grok 用正则模式把一行日志拆成命名捕获组。Nginx 默认 combined 格式的日志行可以这样解析：

filter {
  grok {
    match => {
      "message" => "%{IPORHOST:client_ip} - %{DATA:ident} \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{DATA:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:status:int} %{NUMBER:bytes:int}"
    }
  }
}

其中 %{NUMBER:status:int} 的 :int 后缀表示把捕获值转换为整数，避免后续在 Elasticsearch 中做数值比较时因字符串类型失效。grok 的常见性能陷阱是模式过于宽泛或回溯过多，例如用 .* 匹配长文本。应优先使用内置模式（IPORHOST、HTTPDATE、NUMBER 等），它们经过优化且可读性更好。若单条 grok 无法覆盖多种日志格式，可用多个 match 分支或 break_on_match 控制匹配顺序，但分支越多，单事件 CPU 开销越高。

mutate：字段的整理与类型规整

grok 解析后，字段名往往带有原始日志的痕迹，需要 mutate 做统一。常见操作包括 rename、remove、convert、lowercase、strip 和 gsub。例如把 client_ip 重命名为 source.ip，删除无用的 ident 字段，并把 request 中的查询串剥离：

filter {
  mutate {
    rename => { "client_ip" => "source.ip" }
    remove_field => [ "ident", "http_version" ]
    gsub => [ "request", "\?.*", "" ]
    convert => { "bytes" => "integer" }
  }
}

mutate 的执行顺序是固定的：coerce、rename、update、replace、convert、gsub、uppercase、lowercase、strip、remove、split、join、merge、copy。如果同一字段既要 rename 又要 convert，必须清楚 rename 先于 convert 执行，否则会因字段名不匹配而静默失败。mutate 是轻量操作，但字段数量多时仍会累积开销，建议只保留检索和聚合真正需要的字段。

date：让时间戳可被正确排序

日志中的时间字符串必须被解析为 @timestamp，否则 Elasticsearch 会使用 Logstash 接收时间，导致历史回放或延迟数据的时间轴错乱。Nginx 的 HTTPDATE 格式对应如下配置：

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

date 过滤器解析失败时默认不会丢弃事件，而是把 _dateparsefailure 标签写入 tags。应在 output 前检查该标签，或使用 tag_on_failure 自定义标签名，便于后续分流。时区必须显式指定，否则 Logstash 按系统时区解释，跨区域部署时会出现数小时的偏移。

geoip：把 IP 补全为地理信息

geoip 根据 IP 查询 MaxMind 数据库，补充国家、城市、经纬度等字段。它通常接在 source.ip 之后：

filter {
  geoip {
    source => "source.ip"
    target => "geo"
    fields => [ "country_name", "city_name", "location" ]
  }
}

geoip 是查表操作，单次开销不大，但每条事件都查会显著增加 CPU。若只需国家维度，应通过 fields 限制返回字段，减少事件体积。数据库文件需定期更新，否则新分配的 IP 段会解析为空。对于内网 IP，geoip 会返回空结果，可先用条件判断跳过：

if [source.ip] !~ /^(10\.|172\.(1[6-9]|2[0-9]|3[01])\.|192\.168\.)/ {
  geoip { source => "source.ip" target => "geo" }
}

过滤器组合与性能注意事项

多个 filter 按声明顺序串行执行，前一个的输出是后一个的输入。合理的顺序是：先 grok 解析，再 mutate 规整字段，然后 date 修正时间，最后 geoip 补全。若把 geoip 放在 grok 之前，source.ip 尚不存在，geoip 会直接跳过。

性能方面，filter 是管道中最容易成为瓶颈的阶段。grok 的正则回溯、geoip 的逐条查表、mutate 的大量字段操作都会消耗 CPU。可通过以下方式控制：一是用条件判断缩小 filter 作用范围，例如只对 [type] == "nginx" 的事件执行 grok；二是减少不必要的字段，output 前用 mutate 删除中间字段；三是把复杂解析尽量前移到采集端，Filebeat 的 processors 可以完成部分字段裁剪；四是监控 Logstash 的 logstash_filter_duration_seconds 指标，定位耗时最高的过滤器。

output：投递与索引策略

output 决定事件的最终去向。写入 Elasticsearch 时，索引名通常按日期或类型划分：

output {
  elasticsearch {
    hosts => ["http://es-node:9200"]
    index => "nginx-access-%{+YYYY.MM.dd}"
    user => "logstash_writer"
    password => "${ES_PWD}"
  }
}

索引名中的日期来自 @timestamp，因此 date 过滤器是否正确直接决定索引是否落到预期日期。若同一管道需要分流，可用条件判断：

output {
  if [status] >= 500 {
    elasticsearch { index => "nginx-error-%{+YYYY.MM.dd}" }
  } else {
    elasticsearch { index => "nginx-access-%{+YYYY.MM.dd}" }
  }
}

output 阶段同样有性能考量：批量写入的 flush_size 与 idle_flush_time 影响吞吐与延迟；过多 output 分支会增加事件复制与网络开销。生产环境建议只保留必要的 output，调试用的 stdout 在验证完成后移除。

把三阶段串起来看，input 保证数据进得来，filter 保证字段对得上，output 保证结果落得准。grok、mutate、date、geoip 的组合顺序与作用范围，决定了管道的正确性与吞吐能力；而条件判断、字段裁剪与指标监控，则是让管道在数据量增长后仍然稳定的关键手段。