Анализ события JSON в Logstash

У меня есть журнал в следующем формате, это обычный 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, чтобы увидеть все поля json в Kibana как отдельные (проанализированные) поля? У меня проблема с полем "message", в которое вложены поля json. У меня нет проблем, чтобы разобрать событие, которое имеет строку в "сообщении", но не JSON.

Мои попытки:

1 Я пытался сказать Filebeat, что это JSON со следующей конфигурацией: (и ничего не делать на стороне LS)

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

Результат для меня странный, потому что я получил "message" в виде строки в Kibana, где все: заменены на =>

{
    "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"    
} 

Логи вообще не появились в Кибане, я получил следующие ошибки в ЛС:

[2019-03-08T09: 55: 47,484][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"=>"ERS6XGkBgE-US7A6Mvt", "status"=>400, "error"=>{"type"=>"mapper_parsing_exception", "reason"=>" не удалось проанализировать поле [json.message] типа [ключевое слово ]", " reason_by"=>{"type"=>"invalid_state_exception "," reason "=>" Не удается получить текст на START_OBJECT в 1:461"}}}}} [2019-03-08T09:55:47,485][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"=>" не удалось проанализировать поле [json.message] типа [ключевое слово ]", " reason_by"=>{"type"=>"invalid_state_exception "," reason "=>" Невозможно получить текст на START_OBJECT в 1: 461 "}}}}}

Этот фильтр хорошо работает для меня, если поле "сообщение" является строкой (не JSON).

Любые идеи, как разобрать вложенный JSON в "поле сообщения"?