Combining the logs in logstash

Elastic Stack Logstash
1

Hi,

I need to combine the two log lines using the message id i tried aggregate filter plugin but i cant achieve it.Please guide

Jun  5 19:24:18 gpigw11 postfix/qmgr[37531]: 3077F92E45: from=<******>, size=2801, nrcpt=1 (queue active)

Jun 5 19:24:18 gpigw11 postfix/lmtp[99429]: 3077F92E45: to=<******>, delay=0.2, delays=0.01/0/0.05/0.14, dsn=2.1.5, status=sent (250 2.1.5 Delivery OK)

2

The following would do it. Note that it drops the original events, so the only fields that will be on the final event are the ones that were stored in the map.

Also, the -> in the dissect filter has the effect of compacting the double space down to a single space, so you do not need a second pattern in the date filter to hand the double space case for the first 9 days of the month.

You may need a longer timeout than 10 seconds.

    dissect { mapping => { "message" => "%{[@metadata][ts]->} %{+[@metadata][ts]} %{+[@metadata][ts]} %{theHost} %{}/%{}[%{}]: %{task}: %{[@metadata][kvStuff]}" } }
    date { match => [ "[@metadata][ts]", "MMM dd HH:mm:ss" ] }
    kv { source => "[@metadata][kvStuff]" target => "kvData" field_split => ", " }
    aggregate {
        task_id => "%{task}"
        push_map_as_event_on_timeout => true
        inactivity_timeout => 10
        timeout_task_id_field => "correlationId"
        code => '
            map["kvData"] ||= {}
            map["kvData"] =  map["kvData"].merge(event.get("kvData"))
            map["theHost"] = event.get("theHost")
            map["@timestamp"] = event.get("@timestamp")
            event.cancel
        '
    }
3

Thanks a lot will check and let u know..

4

hi can u please explain how this works...i have seen this dissect for the first time

5

and i m getting this warning [2019-07-17T12:18:31,365][WARN ][org.logstash.dissect.Dissector] Dissector mapping, field found in event but it was empty {"field"=>"message", "event"=>{"host"=>"localhost.localdomain", "message"=>"", "tags"=>["_dissectfailure"], "@version"=>"1", "@timestamp"=>2019-07-17T06:48:31.261Z}} and this is also getting stored in elasticindex and displaying in kibana...

6

If you want to understand dissect better I would start with the blog post that introduced it and the documentation.

The warning occurs because the message field is an empty string ("message"=>""), so it does not match the dissect mapping. The documentation explains that conditional processing may be needed to avoid these warnings.

7

Thanks a lot....Will read the documents

8

i have read the docs and cleared the warnings....it is storing all the combined logs in kvData format which is ok...there is a small problem in storing the fields if u see my logs the last field status has some line in the brackets for e.g status=sent (250 2.1.5 Delivery OK) while storing the value in kvData it is storing it as kvData.status=sent but the lines in bracket are not included this is because we have given the field_split => ", " so how to store the whole status field...

9

Remove the space from field_split and add trim_key => " "

10

ok thanks

11

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