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

全链路四步：Filebeat 用 filestream 的 ndjson parser 把 JSON 解析到 app 命名空间 → 发到 Logstash（beats input）→ Logstash 转发到 Elasticsearch → Kibana 创建索引模式查看。

第一步：Filebeat 配置

filebeat.inputs:
  - type: filestream
    id: app-json
    paths:
      - /path/to/your/app.json
    parsers:
      - ndjson:
          target: app
          add_error_key: true

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

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

第二步：Logstash 配置

input {
  beats {
    port => 5044
  }
}

filter {
  # JSON 已在 Filebeat 解析，UNIX ts 转为事件时间
  date { match => ["[app][ts]", "UNIX"] }
}

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

第三步：Kibana 查看

Stack Management → Data Views → 创建 app-logs-*，时间字段选 @timestamp，再到 Discover 检索。示例 ts 是历史时间，需选择对应时间范围；上例明确使用普通每日索引，生产若启用 ILM 或 data stream 应改用配套管理方式。

常见问题排查

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

常见问题（FAQ）

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

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

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