Logstash —— 服务端日志 ETL 与解析管道

分类：04 日志监控与分析 适用：日志集中处理层；日均 GB~TB 级、需复杂解析/富化/多路分发；Linux/容器/K8s 一句话：基于 JVM 的开源数据处理管道，用 input → filter → output 把原始日志清洗成结构化事件再送往 ES/Kafka 官方：https://www.elastic.co/logstash/ 授权：Elastic License 2.0（8.x 起；免费含全部 filter/output 插件、持久化队列与监控；部分集中管理属付费）

1. 它解决什么问题

采集器（Beats）只负责"搬运"，原始日志仍是一行行文本：字段混在字符串里、格式各异、时间格式不统一、IP 无地理信息、需要按业务分流。Logstash 提供：

• 结构化解析：Grok/Dissect/JSON/KV 把非结构化日志拆成字段（状态码、URL、耗时、客户端 IP）。
• 清洗与富化：mutate 改名/改类型/删字段，date 统一时间，geoip 补地理位置，fingerprint 生成去重 ID。
• 缓冲削峰：持久化队列（PQ）把在途事件落盘，下游 ES 抖动时不丢数据、不压垮上游。
• 多路分发：一个输入按条件写多个 ES 索引 / Kafka topic，或用多 pipeline 做故障隔离。
• 统一接入层：收 Beats(5044)、Kafka、JDBC、Syslog 等，屏蔽下游差异。

它重（JVM、数百 MB 内存），不适合部署在每台业务主机（那是 Filebeat/Fluent Bit 的活）；也不做长期存储与检索（那是 ES/Loki 的活）。

2. 核心概念与架构

术语表

两套链路中的位置与完整数据流（ASCII）

【ELK / EFK 链路 —— Logstash 是"集中解析/缓冲层"】
  应用写日志文件 / 容器 stdout
        │ ① 采集：Filebeat / Fluentd(F) / Fluent Bit，记录 offset
        ▼
  Filebeat ──► [可选 Kafka/Redis 二级缓冲] ──► ★ Logstash ★
                                              input(beats:5044)
                                              → filter(grok/mutate/date/geoip)
                                              → output(elasticsearch / kafka)
                                              │ PQ 落盘削峰、至少一次投递
                                              ▼
                                       Elasticsearch(存储检索层) → refresh(1s) 后可检索
                                              ▼
                                       Kibana: Discover / Dashboard / Alerting

【PLG 链路 —— 不含 Logstash】
  文件 / stdout / journal ─► Alloy(推荐) / Promtail(已 EOL) ─► Loki(仅标签+压缩块) ─► Grafana(LogQL)
  对照：PLG 用 Alloy 的轻量 pipeline_stages 替代 Logstash 重解析；需要重 ETL 才前置 Logstash，属非典型。

日志从文件到可检索的完整路径（ELK）：

① 文件产生 → ② Beats 采集(记录位移) → ③ 送 Logstash beats input(5044)
→ ④ 进入 PQ 落盘 → ⑤ filter 解析成结构化 JSON(@timestamp、status_code、geo…)
→ ⑥ output 批量写 ES → ⑦ ES refresh 后可被 Query DSL 检索 → ⑧ Kibana 检索/聚合/告警

PLG 等价路径：文件 → Alloy 解析+加标签 → Loki 按标签分块 → Grafana 用 LogQL 查询。差异：ELK 写入前做重解析并全字段索引；PLG 只加少量标签、检索时才做行过滤。

3. 安装与部署

版本差异：8.x 自带 bundled JDK、ES 8.x 默认开启安全与 TLS，因此 elasticsearch output 必须配认证（user/password）与 CA 证书，否则写入 401/SSL 失败；7.x 无此要求。TLS 参数名随 output 插件版本变化，以官方插件文档为准。具体版本号以官方最新稳定版为准。

3.1 环境准备

3.2 多种安装方式

A. 官方 APT/YUM 仓库（生产首选）

# Debian/Ubuntu（与 ES 共用 GPG key 与 8.x 仓库）
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo gpg --dearmor -o /usr/share/keyrings/elastic-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/elastic-keyring.gpg] https://artifacts.elastic.co/packages/8.x/apt stable main" | sudo tee /etc/apt/sources.list.d/elastic-8.x.list
sudo apt update && sudo apt install -y logstash
# RHEL/Rocky：复用 ES 的 elasticsearch.repo（见 elasticsearch.md 3.2），sudo dnf install -y logstash

B. Docker / docker compose

docker run --rm -it --name ls --net elastic -p 5044:5044 -p 9600:9600 \
  -v "$PWD/pipeline:/usr/share/logstash/pipeline:ro" \
  -v "$PWD/logstash.yml:/usr/share/logstash/config/logstash.yml:ro" \
  -e LS_JAVA_OPTS="-Xms2g -Xmx2g" \
  docker.elastic.co/logstash/logstash:<版本>          # tag 以官方最新稳定版为准

C. 二进制 tar.gz + systemd

tar -xzf logstash-<版本>-linux-x86_64.tar.gz -C /opt
useradd -r logstash && chown -R logstash:logstash /opt/logstash-*
sudo -u logstash /opt/logstash-*/bin/logstash -f /opt/logstash-*/pipeline/ --config.reload.automatic
# 生产写 /etc/systemd/system/logstash.service：User=logstash、LimitNOFILE=65535、Restart=on-failure

D. K8s（Helm / Operator）：官方 elastic/helm-charts 提供 logstash chart；也常以独立 Deployment + ConfigMap 挂 pipeline + PVC 承载 PQ（ECK Operator 侧重 ES/Kibana/Beats）。具体 chart 版本以官方文档为准。

3.3 首次启动与验证

sudo systemctl daemon-reload && sudo systemctl enable --now logstash
sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/ --config.test_and_exit   # -t 校验配置
curl -s http://localhost:9600/?pretty ; curl -s :9600/_node/stats/pipelines?pretty | jq '.pipelines.main.events'
/usr/share/logstash/bin/logstash -e 'input { stdin {} } output { stdout { codec => rubydebug } }'   # 键入一行验证
tail -f /var/log/logstash/logstash-plain.log

4. 配置文件详解

路径：主配置 /etc/logstash/logstash.yml；管道 /etc/logstash/conf.d/*.conf 或 /etc/logstash/pipelines.yml；JVM /etc/logstash/jvm.options；日志 /var/log/logstash/logstash-plain.log；数据/PQ /var/lib/logstash。

完整可运行最小示例（pipeline conf：Grok + 富化 + ES 输出）：

# /etc/logstash/conf.d/nginx.conf
input { beats { port => 5044 } }                     # 接收 Filebeat；调试可临时加 stdin {}
filter {
  if "nginx" in [log][file][path] {
    grok { match => { "message" => "%{COMBINEDAPACHELOG}" } }   # 内置模式产出 clientip/timestamp/verb/request/response/bytes/agent
    # 自定义等价：match => { "message" => "%{IPORHOST:client.ip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:ts}\] \"%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:status:int} %{NUMBER:bytes:int}" }
    date { match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ] timezone => "Asia/Shanghai" target => "@timestamp" remove_field => ["timestamp"] }
    mutate {
      convert => { "bytes" => "integer" }
      rename  => { "clientip" => "client.ip" }
      add_field => { "[@metadata][target_index]" => "logs-nginx-%{+YYYY.MM.dd}" }
      remove_field => [ "ident", "auth", "agent", "referrer", "host" ]
    }
    geoip { source => "client.ip" target => "geo" }   # 需 GeoLite2 数据库
    if "_grokparsefailure" in [tags] { drop {} }      # 或路由到死信索引
  }
}
output {
  elasticsearch {
    hosts => ["https://<es_host>:9200"]
    user => "<es_user>"  password => "<es_password>"   # 8.x 安全集群必填
    index => "%{[@metadata][target_index]}"
    # 8.x 安全集群还需 TLS/CA 相关项，参数名随 output 插件版本变化，见官方插件文档
  }
}

多 pipeline（pipelines.yml） —— 为不同业务隔离队列与故障域：

# /etc/logstash/pipelines.yml
- pipeline.id: main
  path.config: "/etc/logstash/conf.d/*.conf"
  pipeline.workers: 4
  queue.type: persisted
  queue.max_bytes: 8gb
- pipeline.id: nginx
  path.config: "/etc/logstash/nginx.d/*.conf"
  pipeline.workers: 2
  queue.type: persisted
  queue.max_bytes: 4gb
- pipeline.id: audit
  config.string: "input { pipeline { address => audit } } output { elasticsearch { hosts => [\"https://<es_host>:9200\"] index => \"audit-%{+YYYY.MM.dd}\" } }"

5. 核心使用方法（CLI / 过滤插件 / 多 pipeline / 监控 API）

A. CLI 与插件管理

bin/logstash -e 'input { stdin {} } output { stdout { codec => rubydebug } }'   # 内联最小管道
bin/logstash -f /etc/logstash/conf.d/ --config.test_and_exit                    # -t 校验语法
bin/logstash -f /etc/logstash/conf.d/ --config.reload.automatic                  # 自动重载
bin/logstash -f /etc/logstash/conf.d/ --path.settings /etc/logstash --verbose    # 指定配置目录
bin/logstash --version ; systemctl start|stop|restart|status logstash ; bin/logstash-plugin list ; bin/logstash-plugin list --group filter
bin/logstash-plugin install logstash-filter-dissect ; bin/logstash-plugin update logstash-filter-grok ; bin/logstash-plugin remove logstash-output-email

B. Grok 模式（解析核心）

# 常用内置：%{IPORHOST} %{WORD} %{NUMBER} %{INT} %{DATA} %{GREEDYDATA} %{HTTPDATE} %{TIMESTAMP_ISO8601} %{LOGLEVEL}；组合：%{COMBINEDAPACHELOG} %{SYSLOGLINE}；语法 %{PATTERN:字段名:类型}
grok {
  match => { "message" => "%{TIMESTAMP_ISO8601:log_time} %{LOGLEVEL:log.level} \[%{DATA:trace_id}\] %{GREEDYDATA:msg}" }
  # 自定义命名捕获：match => { "message" => "code=(?<code>\d+) cost=(?<cost_ms>\d+)" }；patterns_dir 放自定义模式文件
  # break_on_match => true ; tag_on_failure => ["_grokparsefailure"]
}
dissect { mapping => { "message" => "%{ts} %{+ts} %{level} %{msg}" } }   # 分隔符固定时比 Grok 更快

C. mutate / date / geoip / 其它过滤器

mutate {
  rename => { "host" => "host.name" }
  convert => { "status" => "integer"; "cost_ms" => "float" }
  add_field => { "env" => "prod"; "[@metadata][idx]" => "app-%{+YYYY.MM.dd}" }
  remove_field => [ "agent", "ecs", "input", "log", "@version" ]
  gsub => [ "request", "\?.*$", "" ] ; split => { "tags" => "," } ; lowercase => [ "log.level" ]
}
date { match => [ "log_time", "ISO8601", "yyyy-MM-dd HH:mm:ss.SSS" ] timezone => "Asia/Shanghai" target => "@timestamp" }
geoip { source => "client.ip" target => "geo" }                       # 输出 geo.country_name / geo.location
fingerprint { source => ["host.name","trace_id","@timestamp"] method => "SHA1" target => "[@metadata][fp]" }
json { source => "message" target => "json" }                         # 应用输出 JSON 日志时先解析
kv { source => "msg" field_split => " " value_split => "=" }           # key=value 串
prune { whitelist_names => ["@timestamp","message","status","trace_id"] }   # 精简字段防映射爆炸
if [log.level] == "DEBUG" { drop {} }

D. 多 pipeline 与管道间分发

# 分发器：按类型把事件送到不同下游管道，各管道有独立 PQ（官方 output isolator 模式）
input { beats { port => 5044 } }
output {
  if [event][module] == "nginx" { pipeline { send_to => [nginx_pipe] } }
  else { pipeline { send_to => [app_pipe] } }
}
# 下游：input { pipeline { address => nginx_pipe } }

curl -s :9600/_node/pipelines?pretty | jq 'keys'                      # 列出 pipeline.id
curl -s :9600/_node/stats/pipelines?pretty | jq '.pipelines|to_entries[]|{id:.key,in:.value.events.in,out:.value.events.out,q:.value.queue}'

E. 监控 API / 排查

curl -s :9600/?pretty ; curl -s :9600/_node?pretty                    # 节点与版本、OS/JVM
curl -s :9600/_node/stats?pretty                                      # events in/out/filtered、JVM、PQ
curl -s :9600/_node/hot_threads?pretty                                 # 卡住的线程
# 关注：events.in 与 events.out 差值（积压）、queue 使用率、reloads.failures、_grokparsefailure 计数

上述 CLI、插件名、queue.* 与 filter 选项均为官方稳定用法；pipeline input/output 为默认自带插件。

6. 告警与通知

Logstash 自身不产生告警，它暴露 9600 指标与管道事件，由外部告警：

告警条件示例：events.out 5 分钟环比降 > 80%（管道卡死）、PQ 使用率 > 80% 持续 10 分钟（积压）、ES 中 tags:_grokparsefailure 5 分钟 > 100（格式变更）。

通知渠道：Grafana Contact Point 支持 Email / Webhook / 钉钉 / 企业微信（机器人）/ Slack；企微/钉钉把 {"msgtype":"text","text":{"content":"告警内容"}} 发到 <机器人 webhook>。Elastic 侧用 Kibana Connector（Email/SMTP、Webhook、Slack、Teams、PagerDuty），企微/钉钉经 Webhook Connector 转发。

7. 实战场景

7.1 Nginx/应用日志接入并做状态码统计

目标：Nginx access log → Filebeat → Logstash Grok 结构化 → ES → 按状态码统计。

步骤 + 命令/配置：

# ① Filebeat 送 Logstash（/etc/filebeat/filebeat.yml）
# filebeat.inputs: - type: filestream  id: nginx  paths: [/var/log/nginx/access.log]
# output.logstash: hosts: ["<logstash>:5044"]
# ② Logstash 用第 4 章 nginx.conf（grok COMBINEDAPACHELOG + date + mutate）
bin/logstash -f /etc/logstash/conf.d/ --config.test_and_exit && systemctl restart logstash
# ③ 端到端写入验证
echo '127.0.0.1 - - [05/Nov/2026:10:00:00 +0800] "GET /api HTTP/1.1" 500 123 "-" "curl"' >> /var/log/nginx/access.log
curl -s -u elastic:<密码> --cacert <http_ca.crt> "https://<es>:9200/logs-nginx-*/_search?pretty"

状态码统计（ES 侧聚合 → Kibana 可视化）：

curl -s -u elastic:<密码> --cacert <ca> -X POST "https://<es>:9200/logs-nginx-*/_search?size=0&pretty" \
  -H 'Content-Type: application/json' \
  -d '{"query":{"range":{"@timestamp":{"gte":"now-24h"}}},"aggs":{"by_status":{"terms":{"field":"status","size":10}},"trend":{"date_histogram":{"field":"@timestamp","fixed_interval":"1h"}}}}'

验证：:9600/_node/stats/pipelines 的 events.out 递增；ES _search 有 status 字段；Kibana 可过滤 status:500。 回滚：停 Logstash 前把 Filebeat output.logstash 指回备用管道或改 output.elasticsearch；删索引 DELETE /logs-nginx-* 前先快照 ⚠️。

7.2 日志暴增导致磁盘打满的处置

目标：PQ/输出磁盘被打满 → 止血 → 恢复 → 根治。

# ① 定位 PQ 使用率、事件积压、磁盘
curl -s :9600/_node/stats/pipelines?pretty | jq '.pipelines.main.queue' ; df -h /var/lib/logstash ; du -sh /var/lib/logstash/queue/*
# ② 止血：临时丢弃低价值日志 filter { if [log.level]=="DEBUG" { drop {} } }；③ 给 ES 索引降副本腾空间
curl -s -u elastic:<密码> --cacert <ca> -X PUT "https://<es>:9200/logs-2025-*/_settings" -d '{"index.number_of_replicas":0}'
# ④ PQ 自身撑满：停机 → 扩容 → 调大 queue.max_bytes(<磁盘容量) → 启动排空（queue.drain: true）；⑤ 根治：采集端限速、ILM 缩短留存

验证：df -h 回落；events.in ≈ events.out；ES _cluster/health 回 green。 回滚：恢复 drop 规则与副本数 PUT /logs-*/_settings {"index.number_of_replicas":1}；⚠️ 已删索引只能从快照恢复。

7.3 多 pipeline 隔离 + 输出到 Kafka 削峰

目标：关键业务与普通日志隔离队列；下游 ES 维护时先写 Kafka 不丢数据。

# pipelines.yml：nginx 与 audit 各自 persisted 队列，互不影响（同上第 4 章结构）
- pipeline.id: nginx
  path.config: "/etc/logstash/nginx.d/*.conf"
  queue.type: persisted
  queue.max_bytes: 8gb
  pipeline.workers: 4

# nginx.d/90-output.conf：主写 ES，同时镜像一份到 Kafka 便于重放
output {
  elasticsearch { hosts => ["https://<es>:9200"] user => "<u>" password => "<p>" index => "logs-nginx-%{+YYYY.MM.dd}" }
  kafka { topic_id => "logs-raw" bootstrap_servers => "<kafka>:9092" codec => json }
}

验证：:9600/_node/pipelines 存在 nginx/audit 两个 id；停 audit 的 ES 输出后 nginx 的 events.out 不受影响。 回滚：pipelines.yml 改回单管道（或注释 kafka output）并 restart；⚠️ 重启前确认 PQ 已排空，避免重复投递。

8. 日常运维与故障排查

故障「现象 → 原因 → 处置」表

9. 性能、容量与高可用

容量估算公式（官方 PQ 公式）：

Bytes/s = EPS × 单事件字节 ； Bytes/h = Bytes/s × 3600
所需 PQ 容量 = Bytes/h × 可容忍下游不可用小时数 × 放大系数(起始 1.10)，例：1000 EPS × 1KB ≈ 3.5GB/h，容忍 12h → ≈ 47GB
约束：所有 pipeline 的 queue.max_bytes 之和 < 磁盘容量；queue.max_bytes ≥ queue.page_capacity(64mb)

Logstash 自身无分片/副本概念：分片与副本在下游 ES 规划（单分片 30–50GB、1 主 1 副、ILM rollover，见 elasticsearch.md）；Logstash 侧对应 PQ 容量 + workers + 多实例 HA。

10. 安全加固、备份与升级

认证鉴权 / TLS / 最小权限

# ES output 用最小权限专用用户（见 elasticsearch.md 10 建 logs_writer 角色），勿用 elastic 超管
elasticsearch { hosts => ["https://<es>:9200"] user => "logstash_svc" password => "${ES_PWD}" index => "logs-%{+YYYY.MM.dd}" }
# TLS/CA 配置项名称随插件版本变化，按官方 output-elasticsearch 插件文档设置
# beats input 启用 TLS 的参数名以官方 input-beats 插件文档为准

• 密码勿写明文：用 keystore（bin/logstash-keystore create / add ES_PWD，配置里用 ${ES_PWD}）。
• 最小权限：仅 create_index + write + view_index_metadata；禁用不必要插件；9600 仅内网可达。

备份：Logstash 无状态（除 PQ 在途数据），备份对象是配置与 keystore：/etc/logstash/、pipelines.yml、conf.d/、keystore；数据最终在 ES，用 SLM 快照备份恢复（PUT /_slm/policy/... + _restore，见 elasticsearch.md 10），⚠️ 恢复前先 _close 目标索引。

滚动升级顺序与回滚：① 备份配置与 keystore → ② 逐台停（先确认 PQ 排空或 queue.drain: true）→ ③ 升级包/镜像 → ④ logstash -t 校验 → ⑤ 启动看 9600 与 events.out 恢复 → ⑥ 下一台。回滚：降回旧版本包并恢复备份配置（PQ 页文件跨版本可能不兼容，回滚前另存/清空 path.queue ⚠️）；下游 ES 升级另按其滚动顺序。

11. 常见坑

12. 速查表

13. 选型建议与同类对比

何时选 Logstash：异构日志格式多、需要 Grok/富化/条件路由/至少一次投递缓冲、已有 ES 且能承担 JVM 运维。何时不选：只需"采集+轻量解析"→ Filebeat/Fluent Bit/Alloy 直接入 ES/Loki，省掉一层；成本敏感且已用 Grafana → PLG；想要单机轻量替代 → Vector。边界：Logstash 在 ELK/EFK 是解析/缓冲层，PLG 链路默认不含 Logstash（由 Alloy 承担解析），二者不是串联关系。

14. 参考链接

• 官网：https://www.elastic.co/logstash/
• 文档（当前版）：https://www.elastic.co/guide/en/logstash/current/index.html
• 持久化队列 PQ：https://www.elastic.co/guide/en/logstash/current/persistent-queues.html
• 多 pipeline：https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html
• Grok filter：https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html
• mutate filter / beats input：https://www.elastic.co/guide/en/logstash/current/plugins-filters-mutate.html 、https://www.elastic.co/guide/en/logstash/current/plugins-inputs-beats.html
• elasticsearch output：https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html
• 监控 API / 源码：https://www.elastic.co/guide/en/logstash/current/monitoring-logstash.html 、https://github.com/elastic/logstash
• 包仓库：https://artifacts.elastic.co/packages/8.x/apt

8.x 起 ES 安全/TLS 默认开启，Logstash output 需认证；PQ 的 queue.compression(9.2+)、queue.checkpoint.interval(9.1+ 弃用) 随版本变化。具体版本与参数以官方最新稳定版文档为准。