Search code examples
elasticsearchsyntaxlogstashkibana-4

Assigning tags for elapsed in logstash


I have log file that look for example like this:

2015-12-05 05:00:51 prefix1_sent 1
2015-12-05 05:00:52 prefix2_sent 2
2015-12-05 05:00:53 prefix1_received 1
2015-12-05 05:00:54 prefix2_received 2

I want to find out how much time particular event took. I am using elapsed plugin in logstash like this:

....

grok {
        patterns_dir => "path"
        match => { 
            "message" => "%{TIMESTAMP:logtime}, %{EVENT:event}, %{SPLIT:ID}"
        }
    }

filter{
    if [event] == "(.)*sent" {
        if [event] == "prefix1(.)*" {
            kv {
                add_tag => [ "prefix1_sent" ]
            }
        } else if [event] == "prefix2(.)*" {
            kv {
                add_tag => [ "prefix2_sent" ]
            }
    } else if [event] == "(.)*received" {
        if [event] == "prefix1(.)*" {
            kv {
                add_tag => [ "prefix1_received" ]
            }
        } else if [event] == "prefix2(.)*" {
            kv {
                add_tag => [ "prefix2_received" ]
            }
        }
    }
}
filter{
    elapsed {
        start_tag => "prefix1_sent"
        end_tag => "prefix1_received"
        unique_id_field => "ID"
        new_event_on_match => false
    }
}
filter{
    elapsed {
        start_tag => "prefix2_sent"
        end_tag => "prefix2_received"
        unique_id_field => "ID"
        new_event_on_match => false
    }
}

....

I am not sure if my syntax of nested if statement is correct of if I understand the use of "kv". I am new to logstash. However I can't find any tag in kibana with this configuration. Where am I making mistake?

Thanks


Solution

  • Several comments:

    • Your nested conditionals are fine.
    • Your comparisons using regexps are wrong. Use "=~" instead of "==".
    • add_tag only works when the underlying filter works. While your use of kv{} would be fine, you're not actually doing anything with key/values, so it can be confusing. Most people use mutate{} for things like this.

    Those changes would probably get you on track.

    For some unsolicited advice, take a look at your workflow. First, you apply a regexp with grok to make three fields. Then you regexp one of them again (2-3 times) before you do the add_field. Try this:

    • in the original grok{}, split the event field into the two pieces. A pattern like this should do it: %{DATA:event_prefix}_%{DATA:event_verb}
    • rewrite your conditionals to be exact matches against these two fields.

    Not only is it more "grok-ish", but it would be faster when applied to a large data set.