From cbc907dab8a7bfdd47b0348a68c368bea6ef9038 Mon Sep 17 00:00:00 2001
From: Andrew Kroh <andrew.kroh@elastic.co>
Date: Thu, 18 Mar 2021 09:35:24 -0400
Subject: [PATCH] Set event.type for Packetbeat flow events (#24564)

Add `event.type: [connection]` to flow events and include `end` for final flows.

(cherry picked from commit 7f35a782d49bdb586d6759cec520e7a3b8b256ed)
---
 CHANGELOG.next.asciidoc                    | 1 +
 packetbeat/flows/worker.go                 | 6 ++++++
 packetbeat/tests/system/test_0060_flows.py | 1 +
 3 files changed, 8 insertions(+)

diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 392867e5630..eda9e956a01 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -674,6 +674,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 
 - Upgrade to ECS 1.8.0. {pull}23783[23783]
+- Add `event.type: [connection]` to flow events and include `end` for final flows. {pull}24564[24564]
 
 *Functionbeat*
 
diff --git a/packetbeat/flows/worker.go b/packetbeat/flows/worker.go
index eebc3c260b5..1c9db02ff6f 100644
--- a/packetbeat/flows/worker.go
+++ b/packetbeat/flows/worker.go
@@ -220,6 +220,12 @@ func createEvent(
 		"category": []string{"network_traffic", "network"},
 		"action":   "network_flow",
 	}
+	eventType := []string{"connection"}
+	if isOver {
+		eventType = append(eventType, "end")
+	}
+	event["type"] = eventType
+
 	flow := common.MapStr{
 		"id":    common.NetString(f.id.Serialize()),
 		"final": isOver,
diff --git a/packetbeat/tests/system/test_0060_flows.py b/packetbeat/tests/system/test_0060_flows.py
index ab2e91bfbbc..ae6a4f923fb 100644
--- a/packetbeat/tests/system/test_0060_flows.py
+++ b/packetbeat/tests/system/test_0060_flows.py
@@ -47,6 +47,7 @@ def test_mysql_flow(self):
             'source.bytes': 1480,
             'destination.packets': 10,
             'destination.bytes': 181133,
+            'event.type': ['connection', 'end'],
         })
 
         start_ts = parse_timestamp(objs[0]['event.start'])