elasticsearch - 如何使用json过滤器将我的json日志文件存储到logstash

Question

这是我的json日志文件。我正在尝试通过logstash将文件存储到我的Elastic-Search中。

{ "id": "135569", "title" : "Star Trek Beyond", "year":2016 , "genre": 
["Action", "Adventure", "Sci-Fi"] }

将数据存储到elasticSearch中之后，我的结果如下

{
    "_index": "filebeat-6.2.4-2018.11.09",
    "_type": "doc",
    "_id": "n-J39mYB6zb53NvEugMO",
    "_score": 1,
    "_source": {
      "@timestamp": "2018-11-09T03:15:32.262Z",
      "source": "/Users/jinwoopark/Jin/json_files/testJson.log",
      "offset": 106,
      "message": """{ "id": "135569", "title" : "Star Trek Beyond", "year":2016 , "genre":["Action", "Adventure", "Sci-Fi"] }""",
      "id": "%{id}",
      "@version": "1",
      "host": "Jinui-MacBook-Pro.local",
      "tags": [
        "beats_input_codec_plain_applied"
      ],
      "prospector": {
        "type": "log"
      },
      "title": "%{title}",
      "beat": {
        "name": "Jinui-MacBook-Pro.local",
        "hostname": "Jinui-MacBook-Pro.local",
        "version": "6.2.4"
      }
    }
  }

我想做的是

我只想将“类型值”存储到消息字段中，并将其他值(例如id，title)存储到额外的字段(创建的字段，即id和title字段)中。但是多余的字段存储为空值(％{id}，％{title})。似乎我需要修改我的logstash json过滤器，但是在这里我需要您的帮助。

我当前的logstash配置如下

input {
    beats {
     port => 5044
    }
}

filter {
    json {
            source => "genre" //want to store only genre (from json log) into message field 
    }
    mutate {
            add_field => {
                    "id" => "%{id}" // want to create extra field for id value from log file
                    "title" => "%{title}" // want to create extra field for title value from log file
            }
    }
    date {
            match => [ "timestamp", "dd/MM/yyyy:HH:mm:ss Z" ]
    }
}
output {
    elasticsearch {
            hosts => ["http://localhost:9200"]
            index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    }
    stdout {
            codec => rubydebug
    }
}

Answer 1

当您告诉json过滤器source是genre时，它应该忽略文档的其余部分，这将解释为什么您没有id或title。

似乎您应该解析整个json文档，并使用mutate->replace plugin将genre的内容移动到message。