added tracking of the number of start-, drop, and end-events in an aggregation window

Signed-off-by: Dmitri <dmitri.external@netbird.io>

client/internal/netflow/store/memory.go

@@ -91,6 +91,14 @@ func (am *AggregatingMemory) GetAggregatedEvents() []*types.Event {
 				aggregatedEvent.RxPackets += v.RxPackets
 				aggregatedEvent.TxBytes += v.TxBytes
 				aggregatedEvent.TxPackets += v.TxPackets
+				switch v.Type {
+				case types.TypeStart:
+					aggregatedEvent.NumOfStarts += 1
+				case types.TypeDrop:
+					aggregatedEvent.NumOfDrops += 1
+				case types.TypeEnd:
+					aggregatedEvent.NumOfEnds += 1
+				}
 				if aggregatedEvent.Timestamp.Compare(v.Timestamp) > 0 {
 					aggregatedEvent.Timestamp = v.Timestamp
 					aggregatedEvent.ID = v.ID
@@ -103,10 +111,19 @@ func (am *AggregatingMemory) GetAggregatedEvents() []*types.Event {
 		} else {
 			switch v.Protocol {
 			case types.ICMP, types.ICMPv6, types.TCP, types.UDP:
-				aggregated[lookupKey] = v
+				event := v.Clone()
+				aggregated[lookupKey] = event
+				switch event.Type {
+				case types.TypeStart:
+					event.NumOfStarts += 1
+				case types.TypeDrop:
+					event.NumOfDrops += 1
+				case types.TypeEnd:
+					event.NumOfEnds += 1
+				}
 			default:
 				lookupKey.ts = time.Now().UnixNano() // to make the lookup key unique so we don't aggregate on it
-				aggregated[lookupKey] = v
+				aggregated[lookupKey] = v.Clone()
 			}
 		}
 	}

client/internal/netflow/store/tcp_aggregation_test.go

@@ -88,6 +88,9 @@ func TestTcpAggregation(t *testing.T) {
 						TxPackets:        60,
 						RxBytes:          40000,
 						TxBytes:          60000,
+						NumOfStarts:      1,
+						NumOfEnds:        1,
+						NumOfDrops:       0,
 					}},
 			},
 		},
@@ -155,6 +158,9 @@ func TestTcpAggregation(t *testing.T) {
 						TxPackets:        60,
 						RxBytes:          40000,
 						TxBytes:          60000,
+						NumOfStarts:      1,
+						NumOfEnds:        0,
+						NumOfDrops:       1,
 					}},
 			},
 		},
@@ -202,6 +208,9 @@ func TestTcpAggregation(t *testing.T) {
 						TxPackets:        20,
 						RxBytes:          10000,
 						TxBytes:          20000,
+						NumOfStarts:      1,
+						NumOfEnds:        0,
+						NumOfDrops:       0,
 					}},
 			},
 		},
@@ -213,7 +222,7 @@ func TestTcpAggregation(t *testing.T) {
 					Timestamp: time.Unix(100, 100).Add(time.Second),
 					EventFields: types.EventFields{
 						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeEnd,
+						Type:             types.TypeDrop,
 						RuleID:           []byte("rule-id-1"),
 						Direction:        types.Egress,
 						Protocol:         types.TCP,
@@ -235,7 +244,7 @@ func TestTcpAggregation(t *testing.T) {
 					Timestamp: time.Unix(100, 100).Add(time.Second),
 					EventFields: types.EventFields{
 						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeEnd,
+						Type:             types.TypeDrop,
 						RuleID:           []byte("rule-id-1"),
 						Direction:        types.Egress,
 						Protocol:         types.TCP,
@@ -249,6 +258,9 @@ func TestTcpAggregation(t *testing.T) {
 						TxPackets:        40,
 						RxBytes:          30000,
 						TxBytes:          40000,
+						NumOfStarts:      0,
+						NumOfEnds:        0,
+						NumOfDrops:       1,
 					}},
 			},
 		}}

client/internal/netflow/types/types.go

@@ -2,6 +2,7 @@ package types
 
 import (
 	"net/netip"
+	"slices"
 	"strconv"
 	"time"
 
@@ -92,6 +93,17 @@ type EventFields struct {
 	TxPackets        uint64
 	RxBytes          uint64
 	TxBytes          uint64
+	NumOfStarts      uint64
+	NumOfEnds        uint64
+	NumOfDrops       uint64
+}
+
+func (e *Event) Clone() *Event {
+	toret := *e
+	toret.RuleID = slices.Clone(e.RuleID)
+	toret.SourceResourceID = slices.Clone(e.SourceResourceID)
+	toret.DestResourceID = slices.Clone(e.DestResourceID)
+	return &toret
 }
 
 type FlowConfig struct {