Logstash kafka input plugin mutate problem

Dears,

I'm trying to load data from kafka topic but have some problem with if statement in filter.
This part of the setup doesn't work and I don't know why:

replace => { "[@metadata][index_prefix]" => "kafka-%{+YYYY.MM.dd}" }

All messages from Kafka are loading to index called other-*

This is my almost whole config of logstash:

input {
  kafka {
    bootstrap_servers => "10.10.131.18:9093"
    type => "kafka_type"
    topics => "TEST_ELK_TOPICS"
    decorate_events => true
    codec => "json"
    # ssl => true
    ssl_truststore_location => "/etc/logstash/certs/KafkaTruststore.p12"
    ssl_truststore_password => "elasticpwd"
    sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='elastic' password='elastic123';"
    sasl_mechanism => "SCRAM-SHA-512"
    security_protocol => "SASL_SSL"
 }
}

filter {
  if "tomcat" not in [tags] or "app" not in [tags] {
  xml {
    source => "message"
    store_xml => false
    force_array => false
    xpath => [
      "/log//isomsg/field[@id='37']/@value", "hi.rrn",
      "/log//isomsg/field[@id='0']/@value", "hi.mti",
      "/log//isomsg/field[@id='39']/@value", "hi.rc",
      "/log//*[contains(name(),'exception')]/@name", "hi.exception_name",
      "/log//*[contains(name(),'exception')]/text()", "hi.exception",
      "/log//error/text()", "hi.error",
      "/log/@realm", "hi.realm",
      "/log/@at", "hi.xml_csshi_date",
      "/log//routing/dst-iface/text()", "hi.dst-iface"
    ]
  }
  }

  if "kafka_type" in [type] {
    json {
      source => "message"
    }
    mutate {
      add_field => { "kafka" => "%{[@metadata][kafka]}" }
      replace => { "[@metadata][index_prefix]" => "kafka-%{+YYYY.MM.dd}" }
      replace => { "[@metadata][_id]" => "%{[@metadata][kafka][offset]}%{[@metadata][kafka][timestamp]}" }
    }
  }

....
else if "webbapisrv" in [host][hostname] {
    mutate {
      replace => { "[@metadata][index_prefix]" => "webapi-%{+YYYY.MM.dd}" }
    }
  }
  else {
    mutate {
      replace => { "[@metadata][index_prefix]" => "other-%{+YYYY.MM.dd}" }
    }
  }
}

output {
  elasticsearch {
    hosts => ["https://${HOSTNAME}:9200"]
    cacert => '/etc/logstash/certs/ca.crt'
    user => 'logstash_internal'
    password => '${ES_PWD}'
    ilm_enabled => false
    document_id => "%{[@metadata][_id]}"
    index => "%{[@metadata][index_prefix]}"
  }
}

Could you advise me what is wrong, please? Thanks

Best Regards,
Dan

A problematic condition was placed in bad place in configuration. The correction of config file solved the issue. Thanks.

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.