我在使用 Logstah 处理数据时，在 logstash-plain.log 文件中发现了以下的错误信息：

1
2
3
4
[2023-03-31T00:10:28,949][ERROR][logstash.filters.ruby    ][main][d6838068510d1ed4e2d1025930d8680ca59bdef970aaa32f4e2d8d28a09ee6d3] Ruby exception occurred: unexpected token at '' {:class=>"JSON::ParserError", :backtrace=>["json/ext/Parser.java:238:in `parse'", "/usr/share/logstash/vendor/bundle/jruby/2.6.0/gems/json-2.6.3-java/lib/json/common.rb:216:in `parse'", "(ruby filter code):3:in `block in filter_method'", "/usr/share/logstash/vendor/bundle/jruby/2.6.0/gems/logstash-filter-ruby-3.1.8/lib/logstash/filters/ruby.rb:96:in `inline_script'", "/usr/share/logstash/vendor/bundle/jruby/2.6.0/gems/logstash-filter-ruby-3.1.8/lib/logstash/filters/ruby.rb:89:in `filter'", "/usr/share/logstash/logstash-core/lib/logstash/filters/base.rb:159:in `do_filter'", "/usr/share/logstash/logstash-core/lib/logstash/filters/base.rb:178:in `block in multi_filter'", "org/jruby/RubyArray.java:1865:in `each'", "/usr/share/logstash/logstash-core/lib/logstash/filters/base.rb:175:in `multi_filter'", "org/logstash/config/ir/compiler/AbstractFilterDelegatorExt.java:134:in `multi_filter'", "/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:301:in `block in start_workers'"]}
[2023-03-31T00:10:28,981][WARN ][logstash.filters.split   ][main][22c40e0aa24b6a13b219d12e11012938eec3bbf9da6d4f2672324181399c74af] Only String and Array types are splittable. field:data_arr is of type = NilClass
[2023-03-31T00:10:29,008][WARN ][logstash.filters.json    ][main][b38819c35002a46187eeb47979e43d989d40afd9c95cdea1b6ac74cfc08927e6] Error parsing json {:source=>"@tmp", :raw=>"%{data_arr}", :exception=>#<LogStash::Json::ParserError: Unexpected character ('%' (code 37)): expected a valid value (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
 at [Source: (byte[])"%{data_arr}"; line: 1, column: 2]>}

从该异常信息可以知道是 filter 的 ruby 这里出错了——JSON 转换错误，但是并没告诉我出错的内容是什么。因为这里处理的是 APP 用户行为的埋点日志，数据量很大，我无法快速确定 APP 端到底上报了什么内容导致出现该异常，从而无法快速通知相关人员及时排查和修改。我需要调整 Logstash 哪些配置以便在出现解析失败时可以记录原始数据？

我的 Logstash 配置文件如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
input {
    beats {
        port => "8085"
    }
}

filter {
    json {
        source => "message"
    }

    ruby {
        code => "
            event.set('data_arr', JSON.parse(event.get('data')))
            event.set('logstash_time', Time.now.to_i)
            "
        remove_field => ["data"]
    }

    split {
        field => "data_arr"
        remove_field => "tags"
    }

    mutate {
        add_field => {"@tmp" => "%{data_arr}"}
    }

    json {
        source => "@tmp"
    }
}

output {
    kafka {
        codec => json
        topic_id => "behavior-log"
        bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092"
        client_id => "logstash-behavior-log-01"
        acks => "all"
    }
}

你可以在出现异常时，把原始数据记录输出到文件上。fiter 的 ruby 代码出错时，默认情况下会把 _rubyexception 添加到事件的 tags 字段上。例如在 ruby 代码中将空字符串传给 JSON.parse() 方法，事件 tags 字段中会有以下的错误信息：

1
2
3
4
"tags":[
    "_rubyexception: unexpected token at ''",
    "_rubyexception"
]

我们可以通过判断 tags 数组中是否有 _rubyexception 值，如果有，说明 ruby 代码有异常，这个时候可以在输出的时候把该数据写到文件上（也可以选择其他存储，如 Elasticsearch）。

根据你的配置，我添加了以下修改，你可以根据自己的实际情况再做修改：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
input {
    beats {
        port => "8085"
    }
}

filter {
    json {
        source => "message"
    }

    ruby {
        code => "
            event.set('data_arr', JSON.parse(event.get('data')))
            event.set('logstash_time', Time.now.to_i)
            "
        remove_field => ["data"]
        tag_with_exception_message => true
    }

    split {
        field => "data_arr"
        remove_field => "tags"
    }

    mutate {
        add_field => {"@tmp" => "%{data_arr}"}
    }

    json {
        source => "@tmp"
    }
}

output {
    kafka {
        codec => json
        topic_id => "behavior-log"
        bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092"
        client_id => "logstash-behavior-log-01"
        acks => "all"
    }
    if "_rubyexception" in [tags] {
        file {
            path => "/var/log/logstash/error.log"
        }
    }
}

（END）