# Merge message with aggregate filter **URL:** \ **Category:** Logstash\ **Created:** [April 2, 2021, 3:44pm UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115 "2021-04-02T15:44:12Z")\ **Posts on this page:** 11\ **Page:** 1
**Author:** ![MalfuncEddie](https://avatars.discourse-cdn.com/v4/letter/m/3da27b/32.png) [@MalfuncEddie](https://discuss.elastic.co/u/MalfuncEddie)\ **Post date:** [April 2, 2021, 3:44pm UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115/1 "2021-04-02T15:44:12Z")
Hi, I am trying to merge 2 log lines with the aggregate filter and after a week browsing the internet and this forum I still cannot get this working. Below you have 2 loglines with an identifier ExchangeId. What I want to achieve is one document that has parsed both json lines REQ\_OUT in the req-out nested object and the RESP\_IN in the resp-in nested object. How it is now is that I get 2 documents and the objects do not exists. The logline is passing the json filter since I do have a meta object. Can anyone help. sample input: ```auto {"timestamp":"2021-04-02T05:49:45.251Z","severity":"INFO","Type":"REQ_OUT","Address":"https://a-link.com","ExchangeId":"260e06a3-9cb5-4154-bf97-637e929fa4c2",REQ_OUT:"test1"} {"timestamp":"2021-04-02T05:50:45.534Z","severity":"INFO","Type":"RESP_IN","ResponseCode":"200","Address":"https://a-link.com","ExchangeId":"260e06a3-9cb5-4154-bf97-637e929fa4c2",RESP_IN:"test2"} ``` logstash filter: ```auto filter { json { source => "message" skip_on_invalid_json => "true" target => "meta" } if [meta][ExchangeId] { mutate { add_field => { "ExchangeId" => "%{[meta][ExchangeId]}" } } if [meta][Type] =~ /REQ_OUT/ { aggregate { task_id => "%{ExchangeId}" #first log line so we save the message to REQ_OUT code => " map['REQ_OUT'] ||= event.get('message') " map_action => "create" timeout => 15 push_map_as_event_on_timeout => true timeout_tags => ["aggregate_timeout"] } } if [meta][Type] =~ /RESP_IN/ { aggregate { task_id => "%{ExchangeId}" #2nd log line so we save the message to RESP_IN code => "map['RESP_IN'] ||= event.get('message')" map_action => "update" end_of_task => true } } } #decode de json of the messages place in REQ_OUT and RESP_IN by aggregate filter json { source => "REQ_OUT" skip_on_invalid_json => "true" target => "req-out" } json { source => "RESP_IN" skip_on_invalid_json => "true" target => "resp-in" } } ``` Logstash does not give an error ```auto [DEBUG] 2021-04-02 15:40:45.004 [[main]>worker0] aggregate - Aggregate successful filter code execution {:code=>"\n map['REQ_OUT'] ||= event.get('message')\n "} [DEBUG] 2021-04-02 15:40:45.005 [[main]>worker0] aggregate - Aggregate successful filter code execution {:code=>"map['RESP_IN'] ||= event.get('message')"} ``` ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [April 2, 2021, 4:44pm UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115/2 "2021-04-02T16:44:51Z")
> [@MalfuncEddie](#): > > ``` > ..."ExchangeId":"260e06a3-9cb5-4154-bf97-637e929fa4c2",RESP_IN:"test2" > > ``` logstash does not log an error because you set skip\_on\_invalid\_json to true. That is invalid JSON, so it does not get parsed. If you had ``` ... "ExchangeId":"260e06a3-9cb5-4154-bf97-637e929fa4c2","RESP_IN":"test2" ``` Then you could use ``` json { source => "message" target => "meta" } aggregate { task_id => "%{[meta][ExchangeId]}" timeout => 15 push_map_as_event_on_timeout => true code => ' map["Address"] ||= event.get("[meta][Address]") map["RESP_IN"] ||= event.get("[meta][RESP_IN]") map["REQ_OUT"] ||= event.get("[meta][REQ_OUT]") map["timestamp"] ||= event.get("[meta][timestamp]") map["ResponseCode"] ||= event.get("[meta][ResponseCode]") event.cancel # If you do not want the source events ' timeout_task_id_field => "ExchangeId" } ``` and get ``` { "Address" => "https://a-link.com", "RESP_IN" => "test2", "@timestamp" => 2021-04-02T16:42:55.594Z, "timestamp" => "2021-04-02T05:49:45.251Z", "REQ_OUT" => "test1", "@version" => "1", "ExchangeId" => "260e06a3-9cb5-4154-bf97-637e929fa4c2", "ResponseCode" => "200" } ``` Note that when using push\_map\_as\_event\_on\_timeout the _only_ things in the pushed event are what you added to the map. ---
**Author:** ![MalfuncEddie](https://avatars.discourse-cdn.com/v4/letter/m/3da27b/32.png) [@MalfuncEddie](https://discuss.elastic.co/u/MalfuncEddie)\ **Post date:** [April 2, 2021, 5:24pm UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115/3 "2021-04-02T17:24:07Z")
Hi, I'll test this tomorrow. But what do you do when you don't know all the fields in the json and want to put the fields in an object? eg: I want something like this output: ``` { "@timestamp": "2021-04-02T16:42:55.594Z", "@version": "1", "ExchangeId": "260e06a3-9cb5-4154-bf97-637e929fa4c2", "resp-in": { "Address": "https://a-link.com", "RESP_IN": "test2", "timestamp": "2021-04-02T05:49:45.251Z", "fieldx": "something", "fieldy": "something" }, "req-out": { "Address": "https://a-link.com", "REQ_OUT": "test1", "timestamp": "2021-04-02T05:49:45.251Z", "fieldA": "something", "fieldB": "something", "ResponseCode": "200" } } ``` ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [April 2, 2021, 5:38pm UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115/4 "2021-04-02T17:38:33Z")
> [@MalfuncEddie](#): > > I want something like this output I have not tested it, but something like ``` code => ' type = event.get("Type") if type == "REQ_OUT" dst = "req-out" elsif type == "RESP_IN" dst = "resp-in" end if dst event.to_hash.each { |k, v| map[dst] = {} unless ["host", "path", "ExchangeId"].include? k map[dst][k] = v end } end ' ``` ---
**Author:** ![MalfuncEddie](https://avatars.discourse-cdn.com/v4/letter/m/3da27b/32.png) [@MalfuncEddie](https://discuss.elastic.co/u/MalfuncEddie)\ **Post date:** [April 13, 2021, 1:38pm UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115/5 "2021-04-13T13:38:18Z")
Hi, I do get a Aggregate successful filter code execution. But I just get 2 documents only with the meta object. Trying to debug ``` filter { json { source => "message" target => "meta" } aggregate { task_id => "%{[meta][ExchangeId]}" timeout => 15 push_map_as_event_on_timeout => true code => ' type = event.get("Type") if type == "REQ_OUT" dst = "req-out" elsif type == "RESP_IN" dst = "resp-in" end if dst event.to_hash.each { |k, v| map[dst] = {} unless ["host", "path", "ExchangeId"].include? k map[dst][k] = v end } end ' timeout_task_id_field => "ExchangeId" timeout_tags => ['_aggregatetimeout'] } ``` } ---
**Author:** ![MalfuncEddie](https://avatars.discourse-cdn.com/v4/letter/m/3da27b/32.png) [@MalfuncEddie](https://discuss.elastic.co/u/MalfuncEddie)\ **Post date:** [April 14, 2021, 6:59am UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115/6 "2021-04-14T06:59:13Z")
Did some more debugging and for some reason the aggregate filter is not working it is not matching on the "ExchangeId" and just produces 2 documents. I've tried to simplify the code but with no success. The most annoying thing of all is that when I start logstash in debug mode it does give a `DEBUG] 2021-04-14 06:57:14.170 [[main]>worker0] aggregate - Aggregate successful filter code execution {:code=>"\n type = event.get(\"Type\")\n map[type] = {}\n event.to_hash.each { |k, v|\n map[type][k] = v\n } \n "}` event. So it does do something. Is this a bug? > ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [April 14, 2021, 1:36pm UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115/7 "2021-04-14T13:36:20Z")
If test.txt only has two lines why would expect more than two events? ---
**Author:** ![MalfuncEddie](https://avatars.discourse-cdn.com/v4/letter/m/3da27b/32.png) [@MalfuncEddie](https://discuss.elastic.co/u/MalfuncEddie)\ **Post date:** [April 14, 2021, 1:50pm UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115/8 "2021-04-14T13:50:40Z")
Hi, maybe I am on the completely wrong track but I would expect 1 document: (when event cancel is active I get nothing) ``` { "Type" => "REQ_OUT", "@version" => "1", "host" => "LGS02", "ExchangeId" => "260e06a3-9cb5-4154-bf97-637e929fa4c2", "@timestamp" => 2021-04-14T06:46:29.221Z, "path" => "/tmp/test.txt" "REQ_OUT" => { "timestamp" => "2021-04-02T05:50:45.534Z", "severity" => "INFO", ... } "RESP_IN" => { "timestamp" => "2021-04-02T05:50:44.251Z", "severity" => "INFO", ResponseCode" => "200", ... } "@version" => "1", "host" => "LGS02", "ExchangeId" => "260e06a3-9cb5-4154-bf97-637e929fa4c2", "@timestamp" => 2021-04-14T06:46:29.247Z, "path" => "/tmp/test.txt" ``` } ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [April 14, 2021, 3:00pm UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115/9 "2021-04-14T15:00:17Z")
Using ``` json { source => "message" target => "[@metadata][data]" remove_field => ["message"] } aggregate { task_id => "%{[@metadata][data][ExchangeId]}" timeout => 5 push_map_as_event_on_timeout => true code => ' # we want 2 nested objects in the document object REQ_OUT and RESP_IN type = event.get("[@metadata][data][Type]") map[type] = {} event.get("[@metadata][data]").each { |k, v| unless ["ExchangeId"].include? k map[type][k] = v end } event.to_hash.each { |k, v| unless ["@version"].include? k map[type][k] = v end } event.cancel ' timeout_task_id_field => "ExchangeId" timeout_tags => ['_aggregatetimeout'] } ``` I get ``` { "RESP_IN" => { "RESP_IN" => "test2", "severity" => "INFO", "path" => "/home/user/foo.txt", "Type" => "RESP_IN", "ResponseCode" => "200", "@timestamp" => 2021-04-14T14:57:51.115Z, "Address" => "https://a-link.com", "host" => "...", "timestamp" => "2021-04-02T05:50:45.534Z" }, "REQ_OUT" => { "severity" => "INFO", "path" => "/home/user/foo.txt", "Type" => "REQ_OUT", "REQ_OUT" => "test1", "@timestamp" => 2021-04-14T14:57:51.115Z, "Address" => "https://a-link.com", "host" => "...", "timestamp" => "2021-04-02T05:50:44.251Z" }, "@timestamp" => 2021-04-14T14:58:01.101Z, "@version" => "1", "ExchangeId" => "260e06a3-9cb5-4154-bf97-637e929fa4c2", "tags" => [ [0] "_aggregatetimeout" ] } ``` Which version of logstash are you running? I wonder if you are hitting the issue mentioned in [this](https://discuss.elastic.co/t/logstash-dash-hyphen-problem-with-aggregation/251947/4) thread. ---
**Author:** ![MalfuncEddie](https://avatars.discourse-cdn.com/v4/letter/m/3da27b/32.png) [@MalfuncEddie](https://discuss.elastic.co/u/MalfuncEddie)\ **Post date:** [April 14, 2021, 3:14pm UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115/10 "2021-04-14T15:14:14Z")
Hi, I'm running 7.8.0 so I guess I hit that bug. I'll try to set the pipeline.java\_execution or just upgrade to the latest version. Pretty sure this is the sollution. You are a god among men! 😃 ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [May 12, 2021, 3:15pm UTC](https://discuss.elastic.co/t/merge-message-with-aggregate-filter/269115/11 "2021-05-12T15:15:12Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.