📚 课程目录第3章 Logstash配置与日志收集

第3章：Logstash配置与日志收集

讲真，Logstash是ELK里最灵活，也最容易翻车的组件。别怕，我带你理顺它。

Logstash本质上就是一个数据管道。它从上游拿数据，做几手加工，再送到下游去。这个过程的三个核心步骤就是：输入、过滤、输出。今天，我们就从这几个点展开，撸一遍配置实战。

输入插件——怎么把日志“吃”进来

Logstash支持数十种输入插件。但实战中，常用的也就那十来个。我习惯把它们分为两类：主动拉取和被动接收。

类型	插件	常见场景
主动拉取	file	读取本地日志文件
主动拉取	beats	接收Filebeat、Winlogbeat等
被动接收	syslog	接收网络设备的syslog
被动接收	tcp/udp	自定义协议发送日志

举个最常见的例子。如果你需要收集Nginx的访问日志，配置大概是这样：

input {
  file {
    path => "/var/log/nginx/access.log*"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}

这里有个小细节——sincedb_path。在生产环境里，我建议保留默认值，让Logstash记录哪些行已经读过。但如果你像我一样在测试环境反复清数据，把它指向/dev/null就能让Logstash每次都从头读。

注意，别小看这个glob模式。文件路径写不好，会导致日志重复采集。我曾经就因为path => "/var/log/nginx/access.*"没加log后缀，把压缩包也读进去了。

过滤插件——日志的灵魂加工厂

过滤是最麻烦，也最好玩的一步。说到底，就是把乱七八糟的文本拆成结构化字段。

最常用的过滤插件，我列个表：

插件	用途	性能开销
grok	正则解析非结构化消息	中
mutate	重命名、删除、类型转换	低
date	将日志时间设为@timestamp	低
geoip	根据IP查地理位置	低
useragent	解析浏览器UA	低
kv	解析key=value格式	低

grok几乎是必用的。它的工作很简单——通过正则把字符串里的有效信息提取出来。比如解析Apache的通用日志格式：

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

COMBINEDAPACHELOG是Logstash自带的grok模式，但很多情况下需要自己写模式。嗯，这里要注意，grok性能并不算特别高。一条日志如果匹配了太多模式，吞吐量会明显下降。

经验之谈： 我会用Grok Debugger先离线测试表达式，别直接怼到生产环境再试。你想想看，线上调试正则，怕不是要跪键盘。

输出插件——日志去哪里了

输出是管道的终点。最常用的就是Elasticsearch。但偶尔也会做多路输出，比如同时写到Elasticsearch和Redis。

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

我个人习惯在调试阶段加上stdout输出，方便看日志处理结果。rubydebug格式会把字段摊开，一目了然。等验证没问题了，再删掉这个输出，不然会刷屏。

输出插件配置不当，会导致Logstash直接把CPU打满。为什么？因为Elasticsearch写入索引时遇到映射冲突，Logstash会不断重试。解决办法是在输出段加retry_on_conflict参数，或者提前在ES里设计好索引模板。

管道配置实战——把一切串起来

一个完整的Logstash配置文件就是一个.conf文件，里面包含input、filter、output三部分。

我给你们整一个完整的实战例子。假设我们要收集Spring Boot应用的JSON日志：

input {
  beats {
    port => 5044
  }
}
filter {
  if [fields][type] == "spring-boot" {
    json {
      source => "message"
    }
    date {
      match => ["timestamp", "ISO8601"]
      target => "@timestamp"
    }
    mutate {
      remove_field => ["message", "original"]
    }
  }
}
output {
  if [fields][type] == "spring-boot" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "spring-boot-%{+YYYY.MM.dd}"
    }
  }
}

这个配置做了几件事：
1. 从Filebeat接收数据。
2. 判断日志类型是否为spring-boot。
3. 将JSON格式的message解析成结构化字段。
4. 提取时间戳并覆盖@timestamp。
5. 删除原始message字段，节省存储。
6. 按日期写入ES索引。

避坑指南： 我曾经在生产环境直接复制粘贴了旧配置文件去跑，结果忘记改索引名，导致日志全写入了同一个索引，排查了整整一下午。每次新建管道前，都看一眼索引名。

调试方法与性能监控

配置写完了，怎么验证呢？别急着上生产。我推荐几个调试方式：

使用stdout：在output里加stdout { codec => rubydebug }，直接看控制台输出。
使用--config.test_and_exit：这个参数会检查配置语法，但不启动Logstash。
使用http输入插件：临时启用一个HTTP端口，手动发请求测管道。
使用sinewave或generator：模拟输入数据，测试过滤器的效果。

性能监控方面，Logstash暴露了非常多的指标。最直接的方式是访问它的监控API：

curl http://localhost:9600/_node/stats

这个接口会返回大量的信息：

event.in：输入的事件数
event.out：输出的事件数
event.duration_in_millis：事件处理总耗时（毫秒）
pipeline.workers：管道并行数
pipeline.batch.size：每次批量处理的条数

我常用的性能调优命令就这几个：

# 查看管道堆积情况
curl -s http://localhost:9600/_node/stats | jq '.pipelines.main'
# 实时查看事件吞吐
watch -n 1 'curl -s http://localhost:9600/_node/stats | jq ".jvm.threads"'

如果发现日志堆积严重，我会先调大pipeline.batch.size和pipeline.workers。但别盲目调大，CPU和内存不够的话，调了反而更慢。

超时与阻塞处理

Logstash默认没有严格的超时机制。如果你有输出阻塞，日志会堆积在内存中，最终OOM。

解决方法有两个：

在output段设置flush_size和idle_flush_time，控制写入节奏。
在启动参数中加--queue.type，使用持久化队列。

持久化队列的好处是——Logstash挂掉后，日志不丢。缺点是需要额外的磁盘空间。

pipeline:
  batch:
    size: 125
    delay: 5
  workers: 4
queue:
  type: persisted
  max_bytes: 2gb

这段配置是在pipelines.yml里写的。持久化队列的max_bytes设多大？我建议按总日志量的1.5倍来估算。

  核心观点： 调试靠stdout，压测靠generator，监控靠API。这三板斧吃透了，Logstash配置就没什么大坑了。

好了，这章的内容差不多就是这些。Logstash本身不复杂，但细节很多。配置写完后，多测几次，多观察指标，你会发现很多问题。下次再聊其他组件的联动。