Sending mutiple events to single event using multiline codec for email output?

Yaswanth · May 10, 2017, 3:50pm

I am using logstash 2.4.0

My config is like this:

        input {
      file {
        path => "F:\logstash-2.4.0\logstash-2.4.0\bin\slowlog.txt"
        start_position => "beginning"
        codec => multiline {
          # Grok pattern names are valid! :)
          pattern => "^%{TIMESTAMP_ISO8601} "
          what => previous
        }
      }
    }
    
    filter {
        grok {
           match => [ "message", "\[%{TIMESTAMP_ISO8601:TIMESTAMP}\]\[%{LOGLEVEL:LEVEL}%{SPACE}\]\[%{DATA:QUERY}\]%{SPACE}\[%{DATA:QUERY1}\]%{SPACE}\[%{DATA:INDEX-NAME}\]\[%{DATA:SHARD}\]%{SPACE}took\[%{DATA:TOOK}\],%{SPACE}took_millis\[%{DATA:TOOKM}\], types\[%{DATA:types}\], stats\[%{DATA:stats}\], search_type\[%{DATA:search_type}\], total_shards\[%{NUMBER:total_shards}\], source\[%{DATA:source_query}\], extra_source\[%{DATA:extra_source}\],"]
        }
    
        # ==> add this filter to convert TOOKM to integer
        mutate {
            convert => { "TOOKM" => "integer" }
        }
    
        # ==> use TOOKM field instead
        if [TOOKM] > 30 {
            
        } else {
            drop { }
        }
    }
    output {
       stdout { codec => rubydebug }
    }

My output is like this:

{
      "@timestamp" => "2017-05-10T18:14:47.269Z",
         "message" => "[2017-01-14 10:59:58,591][WARN ][index.search.slowlog.query] [yaswanth] [bank][3] took[50ms], took_millis[50], types[details], stats[], search_type[QUERY_THEN_FETCH], total_shards[5], source[{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}], extra_source[], \r",
        "@version" => "1",
            "path" => "F:\\logstash-2.4.0\\logstash-2.4.0\\bin\\picaso.txt",
            "host" => "yaswanth",
       "TIMESTAMP" => "2017-01-14 10:59:58,591",
           "LEVEL" => "WARN",
           "QUERY" => "index.search.slowlog.query",
          "QUERY1" => "yaswanth",
      "INDEX-NAME" => "bank",
           "SHARD" => "3",
            "TOOK" => "50ms",
           "TOOKM" => 50,
           "types" => "details",
     "search_type" => "QUERY_THEN_FETCH",
    "total_shards" => "5",
    "source_query" => "{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}"
}
{
      "@timestamp" => "2017-05-10T18:14:47.270Z",
         "message" => "[2017-01-14 10:59:58,591][WARN ][index.search.slowlog.query] [yaswanth] [bank][2] took[50.2ms], took_millis[50], types[details], stats[], search_type[QUERY_THEN_FETCH], total_shards[5], source[{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}], extra_source[], \r",
        "@version" => "1",
            "path" => "F:\\logstash-2.4.0\\logstash-2.4.0\\bin\\picaso.txt",
            "host" => "yaswanth",
       "TIMESTAMP" => "2017-01-14 10:59:58,591",
           "LEVEL" => "WARN",
           "QUERY" => "index.search.slowlog.query",
          "QUERY1" => "yaswanth",
      "INDEX-NAME" => "bank",
           "SHARD" => "2",
            "TOOK" => "50.2ms",
           "TOOKM" => 50,
           "types" => "details",
     "search_type" => "QUERY_THEN_FETCH",
    "total_shards" => "5",
    "source_query" => "{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}"
}

But what i want is like this

{
          "@timestamp" => "2017-05-10T18:14:47.269Z",
             "message" => "[2017-01-14 10:59:58,591][WARN ][index.search.slowlog.query] [yaswanth] [bank][3] took[50ms], took_millis[50], types[details], stats[], search_type[QUERY_THEN_FETCH], total_shards[5], source[{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}], extra_source[], \r",[2017-01-14 10:59:58,591][WARN ][index.search.slowlog.query] [yaswanth] [bank][2] took[50.2ms], took_millis[50], types[details], stats[], search_type[QUERY_THEN_FETCH], total_shards[5], source[{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}], extra_source[], \r"
            "@version" => "1",
                "path" => "F:\\logstash-2.4.0\\logstash-2.4.0\\bin\\picaso.txt",
                "host" => "yaswanth",
           "TIMESTAMP" => "2017-01-14 10:59:58,591",
               "LEVEL" => "WARN",
               "QUERY" => "index.search.slowlog.query",
              "QUERY1" => "yaswanth",
          "INDEX-NAME" => "bank",
               "SHARD" => "3",
                "TOOK" => "50ms",
               "TOOKM" => 50,
               "types" => "details",
         "search_type" => "QUERY_THEN_FETCH",
        "total_shards" => "5",
        "source_query" => "{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}"
    }

I want to send all the message fields from multiple events to a single event for sending email .

Is there anything wrong in the above config ?

Thanks

Xavy · May 10, 2017, 11:26pm

Hello @Yaswanth:

I have realized that using filter plugin aggregate may serve to your intentions.

https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html

The aim of this filter is to aggregate information available among several events (typically log lines) belonging to a same task, and finally push aggregated information into final task event.

Yaswanth · May 11, 2017, 2:12am

Thanks @Xavy . Based on your suggestion i had updated my config like this:

input {
  file {
    path => "F:\logstash-2.4.0\logstash-2.4.0\bin\picaso.txt"
    start_position => "beginning"
  }
}

filter {
    grok {
       match => [ "message", "\[%{TIMESTAMP_ISO8601:TIMESTAMP}\]\[%{LOGLEVEL:LEVEL}%{SPACE}\]\[%{DATA:QUERY}\]%{SPACE}\[%{DATA:QUERY1}\]%{SPACE}\[%{DATA:INDEX-NAME}\]\[%{DATA:SHARD}\]%{SPACE}took\[%{DATA:TOOK}\],%{SPACE}took_millis\[%{DATA:TOOKM}\], types\[%{DATA:types}\], stats\[%{DATA:stats}\], search_type\[%{DATA:search_type}\], total_shards\[%{NUMBER:total_shards}\], source\[%{DATA:source_query}\], extra_source\[%{DATA:extra_source}\],"]
    }

    # ==> add this filter to convert TOOKM to integer
    mutate {
        convert => { "TOOKM" => "integer" }
    }

    # ==> use TOOKM field instead
    if [TOOKM] > 30 {
     aggregate {
      task_id => "%{message}"
      code => "event.set('message')"
      end_of_task => true
       timeout => 120
      }
        
    } else {
        drop { }
    }
}
output {
   stdout { codec => rubydebug }
}

My output in the screen is like this:

  "message" => "[2017-01-14 10:59:58,591][WARN ][index.search.slowlog.query] [yaswanth] [bank][2] took[50.2ms], took_millis[50], types[details], stats[], search_type[QUERY_THEN_FETCH], total_shards[5], source[{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}], extra_source[], \r",
        "@version" => "1",
      "@timestamp" => "2017-05-11T03:13:53.563Z",
            "path" => "F:\\logstash-2.4.0\\logstash-2.4.0\\bin\\picaso.txt",
            "host" => "yaswanth",
       "TIMESTAMP" => "2017-01-14 10:59:58,591",
           "LEVEL" => "WARN",
           "QUERY" => "index.search.slowlog.query",
          "QUERY1" => "yaswanth",
      "INDEX-NAME" => "bank",
           "SHARD" => "2",
            "TOOK" => "50.2ms",
           "TOOKM" => 50,
           "types" => "details",
     "search_type" => "QUERY_THEN_FETCH",
    "total_shards" => "5",
    "source_query" => "{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}",
            "tags" => [
        [0] "_aggregateexception"
    ]
}
{
         "message" => "[2017-01-14 10:59:58,593][WARN ][index.search.slowlog.query] [yaswanth] [bank][1] took[52.2ms], took_millis[52], types[details], stats[], search_type[QUERY_THEN_FETCH], total_shards[5], source[{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}], extra_source[], \r",
        "@version" => "1",
      "@timestamp" => "2017-05-11T03:13:53.564Z",
            "path" => "F:\\logstash-2.4.0\\logstash-2.4.0\\bin\\picaso.txt",
            "host" => "yaswanth",
       "TIMESTAMP" => "2017-01-14 10:59:58,593",
           "LEVEL" => "WARN",
           "QUERY" => "index.search.slowlog.query",
          "QUERY1" => "yaswanth",
      "INDEX-NAME" => "bank",
           "SHARD" => "1",
            "TOOK" => "52.2ms",
           "TOOKM" => 52,
           "types" => "details",
     "search_type" => "QUERY_THEN_FETCH",
    "total_shards" => "5",
    "source_query" => "{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}",
            "tags" => [
        [0] "_aggregateexception"
    ]

What i want in my final event should have all the message fields in the above logs like

{

 "message" => [2017-01-14 10:59:58,591][WARN ][index.search.slowlog.query] [yaswanth] [bank][2] took[50.2ms], took_millis[50], types[details], stats[], search_type[QUERY_THEN_FETCH], total_shards[5], source[{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}], extra_source[], \r",[2017-01-14 10:59:58,593][WARN ][index.search.slowlog.query] [yaswanth] [bank][1] took[52.2ms], took_millis[52], types[details], stats[], search_type[QUERY_THEN_FETCH], total_shards[5], source[{\"sort\":[{\"balance\":{\"order\":\"asc\"}}]}], extra_source[], \r"

}

Whether my approach for the scenario is correct? I dont know exactly what to keep in task_id and code in aggregate filter to produce desired result.
Thanks

Xavy · May 11, 2017, 6:32am

Hello:

Your new config is not working as it is triggering an aggregate exception:

if code execution raises an exception, the error is logged and event is tagged _aggregateexception

Unfortunately I don't have here a 2.3 scenario where I can test your use case, but I think that something similar to the example in doc should work:

code => "map['aggr_message'] += event.get('message')"

This should create a new field aggr_message

Regarding the task_id, it may be whatever you want ("%{QUERY}" sounds OK to me)

Actually, if you look at the doc, "Example #4" looks quite similar to your needs (it aggregates a text field from several events into just one), with the only difference that the example aggregates country_name field, while you want to aggregate the message one.

Hope this brings some light on how you may achieve your needs

Yaswanth · May 11, 2017, 7:00am

Thanks so much @Xavy

Even i tried with by giving this

if [TOOKM] > 15 {
    aggregate {
      task_id => "%{QUERY}"
      code => "map['aggr_message'] += event.get('message')"
      end_of_task => true
       timeout => 120
      }

    } else {
        drop { }
    }

It is throwing error:

Aggregate exception occurred {:error=>#<NoMethodError: undefined method `+' for nil:NilClass>, :code=>"map['aggr_message'] += event.get('message')", :map=>{}, :event_data=>{"message"=>"[2017-04-25 04:40:05,240][TRACE][index.search.slowlog.query] [data-0] [data-apr-2017][4] took[20.6ms], took_millis[20], types[details], stats[], search_type[QUERY_THEN_FETCH], total_shards[5], source[{\"query\":{\"bool\":{\"must\":[{\"terms\":{\"articleId\":[316249486]}}]}}}], extra_source[],", "@version"=>"1", "@timestamp"=>"2017-05-11T06:54:50.932Z", "path"=>"/home/itadmin/logstash/logstash-2.4.1/slow.txt", "host"=>"kibana", "TIMESTAMP"=>"2017-04-25 04:40:05,240", "LEVEL"=>"TRACE", "QUERY"=>"index.search.slowlog.query", "QUERY1"=>"data-0", "INDEX-NAME"=>"data-apr-2017", "SHARD"=>"4", "TOOK"=>"20.6ms", "TOOKM"=>20, "types"=>"publishedarticle", "search_type"=>"QUERY_THEN_FETCH", "total_shards"=>"5", "source_query"=>"{\"query\":{\"bool\":{\"must\":[{\"terms\":{\"articleId\":[316249486]}}]}}}", "@metadata"=>{"path"=>"/home/itadmin/logstash/logstash-2.4.1/slow.txt"}}, :level=>:error}
Aggregate exception occurred {:error=>#<NoMethodError: undefined method `+' for nil:NilClass>, :code=>"map['aggr_message'] += event.get('message')", :map=>{}, :event_data=>{"message"=>"[2017-04-25 05:47:02,335][TRACE][index.search.slowlog.query] [data-0] [data-apr-2017][4] took[20.8ms], took_millis[20], types[details], stats[], search_type[QUERY_THEN_FETCH], total_shards[5], source[{\"query\":{\"bool\":{\"must\":[{\"terms\":{\"articleId\":[316252085]}}]}}}], extra_source[],",

Thanks

Xavy · May 11, 2017, 7:36am

Hello:

It looks the code does not create the field by itself, so I would try using message to store the final message field (something like this:

  code => "map['message'] += event.get('message')"

Yaswanth · May 11, 2017, 7:54am

After using too, It is showing the same error

Thanks

Yaswanth · May 11, 2017, 11:29am

Thanks @Xavy

You was very helpful . As you said i interpreted #ex 4 in documentation which is similar to my scenario and used it .

  aggregate {
        task_id => "%{LEVEL}"
        code => "
          map['LEVEL'] = event.get('LEVEL')
          map['messages'] ||= []
          map['messages'] << {'message' => event.get('message')}
          event.cancel()
        "
        push_previous_map_as_event => true
        timeout => 3
      }

It worked fine . I used against the LEVEL field.

Thanks

Xavy · May 11, 2017, 12:49pm

You're welcome If any of my posts particularly helped you inf finding the solution, I would thank you if you could mark it as "Solution"

Glad of being of any help

system · June 8, 2017, 12:52pm

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

Topic		Replies	Views
Multiline logs into one event using logstash? Logstash	10	262	December 30, 2022
Multiline Codec Issues Logstash	4	856	July 6, 2017
Using multiline codec for python stacktrace Logstash	11	1123	October 5, 2019
Multiple logs are concatenated to single event by Logstash Logstash	3	695	October 19, 2020
Sending multiple events using a single POST with http output codec Logstash	11	45	September 3, 2024

Sending mutiple events to single event using multiline codec for email output?

Related Topics