Set event.type for Packetbeat flow events (elastic#24564)

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

CHANGELOG.next.asciidoc

@@ -1034,6 +1034,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Tuned the internal queue size to reduce the chances of events being dropped. {pull}22650[22650]
 - Add support for "http.request.mime_type" and "http.response.mime_type". {pull}22940[22940]
 - 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*
 

packetbeat/flows/worker.go

@@ -220,6 +220,12 @@ func createEvent(
 		"category": []string{"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,

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'])