net (04/12): Wire controller summary and diagnostic transports

cmd/unbounded-net-controller/cluster_status.go

@@ -18,6 +18,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"github.com/Azure/unbounded/internal/net/controller"
+	statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1"
 	"github.com/Azure/unbounded/internal/version"
 )
 
@@ -236,6 +237,13 @@ func fetchClusterStatus(ctx context.Context, health *healthState, pullEnabled bo
 	}
 
 	cachedStatuses := health.statusCache.GetAll()
+	status.NodeOverviews = make(map[string]*statusv1alpha1.NodeStatusOverview)
+
+	for name, cached := range cachedStatuses {
+		if cached.Overview != nil {
+			status.NodeOverviews[name] = cached.Overview
+		}
+	}
 
 	type pullNode struct{ nodeName, nodeIP string }
 
@@ -371,6 +379,7 @@ func fetchClusterStatus(ctx context.Context, health *healthState, pullEnabled bo
 			} else if result.status != nil {
 				result.status.StatusSource = "pull"
 				cachedResults[result.nodeName] = *result.status
+				delete(status.NodeOverviews, result.nodeName)
 			}
 		}
 	}
@@ -447,6 +456,9 @@ func fetchClusterStatus(ctx context.Context, health *healthState, pullEnabled bo
 			if pubKey := node.Annotations[controller.WireGuardPubKeyAnnotation]; pubKey != "" {
 				if nodeStatus.NodeInfo.WireGuard == nil {
 					nodeStatus.NodeInfo.WireGuard = &WireGuardStatusInfo{}
+				} else {
+					wireguard := *nodeStatus.NodeInfo.WireGuard
+					nodeStatus.NodeInfo.WireGuard = &wireguard
 				}
 
 				nodeStatus.NodeInfo.WireGuard.PublicKey = pubKey
@@ -809,6 +821,18 @@ func collectClusterProblems(status *ClusterStatusResponse) []StatusProblem {
 			appendProblem("node", nodeName, summary)
 		}
 
+		if overview := status.NodeOverviews[node.NodeInfo.Name]; overview != nil {
+			if overview.RouteMismatch {
+				appendProblem("node", nodeName, "Route next-hop mismatches (expected vs present)")
+			}
+
+			if unhealthy := overview.PeerCount - overview.HealthyPeers; unhealthy > 0 {
+				appendProblem("node", nodeName, fmt.Sprintf("%d peers are not healthy", unhealthy))
+			}
+
+			continue
+		}
+
 		if mismatchCount := routeMismatchCount(node); mismatchCount > 0 {
 			appendProblem("node", nodeName, fmt.Sprintf("%d route next-hop mismatches (expected vs present)", mismatchCount))
 		}

cmd/unbounded-net-controller/cluster_status_cache.go

@@ -5,6 +5,7 @@ package main
 
 import (
 	"context"
+	"maps"
 	"reflect"
 	"slices"
 	"sync"
@@ -15,6 +16,8 @@ import (
 
 	unboundednetv1alpha1 "github.com/Azure/unbounded/api/net/v1alpha1"
 	"github.com/Azure/unbounded/internal/net/controller"
+	statuspkg "github.com/Azure/unbounded/internal/net/status"
+	statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1"
 )
 
 // ClusterStatusCache maintains a pre-built ClusterStatusResponse in memory,
@@ -115,6 +118,15 @@ func (c *ClusterStatusCache) Rebuild(ctx context.Context) {
 // PatchNode updates a single node's cached status in-place without a full
 // rebuild.
 func (c *ClusterStatusCache) PatchNode(nodeName string, nodeStatus NodeStatusResponse) {
+	c.patchNode(nodeName, nodeStatus, nil)
+}
+
+// PatchOverview updates metadata and observed facts without collecting details.
+func (c *ClusterStatusCache) PatchOverview(nodeName string, overview statusv1alpha1.NodeStatusOverview) {
+	c.patchNode(nodeName, statuspkg.OverviewMetadata(overview), &overview)
+}
+
+func (c *ClusterStatusCache) patchNode(nodeName string, nodeStatus NodeStatusResponse, overview *statusv1alpha1.NodeStatusOverview) {
 	now := time.Now()
 	nodeStatus.NodeInfo.ExternalIPs = c.resolveNodeExternalIPs(nodeName, now)
 
@@ -125,6 +137,16 @@ func (c *ClusterStatusCache) PatchNode(nodeName string, nodeStatus NodeStatusRes
 		return
 	}
 
+	if overview == nil {
+		delete(c.status.NodeOverviews, nodeName)
+	} else {
+		if c.status.NodeOverviews == nil {
+			c.status.NodeOverviews = make(map[string]*statusv1alpha1.NodeStatusOverview)
+		}
+
+		c.status.NodeOverviews[nodeName] = overview
+	}
+
 	if i, ok := c.nodeIndex[nodeName]; ok && i < len(c.status.Nodes) {
 		// Preserve controller-enriched fields across node-agent status updates.
 		existing := c.status.Nodes[i]
@@ -225,13 +247,21 @@ func (c *ClusterStatusCache) MarkFullRebuildNeeded() {
 	}
 }
 
-// Get returns the current pre-built status (read-locked, fast).
+// Get snapshots mutable containers; nested node data remains immutable and shared.
 // Returns nil if the status has not been built yet.
 func (c *ClusterStatusCache) Get() *ClusterStatusResponse {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 
-	return c.status
+	if c.status == nil {
+		return nil
+	}
+
+	snapshot := *c.status
+	snapshot.Nodes = slices.Clone(c.status.Nodes)
+	snapshot.NodeOverviews = maps.Clone(c.status.NodeOverviews)
+
+	return &snapshot
 }
 
 // GetSeq returns the current sequence number.

cmd/unbounded-net-controller/detail_dispatch.go

@@ -0,0 +1,101 @@
+// Copyright (c) Microsoft Corporation.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"context"
+	"time"
+
+	statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1"
+)
+
+type nodeWSConnection struct {
+	cancel context.CancelFunc
+	send   func(context.Context, statusv1alpha1.DetailRequest) error
+}
+
+func (h *healthState) markNodeWSStale(nodeName string, connection *nodeWSConnection, source string) {
+	h.nodeWSMu.Lock()
+	defer h.nodeWSMu.Unlock()
+
+	if connection != nil && h.nodeWSRegistry[nodeName] == connection {
+		h.statusCache.UpdateSourceIf(nodeName, source, "stale-cache")
+	}
+}
+
+func (h *healthState) setNodeWSDetailSender(nodeName string, connection *nodeWSConnection, send func(context.Context, statusv1alpha1.DetailRequest) error) {
+	h.nodeWSMu.Lock()
+
+	ready := false
+	if connection != nil && h.nodeWSRegistry[nodeName] == connection {
+		ready = connection.send == nil
+		connection.send = send
+	}
+	h.nodeWSMu.Unlock()
+
+	if ready {
+		h.retryNodeDetails(nodeName)
+	}
+}
+
+func (h *healthState) dispatchNodeDetail(ctx context.Context, nodeName string, command statusv1alpha1.DetailRequest) (bool, error) {
+	h.nodeWSMu.Lock()
+
+	var send func(context.Context, statusv1alpha1.DetailRequest) error
+	if connection := h.nodeWSRegistry[nodeName]; connection != nil {
+		send = connection.send
+	}
+	h.nodeWSMu.Unlock()
+
+	if send == nil {
+		return false, nil
+	}
+
+	writeCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+
+	err := send(writeCtx, command)
+
+	return err == nil, err
+}
+
+func (h *healthState) retryNodeDetails(nodeName string) {
+	if manager := h.getDetailRequests(); manager != nil {
+		manager.Retry(nodeName)
+	}
+}
+
+// Retry wakes an existing request after a transport change, retaining its ID,
+// deadline, and one-dispatch-at-a-time ownership.
+func (m *nodeDetailRequests) Retry(nodeName string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	m.expireLocked(time.Now())
+
+	if request := m.active[nodeName]; request != nil && m.ctx.Err() == nil && !m.closed {
+		request.retry = true
+		if !request.dispatching {
+			m.startDispatchLocked(request)
+		}
+	}
+}
+
+func (m *nodeDetailRequests) startDispatchLocked(request *nodeDetailRequest) {
+	request.dispatching = true
+	request.retry = false
+	request.poll = false
+
+	m.workers.Go(func() {
+		m.dispatch(request.ctx, request.nodeName, request.command)
+
+		m.mu.Lock()
+		defer m.mu.Unlock()
+
+		request.dispatching = false
+		if m.active[request.nodeName] == request && request.retry && m.ctx.Err() == nil && !m.closed {
+			m.startDispatchLocked(request)
+		}
+	})
+}