Elapsed plugin modify/adding a field not possible

[kumar-m-kiran (Kiran)]

Consider the below (watered down version ) of the configuration file

---

Configuration file

input {
  tcp {
    port => 5000
    type => syslog
  }
  udp {
    port => 5000
    type => syslog
  }
}

filter {
  if [type] == "syslog" {
    grok {
        match => ["message", "%{TIMESTAMP_ISO8601:timestamp_match}-StartEvent-%{WORD:event_info}.*"]
       	add_tag => [ "PingStart"  ]
       	remove_tag => ["_grokparsefailure"]
    }

    grok {
        match => ["message", "%{TIMESTAMP_ISO8601:timestamp_match}-EndEvent-%{WORD:event_info}.*"]
       	add_tag => [ "PingEnd"  ]
       	remove_tag => ["_grokparsefailure"]
    }
  elapsed {
      unique_id_field => "event_info"
      start_tag => "PingStart"
      end_tag => "PingEnd"
      new_event_on_match => true
      add_tag => ["in2"]
    }
    # Records the execution time of system1
    if "in2" in [tags] and "elapsed" in [tags] {
      aggregate {
        task_id => "%{event_info}"
        code => "map['report'] = [(event.get('elapsed_time')*1000).to_i]"
        map_action => "create"
      }
    }
}
}


output {
       	stdout { codec => rubydebug }
}

---

Input

The port 5000 is used to push the log from remote machine (say 172.19.221.53 in this case)

telnet 198.18.66.198 5000
Trying 198.18.66.198...
Connected to 198.18.66.198.
Escape character is '^]'.
2015-03-13 00:23:37.616-StartEvent-Test1
2015-03-13 00:33:37.616-EndEvent-Test1
^CConnection closed by foreign host

---

Output

{
         "@timestamp" => 2017-03-26T08:25:32.813Z,
         "event_info" => "Test1",
               "port" => 61390,
    "timestamp_match" => "2015-03-13 00:23:37.616",
           "@version" => "1",
               "host" => "172.19.221.53",
            "message" => "2015-03-13 00:23:37.616-StartEvent-Test1\r",
               "type" => "syslog",
               "tags" => [
        [0] "PingStart",
        [1] "_grokparsefailure",
        [2] "in2"
    ]
}
[2017-03-26T08:25:58,624][INFO ][logstash.filters.elapsed ] Elapsed, 'end event' received {:end_tag=>"PingEnd", :unique_id_field=>"event_info"}
{
         "@timestamp" => 2017-03-26T08:25:58.616Z,
         "event_info" => "Test1",
               "port" => 61390,
    "timestamp_match" => "2015-03-13 00:33:37.616",
           "@version" => "1",
               "host" => "172.19.221.53",
            "message" => "2015-03-13 00:33:37.616-EndEvent-Test1\r",
               "type" => "syslog",
               "tags" => [
        [0] "PingEnd",
        [1] "in2"
    ]
}
{
                 "@timestamp" => 2017-03-26T08:25:58.625Z,
                 "event_info" => "Test1",
                   "@version" => "1",
                       "host" => "198-18-66-198.deploy.com",
               "elapsed_time" => 25.803,
                       "tags" => [
        [0] "elapsed",
        [1] "elapsed_match",
        [2] "in2"
    ],
    "elapsed_timestamp_start" => 2017-03-26T08:25:32.813Z
}

---

Question

When the event is matched (elapsed is trigged, the host name recorded is where the logstash is executing (198-18-66-198.deploy.com) and not the host name of the client which pushed the logs (i.e 172.19.221.53 ).

1. How do I modify the document of the elapsed event to reflect the client hostname and not the logstash host name?
2. Is this possible? Or should I look for another workaround?

---

Expected output

{
                 "@timestamp" => 2017-03-26T08:25:58.625Z,
                 "event_info" => "Test1",
                   "@version" => "1",
                       "host" => "172.19.221.53",  //Actual client ip which pushed the log
               "elapsed_time" => 25.803,
                       "tags" => [
        [0] "elapsed",
        [1] "elapsed_match",
        [2] "in2"
    ],
    "elapsed_timestamp_start" => 2017-03-26T08:25:32.813Z
}

Any help is appreciated.

• Kiran

[kumar-m-kiran (Kiran)]

Just to summarize in short, the elapsed plugin filter will create a new event. Now how do I "pass" on certain variables/dynamic values to the elapsed event created anew? I understand that generally @metadata is used for this purpose. However I see that @metadata would be useful only when you share from filter to output.

So what mechanism is available to share data from grok-match to grok-elapsed?