第4章 Logstash深入

第4章：Logstash深入：Pipeline概念、Input/Filter/Output插件、Grok正则解析实战

好，咱们今天来啃一块硬骨头——Logstash。说实话，我刚开始接触ELK的时候，觉得Logstash就是个管道工，把数据从A搬到B。但用久了才发现，这家伙的Pipeline设计，其实藏着不少门道。

4.1 什么是Logstash Pipeline？

Pipeline，说白了就是一条数据流水线。数据从源头进来，经过几道工序处理，最后送到目的地。Logstash的Pipeline由三个阶段组成：

Input：数据从哪里来？文件、网络、消息队列...
Filter：数据怎么处理？解析、转换、丰富...
Output：数据往哪里去？Elasticsearch、文件、Kafka...

我习惯把Pipeline想象成工厂里的传送带。原始日志就是原材料，经过清洗、加工、包装，最后变成可用的成品。每个阶段都可以有多个插件，就像流水线上可以有多道工序。

核心要点：Logstash的Pipeline是事件驱动的。每一条日志进来，都会依次经过Input → Filter → Output。如果Filter处理失败，数据可能就卡在半路了。

4.2 Input插件：数据入口

Input插件决定了数据从哪里来。我见过最常用的就这几个：

插件名称	适用场景	配置示例
file	读取日志文件	`input { file { path => "/var/log/*.log" } }`
beats	接收Filebeat数据	`input { beats { port => 5044 } }`
kafka	从Kafka消费	`input { kafka { topics => ["logs"] } }`
tcp	接收网络数据	`input { tcp { port => 4560 } }`

嗯，这里要注意一点。用file插件时，Logstash会记录文件读取的位置（sincedb文件）。我曾经遇到过一个问题：服务器重启后，Logstash重新读取了所有历史日志，导致数据重复。后来发现是sincedb文件没配置好。

我的建议：生产环境中，尽量用Beats或Kafka作为输入源。它们有更好的数据可靠性保证，而且支持背压机制，不会把Logstash撑爆。

4.3 Filter插件：数据处理核心

Filter是Logstash的灵魂。没有Filter，你拿到的就是一堆原始字符串，毫无价值。我个人最常用的Filter插件有这些：

grok：正则解析，把非结构化文本变成结构化数据
mutate：字段增删改，比如重命名、类型转换
date：时间戳解析，把字符串时间转成标准格式
geoip：IP地址地理定位
useragent：解析浏览器User-Agent

你想想看，如果日志里全是这样的内容：

192.168.1.1 - - [10/Oct/2023:13:55:36 +0800] "GET /index.html HTTP/1.1" 200 2326

不经过Filter处理，你根本没法做分析。但用Grok一解析，立马变成：

{
  "clientip": "192.168.1.1",
  "timestamp": "10/Oct/2023:13:55:36 +0800",
  "method": "GET",
  "path": "/index.html",
  "status": 200,
  "bytes": 2326
}

这才是我们想要的结构化数据。

4.4 Grok正则解析实战

Grok说白了就是正则表达式的封装。它预定义了很多模式，比如%{IP}匹配IP地址，%{NUMBER}匹配数字。但真正用起来，还是得自己写。

我记得第一次写Grok表达式时，折腾了整整一下午。后来总结出一个套路：

先看原始日志长什么样，把关键字段圈出来
用在线调试工具（比如Grok Debugger）测试
从简单到复杂，先匹配整行，再细化字段

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

filter {
  grok {
    match => {
      "message" => "%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:response} (?:%{NUMBER:bytes}|-) %{QS:referrer} %{QS:agent}"
    }
  }
}

这个表达式看起来很长，但拆开看其实很简单。每个%{模式:字段名}就是一个匹配单元。我习惯先匹配整行，确认没问题后，再逐个字段调整。

避坑指南：我曾经在生产环境里用了一个过于复杂的Grok表达式，结果Logstash的CPU直接飙到100%。后来发现是正则回溯导致的。解决办法有两个：一是简化表达式，二是用grok { break_on_match => true }让匹配到就停止。

4.5 Output插件：数据出口

Output决定了数据最终的去向。最常见的配置就是输出到Elasticsearch：

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

但有时候，我们还需要同时输出到多个地方。比如：

一份到Elasticsearch做实时分析
一份到Kafka做数据备份
一份到文件做调试

Logstash支持多Output，数据会复制到每个输出端。我建议在调试阶段，加一个stdout { codec => rubydebug }输出，这样能在控制台看到解析后的数据长什么样。

4.6 Pipeline配置实战

好了，理论说完了，咱们来写一个完整的Pipeline配置。假设我们要处理Nginx访问日志：

input {
  beats {
    port => 5044
  }
}

filter {
  grok {
    match => {
      "message" => "%{COMBINEDAPACHELOG}"
    }
  }
  
  date {
    match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
  }
  
  geoip {
    source => "clientip"
    target => "geo"
  }
  
  useragent {
    source => "agent"
    target => "ua"
  }
  
  mutate {
    remove_field => ["message", "ident", "auth"]
  }
}

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

这个配置做了几件事：

从Filebeat接收数据
用Grok解析日志行
把时间戳转成标准格式
根据IP地址做地理定位
解析浏览器信息
删除原始字段，节省存储空间
输出到Elasticsearch和控制台

嗯，这里有个小技巧。调试的时候，先用stdout输出看看数据对不对，确认没问题了再关掉。不然Elasticsearch里存了一堆脏数据，清理起来很麻烦。

4.7 性能优化建议

最后，分享几个我在项目中踩过的坑：

Pipeline workers：默认是CPU核心数，但I/O密集型场景可以调高到2倍
Batch size：默认125条一批，如果日志量很大，可以调到500-1000
Filter顺序：把耗时的Filter（比如Grok）放在前面，尽早丢弃无效数据
条件判断：用if [type] == "nginx"区分不同来源的日志

我曾经遇到过一个场景：每天处理10TB的日志，Logstash集群扛不住了。后来发现是Grok表达式太复杂，每个事件要匹配几十个模式。优化后只保留最常用的几个模式，性能提升了3倍。

好了，这一章的内容就到这。Pipeline的概念、三个阶段的插件、Grok实战，这些是Logstash的核心。下一章咱们聊聊如何用Logstash做日志聚合和告警，敬请期待。

📚 课程目录