# Aggregation rule
**URL:** https://discuss.elastic.co/t/aggregation-rule/375011
**Category:** Logstash
**Created:** [February 25, 2025, 11:43am UTC](https://discuss.elastic.co/t/aggregation-rule/375011 "2025-02-25T11:43:23Z")
**Posts on this page:** 8
**Page:** 1
### Author:  [@cybersc\_1](https://discuss.elastic.co/u/cybersc_1)
#### Post date: [February 25, 2025, 11:43am UTC](https://discuss.elastic.co/t/aggregation-rule/375011/1 "2025-02-25T11:43:23Z")
Hi there!
I have plenty of events that look like that:
Feb 25 15:05:50 xfirewall CEF:0|infotecs|xf|5.4|62|Non-encrypted forwarded IP packet passed|5|start=1740477743000 end=1740477743000 src=\*\*\* dst=\*\*\* spt=53 dpt=38826 proto=udp cnt=2 in=492 deviceInboundInterface=eth9.2066 cn1=800h cn1Label=Eth protocol hex cs1=NAT no cs2=Drop no cs3=Broadcast no cs4=Forward yes cs5=Encrypted no cs1Label=Flags cs6=unknown cs6Label=Application name deviceDirection=inbound app=unknown suser= deviceExternalId=05bf2548
I need to create an aggregation rule which will aggregate such events by these fields: src, dst, spt, dpt and the one that have "xfirewall" value (don't really know what this field is called)
I tried different things - even used AI, but it won't provide a correct rule. Help, please.
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [February 25, 2025, 1:42pm UTC](https://discuss.elastic.co/t/aggregation-rule/375011/2 "2025-02-25T13:42:32Z")
> [@cybersc\_1](#):
>
> I need to create an aggregation rule
What do you want to aggregate? Do you want to combine all events that have the same src/dst/spt/dpt into a single event?
---
### Author:  [@cybersc\_1](https://discuss.elastic.co/u/cybersc_1)
#### Post date: [February 26, 2025, 10:13am UTC](https://discuss.elastic.co/t/aggregation-rule/375011/4 "2025-02-26T10:13:10Z")
Yeah, exactly
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [February 26, 2025, 5:09pm UTC](https://discuss.elastic.co/t/aggregation-rule/375011/5 "2025-02-26T17:09:30Z")
OK. Your messages appear to be in syslog format with a CEF payload. I'd use two pipelines to process that
```
input { generator { count => 1 lines => [
'Feb 25 15:05:50 xfirewall CEF:0|infotecs|xf|5.4|62|Non-encrypted forwarded IP packet passed|5|start=1740477743000 end=1740477743000 src=127.4.8.23 dst=127.5.1.82 spt=53 dpt=38826 proto=udp cnt=2 in=492 deviceInboundInterface=eth9.2066 cn1=800h cn1Label=Eth protocol hex cs1=NAT no cs2=Drop no cs3=Broadcast no cs4=Forward yes cs5=Encrypted no cs1Label=Flags cs6=unknown cs6Label=Application name deviceDirection=inbound app=unknown suser= deviceExternalId=05bf2548',
'Feb 25 15:05:51 xfirewall CEF:0|infotecs|xf|5.4|62|Non-encrypted forwarded IP packet passed|5|start=1740477743000 end=1740477743000 src=127.4.8.23 dst=127.5.1.82 spt=53 dpt=38826 proto=udp cnt=2 in=492 deviceInboundInterface=eth9.2066 cn1=800h cn1Label=Eth protocol hex cs1=NAT no cs2=Drop no cs3=Broadcast no cs4=Forward yes cs5=Encrypted no cs1Label=Flags cs6=unknown cs6Label=Application name deviceDirection=inbound app=unknown suser= deviceExternalId=05bf2548',
'Feb 25 15:05:52 xfirewall CEF:0|infotecs|xf|5.4|62|Non-encrypted forwarded IP packet passed|5|start=1740477743000 end=1740477743000 src=127.4.8.22 dst=127.5.1.82 spt=53 dpt=55333 proto=udp cnt=2 in=492 deviceInboundInterface=eth9.2066 cn1=800h cn1Label=Eth protocol hex cs1=NAT no cs2=Drop no cs3=Broadcast no cs4=Forward yes cs5=Encrypted no cs1Label=Flags cs6=unknown cs6Label=Application name deviceDirection=inbound app=unknown suser= deviceExternalId=05bf2548' ] } }
output { tcp { host => "127.2.2.2" port => 5678 codec => line { format => "%{message}" } } }
```
and connect that to
```
input { tcp { host => "127.2.2.2" port => 5678 codec => cef { delimiter => "
" ecs_compatibility => "disabled" } } }
output { stdout {} }
filter {
dissect { mapping => { "syslog" => "%{[@metadata][ts]} %{+[@metadata][ts]} %{+[@metadata][ts]} %{hostname}" } }
date { match => ["[@metadata][ts]", "MMM dd HH:mm:ss", "MMM d HH:mm:ss" ] }
aggregate {
task_id => "%{sourceAddress}+%{sourcePort}+%{destinationAddress}+%{destinationPort}"
code => '
map["events"] ||= []
map["events"] << event.to_hash
map["startTime"] ||= event.get("@timestamp")
map["endTime"] = event.get("@timestamp")
event.cancel
'
push_map_as_event_on_timeout => true
timeout => 6
timeout_code => '
'
}
}
```
If you do not include `ecs_compatibility => "disabled"` on the code then the task\_id option would be `%{[source][ip]}+%{[source][port]}+%{[destination][ip]}+%{[destination][port]}`
It's not at all clear to me that an array of events is useful. It may be better to let elasticsearch aggregate over a connection using src/dst ip and port numbers.
---
### Author:  [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)
#### Post date: [February 26, 2025, 6:14pm UTC](https://discuss.elastic.co/t/aggregation-rule/375011/6 "2025-02-26T18:14:59Z")
For curiosity, what is the reason behind this?
While this may be possible, as Badger demonstred, this may seem as unusual as firewall/network devices normally logs one event per line and this can also be extremely inneficient as you woul need to run logstash with just one worker.
Also, this may lead to some events that have nothing to do to being aggregate, as source ports are reused.
---
### Author:  [@cybersc\_1](https://discuss.elastic.co/u/cybersc_1)
#### Post date: [February 27, 2025, 6:37am UTC](https://discuss.elastic.co/t/aggregation-rule/375011/7 "2025-02-27T06:37:53Z")
The thing is i work in a SOC, so we basically catch events from almost every device. Recently, our SIEM system have been dealing with a big load due to these events, so to reduce this load we decided to aggregate such logs. In the end, SIEM would process not a ton of events from XFirewall, but way less.
---
### Author:  [@cybersc\_1](https://discuss.elastic.co/u/cybersc_1)
#### Post date: [February 27, 2025, 6:44am UTC](https://discuss.elastic.co/t/aggregation-rule/375011/8 "2025-02-27T06:44:46Z")
What IP should i put in "host" (in **input** )? Is it my logstash server's ip? or the device that sends these logs?
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [February 27, 2025, 1:30pm UTC](https://discuss.elastic.co/t/aggregation-rule/375011/9 "2025-02-27T13:30:10Z")
127.2.2.2 is an address on the loopback network, just like 127.0.0.1. It will work on every system I am familiar with. It is one of your logstash server's IP addresses.