使用Filebeat提取AWS WAF日志JSON至根节点时Logstash报TypeError

将S3中的AWS WAF日志通过Filebeat+Logstash导入Elasticsearch，初始配置中Filebeat把JSON解码到json字段下，Logstash可正常处理。修改Filebeat配置将日志JSON字段提取到文档根节点后，即便删除Logstash对应处理逻辑，仍触发org.jruby.exceptions.TypeError: (TypeError) no implicit conversion of String into Hash错误，需排查解决。

相关配置信息 #

初始配置 #

Filebeat配置

processors:
    - add_host_metadata:
        when.not.contains.tags: forwarded
    - add_cloud_metadata: ~
    - add_docker_metadata:
        host: "unix:///var/run/docker.sock"
    - add_kubernetes_metadata: ~
    - decode_json_fields:
        fields: ["message"]
        target: "json"

Logstash过滤器配置

filter {
   if ([aws][s3][object][key]){
        if "my-waf" in [aws][s3][bucket][name] {
            grok {
                match => { "[aws][s3][bucket][name]" => "%{GREEDYDATA:servicename}"}
            }
            date { 
                match => [ "[json][timestamp]", "UNIX_MS" ]
            }
            if ([json][httpRequest][headers]){
                ruby { 
                    code => ' 
                        event.get("[json][httpRequest][headers]").each { |a| 
                            name = a["name"]
                            value = a["value"]
                            event.set("[json][httpRequest][headersFlattened][#{name}]", value)
                        } 
                    ' 
                }
            }
        } 
    mutate {
        rename => {
            "servicename" => "[service][name]"
        }
    }
}

修改后的配置 #

Filebeat配置

processors:
    - add_host_metadata:
        when.not.contains.tags: forwarded
    - add_cloud_metadata: ~
    - add_docker_metadata:
        host: "unix:///var/run/docker.sock"
    - add_kubernetes_metadata: ~
    - decode_json_fields:
        fields: ["message"]
        target: ''

Logstash过滤器配置（删除部分逻辑后仍报错）

date { 
    match => [ "[timestamp]", "UNIX_MS" ]
}
if ([httpRequest][headers]){
    ruby { 
        code => ' 
            event.get("[httpRequest][headers]").each { |a| 
                name = a["name"]
                value = a["value"]
                event.set("[httpRequest][headersFlattened][#{name}]", value)
            } 
        ' 
    }
}

错误日志 #

[2024-06-06T07:33:39,035][INFO ][org.logstash.beats.BeatsHandler][main][3de0d69f427460aef4da30cf326fa3108ef7e08a9518d98711813b321a0d15fa] [local: XXXX, remote: XXXXX] Handling exception: org.jruby.exceptions.TypeError: (TypeError) no implicit conversion of String into Hash (caused by: org.jruby.exceptions.NoMethodError: (NoMethodError) undefined method `accept' for nil:NilClass) [2024-06-06T07:33:39,036][WARN ][io.netty.channel.DefaultChannelPipeline][main][3de0d69f427460aef4da30cf326fa3108ef7e08a9518d98711813b321a0d15fa] An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception. org.jruby.exceptions.TypeError: (TypeError) no implicit conversion of String into Hash
        at org.jruby.RubyHash.merge!(org/jruby/RubyHash.java:2097) ~[jruby.jar:?]
        at org.jruby.RubyHash.merge(org/jruby/RubyHash.java:2129) ~[jruby.jar:?]
        at usr.share.logstash.vendor.bundle.jruby.$3_dot_1_dot_0.gems.logstash_minus_input_minus_beats_minus_6_dot_6_dot_3_minus_java.lib.logstash.inputs.beats.decoded_event_transform.transform(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-input-beats-6.6.3-java/lib/logstash/inputs/beats/decoded_event_transform.rb:16) ~[?:?]
        at org.jruby.RubyHash.each(org/jruby/RubyHash.java:1587) ~[jruby.jar:?]
        at org.jruby.java.proxies.MapJavaProxy.each(org/jruby/java/proxies/MapJavaProxy.java:587) ~[jruby.jar:?]
        at usr.share.logstash.vendor.bundle.jruby.$3_dot_1_dot_0.gems.logstash_minus_input_minus_beats_minus_6_dot_6_dot_3_minus_java.lib.logstash.inputs.beats.decoded_event_transform.transform(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-input-beats-6.6.3-java/lib/logstash/inputs/beats/decoded_event_transform.rb:12) ~[?:?]
        at usr.share.logstash.vendor.bundle.jruby.$3_dot_1_dot_0.gems.logstash_minus_input_minus_beats_minus_6_dot_6_dot_3_minus_java.lib.logstash.inputs.beats.codec_callback_listener.process_event(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-input-beats-6.6.3-java/lib/logstash/inputs/beats/codec_callback_listener.rb:22) ~[?:?]
        at usr.share.logstash.vendor.bundle.jruby.$3_dot_1_dot_0.gems.logstash_minus_input_minus_beats_minus_6_dot_6_dot_3_minus_java.lib.logstash.inputs.beats.patch.accept(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-input-beats-6.6.3-java/lib/logstash/inputs/beats/patch.rb:10) ~[?:?]
        at usr.share.logstash.vendor.bundle.jruby.$3_dot_1_dot_0.gems.logstash_minus_codec_minus_plain_minus_3_dot_1_dot_0.lib.logstash.codecs.plain.decode(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-codec-plain-3.1.0/lib/logstash/codecs/plain.rb:54) ~[?:?]
        at usr.share.logstash.vendor.bundle.jruby.$3_dot_1_dot_0.gems.logstash_minus_input_minus_beats_minus_6_dot_6_dot_3_minus_java.lib.logstash.inputs.beats.patch.accept(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-input-beats-6.6.3-java/lib/logstash/inputs/beats/patch.rb:9) ~[?:?]
        at usr.share.logstash.vendor.jruby.lib.ruby.stdlib.delegate.method_missing(/usr/share/logstash/vendor/jruby/lib/ruby/stdlib/delegate.rb:87) ~[?:?]
        at usr.share.logstash.vendor.bundle.jruby.$3_dot_1_dot_0.gems.logstash_minus_input_minus_beats_minus_6_dot_6_dot_3_minus_java.lib.logstash.inputs.beats.message_listener.onNewMessage(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-input-beats-6.6.3-java/lib/logstash/inputs/beats/message_listener.rb:52) ~[?:?] Caused by: org.jruby.exceptions.NoMethodError: (NoMethodError) undefined method `accept' for nil:NilClass
        ... 1 more

排查与解决 #

原因分析 #

错误源于Logstash的Beats输入插件合并事件字段时，日志字段与Logstash/Filebeat自动添加的元数据字段重名且类型冲突：

• 比如AWS WAF日志中的tags字段为字符串/数组类型，但Logstash Beats插件默认会添加tags数组字段，合并时字符串无法转哈希触发错误；
• Filebeat将JSON解码到根节点时，若日志包含host、tags等与Filebeat内置元数据同名的键，会覆盖或导致类型不兼容。

解决步骤 #

1. 避免字段重名：修改Filebeat的decode_json_fields配置，排除冲突字段或禁止覆盖已有元数据：

- decode_json_fields:
    fields: ["message"]
    target: ''
    overwrite_keys: false  # 禁止覆盖Filebeat原有元数据字段
    exclude_fields: ["tags", "host"]  # 排除与元数据冲突的日志字段

若必须解码到根节点，可开启overwrite_keys: true，但需确保冲突字段类型与元数据一致。

2. 验证事件结构：在Logstash中添加stdout { codec => rubydebug }输出原始事件，检查是否存在字段类型混合（如同一字段既是字符串又是哈希）。

3. 修复Ruby逻辑：若保留HTTP头处理代码，需先判断字段类型，避免遍历非数组/哈希数据：

ruby { 
    code => ' 
        headers = event.get("[httpRequest][headers]")
        if headers.is_a?(Array)
            headers.each { |a| 
                name = a["name"]
                value = a["value"]
                event.set("[httpRequest][headersFlattened][#{name}]", value)
            } 
        end
    ' 
}

4. 临时恢复方案：紧急情况下改回Filebeat初始配置（解码到json字段），确保日志正常导入后再逐步排查冲突。

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