Logstash kafka input plugin mutate problem

[d.silwon (Daniel)]

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

[d.silwon (Daniel)]

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