add tracking of window starts and ends

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

client/internal/netflow/manager.go

@@ -299,9 +299,11 @@ func (m *Manager) send(event *nftypes.Event) error {
 
 func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 	protoEvent := &proto.FlowEvent{
-		EventId:   event.ID[:],
-		Timestamp: timestamppb.New(event.Timestamp),
-		PublicKey: publicKey,
+		EventId:     event.ID[:],
+		Timestamp:   timestamppb.New(event.Timestamp),
+		PublicKey:   publicKey,
+		WindowStart: timestamppb.New(event.WindowStart),
+		WindowEnd:   timestamppb.New(event.WindowEnd),
 		FlowFields: &proto.FlowFields{
 			FlowId:           event.FlowID[:],
 			RuleId:           event.RuleID,

client/internal/netflow/store/event_aggregation_test.go

@@ -50,10 +50,12 @@ func TestFlowAggregation(t *testing.T) {
 		for _, tt := range tests {
 			t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
 				store := NewAggregatingMemoryStore()
+				store.WindowEnd = time.Now().Add(5 * time.Second)
+
 				allExpected := make([]*types.Event, 0)
 
 				for _, srcAndDst := range tt.addresses {
-					inEvents, expected := generateEvents(srcAndDst[0], srcAndDst[1], tt.dstPort, tt.eventTypes, protocol, types.Ingress, 0)
+					inEvents, expected := generateEvents(srcAndDst[0], srcAndDst[1], tt.dstPort, tt.eventTypes, protocol, types.Ingress, 0, store.WindowStart, store.WindowEnd)
 					for _, e := range inEvents {
 						store.StoreEvent(e)
 					}
@@ -101,9 +103,11 @@ func TestIcmpEventAggregation(t *testing.T) {
 		for _, tt := range tests {
 			t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
 				store := NewAggregatingMemoryStore()
+				store.WindowEnd = time.Now().Add(5 * time.Second)
+
 				allExpected := make([]*types.Event, 0)
 				for _, icmpType := range icmpTypes {
-					events, expected := generateEvents(tt.addresses[0][0], tt.addresses[0][1], 0, tt.eventTypes, protocol, types.Ingress, icmpType)
+					events, expected := generateEvents(tt.addresses[0][0], tt.addresses[0][1], 0, tt.eventTypes, protocol, types.Ingress, icmpType, store.WindowStart, store.WindowEnd)
 					for _, e := range events {
 						store.StoreEvent(e)
 					}
@@ -122,7 +126,8 @@ func ipAddr(a string) netip.Addr {
 	return addr
 }
 
-func generateEvents(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.Type, protocol types.Protocol, direction types.Direction, icmpType uint8) ([]*types.Event, *types.Event) {
+func generateEvents(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.Type, protocol types.Protocol,
+	direction types.Direction, icmpType uint8, windowStart, windowEnd time.Time) ([]*types.Event, *types.Event) {
 	var rxPackets, txPackets, rxBytes, txBytes uint64
 	inEvents := make([]*types.Event, 0)
 	ts := time.Now()
@@ -172,8 +177,10 @@ func generateEvents(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.
 		}
 	}
 	aggregatedEvent := &types.Event{
-		ID:        inEvents[0].ID,
-		Timestamp: inEvents[0].Timestamp,
+		ID:          inEvents[0].ID,
+		Timestamp:   inEvents[0].Timestamp,
+		WindowStart: windowStart,
+		WindowEnd:   windowEnd,
 		EventFields: types.EventFields{
 			FlowID:           flowId,
 			Type:             types.TypeUnknown,

client/internal/netflow/store/memory.go

@@ -24,6 +24,8 @@ type Memory struct {
 
 type AggregatingMemory struct {
 	Memory
+	WindowStart time.Time
+	WindowEnd   time.Time
 }
 
 func (m *Memory) StoreEvent(event *types.Event) {
@@ -57,15 +59,17 @@ func (m *Memory) DeleteEvents(ids []uuid.UUID) {
 }
 
 func NewAggregatingMemoryStore() *AggregatingMemory {
-	return &AggregatingMemory{Memory{events: make(map[uuid.UUID]*types.Event)}}
+	return &AggregatingMemory{WindowStart: time.Now(), Memory: Memory{events: make(map[uuid.UUID]*types.Event)}}
 }
 
 func (am *AggregatingMemory) ResetAggregationWindow() types.FlowEventAggregator {
 	am.mux.Lock()
 	defer am.mux.Unlock()
 
-	toret := AggregatingMemory{Memory: Memory{events: am.events}}
+	toret := AggregatingMemory{WindowStart: am.WindowStart, WindowEnd: time.Now(), Memory: Memory{events: am.events}}
+
 	am.events = make(map[uuid.UUID]*types.Event)
+	am.WindowStart = time.Now()
 
 	return &toret
 }
@@ -110,6 +114,8 @@ func (am *AggregatingMemory) GetAggregatedEvents() []*types.Event {
 			// Please note that ICMPCode field isn't propagated by the manager (see flow/proto/flow.pb.go, FlowFields struct)
 			// so the field value in an icmp event in the "aggregated" doesn't matter
 
+			event.WindowStart = am.WindowStart
+			event.WindowEnd = am.WindowEnd
 			continue
 		}
 

client/internal/netflow/types/types.go

@@ -70,8 +70,10 @@ const (
 )
 
 type Event struct {
-	ID        uuid.UUID
-	Timestamp time.Time
+	ID          uuid.UUID
+	Timestamp   time.Time
+	WindowStart time.Time
+	WindowEnd   time.Time
 	EventFields
 }