Filebeat 如何从日志行（message）中解析出字段？

三种方式按日志格式选：结构化 JSON 日志用 decode_json_fields；格式固定的文本用 dissect（快）；复杂不规则文本用 Elasticsearch ingest pipeline 或 Logstash 的 grok（Filebeat 本身没有该处理器）。

方式一：JSON 日志——decode_json_fields

processors:
  - decode_json_fields:
      fields: ["message"]
      target: "parsed"  # 解析到独立对象，避免覆盖元数据
      overwrite_keys: false

日志 {"user":"tom","action":"login"} 解析后 parsed.user、parsed.action 成为独立字段，可直接检索。

方式二：固定格式——dissect

processors:
  - dissect:
      tokenizer: "%{timestamp} %{level} %{service} %{msg}"
      field: "message"
      target_prefix: "app"

2024-01-01 INFO order 下单成功 → app.level=INFO、app.service=order。dissect 按分隔符切分；性能应按实际规则与负载测量。

方式三：复杂文本——grok

先在 Elasticsearch 中创建 ingest pipeline（Dev Tools）：

PUT _ingest/pipeline/app-grok
{
  "processors": [{
    "grok": {
      "field": "message",
      "patterns": ["%{IP:client_ip} - %{USER:user} \\[%{HTTPDATE:ts}\\] \"%{WORD:method} %{URIPATH:path}\" %{NUMBER:status:int}"]
    }
  }]
}

Filebeat 的 output.elasticsearch 中设置 pipeline: app-grok。

内置丰富预置模式（IP、WORD、NUMBER、HTTPDATE……），适合 Nginx/Apache 等格式明确的日志。

常见问题（FAQ）

Q：dissect 和 grok 怎么选？ 格式稳定可用 dissect；需要模式复用与类型转换用 grok。

Q：解析失败的日志会怎样？ 行为取决于处理器和配置。decode_json_fields 可设 add_error_key；ingest pipeline 可用 on_failure，未处理的失败可能拒绝文档。不要假设都会带标记继续发送。