Logstash 嵌套 JSON 解析答案

【问题标题】：Logstash nested JSON ParseLogstash 嵌套 JSON 解析
【发布时间】：2017-11-07 16:38:34
【问题描述】：

我有一个 Logstash 配置，它从 Kafka 接收 JSON 输入并将其发送到 Elasticsearch 输出。

接收到的 JSON 结构如下：

TrackingData: {
    "path": "/hello/world",
    "method": "GET",
    "requestDateTime": "2017-11-07T16:12:02.635",
    "duration": 5104684,
    "status": 200
}

我想将status 值添加为一个新字段，以便以后在使用 Kibana 查看 EL 数据时它可以显示在一个新列中。我知道为此我应该在 Logstash 配置中包含一个 mutate filter。

为了将 status 值从 JSON 结构中取出并放入新字段中，Ruby 字符串应该是什么？

我的 Logstash 配置目前如下：

input {
    kafka {
        bootstrap_servers => ["kafka:9092"]
        topics => "test-topic"
        group_id => "test-topic-group"
    }
}
filter {
    mutate {
        add_field => {
            "root_field" => "%{[message]}"
            "status_field" => "%{[message][TrackingData][status]}"
        }
    }
}
output {
    stdout {
        codec => rubydebug
    }   
    elasticsearch {
        hosts => ["elk:9200"]
}

当 Logstash 处理消息时，我可以在 Kibana 中看到 root_field 解析正常（我添加此字段只是为了证明 json 是可解析的）但 status_field 在中显示为 %{[message][TrackingData][status]} Kibana（即解析字符串一定不正确）。

应该如何指定解析字符串以便从上面的示例 JSON 结构中获取 status 值？

【问题讨论】：

标签： json ruby logstash

【解决方案1】：

有两件事是错误的。第一个问题是接收到的 JSON 本身是错误的。 TrackingData 没有被 " 包围。正确的 JSON 应该是这样的：

"TrackingData": {
    "path": "/hello/world",
    "method": "GET",
    "requestDateTime": "2017-11-07T16:12:02.635",
    "duration": 5104684,
    "status": 200
}

其次，我在 Logstash 的配置中添加了 codec => "json" 并删除了 filter 代码块：

input {
    kafka {
        bootstrap_servers => ["kafka:9092"]
        topics => "test-topic"
        group_id => "test-topic-group"
        codec => "json"
    }
}
output {
    stdout {
        codec => rubydebug
    }   
    elasticsearch {
        hosts => ["elk:9200"]
}

现在，接收到的 JSON 的键/值被解析为单独的字段，因此可以根据需要显示在 Kibana 的单独列中（或者如果根本不需要，则将其过滤掉）。

【讨论】：