MQTT消息解析及ELK链路确认求助（ELK+Filebeat环境）

问题1：选择Filebeat Processor还是Logstash处理MQTT字段提取 #

根据你的需求复杂度来选：

• 仅需基础字段解析提取：优先用Filebeat Processor。你的MQTT消息是标准JSON格式，在Filebeat配置里添加json处理器，就能把原始消息解析为结构化数据，POWER1~POWER4这类顶层字段会自动成为独立字段，直接用于Kibana可视化。这种方式链路更短，资源消耗更低。
  示例Filebeat配置片段：

  processors:
    - decode_json_fields:
        fields: ["message"]
        target: ""
        overwrite_keys: true
  

• 需复杂处理（字段转换/条件过滤/多源聚合等）：用Logstash。比如要把POWER状态转成布尔值、按设备类型过滤事件，或和其他数据源关联时，可用Logstash的json过滤器解析消息，再配合mutate等过滤器处理字段。如果要将每个POWER字段拆分为独立事件（比如统计单开关状态变化趋势），还能用到split过滤器。

问题2：确认事件是否经过Logstash #

三种简单验证方法：

• 添加标识字段：在Logstash的filter或output块中插入自定义字段，比如：

  filter {
    mutate {
      add_field => { "processed_by_logstash" => "true" }
    }
  }
  

  之后在Kibana的Discover页面查看事件，若存在processed_by_logstash: true字段，说明事件经过Logstash处理。
• 检查Filebeat配置：确认Filebeat的输出配置指向Logstash，而非直接发送到Elasticsearch。正确配置示例：

  output.logstash:
    hosts: ["your-logstash-ip:5044"]
  

  若配置了output.elasticsearch，事件会直接发往Elasticsearch，跳过Logstash。
• 查看Logstash日志：查看Logstash运行日志（默认路径/var/log/logstash/logstash-plain.log），搜索beats或elasticsearch相关条目，若有Received event或Successfully sent events类日志，说明Logstash在处理事件。

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