Logstash 过滤器中的 Grok 和 mutate 对输出没有影响

Question

Logstash 将通过远程主机上的 Filebeat 收集信息。 我最终的预期结果是创建一个图表来显示来自 HTTP 日志的响应时间。 所以我在日志消息中关注响应时间。

现在从 Kibana 得到的 json 消息是这样的

{
  "_index": "filebeat-7.9.2-2020.10.06-000001",
  "_type": "_doc",
  "_id": "NYaZB3UB-JR2hYtrwawX",
  "_version": 1,
  "_score": null,
  "_source": {
    "@timestamp": "2020-10-08T09:45:17.358Z",
    "agent": {
      "id": "acbc8cf5-7bc1-46ab-a9ad-8ee1fef26183",
      "name": "42dcb5cf4d4a",
      "type": "filebeat",
      "version": "7.9.2",
      "hostname": "42dcb5cf4d4a",
      "ephemeral_id": "e45f89ee-f355-4490-b79b-394a2226ffe7"
    },
    "log": {
      "offset": 5157015,
      "file": {
        "path": "/usr/logs/tomcat/test.com.access.2020-10-08.log"
      }
    },
    "message": "203.149.37.226 - - [08/Oct/2020:11:45:10 +0200] \"POST /TestserviceWarning/get_warning/render?method=get_warnings_onsite_status_excel&subDomain=testClick&webId=1196&scanId=4482 HTTP/1.1\" 200 1151 0.010",
    "tags": [
      "tomcat"
    ],
    "input": {
      "type": "log"
    },
    "ecs": {
      "version": "1.5.0"
    },
    "host": {
      "name": "42dcb5cf4d4a"
    }
  },
  "fields": {
    "@timestamp": [
      "2020-10-08T09:45:17.358Z"
    ]
  },
  "sort": [
    1602150317358
  ]
}

还有我的流水线配置

input {
        beats {
                port => 5000
        }
}

filter {
        if "tomcat" in [tags] {
                grok {
                        match => { "message" => "%{IPORHOST:client} %{DATA} %{DATA:user} \[%{DATA:logtimestamp} %{ISO8601_TIMEZONE:timezone}\] \"%{WORD:method} %{URIPATH:uri_path}(%{URIPARAM:params}|) %{DATA:protocol}\" %{NUMBER:code} (%{NUMBER:bytes}|%{DATA}) %{NUMBER:response_time_sec}"}
                        overwrite => [ "message" ]
                }
                mutate { add_field => {"respones_time" => "%{response_time_sec}"}
                }
        }
}

一旦管道被触发，我没有在我的 json 消息中看到任何额外的字段。结果保持不变。

---

于 2020 年 9 月 10 日编辑

浪费了一整天的时间，我还是做不出来。

所以我打开了 DEBUG 日志级别，看起来 Logstash 能够访问过滤器，但它以某种方式无法将过滤器应用于输出。

 org.logstash.config.ir.compiler.ComputeStepSyntaxElement@8d7da797
 P[filter-grok{"match"=>{"message"=>"%{IPORHOST:client} %{DATA} %{DATA:user} \\[%{DATA:logtimestamp} %{ISO8601_TIMEZONE:timezone}\\] \\\"%{WORD:method} %{URIPATH:uri_path}(%{URIPARAM:params}|) %{DATA:protocol}\\\" %{NUMBER:code} (%{NUMBER:bytes}|%{DATA}) %{NUMBER:response_time_sec}"}}|[file]/usr/share/logstash/pipeline/logstash.conf:9:17:```
grok {
                        match => { "message" => "%{IPORHOST:client} %{DATA} %{DATA:user} \[%{DATA:logtimestamp} %{ISO8601_TIMEZONE:timezone}\] \"%{WORD:method} %{URIPATH:uri_path}(%{URIPARAM:params}|) %{DATA:protocol}\" %{NUMBER:code} (%{NUMBER:bytes}|%{DATA}) %{NUMBER:response_time_sec}"}
                     }
```] 
 into 
 org.logstash.config.ir.compiler.ComputeStepSyntaxElement@8d7da797

那我又看到了新的异常消息

[INFO ] 2020-10-09 06:24:55.568 [Agent thread] agent - Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[INFO ] 2020-10-09 06:24:55.649 [Api Webserver] agent - Successfully started Logstash API endpoint {:port=>9600}
java.lang.UnsupportedOperationException: Reflective setAccessible(true) disabled

和

java.lang.IllegalAccessException: class io.netty.util.internal.PlatformDependent0$6 cannot access class jdk.internal.misc.Unsafe (in module java.base) because module java.base does not export jdk.internal.misc to unnamed module @72da5f0e

Answer 1

您的 logstash 配置看起来正确。下面是 logstash 的输出，应该在 Elastic search 中摄取相同的输出。我假设您在放入最新配置后重新摄取了日志文件。此外，请确保重新加载 logstash 配置并刷新 Kibana 中的索引，以便识别最新字段。

{
    "response_time_sec" => "0.010",
             "timezone" => "+0200",
           "@timestamp" => 2020-10-08T13:10:21.115Z,
                 "host" => "4ddee8887e1b",
               "client" => "203.149.37.226",
        "respones_time" => "0.010",
             "@version" => "1",
                "bytes" => "1151",
                 "path" => "/usr/share/logstash/stack/data/data.log",
         "logtimestamp" => "08/Oct/2020:11:45:10",
               "method" => "POST",
             "uri_path" => "/TestserviceWarning/get_warning/render",
                 "user" => "-",
             "protocol" => "HTTP/1.1",
                 "code" => "200",
               "params" => "?method=get_warnings_onsite_status_excel&subDomain=testClick&webId=1196&scanId=4482",
              "message" => "203.149.37.226 - - [08/Oct/2020:11:45:10 +0200] \"POST /TestserviceWarning/get_warning/render?method=get_warnings_onsite_status_excel&subDomain=testClick&webId=1196&scanId=4482 HTTP/1.1\" 200 1151 0.010"
}

虽然你想要达到的目标可以通过下面的方式简单地完成

filter {
   grok{
     match =>  { "message" => "%{IPORHOST:client} %{DATA} %{DATA:user} \[%{DATA:logtimestamp} %{ISO8601_TIMEZONE:timezone}\] \"%{WORD:method} %{URIPATH:uri_path}(%{URIPARAM:params}|) %{DATA:protocol}\" %{NUMBER:code} (%{NUMBER:bytes}|%{DATA}) %{NUMBER:response_time}"} 
   }
}

输出字段会喜欢

{
                "bytes" => "1151",
             "@version" => "1",
         "logtimestamp" => "08/Oct/2020:11:45:10",
               "method" => "POST",
                 "code" => "200",
               "client" => "203.149.37.226",
                 "host" => "169f8e2acdce",
             "timezone" => "+0200",
               "params" => "?method=get_warnings_onsite_status_excel&subDomain=testClick&webId=1196&scanId=4482",
                 "path" => "/usr/share/logstash/stack/data/data.log",
           "@timestamp" => 2020-10-08T13:15:31.576Z,
             "uri_path" => "/TestserviceWarning/get_warning/render",
                 "user" => "-",
             "protocol" => "HTTP/1.1",
    "response_time_sec" => "0.010",
              "message" => "203.149.37.226 - - [08/Oct/2020:11:45:10 +0200] \"POST /TestserviceWarning/get_warning/render?method=get_warnings_onsite_status_excel&subDomain=testClick&webId=1196&scanId=4482 HTTP/1.1\" 200 1151 0.010"
}