3、日志采集工具（Logstash）：Pipeline 原理与插件实战

说到日志采集，很多人第一反应就是 Filebeat 或者 Fluentd。但如果你需要做复杂的日志清洗、字段提取、格式转换，Logstash 才是真正的重武器。我最早接触 Logstash 是在一次电商大促的日志迁移项目中，那时候每天几十 GB 的访问日志要实时解析成结构化数据，Filebeat 搞不定，最后全靠 Logstash 的 Pipeline 扛了下来。

说白了，Logstash 就是一个数据管道。它把日志从源头拉进来，中间做各种加工，最后送到目的地。这个“拉进来→加工→送出去”的过程，就是它的 Pipeline 模型。

3.1 Logstash Pipeline 原理

Logstash 的 Pipeline 由三个阶段组成：Input、Filter、Output。每个阶段都可以配置多个插件，数据会按顺序流过这些插件。

核心流程： Input → Filter → Output

数据从 Input 进入，经过 Filter 处理，最后由 Output 输出。如果 Filter 阶段没有配置，数据会原样输出。

我习惯把 Pipeline 想象成一条流水线。Input 是上料口，Filter 是加工车间，Output 是打包发货。每个车间可以有多道工序，工序之间可以串联也可以并联。

Logstash Pipeline 数据流 Input 阶段 Filter 阶段 Output 阶段 常见 Input 插件 file / beats / tcp syslog / kafka http / jdbc 常见 Filter 插件 grok / mutate date / geoip kv / json / csv 常见 Output 插件 elasticsearch kafka / file stdout / csv 数据在 Pipeline 中按顺序流经每个阶段，每个阶段可配置多个插件

Pipeline 的配置格式很简单，就是一个配置文件里写三个区块：

input {
  # 输入插件配置
}

filter {
  # 过滤插件配置
}

output {
  # 输出插件配置
}

嗯，这里要注意：Logstash 是按顺序执行 Filter 的。如果你在 Filter 里先用了 grok 提取字段，后面再用 mutate 修改字段，顺序不能乱。我见过有人把 mutate 写在 grok 前面，结果字段还没提取出来就试图修改，自然就失败了。

3.2 Input 插件：从哪里拿数据

Input 插件决定了 Logstash 从哪里读取数据。常用的有这些：

插件名称 适用场景 典型配置
file 读取本地日志文件 path => "/var/log/*.log"
beats 接收 Filebeat 发送的数据 port => 5044
tcp 接收 TCP 协议的数据 port => 4560
syslog 接收系统日志 port => 514
kafka 从 Kafka 消费数据 bootstrap_servers => "localhost:9092"
http 通过 HTTP 接口接收数据 port => 8080

我个人最常用的是 beats 和 kafka。beats 适合从服务器上采集日志文件，kafka 适合在微服务架构中做数据中转。如果你只是本地调试，用 file 插件就够了。

小技巧：调试时可以在 input 里加一个 stdin { }，这样可以直接在终端输入日志，方便测试 Filter 配置是否正确。

3.3 Filter 插件：日志的加工车间

Filter 是 Logstash 的灵魂。没有 Filter，Logstash 就是个搬运工。有了 Filter，它才能变成数据清洗专家。

常用的 Filter 插件有这些：

  • grok：正则解析，把非结构化日志变成结构化字段
  • mutate：修改字段，比如重命名、删除、类型转换
  • date：解析时间戳，把字符串时间转成 Logstash 内部时间
  • geoip：根据 IP 地址获取地理位置信息
  • kv：解析 key=value 格式的日志
  • json：解析 JSON 格式的日志
  • csv：解析 CSV 格式的日志

为什么 Filter 这么重要？你想想看，原始日志长这样：

192.168.1.1 - - [10/Oct/2023:13:55:36 +0800] "GET /api/user HTTP/1.1" 200 1234

经过 Filter 处理后，它应该变成这样：

{
  "clientip": "192.168.1.1",
  "timestamp": "2023-10-10T13:55:36+08:00",
  "method": "GET",
  "path": "/api/user",
  "status": 200,
  "bytes": 1234
}

只有变成结构化数据，Elasticsearch 才能做聚合分析，Kibana 才能做可视化。这就是 Filter 的价值所在。

3.4 Grok 正则解析实战

Grok 是 Logstash 里最强大的 Filter 插件，也是学习曲线最陡的一个。说白了，Grok 就是用正则表达式从日志里提取字段。

Grok 的语法格式是：

%{PATTERN_NAME:field_name}

其中 PATTERN_NAME 是预定义的正则模式，field_name 是你给提取结果起的字段名。

举个例子，解析 Nginx 访问日志：

filter {
  grok {
    match => {
      "message" => "%{IP:clientip} - - \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:status} %{NUMBER:bytes}"
    }
  }
}

这段配置会把 message 字段里的日志解析成 clientip、timestamp、method、request、httpversion、status、bytes 七个字段。

我曾经踩过的坑：Grok 模式匹配失败时，Logstash 不会报错，而是会在日志里加一个 _grokparsefailure 标签。如果你发现数据没有解析成功，记得去 Elasticsearch 里搜这个标签。我当年排查了整整一天，才发现是时间格式写错了。

Grok 内置了 100 多种模式，常用的有：

模式名称 匹配内容 示例
IP IP 地址 192.168.1.1
HTTPDATE HTTP 日期格式 10/Oct/2023:13:55:36 +0800
WORD 单词（不含空格） GET
NUMBER 数字 200
URIPATHPARAM URI 路径和参数 /api/user?id=1
GREEDYDATA 贪婪匹配任意字符 任意文本
DATA 匹配任意字符（非贪婪） 任意文本

如果内置模式不够用，你还可以自定义模式：

filter {
  grok {
    patterns_dir => ["/etc/logstash/patterns"]
    match => {
      "message" => "%{MY_CUSTOM_PATTERN:myfield}"
    }
  }
}

在 /etc/logstash/patterns 目录下创建自定义模式文件，比如：

MY_CUSTOM_PATTERN \d{3}-\d{2}-\d{4}

这样就能匹配类似 123-45-6789 的格式了。

3.5 日志多行合并处理

这是日志采集里最头疼的问题之一。Java 异常栈、Python Traceback、SQL 执行计划……这些日志都是一行开头、多行内容的。如果按行读取，就会把一条完整的日志拆成几十条。

Logstash 提供了 multiline 插件来解决这个问题。它不在 Filter 阶段，而是在 Input 阶段配置。

举个例子，合并 Java 异常栈：

input {
  file {
    path => "/var/log/app/error.log"
    codec => multiline {
      pattern => "^\s"  # 以空格开头的行
      what => "previous"  # 合并到上一行
    }
  }
}

这段配置的意思是：如果某一行以空格开头，就把它合并到上一行。因为 Java 异常栈的后续行都是以空格或 Tab 开头的。

再比如，合并以时间戳开头的多行日志：

input {
  file {
    path => "/var/log/app/app.log"
    codec => multiline {
      pattern => "^\d{4}-\d{2}-\d{2}"  # 以日期开头的行
      what => "previous"
      negate => true  # 取反：匹配到的行是新日志，没匹配到的行是续行
    }
  }
}

避坑指南：multiline 插件会占用内存，因为 Logstash 需要缓存未完成的多行数据。如果日志量很大，建议设置 max_lines 和 max_bytes 参数，防止内存溢出。我曾经在生产环境遇到过因为异常栈太长导致 Logstash OOM 的情况，后来加了 max_lines => 500 才解决。

多行合并的配置参数：

参数 说明 默认值
pattern 匹配模式，决定哪些行需要合并 无
what 合并方向：previous（向上合并）或 next（向下合并） previous
negate 是否取反匹配结果 false
max_lines 最多合并多少行 500
max_bytes 最多合并多少字节 10485760 (10MB)

3.6 Output 插件：数据送到哪里去

Output 是 Pipeline 的最后一站。最常用的 Output 是 Elasticsearch，但也不止于此。

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "nginx-access-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "changeme"
  }
}

除了 Elasticsearch，还有这些常用 Output：

  • stdout：输出到控制台，调试时非常有用
  • file：写入本地文件
  • kafka：发送到 Kafka 消息队列
  • csv：输出为 CSV 文件
  • redis：写入 Redis 队列

我习惯在调试阶段先用 stdout 输出，确认数据格式正确后再切换到 Elasticsearch。这样可以避免把错误数据写入生产索引。

最佳实践：一个 Pipeline 可以配置多个 Output。比如同时输出到 Elasticsearch 做实时分析，输出到文件做归档备份。Logstash 会并行执行所有 Output，互不影响。

3.7 完整 Pipeline 示例

最后，给一个完整的 Pipeline 配置，把今天讲的内容串起来：

input {
  beats {
    port => 5044
  }
}

filter {
  # 解析 Nginx 访问日志
  grok {
    match => {
      "message" => "%{IP:clientip} - - \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:status} %{NUMBER:bytes}"
    }
  }
  
  # 转换时间格式
  date {
    match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
  }
  
  # 删除原始 message 字段，节省存储空间
  mutate {
    remove_field => ["message"]
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "nginx-access-%{+YYYY.MM.dd}"
  }
  
  stdout {
    codec => rubydebug
  }
}

这个配置从 Filebeat 接收日志，用 grok 解析成结构化字段，用 date 插件标准化时间戳，最后同时输出到 Elasticsearch 和控制台。调试时看控制台输出，上线后关掉 stdout 就行。

Logstash 的 Pipeline 模型其实不复杂，复杂的是怎么用好 Filter 插件。尤其是 grok，需要花时间积累模式库。我建议你从最简单的日志开始练手，逐步增加复杂度。等你把 grok 用熟了，大部分日志格式都能搞定。

← 上一章 📖 返回目录 下一章 →