Aggregate filter

sergio_junior · July 15, 2021, 3:40pm

hello guys, i need help with the aggregate filter.
I'm trying to merge these logs' lines into a single event.

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...

{ "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?

Badger · July 15, 2021, 4:32pm

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.

sergio_junior · July 15, 2021, 5:59pm

i'm sorry...
done, can you help me?

Badger · July 15, 2021, 7:02pm

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"

sergio_junior · July 15, 2021, 9:20pm

@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

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?

Badger · July 15, 2021, 9:30pm

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.

sergio_junior · July 16, 2021, 1:18pm

@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.

Badger · July 16, 2021, 1:27pm

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.

system · August 13, 2021, 1:28pm

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.

Topic		Replies	Views
Logstash-filter-aggregate Logstash	1	371	February 20, 2018
Aggregate filter (Combine Logs) Logstash	2	325	April 24, 2018
Combining 3 logs into one Logstash	2	173	August 17, 2023
Logstash aggregation filter not working properly Logstash	3	1521	July 6, 2017
Logstash aggregate filter not working Logstash	2	943	January 11, 2017

Aggregate filter

Related topics