How to Look for Specific Tag in Filter Conditional

I've been playing with building an ELK stack with Kafka between Filebeat and Logstash for a couple weeks. On hosts, I have Filebeat configured with a Tags property like so:

  - type: log
    enabled: true
      - /var/log/kamailio
     - log_topic: "sbc-logs"
   tags: ["kamailio", "sbc"]

  hosts: ["kafka1:9092"]
  topic: "sbc-logs"
    - reachable_only: false
  required_acks: 1
 compression: gzip
 max_message_bytes: 1000000

Kafka is getting everything fine and if I don't bother with any conditional on the Logstash config filter, I get everything without no processing via the stdout rubydebug codec. Everything looks as expected initially, then I started trying to build filters with the intent of having conditionals for the different log files ingested from Kafka. However, what I'm seeing is the conditional is never met based on the syntax

filter {
  if "sbc" in [tags] {
    grok { #do parsing here }

What I get out of Logstash is the same when I wasn't trying to run a filter with the conditional above - it's just unformatted output to stdout:

Aug  3 18:56:53 atl-ls1 logstash: "message" => "{\"@timestamp\":\"2018-08-03T18:54:45.479Z\",\"@metadata\"
{\"name\":\"sbc01\"},\"source\":\"/var/log/kamailio\",\"offset\":27986653,\"message\":\"[2018-08-03-16:16:41] sbc01 kamailio[22423]: INFO: message: Dummy event.\",\"tags\":[\"kamailio\",\"sbc\"],\"input\"
Aug  3 18:56:53 atl-ls1 logstash: "@timestamp" => 2018-08-03T18:56:52.277Z,
Aug  3 18:56:53 atl-ls1 logstash: "@version" => "1"
Aug  3 18:56:53 atl-ls1 logstash: }

So what bothers me is that [tags] exists, but it looks like it's part of [message]. I would expect [tags] to not be nested in [message]. Is this normal given the setup? I've referenced other configs for Logstash performing conditional filtering with a specific tag, so I'm thinking it's likely not a syntax error on my part. If the placement of [tags] is expected, how should I be setting up that conditional to match on a specific tag in the array?

For comparison, I specifically fed in the kamailio log file with no other logs and got rid of the if conditional in filter, and got the expected formatted output I was looking for. Any advice on what I'm not quite getting is greatly appreciated!

The kafka input in your Logstash configuration is missing codec => json.

That...would quite nicely explain everything. Thanks for the quick turnaround!

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