Calculating event.duration not working

[willemdh]

Hello,

Based on https://www.elastic.co/blog/calculating-ingest-lag-and-storing-ingest-time-in-elasticsearch-to-improve-observability

I added a script processor to my pipeline with the goal to calculate the event.duration

{
  "csm-av-clamav-update" : {
    "description" : "CSM ClamAV Update Pipeline",
    "processors" : [
      {
        "set" : {
          "field" : "_index",
          "value" : "ecs-av-30-rollover",
          "description" : "Set _index ecs-av-30-rollover"
        }
      },
      {
        "set" : {
          "field" : "pipeline",
          "value" : "csm-av-clamav-update",
          "description" : "Set pipeline csm-av-clamav-update"
        }
      },
      {
        "set" : {
          "field" : "event.original",
          "value" : "{{message}}",
          "description" : "Set event.original message"
        }
      },
      {
        "set" : {
          "field" : "event.created",
          "value" : "{{@timestamp}}",
          "description" : "Set event.created @timestamp"
        }
      },
      {
        "set" : {
          "field" : "event.module",
          "value" : "clamav",
          "description" : "Set event.module clamav"
        }
      },
      {
        "set" : {
          "field" : "event.dataset",
          "value" : "clamav.update",
          "description" : "Set event.dataset clamav.update"
        }
      },
      {
        "set" : {
          "field" : "service.name",
          "value" : "ClamAV",
          "description" : "Set service.name ClamAV"
        }
      },
      {
        "set" : {
          "field" : "service.type",
          "value" : "antivirus",
          "description" : "Set service.type antivirus"
        }
      },
      {
        "grok" : {
          "field" : "message",
          "patterns" : [
            """(?m)\AStart time: %{TIMESTAMP_ISO8601:av.update.start_time}\n%{SUPERGREEDY:message}\nEnd time: %{TIMESTAMP_ISO8601:av.update.end_time}\Z"""
          ],
          "pattern_definitions" : {
            "CLAMAVTIMESTAMP" : "%{DAY} %{MONTH}%{SPACE}%{MONTHDAY} %{TIME} %{YEAR}",
            "SUPERGREEDY" : """(.|\r|\n)*"""
          },
          "description" : "Grok message",
          "on_failure" : [
            {
              "set" : {
                "field" : "pipeline_error",
                "value" : "{{ _ingest.on_failure_message }}",
                "description" : "Set Grok pipeline_error"
              }
            }
          ]
        }
      },
      {
        "date" : {
          "field" : "av.update.start_time",
          "target_field" : "av.update.start_time",
          "formats" : [
            "yyyy-MM-dd HH:mm:ss,SSS"
          ],
          "timezone" : "Europe/Brussels",
          "description" : "Date av.update.start_time",
          "on_failure" : [
            {
              "set" : {
                "field" : "pipeline_error",
                "value" : "{{ _ingest.on_failure_message }}",
                "description" : "Set Date pipeline_error"
              }
            }
          ]
        }
      },
      {
        "date" : {
          "field" : "av.update.end_time",
          "target_field" : "av.update.end_time",
          "formats" : [
            "yyyy-MM-dd HH:mm:ss,SSS"
          ],
          "timezone" : "Europe/Brussels",
          "description" : "Date av.update.end_time",
          "on_failure" : [
            {
              "set" : {
                "field" : "pipeline_error",
                "value" : "{{ _ingest.on_failure_message }}",
                "description" : "Set Data pipeline_error"
              }
            }
          ]
        }
      },
      {
        "script" : {
          "lang" : "painless",
          "source" : """ 
            if(ctx.containsKey("av.update.start_time") && ctx.containsKey("av.update.end_time")) { 
              ctx['event.duration'] = ChronoUnit.MILLIS.between(ZonedDateTime.parse(ctx['av.update.end_time']), ZonedDateTime.parse(ctx['av.update.start_time']))/1000; 
            } 
        """,
          "on_failure" : [
            {
              "set" : {
                "field" : "pipeline_error",
                "value" : "{{ _ingest.on_failure_message }}",
                "description" : "Set script painless pipeline_error"
              }
            }
          ]
        }
      }
    ],
    "version" : 3
  }
}

But I have probably made a mistake, as there is no event.duration in my data. Also no pipeline_error

Anyone got an idea what might be wrong with my painless script? av.update.start_time and av.update.end_time exist and are date type and format:

Thanks.

Willem

[willemdh]

Ok, found the solution..

{ 
  "script": { 
    "lang": "painless", 
    "source": """
        if (ctx.containsKey('event')) {
          if(ctx['event'].containsKey('start') && ctx['event'].containsKey('end')) { 
          ctx['event.duration'] = ChronoUnit.NANOS.between(ZonedDateTime.parse(ctx['event']['start']), ZonedDateTime.parse(ctx['event']['end']));
          }
        } 
    """,
    "on_failure" : [
      {
        "set" : {
          "field" : "pipeline_error",
          "value" : "{{ _ingest.on_failure_message }}",
          "description" : "Set script painless pipeline_error"
        }
      }
    ]
  }
},

I was wrongly assuming painless recognized dot field notation (eg event.udation)

Grtz

Willem