# Using aggregate to add data to previous event
**URL:** https://discuss.elastic.co/t/using-aggregate-to-add-data-to-previous-event/202904
**Category:** Logstash
**Created:** [October 9, 2019, 7:27pm UTC](https://discuss.elastic.co/t/using-aggregate-to-add-data-to-previous-event/202904 "2019-10-09T19:27:53Z")
**Posts on this page:** 4
**Page:** 1
### Author:  [@Factor3](https://discuss.elastic.co/u/Factor3)
#### Post date: [October 9, 2019, 7:27pm UTC](https://discuss.elastic.co/t/using-aggregate-to-add-data-to-previous-event/202904/1 "2019-10-09T19:27:53Z")
Hi, I have a logfile that looks like this:
```
61f3483e-e7f4-4202-b34a-d85108f0fea9 - some log line
61f3483e-e7f4-4202-b34a-d85108f0fea9 - some other log line ID=123456
61f3483e-e7f4-4202-b34a-d85108f0fea9 - another log line
```
The first part is an execution ID, then there is the log text and one of the logs has an ID that came from the request. I am trying to get that ID and add it as a value on all the events, including the ones before it.
My filter looks like this (the execution ID is dissected before), it is able to add the value to the events after that one but not the ones before.
```
grok {
match => { "message" => "ID=(?\d{6})" }
}
if "_grokparsefailure" in [tags] {
aggregate {
task_id => "%{execid}"
code => "event.set('id', map['id'])"
timeout => 5
push_map_as_event_on_timeout => true
}
} else {
aggregate {
task_id => "%{execid}"
code => "map['id'] = event.get('id')"
}
}
```
If I add a sleep before the second aggregate it kinda works but logstash eventually stalls and stops working correctly due to the high number of events.
Is there a way to make aggregate wait until it gets the event with the ID it needs and then run the code for all the previous events?
Thanks.
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [October 9, 2019, 7:52pm UTC](https://discuss.elastic.co/t/using-aggregate-to-add-data-to-previous-event/202904/2 "2019-10-09T19:52:21Z")
Do it in a single filter. Collect all the lines related to the id, then split them when the timeout occurs.
```
grok { match => { "message" => "^%{NOTSPACE:execid} - " } }
grok { match => { "message" => "ID=(?\d{6})" } }
aggregate {
task_id => "%{execid}"
timeout => 5
push_map_as_event_on_timeout => true
code => '
map["message"] ||= []
map["message"] << event.get("message")
id = event.get("id")
if id
map["id"] = id
end
event.cancel
'
}
split { field => "message" }
```
---
### Author:  [@Factor3](https://discuss.elastic.co/u/Factor3)
#### Post date: [October 10, 2019, 5:37pm UTC](https://discuss.elastic.co/t/using-aggregate-to-add-data-to-previous-event/202904/3 "2019-10-10T17:37:56Z")
That worked perfectly, thank you.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [November 7, 2019, 5:38pm UTC](https://discuss.elastic.co/t/using-aggregate-to-add-data-to-previous-event/202904/4 "2019-11-07T17:38:08Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.