如何用 Filebeat + Logstash + Elasticsearch 把 JSON 日志发送到 Kibana？

全链路四步：Filebeat 采集时用 json.keys_under_root: true 把 JSON 字段提升到顶层 → 发到 Logstash（beats input）→ Logstash 转发到 Elasticsearch → Kibana 创建索引模式查看。

第一步：Filebeat 配置

filebeat.inputs:
  - type: log
    paths:
      - /path/to/your/app.json
    json.keys_under_root: true     # JSON 字段提升到顶层
    json.add_error_key: true       # 解析失败时记录错误

output.logstash:
  hosts: ["localhost:5044"]

假设日志每行是一个 JSON：{"level":"INFO","msg":"用户登录","ts":1696150204}

第二步：Logstash 配置

input {
  beats {
    port => 5044
  }
}

filter {
  # JSON 已在 Filebeat 解析，这里做类型转换等补充处理
  mutate {
    convert => { "status" => "integer" }
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "app-logs-%{+YYYY.MM.dd}"
  }
}

第三步：Kibana 查看

Stack Management → Data Views → 创建 app-logs-* 索引模式 → Discover 中检索。

常见问题排查

• Kibana 里看不到字段：Data View 需要刷新字段列表；
• JSON 解析失败：Filebeat 加了 json.add_error_key 后，失败的事件带 error.json 字段，据此排查格式问题；
• 字段冲突：不同日志源同名字段类型不同会导致 ES 拒绝写入，统一字段类型或按来源分索引。

观测云对照

整条 Filebeat→Logstash→ES→Kibana 链路可以简化为：DataKit 采集 JSON 日志（自动解析字段）→ 观测云平台，日志查看器直接检索，监控器配置告警，无需维护三个组件。

常见问题（FAQ）

Q：Filebeat 解析 JSON 后还需要 Logstash 吗？
A：不需要复杂转换时可直连 ES（Filebeat → ES）。Logstash 的价值在复杂 filter、多输出路由。

Q：JSON 里的时间字段怎么变成 @timestamp？
A：Logstash 的 date filter：date { match => ["ts", "UNIX"] }。

Q：每天新建索引，mapping 怎么管？
A：配置索引模板（index template），新索引自动套用映射。