4. 日志数据管道构建

在 Elastic 可观测性体系中，日志从产生到可检索，通常要经过采集、传输、解析、富化、写入五个阶段。Filebeat 与 Logstash 的协同正是这条链路的骨架：Filebeat 部署在靠近日志源的主机上，负责轻量采集与可靠投递；Logstash 部署在汇聚层，负责重量级的解析、字段加工与路由。理解两者的职责边界，是构建稳定日志管道的第一步。

一、职责分工与协同拓扑

Filebeat 的核心能力是“读文件、记位置、发事件”。它通过 registry 文件记录每个日志文件的读取偏移量，进程重启后从断点续读，避免重复或丢失。Filebeat 内置的 filestream 输入支持多行合并、条件过滤和简单字段添加，但复杂的 Grok 解析、GeoIP 查询、条件分支等能力较弱。Logstash 则拥有完整的 filter 插件生态，适合承担解析与富化。因此典型拓扑是：Filebeat 采集 → 传输（Beats 协议或 Kafka）→ Logstash 解析富化 → Elasticsearch 写入。

两者之间有两种常见连接方式。第一种是 Filebeat 直接通过 output.logstash 把事件推给 Logstash 的 beats 输入，链路短、延迟低，适合中小规模。第二种是 Filebeat 先写入 Kafka，Logstash 再从 Kafka 消费，这样能削峰填谷、解耦采集与处理，适合日志量大或 Logstash 需要维护重启的场景。选择哪种，取决于吞吐量、容错要求和运维复杂度。

二、Grok 与 Dissect：两种解析思路

日志解析的核心目标是把非结构化文本拆成结构化字段。Grok 基于正则表达式，用 %{PATTERN:field} 语法匹配并命名字段，灵活但性能开销较大。Dissect 则基于分隔符定位，不依赖正则，速度快数倍，但要求日志格式严格固定。

以 Nginx 访问日志为例，原始行如下：

192.168.1.10 - - [15/Mar/2025:10:23:45 +0800] "GET /api/users HTTP/1.1" 200 3421

用 Grok 解析：

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

用 Dissect 解析同样的行：

filter {
  dissect {
    mapping => {
      "message" => "%{client_ip} %{ident} %{auth} [%{timestamp}] \"%{method} %{request} HTTP/%{http_version}\" %{status} %{bytes}"
    }
  }
}

选择原则很明确：日志格式稳定、字段位置固定时优先用 Dissect，性能更好；格式多变、需要容错或字段可选时用 Grok。实践中常组合使用——先用 Dissect 快速切分主体，再用 Grok 处理其中某个复杂子字段。无论哪种方式，解析后都应检查 _grokparsefailure 或 _dissectfailure 标签，把解析失败的事件单独路由到死信索引，避免污染主数据。

三、GeoIP 富化与字段加工

解析出 client_ip 后，GeoIP 插件可以把它转换成地理位置信息，为安全分析和访问地图提供维度：

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

GeoIP 依赖 MaxMind 数据库，需定期更新以保持准确。富化后的 geo.location 是经纬度坐标，可直接用于 Kibana 地图可视化。除 GeoIP 外，常见的字段加工还包括：用 date 插件把日志时间戳解析为 @timestamp，用 mutate 做类型转换和字段重命名，用 useragent 解析浏览器信息。这些操作应尽量在 Logstash 中完成，而不是留给 Elasticsearch 的 ingest pipeline，因为 Logstash 的调试和版本管理更直观。

四、多行日志处理

Java 异常堆栈、Python traceback 这类日志天然跨多行，如果按行采集，一条异常会被拆成几十个独立事件，检索时无法还原上下文。多行合并的关键是识别“一行是新事件的开始，还是上一事件的延续”。

Filebeat 的 filestream 输入支持多行配置：

filebeat.inputs:
- type: filestream
  paths:
    - /var/log/myapp/*.log
  multiline:
    type: pattern
    pattern: '^\d{4}-\d{2}-\d{2}'
    negate: true
    match: after
    max_lines: 500
    timeout: 5s

这段配置的含义是：不以日期开头的行，都追加到上一行之后。也就是说，日期开头的行是新事件起点，其余行是延续。max_lines 限制单个事件最多合并 500 行，防止异常刷屏导致内存膨胀；timeout 保证即使没有新行，5 秒后也会把已缓冲的多行事件发出，避免事件长期滞留。

如果多行日志在 Logstash 侧处理，可用 multiline 编解码器，但要注意 Logstash 的 multiline 是单线程的，容易成为瓶颈。更推荐在 Filebeat 侧完成多行合并，让 Logstash 专注于解析和富化。

五、背压控制

背压是日志管道最容易出问题的地方。当 Elasticsearch 写入变慢，Logstash 的处理队列会积压，进而导致 Filebeat 发送受阻。如果没有任何控制机制，Filebeat 会持续读取文件并缓存在内存中，最终 OOM 崩溃。

Filebeat 的背压机制是自动的：当 output.logstash 的发送队列满时，Filebeat 会暂停读取新日志，registry 偏移量不再推进，日志文件继续增长但不会被消费。这保证了不丢数据，但代价是采集延迟增加。可以通过 queue.mem.events 和 bulk_max_size 调整缓冲深度。

Logstash 侧的背压控制更关键。默认的内存队列在压力下会丢事件，生产环境应启用持久化队列：

queue.type: persisted
queue.max_bytes: 8gb
queue.checkpoint.writes: 1024
dead_letter_queue.enable: true
dead_letter_queue.max_bytes: 2gb

持久化队列把事件写到磁盘，Logstash 重启后可以继续处理，避免数据丢失。死信队列则捕获那些被 filter 或 output 拒绝的事件，便于事后排查。此外，Logstash 的 pipeline.workers 应设置为 CPU 核数，pipeline.batch.size 和 pipeline.batch.delay 需要根据吞吐量和延迟要求调优——批量越大吞吐越高，但单批延迟也越大。

六、一个可参考的管道配置骨架

综合以上要点，一个面向 Nginx 日志的 Logstash 管道可以组织为：输入用 beats 接收 Filebeat 事件；filter 阶段先用 Dissect 切分主体，再用 Grok 处理 request 字段，接着 GeoIP 富化 client_ip，date 插件归一化时间戳，最后用条件判断把解析失败的事件打标签；输出阶段按标签路由到不同索引。这个骨架的价值在于把解析、富化、容错、路由四件事分层处理，每一层都可以独立调试和替换。

构建日志管道的本质，是在可靠性、延迟和成本之间找平衡。Filebeat 保证不丢，Logstash 保证可解析可富化，持久化队列和死信队列保证异常可追溯。把这三者配合好，日志数据才能真正成为可观测性的有效输入。