input {
tcp {
port => 5004
codec => multiline {
pattern => "^index::"
negate => true
what => "previous"
}
}
}
filter {
if "index::" in [message] {
grok {
match => { "message" => "%{}" }
}
}
else if "index::" in [message] {
grok {
match => { "message" => "%{}" }
}
}
else {
drop { }
}
}
output {
stdout { codec => rubydebug }
kafka {
bootstrap_servers => ["AUSILKFKWA01.us.er.com:9092,AUSILKFKQB01.us.er.com:9092"]
topic_id => "try1"
compression_type => "snappy"
value_serializer => 'org.apache.kafka.common.serialization.ByteArraySerializer'
}
}
my log looks like this : ::::
i am using logstash 6.5.4 version
"message" => "index::goa_route sourcetype:: rash host::ausflsexsslap23.us.dell.com 2019-04-19 07:59:46,242 INFO - \tFri Apr 19 07:59:46 CDT 2019\tID:<281683.1555678786240.0>\tes_svc7_dr_osb\tes_svc7_dr_osb_ms09\tSiebelSessionAdapter/3_7/ProxyServices/SiebelSessionAdapter\tSIEBEL_WS_REQUEST_HEADER_MSG\t[messageData([UniqueReqId = 2b1d4a3b-68da-40bd-a2a6-cb03250ec390]; [SessionType = Stateless]; [SessionToken = sAiS.bW4zs0kwZaO-8FBXaS.iWgufG7tp58zMfRuX6I2om-Dh5xUBPenApZck2ac9nyouWn9z7usM4ZwdcZuSbMio91F7lJOqJImUjG9mT4NRNCA-jURUreb4600qCBIjlEmwNQqiOhqLMYAW1rBN2vKBGXDbZ5v4hr-vGIMn.PiqLG5HK-SMrNjBEB5AoVSyjPIwbsBRnTgkkfMkWEDQbxe5YEaBH8IVCDGKLCMNgT8Yl3hEgtD4OOzB4mnx.ZkO.bNjk-H8oqLL4jPCvp4yfvn9zdpyoZtvQQGiok.FPHHWPYWTL9qWJ12vG5gbUDC3Scs26nRpJUOb.JlhzhnqA__]; [UsernameToken = ]; [PasswordText = ])]\tms \t",
task : like above message i used to get different type of indexes so each index type i have to route to different kafka topic so please help the filter logic with if condition
looking quick help