logstash with multiple kafka topic input

Question

I would like to start a logstash instance with following setting:

input {
  kafka {
    topic_id => "topic_a"
    .......
  }
  kafka {
    topic_id => "topic_b"
    .......
  }

}
filter {
  json {
    source => "message"
  }
  uuid {
    target => "@uuid"
  }
  mutate {
    replace => { "message" => "%{message}" } # want to get the full json literal but does not work
    add_field => {
      "topic" => "%{topic_id}" # it does not work either
    }
  }

  # logic to apply different filter base on topic_id
  if [topic_id] =~ 'topic_a' { # this block seems never entered        
    mutate {
       replace => { "topic" => "topic_a" }
    }
  } else {
    .....
  }
}
output {
  .....
}

The output on my Kibana would should something like below:

topic : %{topic_id}

It suggested that the configuration above could not extract the topic_id. I have no idea on how to configure the filter part. Could anyone give a hint on this? Thanks.

BTW I'm using logstash-2.2.2

Edit: updated config according to logstash document, result still the same

Кирилл Полищук · Accepted Answer · 2016-03-29 14:30:35Z

By default Kafka Input Plugin do not include metadata info like: topic_id ..
You have to enable decorate_events options:

kafka {
    topic_id => "topic_a"
    decorate_events => true
  }

When this done, you can find your topic_id in: kafka array with topic key.

decorate_events

Value type is boolean Default value is false Option to add Kafka metadata like topic, message size to the event. This will add a field named kafka to the logstash event containing the following attributes: msg_size: The complete serialized size of this message in bytes (including crc, header attributes, etc) topic: The topic this message is associated with consumer_group: The consumer group used to read in this event partition: The partition this message is associated with key: A ByteBuffer containing the message key https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-decorate_events

Alain Collins · Accepted Answer · 2016-03-24 15:18:36Z

1

The documentation for add_field shows a different syntax than the one you're using. You might try it.

filter {
  mutate {
    add_field => { "foo_%{somefield}" => "Hello world, from %{host}" }
  }  
}

answered Mar 24, 2016 at 15:18

Alain Collins

16.4k2 gold badges34 silver badges56 bronze badges

2 Comments

Op Mule Over a year ago

hi, I have updated my question. I was actually using "%" instead of "$", but it doesn't work.

Alain Collins Over a year ago

If the [topic] field is still not being set, it's because either your input is not coming through one of the two inputs you've shown (where topic_id is set), or your conditional (if topic_id = xxx) is not true, or the mutate->replace fails (in which case the add_field will not run). Or something else you're not showing.

Collectives™ on Stack Overflow

logstash with multiple kafka topic input

2 Answers 2

Comments

2 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

2 Comments

Your Answer

Sign up or log in

Post as a guest

Related