[client] Index peer tunnel IPs for faster PeerStateByIP lookup (#6412)

* [client] Index peer tunnel IPs for O(1) PeerStateByIP lookup

Replace the linear scan over all peers with an ipToKey map maintained
by AddPeer/RemovePeer, covering both IPv4 and IPv6 tunnel addresses.

Offline peers are intentionally no longer resolvable by IP: only active
peers can carry traffic, so IdentityForIP and the DNS disconnected-peer
filter now treat them as unknown, same as foreign IPs.

Skip the DNS answer filter for single-record responses; dropping the
only answer was always restored by the empty-answer escape hatch, so
the fast path is behavior-neutral.

* Ensure `ipToKey` entries are only removed if they match the peer being deleted, preventing accidental removal of unrelated mappings.

client/embed/embed.go

@@ -446,8 +446,8 @@ func (c *Client) Expose(ctx context.Context, req ExposeRequest) (*ExposeSession,
 
 // IdentityForIP looks up a remote peer by its tunnel IP using the
 // embedded client's status recorder. Returns the peer's WireGuard public
-// key and FQDN. ok=false means the IP isn't in this client's peer
-// roster — callers should treat that as "unknown peer".
+// key and FQDN. ok=false means the IP doesn't belong to an active peer
+// — offline roster peers are treated as unknown, same as foreign IPs.
 func (c *Client) IdentityForIP(ip netip.Addr) (pubKey, fqdn string, ok bool) {
 	if !ip.IsValid() || c.recorder == nil {
 		return "", "", false

client/internal/dns/local/local.go

@@ -482,7 +482,7 @@ func (d *Resolver) logDNSError(logger *log.Entry, hostname string, qtype uint16,
 // completely when every proxy peer is offline (the upstream may still
 // be reachable some other way, or the peerstore may be stale).
 func (d *Resolver) filterDisconnectedPeerAnswers(logger *log.Entry, question dns.Question, records []dns.RR) []dns.RR {
-	if len(records) == 0 {
+	if len(records) < 2 {
 		return records
 	}
 	d.mu.RLock()

client/internal/dns/local/local_test.go

@@ -2738,6 +2738,17 @@ func TestLocalResolver_FilterDisconnectedPeerAnswers(t *testing.T) {
 			connByIP:    nil,
 			wantInOrder: []string{"100.64.0.10", "100.64.0.11"},
 		},
+		{
+			// A single answer is never filtered: dropping it would only
+			// trigger the empty-answer escape hatch, so the fast path
+			// returns it untouched.
+			name:    "single disconnected answer passes through",
+			records: []nbdns.SimpleRecord{disconnectedRec},
+			connByIP: map[string]ipState{
+				"100.64.0.11": {known: true, connected: false},
+			},
+			wantInOrder: []string{"100.64.0.11"},
+		},
 	}
 
 	for _, tc := range tests {

client/internal/netflow/logger/logger.go

@@ -27,7 +27,7 @@ type Logger struct {
 	wgIfaceNetV6       netip.Prefix
 	dnsCollection      atomic.Bool
 	exitNodeCollection atomic.Bool
-	Store              types.AggregatingStore
+	Store              types.Store
 }
 
 func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix) *Logger {
@@ -35,7 +35,7 @@ func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix)
 		statusRecorder: statusRecorder,
 		wgIfaceNet:     wgIfaceIPNet,
 		wgIfaceNetV6:   wgIfaceIPNetV6,
-		Store:          store.NewAggregatingMemoryStore(),
+		Store:          store.NewMemoryStore(),
 	}
 }
 
@@ -125,10 +125,6 @@ func (l *Logger) stop() {
 	l.mux.Unlock()
 }
 
-func (l *Logger) ResetAggregationWindow() types.FlowEventAggregator {
-	return l.Store.ResetAggregationWindow()
-}
-
 func (l *Logger) GetEvents() []*types.Event {
 	return l.Store.GetEvents()
 }

client/internal/netflow/manager.go

@@ -9,14 +9,12 @@ import (
 	"sync"
 	"time"
 
-	"github.com/cenkalti/backoff/v4"
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/netbirdio/netbird/client/internal/netflow/conntrack"
 	"github.com/netbirdio/netbird/client/internal/netflow/logger"
-	"github.com/netbirdio/netbird/client/internal/netflow/store"
 	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/flow/client"
@@ -25,16 +23,14 @@ import (
 
 // Manager handles netflow tracking and logging
 type Manager struct {
-	mux               sync.Mutex
-	shutdownWg        sync.WaitGroup
-	logger            nftypes.FlowLogger
-	flowConfig        *nftypes.FlowConfig
-	conntrack         nftypes.ConnTracker
-	receiverClient    *client.GRPCClient
-	eventsWithoutAcks nftypes.Store
-	publicKey         []byte
-	cancel            context.CancelFunc
-	retryInterval     time.Duration
+	mux            sync.Mutex
+	shutdownWg     sync.WaitGroup
+	logger         nftypes.FlowLogger
+	flowConfig     *nftypes.FlowConfig
+	conntrack      nftypes.ConnTracker
+	receiverClient *client.GRPCClient
+	publicKey      []byte
+	cancel         context.CancelFunc
 }
 
 // NewManager creates a new netflow manager
@@ -52,11 +48,9 @@ func NewManager(iface nftypes.IFaceMapper, publicKey []byte, statusRecorder *pee
 	}
 
 	return &Manager{
-		logger:            flowLogger,
-		conntrack:         ct,
-		publicKey:         publicKey,
-		retryInterval:     time.Second,
-		eventsWithoutAcks: store.NewMemoryStore(),
+		logger:    flowLogger,
+		conntrack: ct,
+		publicKey: publicKey,
 	}
 }
 
@@ -113,7 +107,7 @@ func (m *Manager) resetClient() error {
 	ctx, cancel := context.WithCancel(context.Background())
 	m.cancel = cancel
 
-	m.shutdownWg.Add(3)
+	m.shutdownWg.Add(2)
 	go func() {
 		defer m.shutdownWg.Done()
 		m.receiveACKs(ctx, flowClient)
@@ -122,10 +116,6 @@ func (m *Manager) resetClient() error {
 		defer m.shutdownWg.Done()
 		m.startSender(ctx)
 	}()
-	go func() {
-		defer m.shutdownWg.Done()
-		m.startRetries(ctx)
-	}()
 
 	return nil
 }
@@ -217,15 +207,13 @@ func (m *Manager) startSender(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			collectedEvents := m.logger.ResetAggregationWindow()
-			events := collectedEvents.GetAggregatedEvents()
+			events := m.logger.GetEvents()
 			for _, event := range events {
 				if err := m.send(event); err != nil {
 					log.Errorf("failed to send flow event to server: %v", err)
-				} else {
-					log.Tracef("sent flow event: %s", event.ID)
+					continue
 				}
-				m.eventsWithoutAcks.StoreEvent(event)
+				log.Tracef("sent flow event: %s", event.ID)
 			}
 		}
 	}
@@ -239,7 +227,7 @@ func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
 			return nil
 		}
 		log.Tracef("received flow event ack: %s", id)
-		m.eventsWithoutAcks.DeleteEvents([]uuid.UUID{id})
+		m.logger.DeleteEvents([]uuid.UUID{id})
 		return nil
 	})
 
@@ -248,41 +236,6 @@ func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
 	}
 }
 
-func (m *Manager) startRetries(ctx context.Context) {
-	ticker := time.NewTimer(m.retryInterval)
-	retryBackoff := backoff.WithContext(&backoff.ExponentialBackOff{
-		InitialInterval:     1 * time.Second,
-		RandomizationFactor: 0.5,
-		Multiplier:          1.7,
-		MaxInterval:         m.flowConfig.Interval / 2,
-		MaxElapsedTime:      3 * 30 * 24 * time.Hour, // 3 months
-		Stop:                backoff.Stop,
-		Clock:               backoff.SystemClock,
-	}, ctx)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-ticker.C:
-			for _, e := range m.eventsWithoutAcks.GetEvents() {
-				if e.Timestamp.Add(time.Second).After(time.Now()) {
-					// grace period on retries to avoid early retries
-					// do not retry if the event is less than 1 sec old
-					continue
-				}
-				if err := m.send(e); err != nil {
-					ticker = time.NewTimer(retryBackoff.NextBackOff()) //nolint:staticcheck,wastedassign
-					break
-				}
-			}
-			retryBackoff.Reset()
-			ticker = time.NewTimer(time.Second)
-		}
-	}
-}
-
 func (m *Manager) send(event *nftypes.Event) error {
 	m.mux.Lock()
 	client := m.receiverClient

client/internal/netflow/manager_integration_test.go

@@ -1,291 +0,0 @@
-package netflow
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"net"
-	"net/netip"
-	"slices"
-	"testing"
-	"time"
-
-	"github.com/google/uuid"
-	"github.com/netbirdio/netbird/client/iface/wgaddr"
-	"github.com/netbirdio/netbird/client/internal/netflow/types"
-	"github.com/netbirdio/netbird/flow/proto"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"google.golang.org/grpc"
-)
-
-type testServer struct {
-	proto.UnimplementedFlowServiceServer
-	events         chan *proto.FlowEvent
-	acks           chan *proto.FlowEventAck
-	grpcSrv        *grpc.Server
-	addr           string
-	handlerDone    chan struct{} // signaled each time Events() exits
-	handlerStarted chan struct{} // signaled each time Events() begins
-}
-
-func newTestServer(t *testing.T) *testServer {
-	listener, err := net.Listen("tcp", "127.0.0.1:0")
-	require.NoError(t, err)
-
-	s := &testServer{
-		events:         make(chan *proto.FlowEvent, 100),
-		acks:           make(chan *proto.FlowEventAck, 100),
-		grpcSrv:        grpc.NewServer(),
-		addr:           listener.Addr().String(),
-		handlerDone:    make(chan struct{}, 10),
-		handlerStarted: make(chan struct{}, 10),
-	}
-
-	proto.RegisterFlowServiceServer(s.grpcSrv, s)
-
-	go func() {
-		if err := s.grpcSrv.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
-			t.Logf("server error: %v", err)
-		}
-	}()
-
-	t.Cleanup(func() {
-		s.grpcSrv.Stop()
-	})
-
-	return s
-}
-
-func (s *testServer) Events(stream proto.FlowService_EventsServer) error {
-	defer func() {
-		select {
-		case s.handlerDone <- struct{}{}:
-		default:
-		}
-	}()
-
-	err := stream.Send(&proto.FlowEventAck{IsInitiator: true})
-	if err != nil {
-		return err
-	}
-
-	select {
-	case s.handlerStarted <- struct{}{}:
-	default:
-	}
-
-	ctx, cancel := context.WithCancel(stream.Context())
-	defer cancel()
-
-	go func() {
-		defer cancel()
-		for {
-			event, err := stream.Recv()
-			if err != nil {
-				return
-			}
-
-			if !event.IsInitiator {
-				select {
-				case s.events <- event:
-				case <-ctx.Done():
-					return
-				}
-			}
-		}
-	}()
-
-	for {
-		select {
-		case ack := <-s.acks:
-			if err := stream.Send(ack); err != nil {
-				return err
-			}
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
-}
-
-func TestSendEventReceiveAck(t *testing.T) {
-	_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	t.Cleanup(cancel)
-
-	server := newTestServer(t)
-	manager := createManager(t, server.addr, 60*time.Second) // set high to prevent retries in this test
-	defer manager.Close()
-
-	assert.Eventually(t, func() bool {
-		select {
-		case <-server.handlerStarted:
-			return true
-		default:
-			return false
-		}
-	}, 3*time.Second, 100*time.Millisecond)
-
-	event1 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.2"),
-		DestPort:  2345,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event1)
-	event2 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.1"),
-		DestPort:  1234,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event2)
-
-	// verify the server received logged events
-	serverSideEvents := make([]*proto.FlowEvent, 0)
-	assert.Eventually(t, func() bool {
-		select {
-		case event := <-server.events:
-			serverSideEvents = append(serverSideEvents, event)
-			if len(serverSideEvents) == 2 {
-				return true
-			}
-		default:
-			if len(serverSideEvents) == 2 {
-				return true
-			}
-		}
-		return false
-	}, 5*time.Second, 100*time.Millisecond)
-
-	serverSideFlowIds := make([]uuid.UUID, 0, 2)
-	slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
-		id, err := uuid.FromBytes(e.FlowFields.FlowId)
-		assert.NoError(t, err)
-		serverSideFlowIds = append(serverSideFlowIds, id)
-		return true
-	})
-	assert.ElementsMatch(t, []uuid.UUID{event1.FlowID, event2.FlowID}, serverSideFlowIds)
-
-	// verify the manager tracks un-acked events
-	unackedEvents := manager.eventsWithoutAcks.GetEvents()
-	assert.Len(t, unackedEvents, 2)
-	flowIds := make([]uuid.UUID, 0)
-	slices.Values(unackedEvents)(func(e *types.Event) bool {
-		flowIds = append(flowIds, e.FlowID)
-		return true
-	})
-	assert.ElementsMatch(t, flowIds, []uuid.UUID{event1.FlowID, event2.FlowID})
-}
-
-// verify handling of retries:
-//   - unacked events are retried
-//   - when acks arrive, events are removed from the un-acked event tracker
-func TestRetryEvents(t *testing.T) {
-	_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	t.Cleanup(cancel)
-
-	server := newTestServer(t)
-	manager := createManager(t, server.addr, time.Second) // set low to start retries sooner
-	defer manager.Close()
-
-	assert.Eventually(t, func() bool {
-		select {
-		case <-server.handlerStarted:
-			return true
-		default:
-			return false
-		}
-	}, 3*time.Second, 100*time.Millisecond)
-
-	event1 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.2"),
-		DestPort:  2345,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event1)
-	event2 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.1"),
-		DestPort:  1234,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event2)
-
-	// verify the server received retries of logged events
-	serverSideEvents := make([]*proto.FlowEvent, 0)
-	func() {
-		c := time.After(2500 * time.Millisecond)
-		for {
-			select {
-			case event := <-server.events:
-				serverSideEvents = append(serverSideEvents, event)
-			case <-c:
-				return
-			}
-		}
-	}()
-	assert.True(t, len(serverSideEvents) > 2) // must see retries
-
-	uniqueServerSideEvents := make(map[uuid.UUID]*proto.FlowEvent)
-	slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
-		id, err := uuid.FromBytes(e.FlowFields.FlowId)
-		assert.NoError(t, err)
-		uniqueServerSideEvents[id] = e
-		return true
-	})
-	assert.Contains(t, uniqueServerSideEvents, event1.FlowID)
-	assert.Contains(t, uniqueServerSideEvents, event2.FlowID)
-
-	// ack events
-	server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event1.FlowID].EventId}
-	server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event2.FlowID].EventId}
-
-	assert.EventuallyWithT(t, func(c *assert.CollectT) {
-		unackedEvents := manager.eventsWithoutAcks.GetEvents()
-		assert.Empty(c, unackedEvents)
-
-	}, 3*time.Second, 100*time.Millisecond)
-}
-
-func createManager(t *testing.T, serverAddr string, retryInterval time.Duration) *Manager {
-	t.Helper()
-
-	mockIFace := &mockIFaceMapper{
-		address: wgaddr.Address{
-			Network: netip.MustParsePrefix("192.168.1.1/32"),
-		},
-		isUserspaceBind: true,
-	}
-
-	publicKey := []byte("test-public-key")
-	manager := NewManager(mockIFace, publicKey, nil)
-	manager.retryInterval = retryInterval
-
-	initialConfig := &types.FlowConfig{
-		Enabled:        true,
-		URL:            fmt.Sprintf("http://%s", serverAddr),
-		TokenPayload:   "initial-payload",
-		TokenSignature: "initial-signature",
-		Interval:       500 * time.Millisecond,
-	}
-
-	err := manager.Update(initialConfig)
-	require.NoError(t, err)
-
-	return manager
-}
-
-func ipAddr(a string) netip.Addr {
-	addr, _ := netip.ParseAddr(a)
-	return addr
-}

client/internal/netflow/store/memory.go

@@ -1,13 +1,10 @@
 package store
 
 import (
-	"maps"
-	"net/netip"
-	"slices"
 	"sync"
-	"time"
 
 	"github.com/google/uuid"
+
 	"github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
@@ -22,10 +19,6 @@ type Memory struct {
 	events map[uuid.UUID]*types.Event
 }
 
-type AggregatingMemory struct {
-	Memory
-}
-
 func (m *Memory) StoreEvent(event *types.Event) {
 	m.mux.Lock()
 	defer m.mux.Unlock()
@@ -55,78 +48,3 @@ func (m *Memory) DeleteEvents(ids []uuid.UUID) {
 		delete(m.events, id)
 	}
 }
-
-func NewAggregatingMemoryStore() *AggregatingMemory {
-	return &AggregatingMemory{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}}
-	am.events = make(map[uuid.UUID]*types.Event)
-
-	return &toret
-}
-
-type aggregationKey struct {
-	destAddr netip.Addr
-	destPort uint16
-	protocol uint8
-	icmpType uint8
-	unique   int64 // used to prevent aggregation on non icmp/udp/tcp events
-}
-
-func (am *AggregatingMemory) GetAggregatedEvents() []*types.Event {
-	aggregated := make(map[aggregationKey]*types.Event)
-	for _, v := range am.events {
-		lookupKey := aggregationKey{destAddr: v.DestIP, destPort: v.DestPort, protocol: uint8(v.Protocol), icmpType: v.ICMPCode}
-		if _, ok := aggregated[lookupKey]; !ok {
-			aggregated[lookupKey] = v.Clone()
-			event := aggregated[lookupKey]
-
-			if event.Protocol != types.ICMP && event.Protocol != types.ICMPv6 && event.Protocol != types.UDP && event.Protocol != types.TCP {
-				lookupKey.unique = time.Now().UnixNano() // to make the lookup key unique so we don't aggregate on it
-				continue
-			}
-
-			switch event.Type {
-			case types.TypeStart:
-				event.NumOfStarts += 1
-			case types.TypeDrop:
-				event.NumOfDrops += 1
-			case types.TypeEnd:
-				event.NumOfEnds += 1
-			}
-			continue
-		}
-
-		aggregatedEvent := aggregated[lookupKey]
-		if aggregatedEvent.Protocol != types.ICMP && aggregatedEvent.Protocol != types.ICMPv6 && aggregatedEvent.Protocol != types.UDP && aggregatedEvent.Protocol != types.TCP {
-			continue // we don't aggregate this type of events; shouldn't ever get here
-		}
-
-		// track the number of connections, duration?, open and close events?
-		aggregatedEvent.RxBytes += v.RxBytes
-		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
-			aggregatedEvent.Type = v.Type
-		}
-		// do we aggregate icmp by code?
-	}
-
-	return slices.Collect(maps.Values(aggregated)) // could return an iterator instead here
-}

client/internal/netflow/store/tcp_aggregation_test.go

@@ -1,281 +0,0 @@
-package store
-
-import (
-	"net/netip"
-	"testing"
-	"time"
-
-	"github.com/google/uuid"
-	"github.com/netbirdio/netbird/client/internal/netflow/types"
-	"github.com/stretchr/testify/assert"
-)
-
-var pregeneratedUUIDs = func() []uuid.UUID {
-	toret := make([]uuid.UUID, 0)
-	for range make([]int, 10) {
-		toret = append(toret, uuid.New())
-	}
-	return toret
-}()
-
-func TestFlowAggregation(t *testing.T) {
-	var protocols = []types.Protocol{types.ICMP, types.ICMPv6, types.TCP, types.UDP}
-	var tests = []struct {
-		description string
-		events      []*types.Event
-		expected    []*types.Event
-	}{
-		{
-			description: "start and stop",
-			events: []*types.Event{
-				{
-					ID:        pregeneratedUUIDs[0],
-					Timestamp: time.Unix(100, 100),
-					EventFields: types.EventFields{
-						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeStart,
-						RuleID:           []byte("rule-id-1"),
-						Direction:        types.Egress,
-						SourceIP:         ipAddr("1.1.1.1"),
-						SourcePort:       1234,
-						DestIP:           ipAddr("2.2.2.2"),
-						DestPort:         443,
-						SourceResourceID: []byte("source-resource-id"),
-						DestResourceID:   []byte("dest-resource-id"),
-						RxPackets:        10,
-						TxPackets:        20,
-						RxBytes:          10000,
-						TxBytes:          20000,
-					}},
-				{
-					ID:        pregeneratedUUIDs[2],
-					Timestamp: time.Unix(100, 100).Add(time.Second),
-					EventFields: types.EventFields{
-						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeEnd,
-						RuleID:           []byte("rule-id-1"),
-						Direction:        types.Egress,
-						SourceIP:         ipAddr("1.1.1.1"),
-						SourcePort:       1234,
-						DestIP:           ipAddr("2.2.2.2"),
-						DestPort:         443,
-						SourceResourceID: []byte("source-resource-id"),
-						DestResourceID:   []byte("dest-resource-id"),
-						RxPackets:        30,
-						TxPackets:        40,
-						RxBytes:          30000,
-						TxBytes:          40000,
-					}},
-			},
-			expected: []*types.Event{
-				{
-					ID:        pregeneratedUUIDs[0],
-					Timestamp: time.Unix(100, 100),
-					EventFields: types.EventFields{
-						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeStart,
-						RuleID:           []byte("rule-id-1"),
-						Direction:        types.Egress,
-						SourceIP:         ipAddr("1.1.1.1"),
-						SourcePort:       1234,
-						DestIP:           ipAddr("2.2.2.2"),
-						DestPort:         443,
-						SourceResourceID: []byte("source-resource-id"),
-						DestResourceID:   []byte("dest-resource-id"),
-						RxPackets:        40,
-						TxPackets:        60,
-						RxBytes:          40000,
-						TxBytes:          60000,
-						NumOfStarts:      1,
-						NumOfEnds:        1,
-						NumOfDrops:       0,
-					}},
-			},
-		},
-		{
-			description: "start and drop",
-			events: []*types.Event{
-				{
-					ID:        pregeneratedUUIDs[0],
-					Timestamp: time.Unix(100, 100),
-					EventFields: types.EventFields{
-						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeStart,
-						RuleID:           []byte("rule-id-1"),
-						Direction:        types.Egress,
-						SourceIP:         ipAddr("1.1.1.1"),
-						SourcePort:       1234,
-						DestIP:           ipAddr("2.2.2.2"),
-						DestPort:         443,
-						SourceResourceID: []byte("source-resource-id"),
-						DestResourceID:   []byte("dest-resource-id"),
-						RxPackets:        10,
-						TxPackets:        20,
-						RxBytes:          10000,
-						TxBytes:          20000,
-					}},
-				{
-					ID:        pregeneratedUUIDs[2],
-					Timestamp: time.Unix(100, 100).Add(time.Second),
-					EventFields: types.EventFields{
-						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeDrop,
-						RuleID:           []byte("rule-id-1"),
-						Direction:        types.Egress,
-						SourceIP:         ipAddr("1.1.1.1"),
-						SourcePort:       1234,
-						DestIP:           ipAddr("2.2.2.2"),
-						DestPort:         443,
-						SourceResourceID: []byte("source-resource-id"),
-						DestResourceID:   []byte("dest-resource-id"),
-						RxPackets:        30,
-						TxPackets:        40,
-						RxBytes:          30000,
-						TxBytes:          40000,
-					}},
-			},
-			expected: []*types.Event{
-				{
-					ID:        pregeneratedUUIDs[0],
-					Timestamp: time.Unix(100, 100),
-					EventFields: types.EventFields{
-						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeStart,
-						RuleID:           []byte("rule-id-1"),
-						Direction:        types.Egress,
-						SourceIP:         ipAddr("1.1.1.1"),
-						SourcePort:       1234,
-						DestIP:           ipAddr("2.2.2.2"),
-						DestPort:         443,
-						SourceResourceID: []byte("source-resource-id"),
-						DestResourceID:   []byte("dest-resource-id"),
-						RxPackets:        40,
-						TxPackets:        60,
-						RxBytes:          40000,
-						TxBytes:          60000,
-						NumOfStarts:      1,
-						NumOfEnds:        0,
-						NumOfDrops:       1,
-					}},
-			},
-		},
-		{
-			description: "start only",
-			events: []*types.Event{
-				{
-					ID:        pregeneratedUUIDs[0],
-					Timestamp: time.Unix(100, 100),
-					EventFields: types.EventFields{
-						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeStart,
-						RuleID:           []byte("rule-id-1"),
-						Direction:        types.Egress,
-						SourceIP:         ipAddr("1.1.1.1"),
-						SourcePort:       1234,
-						DestIP:           ipAddr("2.2.2.2"),
-						DestPort:         443,
-						SourceResourceID: []byte("source-resource-id"),
-						DestResourceID:   []byte("dest-resource-id"),
-						RxPackets:        10,
-						TxPackets:        20,
-						RxBytes:          10000,
-						TxBytes:          20000,
-					}},
-			},
-			expected: []*types.Event{
-				{
-					ID:        pregeneratedUUIDs[0],
-					Timestamp: time.Unix(100, 100),
-					EventFields: types.EventFields{
-						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeStart,
-						RuleID:           []byte("rule-id-1"),
-						Direction:        types.Egress,
-						SourceIP:         ipAddr("1.1.1.1"),
-						SourcePort:       1234,
-						DestIP:           ipAddr("2.2.2.2"),
-						DestPort:         443,
-						SourceResourceID: []byte("source-resource-id"),
-						DestResourceID:   []byte("dest-resource-id"),
-						RxPackets:        10,
-						TxPackets:        20,
-						RxBytes:          10000,
-						TxBytes:          20000,
-						NumOfStarts:      1,
-						NumOfEnds:        0,
-						NumOfDrops:       0,
-					}},
-			},
-		},
-		{
-			description: "drop only",
-			events: []*types.Event{
-				{
-					ID:        pregeneratedUUIDs[2],
-					Timestamp: time.Unix(100, 100).Add(time.Second),
-					EventFields: types.EventFields{
-						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeDrop,
-						RuleID:           []byte("rule-id-1"),
-						Direction:        types.Egress,
-						SourceIP:         ipAddr("1.1.1.1"),
-						SourcePort:       1234,
-						DestIP:           ipAddr("2.2.2.2"),
-						DestPort:         443,
-						SourceResourceID: []byte("source-resource-id"),
-						DestResourceID:   []byte("dest-resource-id"),
-						RxPackets:        30,
-						TxPackets:        40,
-						RxBytes:          30000,
-						TxBytes:          40000,
-					}},
-			},
-			expected: []*types.Event{
-				{
-					ID:        pregeneratedUUIDs[2],
-					Timestamp: time.Unix(100, 100).Add(time.Second),
-					EventFields: types.EventFields{
-						FlowID:           pregeneratedUUIDs[1],
-						Type:             types.TypeDrop,
-						RuleID:           []byte("rule-id-1"),
-						Direction:        types.Egress,
-						SourceIP:         ipAddr("1.1.1.1"),
-						SourcePort:       1234,
-						DestIP:           ipAddr("2.2.2.2"),
-						DestPort:         443,
-						SourceResourceID: []byte("source-resource-id"),
-						DestResourceID:   []byte("dest-resource-id"),
-						RxPackets:        30,
-						TxPackets:        40,
-						RxBytes:          30000,
-						TxBytes:          40000,
-						NumOfStarts:      0,
-						NumOfEnds:        0,
-						NumOfDrops:       1,
-					}},
-			},
-		}}
-
-	for _, protocol := range protocols {
-		for _, tt := range tests {
-			t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
-				store := NewAggregatingMemoryStore()
-				for _, e := range tt.events {
-					e.Protocol = protocol
-					store.StoreEvent(e)
-				}
-				for _, e := range tt.expected {
-					e.Protocol = protocol
-				}
-				events := store.GetAggregatedEvents()
-				assert.Len(t, events, len(tt.expected))
-				assert.ElementsMatch(t, events, tt.expected)
-			})
-		}
-	}
-}
-
-func ipAddr(a string) netip.Addr {
-	addr, _ := netip.ParseAddr(a)
-	return addr
-}

client/internal/netflow/types/types.go

@@ -2,7 +2,6 @@ package types
 
 import (
 	"net/netip"
-	"slices"
 	"strconv"
 	"time"
 
@@ -93,17 +92,6 @@ 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 {
@@ -126,15 +114,13 @@ type FlowManager interface {
 	GetLogger() FlowLogger
 }
 
-type FlowEventAggregator interface {
-	ResetAggregationWindow() FlowEventAggregator
-	GetAggregatedEvents() []*Event
-}
-
 type FlowLogger interface {
-	ResetAggregationWindow() FlowEventAggregator
 	// StoreEvent stores a flow event
 	StoreEvent(flowEvent EventFields)
+	// GetEvents returns all stored events
+	GetEvents() []*Event
+	// DeleteEvents deletes events from the store
+	DeleteEvents([]uuid.UUID)
 	// Close closes the logger
 	Close()
 	// Enable enables the flow logger receiver
@@ -154,11 +140,6 @@ type Store interface {
 	Close()
 }
 
-type AggregatingStore interface {
-	FlowEventAggregator
-	Store
-}
-
 // ConnTracker defines the interface for connection tracking functionality
 type ConnTracker interface {
 	// Start begins tracking connections by listening for conntrack events.

client/internal/peer/conn_status.go

@@ -26,7 +26,6 @@ type connStatusInputs struct {
 	iceInProgress       bool // a negotiation is currently in flight
 }
 
-
 // ConnStatus describe the status of a peer's connection
 type ConnStatus int32
 

client/internal/peer/status.go

@@ -193,6 +193,7 @@ func (s *StatusChangeSubscription) Events() chan map[string]RouterState {
 type Status struct {
 	mux                   sync.RWMutex
 	peers                 map[string]State
+	ipToKey               map[string]string
 	changeNotify          map[string]map[string]*StatusChangeSubscription // map[peerID]map[subscriptionID]*StatusChangeSubscription
 	signalState           bool
 	signalError           error
@@ -231,6 +232,7 @@ type Status struct {
 func NewRecorder(mgmAddress string) *Status {
 	return &Status{
 		peers:                 make(map[string]State),
+		ipToKey:               make(map[string]string),
 		changeNotify:          make(map[string]map[string]*StatusChangeSubscription),
 		eventStreams:          make(map[string]chan *proto.SystemEvent),
 		eventQueue:            NewEventQueue(eventQueueSize),
@@ -282,6 +284,12 @@ func (d *Status) AddPeer(peerPubKey string, fqdn string, ip string, ipv6 string)
 		Mux:        new(sync.RWMutex),
 	}
 	d.peerListChangedForNotification = true
+	if ipv6 != "" {
+		d.ipToKey[ipv6] = peerPubKey
+	}
+	if ip != "" {
+		d.ipToKey[ip] = peerPubKey
+	}
 	return nil
 }
 
@@ -311,28 +319,22 @@ func (d *Status) PeerByIP(ip string) (string, bool) {
 
 // PeerStateByIP returns the full peer State for the given tunnel IP.
 // Matches against either the IPv4 (State.IP) or IPv6 (State.IPv6) tunnel
-// address so dual-stack peers are reachable on either family. Searches
-// both d.peers and d.offlinePeers — peers that have been moved into
-// the offline slice by ReplaceOfflinePeers are still part of the
-// account's roster and callers (DNS filter, embed.Client.IdentityForIP)
-// need to recognise them rather than treating them as unknown. Returns
-// the zero State and false when no peer matches or the input is empty.
+// address so dual-stack peers are reachable on either family. Only
+// active peers are matched; peers moved into the offline slice by
+// ReplaceOfflinePeers are intentionally treated as unknown.
 func (d *Status) PeerStateByIP(ip string) (State, bool) {
 	if ip == "" {
 		return State{}, false
 	}
 	d.mux.RLock()
 	defer d.mux.RUnlock()
-
-	for _, state := range d.peers {
-		if (state.IP != "" && state.IP == ip) || (state.IPv6 != "" && state.IPv6 == ip) {
-			return state, true
-		}
+	key, ok := d.ipToKey[ip]
+	if !ok {
+		return State{}, false
 	}
-	for _, state := range d.offlinePeers {
-		if (state.IP != "" && state.IP == ip) || (state.IPv6 != "" && state.IPv6 == ip) {
-			return state, true
-		}
+	state, ok := d.peers[key]
+	if ok {
+		return state, true
 	}
 	return State{}, false
 }
@@ -342,12 +344,18 @@ func (d *Status) RemovePeer(peerPubKey string) error {
 	d.mux.Lock()
 	defer d.mux.Unlock()
 
-	_, ok := d.peers[peerPubKey]
+	p, ok := d.peers[peerPubKey]
 	if !ok {
 		return errors.New("no peer with to remove")
 	}
 
 	delete(d.peers, peerPubKey)
+	if mappedKey, exists := d.ipToKey[p.IP]; exists && mappedKey == peerPubKey {
+		delete(d.ipToKey, p.IP)
+	}
+	if mappedKey, exists := d.ipToKey[p.IPv6]; exists && mappedKey == peerPubKey {
+		delete(d.ipToKey, p.IPv6)
+	}
 	d.peerListChangedForNotification = true
 	return nil
 }

client/internal/peer/status_test.go

@@ -90,12 +90,11 @@ func TestStatus_PeerStateByIP_MatchesIPv6(t *testing.T) {
 	req.Equal("pk-1", state.PubKey, "matching state must carry the right pub key")
 }
 
-// TestStatus_PeerStateByIP_MatchesOfflinePeers covers peers that have
-// been moved into the offline slice via ReplaceOfflinePeers. Callers
-// (DNS filter, embed.Client.IdentityForIP) need to treat them as known
-// rather than unknown — otherwise authentication / DNS filtering treats
-// known-but-offline peers as foreign IPs.
-func TestStatus_PeerStateByIP_MatchesOfflinePeers(t *testing.T) {
+// TestStatus_PeerStateByIP_IgnoresOfflinePeers documents that peers
+// moved into the offline slice via ReplaceOfflinePeers are intentionally
+// not resolvable by IP: only active peers can carry traffic, so callers
+// (DNS filter, embed.Client.IdentityForIP) treat them as unknown.
+func TestStatus_PeerStateByIP_IgnoresOfflinePeers(t *testing.T) {
 	status := NewRecorder("https://mgm")
 	req := require.New(t)
 
@@ -103,13 +102,31 @@ func TestStatus_PeerStateByIP_MatchesOfflinePeers(t *testing.T) {
 		{PubKey: "pk-offline", FQDN: "offline.netbird", IP: "100.64.0.20", IPv6: "fd00::20"},
 	})
 
-	state, ok := status.PeerStateByIP("100.64.0.20")
-	req.True(ok, "offline peer must resolve by IPv4 tunnel address")
-	req.Equal("pk-offline", state.PubKey, "matching state must carry the offline peer's pub key")
+	_, ok := status.PeerStateByIP("100.64.0.20")
+	req.False(ok, "offline peer must not resolve by IPv4 tunnel address")
 
-	state, ok = status.PeerStateByIP("fd00::20")
-	req.True(ok, "offline peer must resolve by IPv6 tunnel address")
-	req.Equal("pk-offline", state.PubKey, "IPv6 match must carry the offline peer's pub key")
+	_, ok = status.PeerStateByIP("fd00::20")
+	req.False(ok, "offline peer must not resolve by IPv6 tunnel address")
+}
+
+// TestStatus_PeerStateByIP_RemovedPeer verifies RemovePeer drops the
+// IP index entries for both address families.
+func TestStatus_PeerStateByIP_RemovedPeer(t *testing.T) {
+	status := NewRecorder("https://mgm")
+	req := require.New(t)
+
+	req.NoError(status.AddPeer("pk-1", "peer-1.netbird", "100.64.0.10", "fd00::1"))
+
+	_, ok := status.PeerStateByIP("100.64.0.10")
+	req.True(ok, "active peer must resolve before removal")
+
+	req.NoError(status.RemovePeer("pk-1"))
+
+	_, ok = status.PeerStateByIP("100.64.0.10")
+	req.False(ok, "removed peer must not resolve by IPv4 tunnel address")
+
+	_, ok = status.PeerStateByIP("fd00::1")
+	req.False(ok, "removed peer must not resolve by IPv6 tunnel address")
 }
 
 func TestStatus_UpdatePeerFQDN(t *testing.T) {

flow/proto/flow.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.36.11
-// 	protoc        v7.34.1
+// 	protoc-gen-go v1.26.0
+// 	protoc        v3.21.9
 // source: flow.proto
 
 package proto
@@ -12,7 +12,6 @@ import (
 	timestamppb "google.golang.org/protobuf/types/known/timestamppb"
 	reflect "reflect"
 	sync "sync"
-	unsafe "unsafe"
 )
 
 const (
@@ -126,24 +125,27 @@ func (Direction) EnumDescriptor() ([]byte, []int) {
 }
 
 type FlowEvent struct {
-	state protoimpl.MessageState `protogen:"open.v1"`
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
 	// Unique client event identifier
 	EventId []byte `protobuf:"bytes,1,opt,name=event_id,json=eventId,proto3" json:"event_id,omitempty"`
 	// When the event occurred
 	Timestamp *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
 	// Public key of the sending peer
-	PublicKey     []byte      `protobuf:"bytes,3,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"`
-	FlowFields    *FlowFields `protobuf:"bytes,4,opt,name=flow_fields,json=flowFields,proto3" json:"flow_fields,omitempty"`
-	IsInitiator   bool        `protobuf:"varint,5,opt,name=isInitiator,proto3" json:"isInitiator,omitempty"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
+	PublicKey   []byte      `protobuf:"bytes,3,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"`
+	FlowFields  *FlowFields `protobuf:"bytes,4,opt,name=flow_fields,json=flowFields,proto3" json:"flow_fields,omitempty"`
+	IsInitiator bool        `protobuf:"varint,5,opt,name=isInitiator,proto3" json:"isInitiator,omitempty"`
 }
 
 func (x *FlowEvent) Reset() {
 	*x = FlowEvent{}
-	mi := &file_flow_proto_msgTypes[0]
-	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-	ms.StoreMessageInfo(mi)
+	if protoimpl.UnsafeEnabled {
+		mi := &file_flow_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
 }
 
 func (x *FlowEvent) String() string {
@@ -154,7 +156,7 @@ func (*FlowEvent) ProtoMessage() {}
 
 func (x *FlowEvent) ProtoReflect() protoreflect.Message {
 	mi := &file_flow_proto_msgTypes[0]
-	if x != nil {
+	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
 			ms.StoreMessageInfo(mi)
@@ -205,19 +207,22 @@ func (x *FlowEvent) GetIsInitiator() bool {
 }
 
 type FlowEventAck struct {
-	state protoimpl.MessageState `protogen:"open.v1"`
-	// Unique client event identifier that has been ack'ed
-	EventId       []byte `protobuf:"bytes,1,opt,name=event_id,json=eventId,proto3" json:"event_id,omitempty"`
-	IsInitiator   bool   `protobuf:"varint,2,opt,name=isInitiator,proto3" json:"isInitiator,omitempty"`
-	unknownFields protoimpl.UnknownFields
+	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// Unique client event identifier that has been ack'ed
+	EventId     []byte `protobuf:"bytes,1,opt,name=event_id,json=eventId,proto3" json:"event_id,omitempty"`
+	IsInitiator bool   `protobuf:"varint,2,opt,name=isInitiator,proto3" json:"isInitiator,omitempty"`
 }
 
 func (x *FlowEventAck) Reset() {
 	*x = FlowEventAck{}
-	mi := &file_flow_proto_msgTypes[1]
-	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-	ms.StoreMessageInfo(mi)
+	if protoimpl.UnsafeEnabled {
+		mi := &file_flow_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
 }
 
 func (x *FlowEventAck) String() string {
@@ -228,7 +233,7 @@ func (*FlowEventAck) ProtoMessage() {}
 
 func (x *FlowEventAck) ProtoReflect() protoreflect.Message {
 	mi := &file_flow_proto_msgTypes[1]
-	if x != nil {
+	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
 			ms.StoreMessageInfo(mi)
@@ -258,7 +263,10 @@ func (x *FlowEventAck) GetIsInitiator() bool {
 }
 
 type FlowFields struct {
-	state protoimpl.MessageState `protogen:"open.v1"`
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
 	// Unique client flow session identifier
 	FlowId []byte `protobuf:"bytes,1,opt,name=flow_id,json=flowId,proto3" json:"flow_id,omitempty"`
 	// Flow type
@@ -275,7 +283,7 @@ type FlowFields struct {
 	DestIp []byte `protobuf:"bytes,7,opt,name=dest_ip,json=destIp,proto3" json:"dest_ip,omitempty"`
 	// Layer 4 -specific information
 	//
-	// Types that are valid to be assigned to ConnectionInfo:
+	// Types that are assignable to ConnectionInfo:
 	//
 	//	*FlowFields_PortInfo
 	//	*FlowFields_IcmpInfo
@@ -289,18 +297,15 @@ type FlowFields struct {
 	// Resource ID
 	SourceResourceId []byte `protobuf:"bytes,14,opt,name=source_resource_id,json=sourceResourceId,proto3" json:"source_resource_id,omitempty"`
 	DestResourceId   []byte `protobuf:"bytes,15,opt,name=dest_resource_id,json=destResourceId,proto3" json:"dest_resource_id,omitempty"`
-	NumOfStarts      uint64 `protobuf:"varint,16,opt,name=num_of_starts,json=numOfStarts,proto3" json:"num_of_starts,omitempty"`
-	NumOfEnds        uint64 `protobuf:"varint,17,opt,name=num_of_ends,json=numOfEnds,proto3" json:"num_of_ends,omitempty"`
-	NumOfDrops       uint64 `protobuf:"varint,18,opt,name=num_of_drops,json=numOfDrops,proto3" json:"num_of_drops,omitempty"`
-	unknownFields    protoimpl.UnknownFields
-	sizeCache        protoimpl.SizeCache
 }
 
 func (x *FlowFields) Reset() {
 	*x = FlowFields{}
-	mi := &file_flow_proto_msgTypes[2]
-	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-	ms.StoreMessageInfo(mi)
+	if protoimpl.UnsafeEnabled {
+		mi := &file_flow_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
 }
 
 func (x *FlowFields) String() string {
@@ -311,7 +316,7 @@ func (*FlowFields) ProtoMessage() {}
 
 func (x *FlowFields) ProtoReflect() protoreflect.Message {
 	mi := &file_flow_proto_msgTypes[2]
-	if x != nil {
+	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
 			ms.StoreMessageInfo(mi)
@@ -375,27 +380,23 @@ func (x *FlowFields) GetDestIp() []byte {
 	return nil
 }
 
-func (x *FlowFields) GetConnectionInfo() isFlowFields_ConnectionInfo {
-	if x != nil {
-		return x.ConnectionInfo
+func (m *FlowFields) GetConnectionInfo() isFlowFields_ConnectionInfo {
+	if m != nil {
+		return m.ConnectionInfo
 	}
 	return nil
 }
 
 func (x *FlowFields) GetPortInfo() *PortInfo {
-	if x != nil {
-		if x, ok := x.ConnectionInfo.(*FlowFields_PortInfo); ok {
-			return x.PortInfo
-		}
+	if x, ok := x.GetConnectionInfo().(*FlowFields_PortInfo); ok {
+		return x.PortInfo
 	}
 	return nil
 }
 
 func (x *FlowFields) GetIcmpInfo() *ICMPInfo {
-	if x != nil {
-		if x, ok := x.ConnectionInfo.(*FlowFields_IcmpInfo); ok {
-			return x.IcmpInfo
-		}
+	if x, ok := x.GetConnectionInfo().(*FlowFields_IcmpInfo); ok {
+		return x.IcmpInfo
 	}
 	return nil
 }
@@ -442,27 +443,6 @@ func (x *FlowFields) GetDestResourceId() []byte {
 	return nil
 }
 
-func (x *FlowFields) GetNumOfStarts() uint64 {
-	if x != nil {
-		return x.NumOfStarts
-	}
-	return 0
-}
-
-func (x *FlowFields) GetNumOfEnds() uint64 {
-	if x != nil {
-		return x.NumOfEnds
-	}
-	return 0
-}
-
-func (x *FlowFields) GetNumOfDrops() uint64 {
-	if x != nil {
-		return x.NumOfDrops
-	}
-	return 0
-}
-
 type isFlowFields_ConnectionInfo interface {
 	isFlowFields_ConnectionInfo()
 }
@@ -483,18 +463,21 @@ func (*FlowFields_IcmpInfo) isFlowFields_ConnectionInfo() {}
 
 // TCP/UDP port information
 type PortInfo struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	SourcePort    uint32                 `protobuf:"varint,1,opt,name=source_port,json=sourcePort,proto3" json:"source_port,omitempty"`
-	DestPort      uint32                 `protobuf:"varint,2,opt,name=dest_port,json=destPort,proto3" json:"dest_port,omitempty"`
-	unknownFields protoimpl.UnknownFields
+	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	SourcePort uint32 `protobuf:"varint,1,opt,name=source_port,json=sourcePort,proto3" json:"source_port,omitempty"`
+	DestPort   uint32 `protobuf:"varint,2,opt,name=dest_port,json=destPort,proto3" json:"dest_port,omitempty"`
 }
 
 func (x *PortInfo) Reset() {
 	*x = PortInfo{}
-	mi := &file_flow_proto_msgTypes[3]
-	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-	ms.StoreMessageInfo(mi)
+	if protoimpl.UnsafeEnabled {
+		mi := &file_flow_proto_msgTypes[3]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
 }
 
 func (x *PortInfo) String() string {
@@ -505,7 +488,7 @@ func (*PortInfo) ProtoMessage() {}
 
 func (x *PortInfo) ProtoReflect() protoreflect.Message {
 	mi := &file_flow_proto_msgTypes[3]
-	if x != nil {
+	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
 			ms.StoreMessageInfo(mi)
@@ -536,18 +519,21 @@ func (x *PortInfo) GetDestPort() uint32 {
 
 // ICMP message information
 type ICMPInfo struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	IcmpType      uint32                 `protobuf:"varint,1,opt,name=icmp_type,json=icmpType,proto3" json:"icmp_type,omitempty"`
-	IcmpCode      uint32                 `protobuf:"varint,2,opt,name=icmp_code,json=icmpCode,proto3" json:"icmp_code,omitempty"`
-	unknownFields protoimpl.UnknownFields
+	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	IcmpType uint32 `protobuf:"varint,1,opt,name=icmp_type,json=icmpType,proto3" json:"icmp_type,omitempty"`
+	IcmpCode uint32 `protobuf:"varint,2,opt,name=icmp_code,json=icmpCode,proto3" json:"icmp_code,omitempty"`
 }
 
 func (x *ICMPInfo) Reset() {
 	*x = ICMPInfo{}
-	mi := &file_flow_proto_msgTypes[4]
-	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-	ms.StoreMessageInfo(mi)
+	if protoimpl.UnsafeEnabled {
+		mi := &file_flow_proto_msgTypes[4]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
 }
 
 func (x *ICMPInfo) String() string {
@@ -558,7 +544,7 @@ func (*ICMPInfo) ProtoMessage() {}
 
 func (x *ICMPInfo) ProtoReflect() protoreflect.Message {
 	mi := &file_flow_proto_msgTypes[4]
-	if x != nil {
+	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
 			ms.StoreMessageInfo(mi)
@@ -589,83 +575,102 @@ func (x *ICMPInfo) GetIcmpCode() uint32 {
 
 var File_flow_proto protoreflect.FileDescriptor
 
-const file_flow_proto_rawDesc = "" +
-	"\n" +
-	"\n" +
-	"flow.proto\x12\x04flow\x1a\x1fgoogle/protobuf/timestamp.proto\"\xd4\x01\n" +
-	"\tFlowEvent\x12\x19\n" +
-	"\bevent_id\x18\x01 \x01(\fR\aeventId\x128\n" +
-	"\ttimestamp\x18\x02 \x01(\v2\x1a.google.protobuf.TimestampR\ttimestamp\x12\x1d\n" +
-	"\n" +
-	"public_key\x18\x03 \x01(\fR\tpublicKey\x121\n" +
-	"\vflow_fields\x18\x04 \x01(\v2\x10.flow.FlowFieldsR\n" +
-	"flowFields\x12 \n" +
-	"\visInitiator\x18\x05 \x01(\bR\visInitiator\"K\n" +
-	"\fFlowEventAck\x12\x19\n" +
-	"\bevent_id\x18\x01 \x01(\fR\aeventId\x12 \n" +
-	"\visInitiator\x18\x02 \x01(\bR\visInitiator\"\x82\x05\n" +
-	"\n" +
-	"FlowFields\x12\x17\n" +
-	"\aflow_id\x18\x01 \x01(\fR\x06flowId\x12\x1e\n" +
-	"\x04type\x18\x02 \x01(\x0e2\n" +
-	".flow.TypeR\x04type\x12\x17\n" +
-	"\arule_id\x18\x03 \x01(\fR\x06ruleId\x12-\n" +
-	"\tdirection\x18\x04 \x01(\x0e2\x0f.flow.DirectionR\tdirection\x12\x1a\n" +
-	"\bprotocol\x18\x05 \x01(\rR\bprotocol\x12\x1b\n" +
-	"\tsource_ip\x18\x06 \x01(\fR\bsourceIp\x12\x17\n" +
-	"\adest_ip\x18\a \x01(\fR\x06destIp\x12-\n" +
-	"\tport_info\x18\b \x01(\v2\x0e.flow.PortInfoH\x00R\bportInfo\x12-\n" +
-	"\ticmp_info\x18\t \x01(\v2\x0e.flow.ICMPInfoH\x00R\bicmpInfo\x12\x1d\n" +
-	"\n" +
-	"rx_packets\x18\n" +
-	" \x01(\x04R\trxPackets\x12\x1d\n" +
-	"\n" +
-	"tx_packets\x18\v \x01(\x04R\ttxPackets\x12\x19\n" +
-	"\brx_bytes\x18\f \x01(\x04R\arxBytes\x12\x19\n" +
-	"\btx_bytes\x18\r \x01(\x04R\atxBytes\x12,\n" +
-	"\x12source_resource_id\x18\x0e \x01(\fR\x10sourceResourceId\x12(\n" +
-	"\x10dest_resource_id\x18\x0f \x01(\fR\x0edestResourceId\x12\"\n" +
-	"\rnum_of_starts\x18\x10 \x01(\x04R\vnumOfStarts\x12\x1e\n" +
-	"\vnum_of_ends\x18\x11 \x01(\x04R\tnumOfEnds\x12 \n" +
-	"\fnum_of_drops\x18\x12 \x01(\x04R\n" +
-	"numOfDropsB\x11\n" +
-	"\x0fconnection_info\"H\n" +
-	"\bPortInfo\x12\x1f\n" +
-	"\vsource_port\x18\x01 \x01(\rR\n" +
-	"sourcePort\x12\x1b\n" +
-	"\tdest_port\x18\x02 \x01(\rR\bdestPort\"D\n" +
-	"\bICMPInfo\x12\x1b\n" +
-	"\ticmp_type\x18\x01 \x01(\rR\bicmpType\x12\x1b\n" +
-	"\ticmp_code\x18\x02 \x01(\rR\bicmpCode*E\n" +
-	"\x04Type\x12\x10\n" +
-	"\fTYPE_UNKNOWN\x10\x00\x12\x0e\n" +
-	"\n" +
-	"TYPE_START\x10\x01\x12\f\n" +
-	"\bTYPE_END\x10\x02\x12\r\n" +
-	"\tTYPE_DROP\x10\x03*;\n" +
-	"\tDirection\x12\x15\n" +
-	"\x11DIRECTION_UNKNOWN\x10\x00\x12\v\n" +
-	"\aINGRESS\x10\x01\x12\n" +
-	"\n" +
-	"\x06EGRESS\x10\x022B\n" +
-	"\vFlowService\x123\n" +
-	"\x06Events\x12\x0f.flow.FlowEvent\x1a\x12.flow.FlowEventAck\"\x00(\x010\x01B\bZ\x06/protob\x06proto3"
+var file_flow_proto_rawDesc = []byte{
+	0x0a, 0x0a, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x04, 0x66, 0x6c,
+	0x6f, 0x77, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x22, 0xd4, 0x01, 0x0a, 0x09, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
+	0x74, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x0c, 0x52, 0x07, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x38, 0x0a, 0x09,
+	0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32,
+	0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75,
+	0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x74, 0x69, 0x6d,
+	0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12, 0x1d, 0x0a, 0x0a, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63,
+	0x5f, 0x6b, 0x65, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x70, 0x75, 0x62, 0x6c,
+	0x69, 0x63, 0x4b, 0x65, 0x79, 0x12, 0x31, 0x0a, 0x0b, 0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x66, 0x69,
+	0x65, 0x6c, 0x64, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x10, 0x2e, 0x66, 0x6c, 0x6f,
+	0x77, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x52, 0x0a, 0x66, 0x6c,
+	0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x12, 0x20, 0x0a, 0x0b, 0x69, 0x73, 0x49, 0x6e,
+	0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x18, 0x05, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69,
+	0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x22, 0x4b, 0x0a, 0x0c, 0x46, 0x6c,
+	0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x41, 0x63, 0x6b, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76,
+	0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x65, 0x76,
+	0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x20, 0x0a, 0x0b, 0x69, 0x73, 0x49, 0x6e, 0x69, 0x74, 0x69,
+	0x61, 0x74, 0x6f, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69, 0x73, 0x49, 0x6e,
+	0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x22, 0x9c, 0x04, 0x0a, 0x0a, 0x46, 0x6c, 0x6f, 0x77,
+	0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x12, 0x17, 0x0a, 0x07, 0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x69,
+	0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x66, 0x6c, 0x6f, 0x77, 0x49, 0x64, 0x12,
+	0x1e, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0a, 0x2e,
+	0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12,
+	0x17, 0x0a, 0x07, 0x72, 0x75, 0x6c, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c,
+	0x52, 0x06, 0x72, 0x75, 0x6c, 0x65, 0x49, 0x64, 0x12, 0x2d, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65,
+	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0f, 0x2e, 0x66, 0x6c,
+	0x6f, 0x77, 0x2e, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x64, 0x69,
+	0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x63, 0x6f, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x63, 0x6f, 0x6c, 0x12, 0x1b, 0x0a, 0x09, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x70,
+	0x18, 0x06, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x08, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x70,
+	0x12, 0x17, 0x0a, 0x07, 0x64, 0x65, 0x73, 0x74, 0x5f, 0x69, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28,
+	0x0c, 0x52, 0x06, 0x64, 0x65, 0x73, 0x74, 0x49, 0x70, 0x12, 0x2d, 0x0a, 0x09, 0x70, 0x6f, 0x72,
+	0x74, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66,
+	0x6c, 0x6f, 0x77, 0x2e, 0x50, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08,
+	0x70, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x2d, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70,
+	0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c,
+	0x6f, 0x77, 0x2e, 0x49, 0x43, 0x4d, 0x50, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x69,
+	0x63, 0x6d, 0x70, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1d, 0x0a, 0x0a, 0x72, 0x78, 0x5f, 0x70, 0x61,
+	0x63, 0x6b, 0x65, 0x74, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x72, 0x78, 0x50,
+	0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x12, 0x1d, 0x0a, 0x0a, 0x74, 0x78, 0x5f, 0x70, 0x61, 0x63,
+	0x6b, 0x65, 0x74, 0x73, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x74, 0x78, 0x50, 0x61,
+	0x63, 0x6b, 0x65, 0x74, 0x73, 0x12, 0x19, 0x0a, 0x08, 0x72, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65,
+	0x73, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x72, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73,
+	0x12, 0x19, 0x0a, 0x08, 0x74, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x0d, 0x20, 0x01,
+	0x28, 0x04, 0x52, 0x07, 0x74, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73, 0x12, 0x2c, 0x0a, 0x12, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69,
+	0x64, 0x18, 0x0e, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x10, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52,
+	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x64, 0x12, 0x28, 0x0a, 0x10, 0x64, 0x65, 0x73,
+	0x74, 0x5f, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x0f, 0x20,
+	0x01, 0x28, 0x0c, 0x52, 0x0e, 0x64, 0x65, 0x73, 0x74, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
+	0x65, 0x49, 0x64, 0x42, 0x11, 0x0a, 0x0f, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f,
+	0x6e, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x22, 0x48, 0x0a, 0x08, 0x50, 0x6f, 0x72, 0x74, 0x49, 0x6e,
+	0x66, 0x6f, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x70, 0x6f, 0x72,
+	0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x50,
+	0x6f, 0x72, 0x74, 0x12, 0x1b, 0x0a, 0x09, 0x64, 0x65, 0x73, 0x74, 0x5f, 0x70, 0x6f, 0x72, 0x74,
+	0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x64, 0x65, 0x73, 0x74, 0x50, 0x6f, 0x72, 0x74,
+	0x22, 0x44, 0x0a, 0x08, 0x49, 0x43, 0x4d, 0x50, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1b, 0x0a, 0x09,
+	0x69, 0x63, 0x6d, 0x70, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52,
+	0x08, 0x69, 0x63, 0x6d, 0x70, 0x54, 0x79, 0x70, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x69, 0x63, 0x6d,
+	0x70, 0x5f, 0x63, 0x6f, 0x64, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x69, 0x63,
+	0x6d, 0x70, 0x43, 0x6f, 0x64, 0x65, 0x2a, 0x45, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x10,
+	0x0a, 0x0c, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00,
+	0x12, 0x0e, 0x0a, 0x0a, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x53, 0x54, 0x41, 0x52, 0x54, 0x10, 0x01,
+	0x12, 0x0c, 0x0a, 0x08, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x45, 0x4e, 0x44, 0x10, 0x02, 0x12, 0x0d,
+	0x0a, 0x09, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x44, 0x52, 0x4f, 0x50, 0x10, 0x03, 0x2a, 0x3b, 0x0a,
+	0x09, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x15, 0x0a, 0x11, 0x44, 0x49,
+	0x52, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10,
+	0x00, 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x4e, 0x47, 0x52, 0x45, 0x53, 0x53, 0x10, 0x01, 0x12, 0x0a,
+	0x0a, 0x06, 0x45, 0x47, 0x52, 0x45, 0x53, 0x53, 0x10, 0x02, 0x32, 0x42, 0x0a, 0x0b, 0x46, 0x6c,
+	0x6f, 0x77, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x33, 0x0a, 0x06, 0x45, 0x76, 0x65,
+	0x6e, 0x74, 0x73, 0x12, 0x0f, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x45,
+	0x76, 0x65, 0x6e, 0x74, 0x1a, 0x12, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x46, 0x6c, 0x6f, 0x77,
+	0x45, 0x76, 0x65, 0x6e, 0x74, 0x41, 0x63, 0x6b, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x42, 0x08,
+	0x5a, 0x06, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
 
 var (
 	file_flow_proto_rawDescOnce sync.Once
-	file_flow_proto_rawDescData []byte
+	file_flow_proto_rawDescData = file_flow_proto_rawDesc
 )
 
 func file_flow_proto_rawDescGZIP() []byte {
 	file_flow_proto_rawDescOnce.Do(func() {
-		file_flow_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_flow_proto_rawDesc), len(file_flow_proto_rawDesc)))
+		file_flow_proto_rawDescData = protoimpl.X.CompressGZIP(file_flow_proto_rawDescData)
 	})
 	return file_flow_proto_rawDescData
 }
 
 var file_flow_proto_enumTypes = make([]protoimpl.EnumInfo, 2)
 var file_flow_proto_msgTypes = make([]protoimpl.MessageInfo, 5)
-var file_flow_proto_goTypes = []any{
+var file_flow_proto_goTypes = []interface{}{
 	(Type)(0),                     // 0: flow.Type
 	(Direction)(0),                // 1: flow.Direction
 	(*FlowEvent)(nil),             // 2: flow.FlowEvent
@@ -696,7 +701,69 @@ func file_flow_proto_init() {
 	if File_flow_proto != nil {
 		return
 	}
-	file_flow_proto_msgTypes[2].OneofWrappers = []any{
+	if !protoimpl.UnsafeEnabled {
+		file_flow_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*FlowEvent); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_flow_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*FlowEventAck); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_flow_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*FlowFields); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_flow_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*PortInfo); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_flow_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ICMPInfo); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	file_flow_proto_msgTypes[2].OneofWrappers = []interface{}{
 		(*FlowFields_PortInfo)(nil),
 		(*FlowFields_IcmpInfo)(nil),
 	}
@@ -704,7 +771,7 @@ func file_flow_proto_init() {
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
-			RawDescriptor: unsafe.Slice(unsafe.StringData(file_flow_proto_rawDesc), len(file_flow_proto_rawDesc)),
+			RawDescriptor: file_flow_proto_rawDesc,
 			NumEnums:      2,
 			NumMessages:   5,
 			NumExtensions: 0,
@@ -716,6 +783,7 @@ func file_flow_proto_init() {
 		MessageInfos:      file_flow_proto_msgTypes,
 	}.Build()
 	File_flow_proto = out.File
+	file_flow_proto_rawDesc = nil
 	file_flow_proto_goTypes = nil
 	file_flow_proto_depIdxs = nil
 }

flow/proto/flow.proto

@@ -75,9 +75,6 @@ message FlowFields {
   bytes source_resource_id = 14;
   bytes dest_resource_id = 15;
 
-  uint64 num_of_starts = 16;
-  uint64 num_of_ends = 17;
-  uint64 num_of_drops = 18;
 }
 
 // Flow event types

flow/proto/flow_grpc.pb.go

@@ -1,8 +1,4 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
-// versions:
-// - protoc-gen-go-grpc v1.6.1
-// - protoc             v7.34.1
-// source: flow.proto
 
 package proto
 
@@ -15,19 +11,15 @@ import (
 
 // This is a compile-time assertion to ensure that this generated file
 // is compatible with the grpc package it is being compiled against.
-// Requires gRPC-Go v1.64.0 or later.
-const _ = grpc.SupportPackageIsVersion9
-
-const (
-	FlowService_Events_FullMethodName = "/flow.FlowService/Events"
-)
+// Requires gRPC-Go v1.32.0 or later.
+const _ = grpc.SupportPackageIsVersion7
 
 // FlowServiceClient is the client API for FlowService service.
 //
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
 type FlowServiceClient interface {
 	// Client to receiver streams of events and acknowledgements
-	Events(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[FlowEvent, FlowEventAck], error)
+	Events(ctx context.Context, opts ...grpc.CallOption) (FlowService_EventsClient, error)
 }
 
 type flowServiceClient struct {
@@ -38,40 +30,54 @@ func NewFlowServiceClient(cc grpc.ClientConnInterface) FlowServiceClient {
 	return &flowServiceClient{cc}
 }
 
-func (c *flowServiceClient) Events(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[FlowEvent, FlowEventAck], error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &FlowService_ServiceDesc.Streams[0], FlowService_Events_FullMethodName, cOpts...)
+func (c *flowServiceClient) Events(ctx context.Context, opts ...grpc.CallOption) (FlowService_EventsClient, error) {
+	stream, err := c.cc.NewStream(ctx, &FlowService_ServiceDesc.Streams[0], "/flow.FlowService/Events", opts...)
 	if err != nil {
 		return nil, err
 	}
-	x := &grpc.GenericClientStream[FlowEvent, FlowEventAck]{ClientStream: stream}
+	x := &flowServiceEventsClient{stream}
 	return x, nil
 }
 
-// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
-type FlowService_EventsClient = grpc.BidiStreamingClient[FlowEvent, FlowEventAck]
+type FlowService_EventsClient interface {
+	Send(*FlowEvent) error
+	Recv() (*FlowEventAck, error)
+	grpc.ClientStream
+}
+
+type flowServiceEventsClient struct {
+	grpc.ClientStream
+}
+
+func (x *flowServiceEventsClient) Send(m *FlowEvent) error {
+	return x.ClientStream.SendMsg(m)
+}
+
+func (x *flowServiceEventsClient) Recv() (*FlowEventAck, error) {
+	m := new(FlowEventAck)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
 
 // FlowServiceServer is the server API for FlowService service.
 // All implementations must embed UnimplementedFlowServiceServer
-// for forward compatibility.
+// for forward compatibility
 type FlowServiceServer interface {
 	// Client to receiver streams of events and acknowledgements
-	Events(grpc.BidiStreamingServer[FlowEvent, FlowEventAck]) error
+	Events(FlowService_EventsServer) error
 	mustEmbedUnimplementedFlowServiceServer()
 }
 
-// UnimplementedFlowServiceServer must be embedded to have
-// forward compatible implementations.
-//
-// NOTE: this should be embedded by value instead of pointer to avoid a nil
-// pointer dereference when methods are called.
-type UnimplementedFlowServiceServer struct{}
+// UnimplementedFlowServiceServer must be embedded to have forward compatible implementations.
+type UnimplementedFlowServiceServer struct {
+}
 
-func (UnimplementedFlowServiceServer) Events(grpc.BidiStreamingServer[FlowEvent, FlowEventAck]) error {
-	return status.Error(codes.Unimplemented, "method Events not implemented")
+func (UnimplementedFlowServiceServer) Events(FlowService_EventsServer) error {
+	return status.Errorf(codes.Unimplemented, "method Events not implemented")
 }
 func (UnimplementedFlowServiceServer) mustEmbedUnimplementedFlowServiceServer() {}
-func (UnimplementedFlowServiceServer) testEmbeddedByValue()                     {}
 
 // UnsafeFlowServiceServer may be embedded to opt out of forward compatibility for this service.
 // Use of this interface is not recommended, as added methods to FlowServiceServer will
@@ -81,22 +87,34 @@ type UnsafeFlowServiceServer interface {
 }
 
 func RegisterFlowServiceServer(s grpc.ServiceRegistrar, srv FlowServiceServer) {
-	// If the following call panics, it indicates UnimplementedFlowServiceServer was
-	// embedded by pointer and is nil.  This will cause panics if an
-	// unimplemented method is ever invoked, so we test this at initialization
-	// time to prevent it from happening at runtime later due to I/O.
-	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
-		t.testEmbeddedByValue()
-	}
 	s.RegisterService(&FlowService_ServiceDesc, srv)
 }
 
 func _FlowService_Events_Handler(srv interface{}, stream grpc.ServerStream) error {
-	return srv.(FlowServiceServer).Events(&grpc.GenericServerStream[FlowEvent, FlowEventAck]{ServerStream: stream})
+	return srv.(FlowServiceServer).Events(&flowServiceEventsServer{stream})
 }
 
-// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
-type FlowService_EventsServer = grpc.BidiStreamingServer[FlowEvent, FlowEventAck]
+type FlowService_EventsServer interface {
+	Send(*FlowEventAck) error
+	Recv() (*FlowEvent, error)
+	grpc.ServerStream
+}
+
+type flowServiceEventsServer struct {
+	grpc.ServerStream
+}
+
+func (x *flowServiceEventsServer) Send(m *FlowEventAck) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func (x *flowServiceEventsServer) Recv() (*FlowEvent, error) {
+	m := new(FlowEvent)
+	if err := x.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
 
 // FlowService_ServiceDesc is the grpc.ServiceDesc for FlowService service.
 // It's only intended for direct use with grpc.RegisterService,

flow/proto/generate.sh

@@ -10,9 +10,8 @@ fi
 
 old_pwd=$(pwd)
 script_path=$(dirname $(realpath "$0"))
-echo "$script_path"
 cd "$script_path"
-#go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.26
-#go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.1
+go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.26
+go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.1
 protoc -I ./ ./flow.proto --go_out=../ --go-grpc_out=../
 cd "$old_pwd"