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