# Aggregate filter
**URL:** https://discuss.elastic.co/t/aggregate-filter/278817
**Category:** Logstash
**Created:** [July 15, 2021, 3:40pm UTC](https://discuss.elastic.co/t/aggregate-filter/278817 "2021-07-15T15:40:22Z")
**Posts on this page:** 9
**Page:** 1
### Author:  [@sergio\_junior](https://discuss.elastic.co/u/sergio_junior)
#### Post date: [July 15, 2021, 3:40pm UTC](https://discuss.elastic.co/t/aggregate-filter/278817/1 "2021-07-15T15:40:22Z")
hello guys, i need help with the aggregate filter.
I'm trying to merge these logs' lines into a single event.
```auto
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Iniciando pesquisa
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o tipo PREFIXO/SUFIXO
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o tipo da pesquisa
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | tipo: TODOS
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando namedQuery (caso exista)
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | namedQuery: null
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o bean do cache
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | bean: null
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Não encontrado no cache: FV600930495FL
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Chamando método: hawbDAO.findByPesquisaInteligente(value.toString(), tipoPesquisaHawb)
2021-07-08 00:00:00,215 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Documentos encontrados: [hawbId = 297281015 | numEncCli = FV600930495FL | numEncTer = null | tipoHawb = HAWB]
2021-07-08 00:00:00,215 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Fim da pesquisa
```
the expected output would be something like this...
```auto
{ "message": "2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Iniciando pesquisa
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o tipo PREFIXO/SUFIXO
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o tipo da pesquisa
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | tipo: TODOS
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando namedQuery (caso exista)
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | namedQuery: null
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o bean do cache
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | bean: null
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Não encontrado no cache: FV600930495FL
2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Chamando método: hawbDAO.findByPesquisaInteligente(value.toString(), tipoPesquisaHawb)
2021-07-08 00:00:00,215 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Documentos encontrados: [hawbId = 297281015 | numEncCli = FV600930495FL | numEncTer = null | tipoHawb = HAWB]
2021-07-08 00:00:00,215 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Fim da pesquisa
", "tipo":"todos", "bean": null, "cache":"FV600930495FL", "methods":"hawbDAO.findByPesquisaInteligente(value.toString(), tipoPesquisaHawb)","found.hawbId": 297281015,"found.numEncCli":"FV600930495FL","found.numEncTer"= nill,"found.tipoHawb": "HAWAB","timestamp":"2021-07-08 00:00:00,181","queryTime": 34, "logType":"INFO","task":"default task-1876","exit":"stdout", "app":" ###HawbPesquisaInteligenteConverter"}
```
can anybody help me?
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 15, 2021, 4:32pm UTC](https://discuss.elastic.co/t/aggregate-filter/278817/2 "2021-07-15T16:32:55Z")
Your post is unreadable. Please edit it, select the log lines and click on \ in the toolbar above the edit pane. Then do the same for the desired output.
---
### Author:  [@sergio\_junior](https://discuss.elastic.co/u/sergio_junior)
#### Post date: [July 15, 2021, 5:59pm UTC](https://discuss.elastic.co/t/aggregate-filter/278817/3 "2021-07-15T17:59:20Z")
i'm sorry...
done, can you help me?
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 15, 2021, 7:02pm UTC](https://discuss.elastic.co/t/aggregate-filter/278817/4 "2021-07-15T19:02:58Z")
Hopefully this will give you some ideas on how to get started....
```
filter {
# Workaround for dissect handling of single space when padding is allowed
# For example, "181 INFO [" works, "181 ERROR [" does not.
mutate { gsub => ["message", "\s+", " "] }
dissect { mapping => { "message" => "%{[@metadata][timestamp]} %{+[@metadata][timestamp]} %{loglevel} [%{exit}] (%{[@metadata][task]}) %{app} | %{[@metadata][restOfLine]}" } }
date { match => ["[@metadata][timestamp]", "YYYY-MM-dd HH:mm:ss,SSS" ] }
grok {
break_on_match => false
match => {
"[@metadata][restOfLine]" => [
"tipo: %{WORD:tipo}",
"bean: %{WORD:bean}"
]
}
}
if [@metadata][restOfLine] =~ /^Documentos encontrados: \[/ {
dissect { mapping => { "[@metadata][restOfLine]" => "Documentos encontrados: [%{[@metadata][kvData]}]" } }
kv { source => "[@metadata][kvData]" prefix => "found." field_split => "|" value_split => "=" trim_key => " " trim_value => " " target => "kvData" }
}
aggregate {
task_id => "%{[@metadata][task]}"
code => '
map["message"] ||= ""
map["message"] += event.get("message") + " "
map["app"] ||= event.get("app")
map["tipo"] ||= event.get("tipo")
map["bean"] ||= event.get("bean")
kvData = event.get("kvData")
if kvData
kvData.each { |k, v| map[k] = v }
end
map["startTime"] ||= event.get("@timestamp")
map["endTime"] = event.get("@timestamp")
event.cancel
'
timeout_code => '
event.set("queryTime", event.get("endTime").to_f - event.get("startTime").to_f)
event.set("@timestamp", event.get("endTime"))
'
push_map_as_event_on_timeout => true
timeout_task_id_field => "task"
timeout => 3
}
}
output { stdout { codec => rubydebug { metadata => true } } }
```
You will probably want to comment out the event.cancel until you have figured out what additional grok patterns you need. As it is, that pipeline will produce
```
"app" => "###HawbPesquisaInteligenteConverter",
"tipo" => "TODOS",
"found.numEncTer" => "null",
"found.numEncCli" => "FV600930495FL",
"message" => "2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Iniciando pesquisa 2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o tipo PREFIXO/SUFIXO 2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o tipo da pesquisa 2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | tipo: TODOS 2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando namedQuery (caso exista) 2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | namedQuery: null 2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o bean do cache 2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | bean: null 2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Não encontrado no cache: FV600930495FL 2021-07-08 00:00:00,181 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Chamando método: hawbDAO.findByPesquisaInteligente(value.toString(), tipoPesquisaHawb) 2021-07-08 00:00:00,215 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Documentos encontrados: [hawbId = 297281015 | numEncCli = FV600930495FL | numEncTer = null | tipoHawb = HAWB] 2021-07-08 00:00:00,215 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Fim da pesquisa ",
"found.hawbId" => "297281015",
"task" => "default task-1876",
"@timestamp" => 2021-07-08T04:00:00.215Z,
"found.tipoHawb" => "HAWB",
"@version" => "1",
"queryTime" => 0.03399991989135742,
"startTime" => 2021-07-08T04:00:00.181Z,
"endTime" => 2021-07-08T04:00:00.215Z,
"bean" => "null"
```
---
### Author:  [@sergio\_junior](https://discuss.elastic.co/u/sergio_junior)
#### Post date: [July 15, 2021, 9:20pm UTC](https://discuss.elastic.co/t/aggregate-filter/278817/5 "2021-07-15T21:20:42Z")
@Badger thank you very much, I have one more question if it's not asking too much. What I'm actually trying to do is collect through triggers, like for example
```auto
2021-07-08 00:01:25,167 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | **Iniciando pesquisa** <- start keyword
2021-07-08 00:01:25,167 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | Recuperando o tipo PREFIXO/SUFIXO
2021-07-08 00:01:25,167 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | Recuperando o tipo da pesquisa
2021-07-08 00:01:25,167 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | tipo: TODOS
2021-07-08 00:01:25,167 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | Recuperando namedQuery (caso exista)
2021-07-08 00:01:25,167 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | namedQuery: null
2021-07-08 00:01:25,167 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | Recuperando o bean do cache
2021-07-08 00:01:25,167 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | bean: br.com.flashcourier.pegasus.managedbeans.RecepcionarHawbBean@7397e68d
2021-07-08 00:01:25,167 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | Não encontrado no cache: 7004151921636
2021-07-08 00:01:25,167 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | Chamando método: hawbDAO.findByPesquisaInteligente(value.toString(), tipoPesquisaHawb)
2021-07-08 00:01:25,186 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | Documentos encontrados: [hawbId = 298470932 | numEncCli = 7004151921636 | numEncTer = null | tipoHawb = HAWB]
2021-07-08 00:01:25,186 INFO [stdout] (default task-1881) ###HawbPesquisaInteligenteConverter | **Fim da pesquisa** <-end keyword
2021-07-08 00:01:25,598 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | **Iniciando pesquisa** <- start keyword
2021-07-08 00:01:25,599 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o tipo PREFIXO/SUFIXO
2021-07-08 00:01:25,599 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o tipo da pesquisa
2021-07-08 00:01:25,599 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | tipo: TODOS
2021-07-08 00:01:25,599 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando namedQuery (caso exista)
2021-07-08 00:01:25,599 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | namedQuery: null
2021-07-08 00:01:25,599 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Recuperando o bean do cache
2021-07-08 00:01:25,599 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | bean: null
2021-07-08 00:01:25,599 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Não encontrado no cache: 02980896179
2021-07-08 00:01:25,599 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Chamando método: hawbDAO.findByPesquisaInteligente(value.toString(), tipoPesquisaHawb)
2021-07-08 00:01:25,605 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | Documentos encontrados: [hawbId = 298089617 | numEncCli = AGSE803000338BR | numEncTer = null | tipoHawb = HAWB]
2021-07-08 00:01:25,605 INFO [stdout] (default task-1876) ###HawbPesquisaInteligenteConverter | **Fim da pesquisa** <- end keyword
```
because there are many logs with the same tasks. Can you help me one more time?
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 15, 2021, 9:30pm UTC](https://discuss.elastic.co/t/aggregate-filter/278817/6 "2021-07-15T21:30:16Z")
I do not understand why you are interested in the start and end. If you have a unique task id it is far simpler to aggregate events using it than to recognize the start and end and maintain state across multiple events.
---
### Author:  [@sergio\_junior](https://discuss.elastic.co/u/sergio_junior)
#### Post date: [July 16, 2021, 1:18pm UTC](https://discuss.elastic.co/t/aggregate-filter/278817/7 "2021-07-16T13:18:02Z")
@badger thanks for your attention. The logs that will be monitored have these two taskIds, what I needed was to separate them into events in the course of which they would appear. As it is, it aggregates everyone from the entire day into a single event, if you know what I mean.
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 16, 2021, 1:27pm UTC](https://discuss.elastic.co/t/aggregate-filter/278817/8 "2021-07-16T13:27:18Z")
> [@sergio\_junior](#):
>
> As it is, it aggregates everyone from the entire day into a single event, if you know what I mean.
No, I do not, since the code I showed will produce a different event for each value of [task], so "default task-1881" and "default task-1876" will be different events.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [August 13, 2021, 1:28pm UTC](https://discuss.elastic.co/t/aggregate-filter/278817/9 "2021-08-13T13:28:07Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.