How do I do conditional processing based on input

Elastic Stack Logstash
1

I have a case where my pipeline contains inputs from two different JDBC sources. In my filter, I want to run some things but not others.

My input config looks like this:

input {
    jdbc {
        jdbc_connection_string => "jdbc:oracle:thin:@192.168.56.101:1521/DB0"
        id => "db0"
        .
        .
        .
    }
    jdbc {
        jdbc_connection_string => "jdbc:oracle:thin:@192.168.56.101:1521/DB1"
        id => "db1"
        .
        .
        .
    }
}

Then I try a filter like so:

filter {

  if [id] == "db0" {
    ruby {
      init => "require 'time';"
      code => "
        puts id;
        start_temp = Time.iso8601(event.get('start_time').to_s);
        ts = LogStash::Timestamp.coerce (start_temp);
        event.set('@timestamp', ts);
      "
    }
}

This fails to run. When I have an event from my db0 source, nothing happens. I don't see the id printed in my log, and the date code (which I already know is correct) doesn't get called (i.e. @timestamp does not get updated).

How can I get the id in the conditional statement in my filter? Obviously whatever id I'm getting in the conditional is not set to either of the values of the id in my JDBC input definitions. (BTW I have exactly no clue about anything Ruby. I started looking at Logstash last week, and I have exactly no experience with Ruby other than this :slight_smile: ).

2

You're on the right track but the id isn't stamped onto all events. Use a tag or a field instead.

I'm pretty sure you don't need the ruby filter. Look into a date filter instead.

4

Ok, I think I understand. I didn't realize that add_field was available on all the filters, inputs etc. I thought add_field only worked for mutate filters.

I tried this and it seems to do what I want:

input {
    jdbc {
        jdbc_connection_string => "jdbc:oracle:thin:@192.168.56.101:1521/DB0"
        id => "db0"
        add_field => { "[@metadata][source_type]" => "db0" }
        .
        .
        .
    }
    jdbc {
        jdbc_connection_string => "jdbc:oracle:thin:@192.168.56.101:1521/DB1"
        add_field => { "[@metadata][source_type]" => "db1" }
        .
        .
        .
    }
}

Then I can call my filter as follows:

filter {
      if [@metadata][source_type] == "db0" {
        ruby {
          init => "require 'time';"
          code => "
            start_temp = Time.iso8601(event.get('start_time').to_s);
            ts = LogStash::Timestamp.coerce (start_temp);
            event.set('@timestamp', ts);
          "
        }
    }

Thanks for the pointer...

1 Like
5

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