Show Kafka topic title as a field in Kibana, logstash add_field?

user_191647 0
Viewed 1393

I have logstash with ElasticSearch & Kibana 7.6.2
I connect logstash to Kafka as follows:

input {
  kafka {
    bootstrap_servers => "******"
    topics_pattern => [".*"]
    decorate_events => true
    add_field => { "[topic_name]" => "%{[@metadata][kafka][topic]}"}
  }
}
filter {
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}
output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "logstash"
    document_type => "logs"
  }
}

It's OK and work. But I field topic_name show as %{[@metadata][kafka][topic]}

kibana-capture-1

How can I fix it?

logstash kibana elasticsearch apache-kafka
2 Answers

The syntax of the sprintf format you are using ( %{[@metadata][kafka][topic]} ) to get the value of that field is correct.

Allegedly there is no such field @metadata.kafka.topic in your document. Therefore the sprintf can't obtain the field value and as a result, the newly created field contains the sprintf call as a string.

However, since you set decorate_events => true, the metadata fields should be available as stated in the documentation (https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html):

Metadata is only added to the event if the decorate_events option is set to true (it defaults to false).

I can imagine that the add_field action set in the input plugin causes the issue. Since the decorate_events option first enables the addition of the metadata fields, the add_field action should come at second place after the input plugin.

Your configuration would then look like this:

input {
  kafka {
    bootstrap_servers => "******"
    topics_pattern => [".*"]
    decorate_events => true
  }
}

filter {
  mutate{
    add_field => { "[topic_name]" => "%{[@metadata][kafka][topic]}"}
  }

  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "logstash"
    document_type => "logs"
  }
}
user_114866700

How about 

add_field => { "topic_name" => "%{[@metadata][kafka][topic]}"}

i.e. [topic_name] -> topic_name

user_81603180
Related