3. 日志采集架构设计：Filebeat 与 Logstash 的 Pipeline 设计

日志采集是 ELK 黄金信号体系的“第一公里”。采集层的可靠性直接决定下游存储与分析的完整性，而低延迟则影响故障发现的时间窗口。本章聚焦 Filebeat 与 Logstash 的 Pipeline 设计，从组件职责、数据流保障、背压机制到生产级配置，系统化构建高可靠、低延迟的日志采集架构。

3.1 采集层组件职责与数据流

在经典 ELK 架构中，Filebeat 作为轻量级 Shipper 部署在业务节点，Logstash 作为集中式 Aggregator 进行解析与转换。两者通过 Beats 协议 通信，形成“边缘采集 + 中心处理”的 Pipeline 模型。

3.2 Filebeat 可靠性保障机制

Filebeat 作为边缘采集器，面临文件轮转、网络抖动、进程重启等风险。其可靠性通过以下机制实现：

3.3 Logstash Pipeline 与持久队列

Logstash 的 Pipeline 由 Input → Filter → Output 三个阶段组成。为保证节点故障不丢数据，必须启用 持久队列（Persistent Queue, PQ）。

3.3.1 Pipeline 阶段与延迟分析

3.3.2 持久队列配置

# logstash.yml
queue.type: persisted
queue.max_bytes: 4gb
queue.checkpoint.writes: 1
queue.drain: true

• queue.type: persisted — 启用磁盘持久化，进程重启后恢复未消费事件
• queue.max_bytes: 4gb — 队列最大容量，超出后触发背压
• queue.checkpoint.writes: 1 — 每次写入都做 checkpoint，保证崩溃不丢数据
• queue.drain: true — 优雅关闭时等待队列排空

3.4 低延迟优化策略

在保证可靠性的前提下，通过以下策略降低端到端延迟：

1. Filebeat 端： 减小 scan_frequency（默认 10s → 1s），提高文件变更检测灵敏度；增大 harvester_buffer_size（默认 16384 → 65536），减少小文件频繁读取。
2. Logstash 端： 使用 pipeline.workers 等于 CPU 核数；启用 pipeline.batch.size（1250~5000）与 pipeline.batch.delay（5~50ms）平衡吞吐与延迟。
3. 网络层： 部署 Filebeat 与 Logstash 同可用区；启用 TCP_NODELAY 减少 Nagle 算法引入的延迟。
4. Filter 优化： 将 Grok 表达式按频率排序，高频匹配放前面；使用 dissect 替代简单 Grok 提升 5~10 倍解析速度。

3.5 生产级 Pipeline 配置示例

以下是一个同时满足可靠性与低延迟的完整配置参考：

# filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/app/*.log
  scan_frequency: 1s
  harvester_buffer_size: 65536
  close_inactive: 5m
  ignore_older: 24h

output.logstash:
  hosts: ["logstash.example.com:5044"]
  worker: 4
  bulk_max_size: 2048
  ttl: 300
  ssl.enabled: true

queue.mem:
  events: 8192
  flush.min_events: 512
  flush.timeout: 1s

# logstash.yml
pipeline.workers: 8
pipeline.batch.size: 4000
pipeline.batch.delay: 10

queue.type: persisted
queue.max_bytes: 8gb
queue.checkpoint.writes: 1
queue.drain: true

path.queue: /data/logstash/queue

# logstash pipeline.conf
input {
  beats {
    port => 5044
    ssl => true
    ssl_certificate => "/etc/logstash/certs/logstash.crt"
    ssl_key => "/etc/logstash/certs/logstash.key"
  }
}

filter {
  if [fields][log_type] == "nginx" {
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
    date {
      match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    }
  } else if [fields][log_type] == "java" {
    multiline {
      pattern => "^%{TIMESTAMP_ISO8601}"
      negate => true
      what => "previous"
    }
  }
}

output {
  elasticsearch {
    hosts => ["https://es-cluster:9200"]
    index => "%{[fields][log_type]}-%{+YYYY.MM.dd}"
    user => "logstash_writer"
    password => "${ES_PWD}"
    ssl => true
    flush_size => 5000
    idle_flush_time => 5
    http_compression => true
  }
}

3.6 可靠性验证与监控

部署完成后，通过以下指标验证采集层的可靠性：