Logstash Beats输入如何同时支持JSON与Plain编解码器？

核心思路 #

别在Beats输入里固定编解码器，转到Logstash的filter阶段动态处理就行。Filebeat发过来的日志原始内容都存在message字段里，我们先试着把它解析成JSON，成功就按JSON逻辑处理，失败就当作纯文本处理。

具体配置示例 #

1. Beats输入配置 #

先把Beats输入里的编解码器去掉（或者保留plain也不影响，后续会重新处理）：

input {
  beats {
    port => 5044
    # 这里不用指定 codec，或者留着 codec => plain 也没问题
  }
}

2. Filter阶段动态解析逻辑 #

用json过滤器配合条件判断，同时覆盖两种情况：

filter {
  # 尝试把message解析成JSON，失败也不报错，只打标签区分
  json {
    source => "message"
    target => "parsed_json"
    add_tag => ["json_parsed"]
    skip_on_invalid_json => true
    tag_on_failure => ["plain_text"]
  }

  # JSON解析成功的情况：可把解析后的字段移到根级别（可选操作）
  if "json_parsed" in [tags] {
    mutate {
      rename => { "[parsed_json][field1]" => "field1" }
      rename => { "[parsed_json][field2]" => "field2" }
      remove_field => ["message", "parsed_json"]
    }
  }

  # 纯文本的情况：直接保留message或者添加自定义标识字段
  if "plain_text" in [tags] {
    mutate {
      add_field => ["log_type", "plain_text"]
    }
  }
}

3. 更高效的优化方案：Filebeat提前标记日志类型 #

如果能在Filebeat里区分不同格式的日志源，直接给不同日志加自定义字段，Logstash处理起来更精准，还省了解析失败的开销：
比如Filebeat配置：

filebeat.inputs:
- type: log
  paths:
    - /path/to/json/logs/*.log
  fields:
    log_format: json

- type: log
  paths:
    - /path/to/plain/logs/*.log
  fields:
    log_format: plain

对应的Logstash filter配置：

filter {
  if [fields][log_format] == "json" {
    json {
      source => "message"
      remove_field => ["message"]
    }
  } else if [fields][log_format] == "plain" {
    # 这里添加纯文本的处理逻辑，比如提取关键字段、过滤无效内容等
    mutate {
      add_field => ["log_content", "%{message}"]
    }
  }
}

两种方案对比 #

• 动态解析方案：不用修改Filebeat配置，适合没法提前区分日志源的场景；
• 提前标记方案：处理效率更高，适合能明确区分日志格式的场景。

内容的提问来源于stack exchange，提问作者user1297406