hubble: Observe node table for peer updates

Project: cilium/cilium | Language: Go | Concepts: observability, rate-limiting, refactoring | Difficulty: mid
Link: https://github.com/cilium/cilium/pull/48076

The problem

The Hubble peer service previously relied on per-client subscriptions to a NodeManager to receive notifications about node (peer) additions, updates, and deletions. This approach was inefficient and made it difficult to manage lifecycle and state consistency, especially as the system scaled or underwent restarts. There was also unnecessary overhead in handling per-client subscriptions and potential for missed or duplicated events.

Why it's non-trivial

A naive fix might simply swap out the notification backend, but would risk:

Losing the ability to distinguish between add, update, and delete events.
Not preserving the semantics of peer address changes (e.g., filtering out updates that don't affect peer address).
Failing to efficiently seed a new stream with the current node state.
Introducing excessive overhead or race conditions if not rate-limited properly.
Leaking goroutines if context handling is not correct.

The approach

1. Remove NodeManager dependency and refactor service construction

The PR removes all references to NodeManager from the hubble peer integration and replaces them with StateDB table and transaction objects:

-	NodeManager nodeManager.NodeManager
+	DB        *statedb.DB
+	Nodes     statedb.Table[*node.Node]

This change means the service now observes the node table directly, rather than subscribing to a manager.

2. Refactor handler methods to not depend on node.Handler interface

The handler methods are renamed and decoupled from the node.Handler interface:

-func (h *handler) NodeAdd(n types.Node) error {
+func (h *handler) nodeAdded(n types.Node) {

Similarly for NodeUpdate and NodeDelete. This makes the handler a plain utility for emitting notifications, not a subscriber to a manager.

3. Use StateDB change iterators and seed from the initial snapshot

The main Notify implementation now seeds each stream with the current table snapshot, then watches for changes:

wtxn := s.db.WriteTxn(s.nodes)
changes, err := s.nodes.Changes(wtxn)
wtxn.Commit()
if err != nil {
	return err
}
defer changes.Close()

This ensures that each client always sees the current state before any updates.

4. Retain previous node state to preserve add, update, and delete notifications

The service tracks previous node state to distinguish between add, update, and delete events, and to filter out irrelevant updates:

previous := map[string]*node.Node{}
for {
	seq, watch := changes.Next(s.db.ReadTxn())
	for change := range seq {
		n := change.Object
		name := n.Fullname()
		if change.Deleted {
			h.nodeDeleted(n.Node)
			delete(previous, name)
		} else if old, found := previous[name]; found {
			h.nodeUpdated(old.Node, n.Node)
			previous[name] = n
		} else {
			h.nodeAdded(n.Node)
			previous[name] = n
		}
	}
	...
}

This logic ensures that notifications are only sent when the peer address actually changes, and that the right event type is used.

5. Rate-limit change notification batches

To avoid excessive processing, the PR adds a rate limiter that batches changes and only processes them every 50 milliseconds:

limiter := rate.NewLimiter(50*time.Millisecond, 1)
defer limiter.Stop()
...
if limiter.Wait(ctx) != nil {
	return nil
}

This reduces overhead and smooths out bursts of changes.

6. Correct context handling to avoid goroutine leaks

The service now derives its context from the gRPC stream, ensuring that goroutines are properly canceled if the client disconnects:

ctx, cancel := context.WithCancel(stream.Context())
defer cancel()

This addresses a review comment about potential goroutine leaks.

7. Update tests to use the new StateDB-based fixture

All tests are updated to use the new table-based notification system, ensuring correctness and coverage.

From the review thread

It found one regression compared to using nodemanager that may be worth addressing. Essentially, if the Cilium agent restarts while a node is removed, nodemanager received a node-deleted event, but StateDB does not. Therefore, Relay can retain a node forever. This involves a larger change so we may want to consider it an edge-case. wdyt?

I don't think there is a guarantee that the relay agent manages to connect at the right time to catch that node-deleted event, so I think that this never worked reliably (or possible never as the timing would really have to hit right to catch the replay from the "nodes.json").

We probably want the relay agent on retry of Notify() to catch any removed nodes. The protocol seems insufficient for this as we don't know when the initial listing is done so we'd likely need to extend it. E.g. add a "PEER_LIST_DONE" that marks when a full set of nodes has been sent over and relay can then clean up its state.

[Later, on context handling:]

It seems this predates this PR, but because we don't derive the context from stream.Context(), we could leak the goroutine if the client disconnects while we are idle. I see we would return on a send that fails below, but while idle we could wait a long time or forever (if no node changes occurs or very rarely).

Fixed.

Takeaways

Use a table-driven, snapshot-plus-changes approach to seed new subscribers with current state and deliver consistent change notifications.
Track previous state and filter out irrelevant updates to avoid unnecessary notifications and maintain correct event semantics.
Always derive goroutine contexts from the request or stream context to ensure proper cancellation and avoid leaks.
Rate-limit change notification processing to avoid excessive overhead and smooth out bursts, especially in high-churn environments.
When replacing subscription-based event systems, carefully consider edge cases such as missed deletions or restarts, and document protocol limitations.