4. Logstash与Elastic Agent数据管道设计

在 Elastic 可观测性体系中，数据从产生到可查询，中间必须经过一条可设计、可观测、可维护的采集管道。Logstash 与 Elastic Agent 是两类承担该职责的核心组件，但它们的定位并不相同：Logstash 是功能完整的服务端数据处理引擎，擅长复杂解析、字段加工与多路输出；Elastic Agent 是部署在主机侧的轻量统一采集器，擅长以集成（Integration）方式快速接入日志、指标与安全数据。理解二者的输入、过滤器、输出与集成策略，是设计稳定数据管道的前提。

Logstash 管道的三段结构

Logstash 的配置文件由 input、filter、output 三个区段组成，事件按顺序流经它们。input 负责接入数据源，常见插件包括 beats（接收 Filebeat/Elastic Agent 转发的事件）、kafka（从消息队列消费）、file（直接读取文件）、tcp/udp（接收网络日志）。filter 负责解析与加工，最常用的是 grok（正则提取字段）、date（时间戳解析）、mutate（类型转换、重命名、删除字段）、dissect（按分隔符切分，性能优于 grok）。output 负责写出，典型目标是 elasticsearch，也可输出到 kafka、file 或 stdout 用于调试。

下面是一个面向 Nginx 访问日志的示例管道，用于说明各段如何协作。示例参数为教学假设值，实际部署需按环境调整：

input {
  beats {
    port => 5044
  }
}

filter {
  grok {
    match => { "message" => "%{IPORHOST:client_ip} - - \[%{HTTPDATE:ts}\] \"%{WORD:method} %{DATA:path} HTTP/%{NUMBER:http_ver}\" %{INT:status} %{INT:bytes}" }
  }
  date {
    match => [ "ts", "dd/MMM/yyyy:HH:mm:ss Z" ]
    target => "@timestamp"
  }
  mutate {
    convert => { "status" => "integer" "bytes" => "integer" }
    remove_field => [ "ts", "message" ]
  }
}

output {
  elasticsearch {
    hosts => ["https://es-node:9200"]
    index => "nginx-access-%{+yyyy.MM.dd}"
  }
}

该管道的关键判断点在于：grok 是否成功匹配。若匹配失败，事件会带上 _grokparsefailure 标签，此时应通过条件判断将其路由到独立索引或死信队列，而不是直接丢弃。例如在 output 中加入 if "_grokparsefailure" in [tags] { elasticsearch { index => "parse-failed-%{+yyyy.MM.dd}" } }，便于后续排查日志格式变化。

Elastic Agent 的统一采集与集成策略

Elastic Agent 通过 Fleet 集中管理，其采集能力由“集成”定义。每个集成对应一类数据源，例如 Nginx、System、Docker、Kubernetes、MySQL 等，集成内部已经封装了输入配置、解析规则、字段映射与默认仪表板。部署时，Agent 从 Fleet Server 拉取策略（Policy），策略中声明启用哪些集成、采集哪些路径、发送到哪个 Elasticsearch 或 Logstash 输出。

与 Logstash 相比，Elastic Agent 的优势在于开箱即用与统一升级，适合大规模主机与容器环境；其局限在于复杂字段加工能力弱于 Logstash。因此常见架构是：Agent 负责采集与轻量预处理，将事件发送到 Logstash 做深度解析，再由 Logstash 写入 Elasticsearch。这种分层既保留了 Agent 的部署效率，又利用了 Logstash 的加工灵活性。

管道设计中的关键参数与判断依据

设计管道时，以下参数直接影响吞吐与稳定性：pipeline.workers 控制 filter+output 的并行线程数，通常设为 CPU 核数；pipeline.batch.size 控制单批事件数，默认 125，增大可提升吞吐但增加延迟；pipeline.batch.delay 控制批等待毫秒数，默认 50。判断管道是否健康，可观察 Logstash 的 events.in、events.out、events.filtered 指标，若 in 与 out 差距持续扩大，说明 filter 存在阻塞或 output 写入变慢。

索引命名与生命周期同样属于管道设计的一部分。示例中采用 nginx-access-%{+yyyy.MM.dd} 按天滚动，配合 ILM（索引生命周期管理）实现热温冷分层与自动删除。若日志量较小，也可按月命名以减少分片数量。判断依据是：单日索引大小建议控制在 10–50 GB，分片数按每分片 10–50 GB 估算。

集成策略与管道选型

选择 Agent 还是 Logstash，取决于数据复杂度与运维规模。若只需采集标准日志与指标，且希望集中管理，优先使用 Elastic Agent 集成；若需要多源聚合、复杂字段加工、条件路由或写入多个目标，则在 Agent 之后串联 Logstash。两者并非互斥，而是可以组合成“Agent 采集 → Logstash 加工 → Elasticsearch 存储”的分层管道。

无论采用哪种组合，都应遵循三条原则：第一，输入端口与协议要明确，避免与其它服务冲突；第二，过滤器必须可验证，grok 模式应先在测试环境用真实样本验证匹配率；第三，输出索引要配合 ILM，避免无限增长。管道设计完成后，应通过 stdout { codec => rubydebug } 在测试环境观察事件结构，确认字段类型与时间戳正确后再切换到生产输出。

综上，Logstash 与 Elastic Agent 分别代表了“灵活加工”与“统一采集”两种能力。实际部署中，应根据数据源数量、解析复杂度、运维成本与扩展需求，选择单层或分层管道，并通过输入、过滤器、输出三段配置与集成策略的合理组合，构建可观测、可维护的数据接入链路。