# Is it possible to aggregate logs from cef input
**URL:** https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976
**Category:** Logstash
**Created:** [July 13, 2018, 4:08pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976 "2018-07-13T16:08:12Z")
**Posts on this page:** 20
**Page:** 1
### Author:  [@saad](https://discuss.elastic.co/u/saad)
#### Post date: [July 13, 2018, 4:08pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/1 "2018-07-13T16:08:12Z")
Hi,
Is it possible to apply aggregate filter on cef logs. Would you please check my conf file if there is something wrong. it should aggregating when name and sourceAddress are common and increase field aggregatedEvents by 1 every time. below are the conf file:
input
{
udp
{
port =\> "516"
type =\> "Forcepoint"
codec =\> "cef"
}
}
filter
{
aggregate {
task\_id =\> "%{name}\_%{sourceAddress}"
code =\> "
map['aggregatedEvents'] ||= 0;
map['aggregatedEvents'] += 1;
"
timeout =\> 30
}
}
output
{
elasticsearch
{
hosts =\> ["10.xx.xx.xx:9200"]
index =\> "logstash-fpnofilter-%{+MM.dd.YYYY}"
}
}
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 13, 2018, 4:33pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/2 "2018-07-13T16:33:39Z")
It's possible, yes. But if I am reading it correctly you add aggregatedEvents to the map, and then expire it 30 seconds later. You don't push the map as an event on timeout, and you have no end-task.
What do you want to do with this aggregatedEvents counter and when do you want to do it?
---
### Author:  [@saad](https://discuss.elastic.co/u/saad)
#### Post date: [July 13, 2018, 5:09pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/3 "2018-07-13T17:09:51Z")
I made as simple as is
aggregatedEvents is to see how many events were creating this aggregated log
for example if its 3 thats mena the 3 logs inlcuded to make the output aggregated log
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 13, 2018, 9:06pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/4 "2018-07-13T21:06:32Z")
You didn't really answer either of my questions, but does this help you?
```
aggregate {
task_id => "%{name}_%{sourceAddress}"
code => "
event.cancel()
map['aggregatedEvents'] ||= 0;
map['aggregatedEvents'] += 1;
"
push_map_as_event_on_timeout => true
timeout_task_id_field => "task_id"
timeout => 5
}
```
---
### Author:  [@saad](https://discuss.elastic.co/u/saad)
#### Post date: [July 14, 2018, 4:33pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/5 "2018-07-14T16:33:24Z")
Thank you so much Badger
I copied the aggregate section and its working and aggregating. However, currently I am geeting fields only task\_id and aggregatedEvents. How I get the fields name and source address separated.
and again thank you for your full support
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 14, 2018, 5:01pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/6 "2018-07-14T17:01:12Z")
> [@saad](#):
>
> How I get the fields name and source address separated.
Add them as additional entries in the map.
---
### Author:  [@saad](https://discuss.elastic.co/u/saad)
#### Post date: [July 14, 2018, 9:26pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/7 "2018-07-14T21:26:25Z")
I did nd successfully seeing the sourceAddress field in Kibana. Please, How can I delete task\_id field. I dont need to save it in index?
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 14, 2018, 11:02pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/8 "2018-07-14T23:02:53Z")
Remove
```
timeout_task_id_field => "task_id"
```
---
### Author:  [@saad](https://discuss.elastic.co/u/saad)
#### Post date: [July 15, 2018, 6:50pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/9 "2018-07-15T18:50:29Z")
I tried to use it but it didnt work
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 15, 2018, 7:18pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/10 "2018-07-15T19:18:02Z")
I said to remove that. That is what causes it to add task\_id to the event.
---
### Author:  [@saad](https://discuss.elastic.co/u/saad)
#### Post date: [July 16, 2018, 2:49pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/11 "2018-07-16T14:49:59Z")
it works, thank you
Another question:
I need to aggrgate bytesIn and bytesOut fields which are integer fields. I mean if there is three logs with byteIn (500,400,600) the final byteIn aggrgated field will be 1500. How I will do this in code section of aggrgate filter
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 16, 2018, 3:10pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/12 "2018-07-16T15:10:22Z")
Very similar to counting events
```
map['totalBytesIn'] ||= 0
map['totalBytesIn] += event.get('bytesIn')
```
---
### Author:  [@saad](https://discuss.elastic.co/u/saad)
#### Post date: [July 16, 2018, 6:37pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/13 "2018-07-16T18:37:33Z")
Thank you,
I am seeing that logs have field tags = \_aggregatefinalflush
What us that mean?
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 16, 2018, 7:10pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/14 "2018-07-16T19:10:38Z")
It tells you those events were emitted due to the pipeline flush when logstash shut down. See this [issue](https://github.com/logstash-plugins/logstash-filter-aggregate/issues/50) for a discussion. My advice is to ignore it.
---
### Author:  [@saad](https://discuss.elastic.co/u/saad)
#### Post date: [July 16, 2018, 7:37pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/15 "2018-07-16T19:37:56Z")
Now the code is doing aggregation for logs with the same task\_id for timeout of 30 secs.
what is I want only max 50 logs aggreagted within 30 secs. how I will do this?
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 16, 2018, 8:28pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/16 "2018-07-16T20:28:46Z")
If you want to ignore everything after the first 50 then wrap some of the += lines with a test for map['aggregatedEvents'] being \<= 50. If you want to push an event for the first 50 and start a new aggregation I am not sure that you can do that.
---
### Author:  [@saad](https://discuss.elastic.co/u/saad)
#### Post date: [July 17, 2018, 3:52am UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/17 "2018-07-17T03:52:43Z")
I am doing now geoip for the destination IP as below. How can I map field geoip.location to use it for map visualize. Would you advise me and thank you again for your full support
---
### Author:  [@saad](https://discuss.elastic.co/u/saad)
#### Post date: [July 17, 2018, 2:30pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/18 "2018-07-17T14:30:12Z")
I found it.
map['destinationGeoIP'] = event.get('geoip')
Thank you
---
### Author:  [@saad](https://discuss.elastic.co/u/saad)
#### Post date: [August 9, 2018, 2:26pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/19 "2018-08-09T14:26:27Z")
Hi Badger,
I am seeing the Emitted events is more than the Received Events.
I think aggregation goal is to decrease the Emitted events in comparsion to the Recevived Events.
I checked the aggregation its working fine. Need your advise.
Thank you
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [August 9, 2018, 3:08pm UTC](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976/20 "2018-08-09T15:08:14Z")
If you have monitoring enabled and drill into a pipeline you will see that inputs emit events, and outputs receive them. To me this is counter-intuitive, I think a pipeline should receive events, process them, and then emit them, but that's not how those terms are used in logstash.
[Next page](https://discuss.elastic.co/t/is-it-possible-to-aggregate-logs-from-cef-input/139976.md?page=2)