# Multiline events based on identical fields
**URL:** https://discuss.elastic.co/t/multiline-events-based-on-identical-fields/235564
**Category:** Logstash
**Created:** [June 3, 2020, 1:52pm UTC](https://discuss.elastic.co/t/multiline-events-based-on-identical-fields/235564 "2020-06-03T13:52:46Z")
**Posts on this page:** 6
**Page:** 1
### Author:  [@WDK](https://discuss.elastic.co/u/WDK)
#### Post date: [June 3, 2020, 1:52pm UTC](https://discuss.elastic.co/t/multiline-events-based-on-identical-fields/235564/1 "2020-06-03T13:52:46Z")
Hi all,
I have a not so straight forward request of a user to process progress logfiles. The catch is that the want the log files multilined in a specific pattern.
All lines that have the exact same timestamp, P-id and T-id should be in a single doc. As you can see from the example below, there are wild combinations of ID's and timestamp. The bold lines for example should be in 1 doc.
> [18/11/12@15:08:49.357+0100] P-035768 T-2833205120 2 AS AS Application Server connected with connection id: REDACTED
>
> [18/11/12@15:08:49.376+0100] P-035768 T-2833205120 1 AS -- Log entry types activated: Db.Connects:2
>
> [18/11/12@16:22:06.352+0100] P-040789 T-2835367808 1 AS -- Logging level set to = 2
>
> **[18/11/12@16:22:06.354+0100] P-040789 T-2835367808 1 AS -- Log entry types activated: ASPlumbing,DB.Connects**
> **[18/11/12@16:22:06.354+0100] P-040789 T-2835367808 2 AS AS Starting application server for REDACTED. (5560)**
> **[18/11/12@16:22:06.354+0100] P-040789 T-2835367808 2 AS AS Application Server Startup. (5473)**
>
> [18/11/12@16:22:06.434+0100] P-040789 T-2835367808 2 AS CONN Database master Options: (12699)
>
> [18/11/12@16:22:06.435+0100] P-040789 T-2835367808 2 AS CONN Connected to database master, user number 88. (9543)
>
> [18/11/12@16:22:06.442+0100] P-040789 T-2835367808 2 AS CONN Database temp-db Options: (12699)
I was thinking of sending the log files with beats to a specific pipeline in logstash with a single worker and use the aggregate filter but I can't wrap my head around where to start.
Is this even possible with the built-in logstash filters?
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [June 3, 2020, 2:04pm UTC](https://discuss.elastic.co/t/multiline-events-based-on-identical-fields/235564/2 "2020-06-03T14:04:24Z")
You should be able to do that with an aggregate filter. If you have already parsed the timestamp, P-id, and T-id you can combine them as the task\_id using sprintf references
```
task_id => "%{timestamp} %{P-id} %{T-id}"
```
or else dissect them off
```
dissect { mapping => { "message" => "%{task_id} %{+task_id} %{+task_id} %{}" } }
```
Then configure the aggregate filter to flush after a timeout, as in [example 3](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html#plugins-filters-aggregate-example3).
In the code option, either create an array
```
code => '
map["messages"] ||= []
map["messages"] << event.get("message")
'
```
or concatenate them. Whatever works for you.
---
### Author:  [@WDK](https://discuss.elastic.co/u/WDK)
#### Post date: [June 8, 2020, 6:41am UTC](https://discuss.elastic.co/t/multiline-events-based-on-identical-fields/235564/3 "2020-06-08T06:41:59Z")
Thanks for the directions! I'll try that.
---
### Author:  [@WDK](https://discuss.elastic.co/u/WDK)
#### Post date: [June 9, 2020, 11:32am UTC](https://discuss.elastic.co/t/multiline-events-based-on-identical-fields/235564/4 "2020-06-09T11:32:59Z")
I finally got the aggregation to "sorta" work with the following:
> aggregate {
> task\_id =\> "%{progress\_timestamp} %{progress\_pid} %{progress\_thread}"
> code =\> "
> map['progress\_message'] ||=
> map['progress\_message'] \<\< event.get('message')
> "
> push\_previous\_map\_as\_event =\> true
> timeout =\> 10
> }
It does seem to trip on the timestamp though since I see logs with different timestamps getting aggregated:
> [20/06/09@13:25:06. **126** +0200] P-082061 T-2238457728 1 AS QRX-DEBUG REDACTED, [20/06/09@13:25:06. **126** +0200] P-082061 T-2238457728 1 AS QRX-DEBUG START HandleRequest: , [20/06/09@13:25:06. **289** +0200] P-082061 T-2238457728 1 AS QRX-DEBUG ------------------------------------------------------------------------------------------------------------------------------, [20/06/09@13:25:06. **289** +0200] P-082061 T-2238457728 1 AS QRX-DEBUG ], "info":{"changesOnly":
Looking at the debug logs, I can see this. Is this because the timestamp varies and the P and T values do not or is there another issue?
> [2020-06-09T13:27:00,329][DEBUG][logstash.filters.aggregate] Aggregate create\_timeout\_event call with task\_id ' **%{progress\_timestamp}** P-124198 T-242648960'
---
### Author:  [@WDK](https://discuss.elastic.co/u/WDK)
#### Post date: [June 10, 2020, 10:44am UTC](https://discuss.elastic.co/t/multiline-events-based-on-identical-fields/235564/5 "2020-06-10T10:44:55Z")
@Badger just FYI.
with the following aggregate block it seems to be solid.
```
aggregate {
task_id => "%{progress_raw_timestamp} %{progress_pid} %{progress_thread}"
code => "
map['progress_message'] ||= []
map['progress_message'] << event.get('message') +10.chr
map['pid'] ||= event.get('progress_pid')
map['thread'] ||= event.get('progress_thread')
map['loglevel'] ||= event.get('progress_loglevel')
map['environment'] ||= event.get('progress_environment')
map['host'] = event.get('host')
map['agent'] = event.get('agent')
map['log'] = event.get('log')
"
push_map_as_event_on_timeout => true
timeout_tags => ['_aggregatetimeout']
timeout => 10
}
if "_aggregatetimeout" not in [tags]
{
drop {}
}
```
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [July 8, 2020, 10:45am UTC](https://discuss.elastic.co/t/multiline-events-based-on-identical-fields/235564/6 "2020-07-08T10:45:00Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.