在 Logstash 中解析 JSON 事件

Question

我已经按以下格式登录，它是一个带有嵌套字段的普通 json。

{
    "level": "info",
    "message": {
        "req": {
            "headers": {
                "host": "localhost:8080",
                "connection": "keep-alive",
                "x-forwarded-for": "192.168.1.1, 1.1.1.1",
                "x-forwarded-proto": "http"
            },
            "url": "/products?userId=493d0aec-a9a7-42a3",
            "method": "GET",
            "originalUrl": "/products?userId=493d0aec-a9a7-42a3",
            "params": {
                "0": "/products"
            },
            "query": {
                "userId": "493d0aec-a9a7-42a3"
            },
            "body": ""
        },
        "res": {
            "headers": {
                "traceid": "ac586e4e924048",
                "x-correlation-id": "57d7920d-b623-48f8",
                "content-type": "application/json;charset=UTF-8",
                "content-length": "2",
                "date": "Fri, 08 Mar 2019 09:55:45 GMT",
                "connection": "close"
            },
            "statusCode": 200,
            "body": "[]"
        },
        "gateway": "internal"
    },
    "correlationId": "57d7920d-b623-48f8",
    "timestamp": "2019-03-08T09:55:45.833Z"
}

如何使用 Filebeat 和 Logstash 正确解析它，以将 Kibana 中的所有 json 字段视为单独的（已解析）字段？我对具有嵌套 json 字段的“消息”字段有疑问。我可以毫无问题地解析“消息”中包含字符串但不包含 json 的事件。

我的尝试：

1. 我尝试告诉 Filebeat 它是一个具有以下配置的 json:（并且在 LS 端不执行任何操作）

filebeat.inputs:
- type: stdin
  json.keys_under_root: true
  json.add_error_key: true

结果对我来说很奇怪，因为我在 Kibana 中收到了字符串“message”，其中所有 : 都被替换为 =>

{
    "req" => {
        "originalUrl" => "/offers", "params" => {
            "0" => "/offers"
        }, "query" => {}, "body" => "", "headers" => {
            "accept-encoding" => "gzip", "user-agent" => "okhttp/3.8.1", "x-consumer-id" => "f2a6e4cd-2224-4535 .....

“消息”之外的其他字段均能正确解析

2. 我在 Filebeat 端什么都没做，只是在 LS 中使用了过滤器：

json {   
    source => "message"    
    target => "message_json"    
} 

Kibana 中根本没有出现日志，我在 LS 中收到以下错误：

[2019-03-08T09：55：47,084] [警告] [logstash.outputs.elasticsearch] 无法将事件索引到 Elasticsearch。 {:status=>400, :action=>["index", {:_id=>nil, :_index=>"filebeat-6.5.0-2019.03.08-sdx", :_type=>"doc", :routing=>nil}, #], :response=>{"index"=>{"_index"=>"filebeat-6.5.0-2019.03.08-sdx", "_type"=>"doc", "id"=>"ERS6XGkBgE-US7A6Mvt", "status"=>400, "error"=>{"type"=>"mapper_parsing_exception", "reason"=>"无法解析类型为 [keyword] 的字段 [json.message]", "caused_by"=>{"type"=>"illegal_state_exception", “原因”=>“无法在 1:461 的 START_OBJECT 上获取文本”}}}}} [2019-03-08T09:55:47,085][WARN][logstash.outputs.elasticsearch] 无法将事件索引到 Elasticsearch。 {:status=>400, :action=>["index", {:_id=>nil, :_index=>"filebeat-6.5.0-2019.03.08-sdx", :_type=>"doc", :routing=>nil}, #], :response=>{"index"=>{"_index"=>"filebeat-6.5.0-2019.03.08-sdx", "_type"=>"doc", "id"=>"EhS6XGkBgE-US7A6Mvt", "status"=>400, "error"=>{"type"=>"mapper_parsing_exception", "reason"=>"无法解析类型为 [keyword] 的字段 [json.message]", "caused_by"=>{"type"=>"illegal_state_exception", “原因”=>“无法在 1:461 的 START_OBJECT 上获取文本”}}}}}

如果“消息”字段是字符串（而不是 json），则此过滤器对我来说工作正常。

关于如何解析“消息字段”中的嵌套 json，有什么想法吗？