You could use an aggregate filter to do that
if [message] =~ /"@Moving":"/ {
grok { match => { "message" => '{"@time":"(?<[@metadata][timestamp]>[^"]+)" "@Moving":"(?<[@metadata][msg]>[^"]+)"' } }
mutate { add_field => { "[@metadata][json]" => '{ "@timestamp": "%{[@metadata][timestamp]}" }' } }
json { source => "[@metadata][json]" }
mutate { add_field => { "[@metadata][taskId]" => "anyConstant" } }
if [@metadata][msg] =~ /^</ {
aggregate {
task_id => "%{[@metadata][taskId]}"
code => 'map["startTime"] = event.get("@timestamp").to_f'
map_action => "create"
}
} else {
aggregate {
task_id => "%{[@metadata][taskId]}"
code => 'event.set("delta", event.get("@timestamp").to_f - map["startTime"])'
map_action => "update"
end_of_task => true
}
}
}
which will produce
"delta" => 181.1056525707245,
for the second pair of lines. Note that using aggregate requires pipeline.workers to be 1 and pipeline.ordered to be true.