Logstash Aggregate filter sometimes does not work

Elastic Stack Logstash
1

I have some integration logs coming from filebeat and I'm trying to use the aggregate filter to calculate the delta of the entire process. The code implemented seems to work properly, but looking at kibana some records are not computed correctly.

I checked the logstash logs but it does not report any error about the aggregation filter. The pipeline.workers configuration is set to 1 in pipeline.yml and pipeline.ordered is set to auto in logstash.yml.

I calculate the process in two points, I have the start, middle, and end event flags. The timestamps are recorded on the map when they occur, but sometimes the timestamps of the start event and middle are not recorded on the map as if they were null even though they exist.

Here is my input conf

input {

  beats {
    id => "beatsInputMaster"
    port => 5044
    ssl_enabled => true
    ssl_certificate => "/path/certs/logstash.crt"
    ssl_key => "/path/certs/logstash.key"
  }
}

output {

  if "all_service_logs" in [tags] {
    pipeline {
      id => "sendToServiceFilterLogs"
      send_to => [LogsService1, LogsService2, LogsService3]
    }
  }

}

Here is my filter.conf

input {

  pipeline {
    id => "IntegracaoFilterLogs"
    address => "LogsService1"
  }

}

filter {

   if "logs_service1" in [tags] and "/log/service1.log" in [log][file][path] {

     if [message] =~ "Environment.Variables.Log4j.LogText" {
        dissect {
           mapping => {
              "message" => "%{log_timestamp->} %{+log_timestamp->} %{log_level->} - {%{log_message}}"
           }
        }
     }
     else {
        dissect {
          mapping => {
            "message" => "%{log_timestamp->} %{+log_timestamp->} %{log_level->} - IN:%{input}|EG:%{eg}|GID:%{gid}|SRV:%{servico}|FLW:%{processo}|NOD:%{etapa}|%{log_message}"
          }
        }
     }

     date {
       match => ["log_timestamp", "ISO8601","yyyy/MM/dd HH:mm:ss,SSS"]
       timezone => "America/Sao_Paulo"
       target => "log_timestamp"
     }

     mutate {
        add_field => {"event.dataset" => "service1"}
     }

     aggregate {
        id => "agg_ini_processo"
        task_id => "%{gid}"
        code => '
              map["delta_total"] ||= 0;
              map["delta_service"] ||= 0;

              if event.get("log_message") == "InicioDoProcesso"
                map["delta_total"] =+ event.get("log_timestamp").to_f
              end

              if event.get("log_message") == "Mensagem a ser enviada para o Servico"
                map["delta_service"] =+ event.get("log_timestamp").to_f
              end
            '
        map_action => "create_or_update"
     }

     if [log_message] == "Mensagem retornada do Servico" {
       aggregate {
         id => "agg_fimprocesso"
         task_id => "%{gid}"
         code => '
                  event.set("process_total_time", ((event.get("log_timestamp").to_f - map["delta_total"])*1000).round(0));
                  event.set("service_time", ((event.get("log_timestamp").to_f - map["delta_service"])*1000).round(0));
                 '
         map_action => "update"
         end_of_task => true
         timeout_task_id_field => "gid"
         timeout_tags => ['_aggregatetimeout']
         timeout => 7200
       }
     }

     mutate {
         remove_field => ["event", "message"]
     }
   }
}


output {

  if "logs_service1" in [tags] and "/log/service1.log" in [log][file][path] {
    elasticsearch {
      id => "elk_output_logs"
      index => "service_logs"
      action => "create"
      hosts => ["https://myelasticendpoint:9200"]
      api_key => "${apikey.ls}"
      manage_template => false
      ssl_enabled => "true"
      ssl_verification_mode => "full"
      ssl_certificate_authorities => "${ls.path.certs}/mycert.pem"
    }

    #stdout { codec => rubydebug { metadata => true } }
  }
}

in Kibana I see most of the records are calculated correctly, however, sometimes it only records the Unix timestamp of the last field of the aggregation block

Captura de tela 2024-07-12 012028

Im using logstash v8.11.3

Thank you in advance for your help