How to index Kafka topic key value as a field using logstash?

Question

I would like to know to include Kafka topic key value along with the message as a separate field for indexing in Elasticsearch using Logstash .. the Kafka 'topic' weather contains the following messages as key value pairs:

BOSTON:99
BOSTON:89
NYC:75
NYC:85

I am using the following Logstash configuration to do indexing:

input {
    kafka { 
    bootstrap_servers => "localhost:9092"
    topics => ["weather"]
    }
}
output {
   elasticsearch {
      hosts => ["localhost:9200"]
      index => "weather"
      workers => 1
    }
}

However, only values are getting indexed, but not the key. The index looks like this:

 {
    "_index" : "weather",
    "_type" : "doc",
    "_id" : "48cGgXMBtkZlibwHwyWW",
    "_score" : 1.0,
    "_source" : {
      "message" : "99",
      "@timestamp" : "2020-07-24T13:32:50.820Z",
      "@version" : "1"
    }
  {
    "_index" : "weather",
    "_type" : "doc",
    "_id" : "48cGgXMBtkZlibwHwyWW",
    "_score" : 1.0,
    "_source" : {
      "message" : "89",
      "@timestamp" : "2020-07-24T13:32:50.820Z",
      "@version" : "1"
    }

I also need to have the city name (key of the topic) in the index. Once I have both, the end goal is to do different types of visualization in Kibana for each city's weather.

Thanks in advance.

user_2462408 · Answer

input {
    kafka { 
    bootstrap_servers => "localhost:9092"
    topics => ["weather"]
    decorate_events => true         (!!!)
    }
}
filter {
    mutate {
        add_field => {
            "key" => %{[@metadata][kafka][key]}" (!!!)
        }
    }
}
output {
elasticsearch {
    hosts => ["localhost:9200"]
    index => "weather"
    workers => 1
    }
}

ref : https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#_metadata_fields

decorate_events option set to 'true'
add 'add_field' filter

user_7304765 · Answer

I included this in the conf file and I am now able to see the key information:

filter {
        mutate {
          copy => { "[@metadata][kafka]" => "kafka" }
        }
}

How to index Kafka topic key value as a field using logstash?

2 Answers