grok解析,出现一个字段值 重复匹配 ,message字段传过来的是正常的

我把docker里面的日志内容通过TCP输出到logstash , 再由logstash输出到kafka , 最后 logstash从kafka取数据 进行数据清洗 存入ES
 
logstash 输出到 kafka是没有做任何数据处理的一切正常, 但 logstash再从kafka拿数据时匹配到的字段,本应是单个值, 但最后的结果是匹配到了多个相同的值 并以集合的方式存入了该字段   我debug时  手动输入参数模拟kafka的消息发送, 这时又是正常的了
 
logstash接收TCP的消息并存入kafka
 
logstash配置文件
 

# 指定syslog的端口号

input {

    syslog{

        # syslog发送消息到本机的9990端口

        port => 9990

    }

}

# 输出消息到kafka

output{

    kafka{

        codec => json

        topic_id => "syslog_test"

        bootstrap_servers => "bigdata01:9092"

    }

#    stdout{codec=>rubydebug}

}

输出信息
 

{

        "@timestamp" => 2019-05-05T09:51:20.147Z,

          "@version" => "1",

              "tags" => [

        [0] "_grokparsefailure_sysloginput"

    ],

          "severity" => 0,

              "host" => "fe80:0:0:0:19b1:e3b0:6bf9:d9e0%2",

          "priority" => 0,

    "severity_label" => "Emergency",

          "facility" => 0,

           "message" => "<30>May  5 17:51:20 **********test********[8685]: 2019-05-05 09:51:20.000  INFO 1 --- [pool-1-thread-1] llerOne$$EnhancerBySpringCGLIB$$6654d185 : clzzName : aaaaaa361,methodName : 691,statuCode : 3301\n",

    "facility_label" => "kernel"

}

{

        "@timestamp" => 2019-05-05T09:51:22.008Z,

          "@version" => "1",

              "tags" => [

        [0] "_grokparsefailure_sysloginput"

    ],

          "severity" => 0,

              "host" => "fe80:0:0:0:19b1:e3b0:6bf9:d9e0%2",

          "priority" => 0,

    "severity_label" => "Emergency",

          "facility" => 0,

           "message" => "<30>May  5 17:51:22 **********test********[8685]: 2019-05-05 09:51:22.000  INFO 1 --- [pool-1-thread-1] llerOne$$EnhancerBySpringCGLIB$$6654d185 : clzzName : aaaaaa4511,methodName : 2431,statuCode : 4641\n",

    "facility_label" => "kernel"

}

logstash接收kafka的消息并进行解析
 
logstash配置文件
 

input {

    kafka {

        bootstrap_servers => ["bigdata01:9092"]

        group_id => "syslogTest"

        #auto_offset_reset => "earliest"

        #consumer_threads => "5"

        decorate_events => "false"

        topics => ["syslog_test"]

        type => "log"

        codec => "json"

   }

}

filter{

  grok{

    match => {

         "message" => "clzzName : %{NOTSPACE:class_name},methodName : %{NOTSPACE:method_name},statuCode : %{NUMBER:statu_code:int}"

    }

  }

 mutate{

#      remove_field => "message"

  }

}

output {

    stdout{codec=>rubydebug}

}

输出信息
 

 
手动模拟kafka输出消息是可以正常解析的
 

 
Grok Debugger也是可以正常解析的