From 7e150d1ecf6bd74a53d7d175dfc98fbe6ab3d40e Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Wed, 16 Sep 2026 21:33:48 +0000 Subject: [PATCH 1/2] feat(net-node): add lightweight overview collection and local endpoint Collect shared metadata, bootstrap errors, aggregate health and exact peer/route summary facts without BPF traversal or outbound detail arrays. Add /status/summary using the existing local status policy and an additive protobuf converter. Leave all publishers and full endpoints unchanged. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- cmd/unbounded-net-node/status_proto.go | 25 ++ cmd/unbounded-net-node/status_server.go | 66 +++- cmd/unbounded-net-node/status_summary.go | 110 ++++++ cmd/unbounded-net-node/status_summary_test.go | 320 ++++++++++++++++++ 4 files changed, 503 insertions(+), 18 deletions(-) create mode 100644 cmd/unbounded-net-node/status_summary.go create mode 100644 cmd/unbounded-net-node/status_summary_test.go diff --git a/cmd/unbounded-net-node/status_proto.go b/cmd/unbounded-net-node/status_proto.go index 256a64711..fa1350f67 100644 --- a/cmd/unbounded-net-node/status_proto.go +++ b/cmd/unbounded-net-node/status_proto.go @@ -18,6 +18,31 @@ func statusUnixNano(t time.Time) int64 { return t.UnixNano() } +func nodeSummaryToProto(summary *NodeStatusOverview) *statusproto.NodeStatusOverview { + if summary == nil { + return nil + } + + result := &statusproto.NodeStatusOverview{ + TimestampUnixNs: statusUnixNano(summary.Timestamp), + NodeInfo: nodeInfoToProto(&summary.NodeInfo), + HealthCheck: healthCheckStatusToProto(summary.HealthCheck), + NodeErrors: nodeErrorsToProto(summary.NodeErrors), + FetchError: summary.FetchError, + StatusSource: summary.StatusSource, + NodePodInfo: nodePodInfoToProto(summary.NodePodInfo), + PeerCount: int32(summary.PeerCount), + HealthyPeers: int32(summary.HealthyPeers), + RouteCount: int32(summary.RouteCount), + RouteMismatch: summary.RouteMismatch, + } + if summary.LastPushTime != nil { + result.LastPushTimeUnixNs = statusUnixNano(*summary.LastPushTime) + } + + return result +} + // nodeStatusToProto converts a Go NodeStatusResponse to the protobuf NodeStatusFull message. func nodeStatusToProto(status *NodeStatusResponse) *statusproto.NodeStatusFull { if status == nil { diff --git a/cmd/unbounded-net-node/status_server.go b/cmd/unbounded-net-node/status_server.go index 7eed77a49..3cf303936 100644 --- a/cmd/unbounded-net-node/status_server.go +++ b/cmd/unbounded-net-node/status_server.go @@ -25,6 +25,7 @@ import ( "github.com/coder/websocket" "github.com/vishvananda/netlink" "golang.org/x/sys/unix" + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" "google.golang.org/protobuf/proto" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/cache" @@ -465,7 +466,7 @@ func (tm *hmacTokenManager) requestToken() error { return fmt.Errorf("all HMAC token endpoints failed: %s", strings.Join(endpointErrors, "; ")) } -func startHealthServer(port int, healthState *nodeHealthState) { +func newHealthMux(healthState *nodeHealthState) *http.ServeMux { mux := http.NewServeMux() metrics.Register(mux) @@ -528,6 +529,13 @@ func startHealthServer(port int, healthState *nodeHealthState) { } }) + // Same local routing and authentication policy as the full status endpoint. + mux.HandleFunc("/status/summary", healthState.handleStatusSummary) + + return mux +} + +func startHealthServer(port int, healthState *nodeHealthState) { addr := fmt.Sprintf(":%d", port) klog.Infof("Starting health server on %s", addr) @@ -535,7 +543,7 @@ func startHealthServer(port int, healthState *nodeHealthState) { server := &http.Server{ Addr: addr, - Handler: httpMiddleware.Wrap("all", mux), + Handler: httpMiddleware.Wrap("all", newHealthMux(healthState)), ReadHeaderTimeout: 10 * time.Second, } if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { @@ -2291,6 +2299,18 @@ type nodeStatusServer struct { // the host kernel's actual routing table. Production callers leave this // nil and the helpers fall back to the real netlink package. netlinkOps statusServerNetlinkOps + + // Optional collector override for tests; summaries never invoke this. + bpfCollector func() []BpfEntry + wireGuardDevice func(*unboundednetnetlink.WireGuardManager) (*wgtypes.Device, error) +} + +func (s *nodeStatusServer) getWireGuardDevice(manager *unboundednetnetlink.WireGuardManager) (*wgtypes.Device, error) { + if s.wireGuardDevice != nil { + return s.wireGuardDevice(manager) + } + + return manager.GetDevice() } // statusServerNetlinkOps abstracts the netlink reads that @@ -2550,7 +2570,7 @@ func (s *nodeStatusServer) inspectNodePeers(visit func(WireGuardPeerStatus)) *no // Get WireGuard device info if available (netlink syscall) if wgManager != nil { - if device, err := wgManager.GetDevice(); err == nil { + if device, err := s.getWireGuardDevice(wgManager); err == nil { status.NodeInfo.WireGuard.ListenPort = device.ListenPort status.NodeInfo.WireGuard.PeerCount = len(device.Peers) @@ -2627,7 +2647,7 @@ func (s *nodeStatusServer) inspectNodePeers(visit func(WireGuardPeerStatus)) *no for _, gw := range gwSnapshots { // Get WireGuard peer info for this gateway interface (netlink syscall) if gw.wgManager != nil { - if device, err := gw.wgManager.GetDevice(); err == nil && len(device.Peers) > 0 { + if device, err := s.getWireGuardDevice(gw.wgManager); err == nil && len(device.Peers) > 0 { wgPeer := device.Peers[0] // Each gateway interface has one peer peer := WireGuardPeerStatus{ @@ -2843,7 +2863,11 @@ func (s *nodeStatusServer) getNodeStatus() *NodeStatusResponse { } // Collect BPF trie entries. - status.BpfEntries = s.collectBpfEntries() + if s.bpfCollector != nil { + status.BpfEntries = s.bpfCollector() + } else { + status.BpfEntries = s.collectBpfEntries() + } return status } @@ -2866,26 +2890,32 @@ func linkStatsWarningsAsNodeErrors(warnings []string, peers []WireGuardPeerStatu } func suppressHealthyWireGuardRxErrors(warning string, peers []WireGuardPeerStatus, now time.Time) bool { - iface, deltas, ok := parseLinkStatsWarning(warning) - if !ok || len(deltas) != 1 || !strings.HasPrefix(deltas[0], "rx_errors +") { - return false - } + return suppressHealthyInterfaceRxErrors(warning, func(iface string) bool { + matched := false - matched := false + for _, peer := range peers { + if peer.Tunnel.Interface != iface { + continue + } - for _, peer := range peers { - if peer.Tunnel.Interface != iface { - continue + matched = true + + if !peerStatusHealthy(peer, now) { + return false + } } - matched = true + return matched + }) +} - if !peerStatusHealthy(peer, now) { - return false - } +func suppressHealthyInterfaceRxErrors(warning string, healthy func(string) bool) bool { + iface, deltas, ok := parseLinkStatsWarning(warning) + if !ok || len(deltas) != 1 || !strings.HasPrefix(deltas[0], "rx_errors +") { + return false } - return matched + return healthy(iface) } func parseLinkStatsWarning(warning string) (string, []string, bool) { diff --git a/cmd/unbounded-net-node/status_summary.go b/cmd/unbounded-net-node/status_summary.go new file mode 100644 index 000000000..5d088de91 --- /dev/null +++ b/cmd/unbounded-net-node/status_summary.go @@ -0,0 +1,110 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "net/http" + "time" + + "k8s.io/klog/v2" + + "github.com/Azure/unbounded/internal/net/routeplan" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +type NodeStatusOverview = statusv1alpha1.NodeStatusOverview + +// getNodeSummary collects overview facts directly. Only route-planning inputs +// survive peer visitation; full peer, route, and BPF snapshots are never built. +func (s *nodeStatusServer) getNodeSummary() *NodeStatusOverview { + summary := &NodeStatusOverview{} + + var routePeers []routeplan.Peer + + interfaceHealthy := make(map[string]bool) + now := time.Now() + facts := s.inspectNodePeers(func(peer WireGuardPeerStatus) { + summary.PeerCount++ + if summaryPeerHealthy(peer, now) { + summary.HealthyPeers++ + } + + previous, seen := interfaceHealthy[peer.Tunnel.Interface] + interfaceHealthy[peer.Tunnel.Interface] = (!seen || previous) && peerStatusHealthy(peer, now) + routePeers = append(routePeers, routeplan.Peer{ + Name: peer.Name, PeerType: peer.PeerType, SiteName: peer.SiteName, + SkipPodCIDRRoutes: peer.SkipPodCIDRRoutes, + Interface: peer.Tunnel.Interface, Endpoint: peer.Tunnel.Endpoint, + PodCIDRGateways: peer.PodCIDRGateways, AllowedIPs: peer.Tunnel.AllowedIPs, + RouteDistances: peer.RouteDistances, + }) + }) + summary.Timestamp = facts.Timestamp + summary.NodeInfo = facts.NodeInfo + summary.NodeErrors = facts.NodeErrors + summary.HealthCheck = facts.HealthCheck + summary.RouteCount, summary.RouteMismatch = s.collectRouteSummary(routePeers, facts.NodeInfo.SiteName) + + if s.state.linkStatsMonitor != nil { + for _, warning := range s.state.linkStatsMonitor.GetWarnings() { + if !suppressHealthyInterfaceRxErrors(warning, func(iface string) bool { return interfaceHealthy[iface] }) { + summary.NodeErrors = append(summary.NodeErrors, NodeError{Type: "link-stats", Message: warning}) + } + } + } + + return summary +} + +// This is deliberately stricter than link-warning suppression: the controller +// counts only "up"/"Up" when enabled, and otherwise uses handshake freshness. +func summaryPeerHealthy(peer WireGuardPeerStatus, now time.Time) bool { + if peer.HealthCheck != nil && peer.HealthCheck.Enabled { + return peer.HealthCheck.Status == "up" || peer.HealthCheck.Status == "Up" + } + + return !peer.Tunnel.LastHandshake.IsZero() && now.Sub(peer.Tunnel.LastHandshake) < 3*time.Minute +} + +func (h *nodeHealthState) getSummarySnapshot() *NodeStatusOverview { + h.mu.RLock() + srv := h.statusServer + + summary := &NodeStatusOverview{ + Timestamp: time.Now(), + NodeInfo: NodeInfo{ + Name: h.nodeName, SiteName: h.siteName, IsGateway: h.isGateway, + PodCIDRs: append([]string(nil), h.podCIDRs...), BuildInfo: nodeAgentBuildInfo(), + }, + } + if h.pubKey != "" { + summary.NodeInfo.WireGuard = &WireGuardStatusInfo{PublicKey: h.pubKey} + } + + cniManaged, cniReady, cniReason := h.cniManaged, h.cniReady, h.cniReason + transientErrors := append([]NodeError(nil), h.transientErrors...) + h.mu.RUnlock() + + if srv != nil { + summary = srv.getNodeSummary() + } + + summary.NodeErrors = mergeNodeErrors(summary.NodeErrors, filterExpiredNodeErrors(transientErrors, time.Now(), time.Minute)) + + summary.NodeErrors = removeNodeErrorsByType(summary.NodeErrors, configPodCIDRGuard) + if cniManaged && !cniReady && cniReason != "" { + summary.NodeErrors = append(summary.NodeErrors, NodeError{Type: configPodCIDRGuard, Message: cniReason}) + } + + return summary +} + +func (h *nodeHealthState) handleStatusSummary(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + + if err := json.NewEncoder(w).Encode(h.getSummarySnapshot()); err != nil { + klog.V(4).Infof("status summary json encode failed: %v", err) + } +} diff --git a/cmd/unbounded-net-node/status_summary_test.go b/cmd/unbounded-net-node/status_summary_test.go new file mode 100644 index 000000000..75cecacfe --- /dev/null +++ b/cmd/unbounded-net-node/status_summary_test.go @@ -0,0 +1,320 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "errors" + "fmt" + "net" + "net/http" + "net/http/httptest" + "reflect" + "sync" + "testing" + "time" + + "github.com/vishvananda/netlink" + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" + "google.golang.org/protobuf/proto" + + "github.com/Azure/unbounded/internal/net/healthcheck" + unboundednetnetlink "github.com/Azure/unbounded/internal/net/netlink" + statusproto "github.com/Azure/unbounded/internal/net/status/proto" +) + +func TestSummaryPeerHealthy(t *testing.T) { + now := time.Now() + for _, tc := range []struct { + name string + health *HealthCheckPeerStatus + handshake time.Time + want bool + }{ + {"up", &HealthCheckPeerStatus{Enabled: true, Status: "up"}, time.Time{}, true}, + {"Up", &HealthCheckPeerStatus{Enabled: true, Status: "Up"}, time.Time{}, true}, + {"uppercase is not counted", &HealthCheckPeerStatus{Enabled: true, Status: "UP"}, now, false}, + {"enabled unknown", &HealthCheckPeerStatus{Enabled: true}, now, false}, + {"enabled down", &HealthCheckPeerStatus{Enabled: true, Status: "down"}, now, false}, + {"disabled uses handshake", &HealthCheckPeerStatus{Status: "down"}, now, true}, + {"missing health uses handshake", nil, now.Add(-time.Minute), true}, + {"missing handshake", nil, time.Time{}, false}, + {"boundary", nil, now.Add(-3 * time.Minute), false}, + {"future handshake", nil, now.Add(time.Minute), true}, + } { + t.Run(tc.name, func(t *testing.T) { + peer := WireGuardPeerStatus{HealthCheck: tc.health, Tunnel: PeerTunnelStatus{LastHandshake: tc.handshake}} + if got := summaryPeerHealthy(peer, now); got != tc.want { + t.Fatalf("healthy=%v, want %v", got, tc.want) + } + }) + } +} + +func TestNodeSummaryParityAndNoBPF(t *testing.T) { + for _, deviceFailure := range []bool{false, true} { + t.Run(map[bool]string{false: "device success", true: "device failure"}[deviceFailure], func(t *testing.T) { + s := summaryRouteFixture() + s.state.siteName = "local" + s.state.nodePodCIDRs = []string{"10.42.0.0/24"} + s.state.nodeInternalIPs = []string{"192.0.2.1"} + s.state.nodeExternalIPs = []string{"198.51.100.1"} + s.state.nodeErrors = []NodeError{{Type: "test", Message: "failure"}} + s.state.wireguardManager = &unboundednetnetlink.WireGuardManager{} + + manager, err := healthcheck.NewManager("local", 0, nil) + if err != nil { + t.Fatal(err) + } + + if err := manager.AddPeer("down", net.ParseIP("10.42.1.1"), healthcheck.DefaultSettings()); err != nil { + t.Fatal(err) + } + + s.state.healthCheckManager = manager + s.state.meshPeerHealthCheckEnabled = map[string]bool{"down": true, "missing": true} + s.state.peers = []meshPeerInfo{ + {Name: "down", WireGuardPublicKey: "down", TunnelProtocol: "GENEVE", InternalIPs: []string{"192.0.2.2"}, PodCIDRs: []string{"10.42.1.0/24"}}, + {Name: "missing", WireGuardPublicKey: "missing", TunnelProtocol: "VXLAN", InternalIPs: []string{"192.0.2.3"}, PodCIDRs: []string{"10.42.2.0/24"}}, + {Name: "no-address", TunnelProtocol: "IPIP"}, + } + s.state.gatewayPeers = []gatewayPeerInfo{ + {Name: "gateway", TunnelProtocol: "IPIP", InternalIPs: []string{"192.0.2.4"}, PodCIDRs: []string{"10.42.3.0/24"}}, + } + s.state.gatewayHealthEndpoints = map[string]string{"wg51821": "10.42.3.1"} + s.state.gatewayNames = map[string]string{"wg51821": "gateway"} + s.state.gatewayWireguardManagers = map[string]*unboundednetnetlink.WireGuardManager{"wg51821": {}} + s.state.linkStatsMonitor = &linkStatsMonitor{warnings: []string{ + "interface /wg51820: rx_errors +24", "interface /gn0: rx_errors +24", "interface /eth0: tx_errors +2", + }} + s.wireGuardDevice = func(*unboundednetnetlink.WireGuardManager) (*wgtypes.Device, error) { + if deviceFailure { + return nil, errors.New("device unavailable") + } + + return &wgtypes.Device{ListenPort: 51820, Peers: []wgtypes.Peer{ + {LastHandshakeTime: time.Now().Add(-time.Minute)}, + }}, nil + } + s.netlinkOps.(*fakeNetlinkOps).mainRoutes = map[int][]netlink.Route{ + netlink.FAMILY_V4: {summaryRoute("10.42.0.0/16", 2, 0, 0)}, + } + bpfCalls := 0 + s.bpfCollector = func() []BpfEntry { bpfCalls++; return nil } + + summary := s.getNodeSummary() + if bpfCalls != 0 || !s.routingTableCachedAt.IsZero() || len(s.routingTableCache.Routes) != 0 { + t.Fatal("summary collected BPF or populated the full route cache") + } + + full := s.getNodeStatus() + + if bpfCalls != 1 { + t.Fatal("legacy full collection no longer collects BPF") + } + + if !reflect.DeepEqual(summary.NodeInfo, full.NodeInfo) || !reflect.DeepEqual(summary.NodeErrors, full.NodeErrors) { + t.Fatalf("metadata/errors differ: summary=%+v full=%+v", summary, full) + } + + wantHealthy := 0 + + for _, peer := range full.Peers { + if summaryPeerHealthy(peer, time.Now()) { + wantHealthy++ + } + } + + if summary.PeerCount != len(full.Peers) || summary.HealthyPeers != wantHealthy || summary.RouteCount != len(full.RoutingTable.Routes) { + t.Fatalf("counts differ: summary=%+v full peers=%+v routes=%+v", summary, full.Peers, full.RoutingTable) + } + + wantPeers, wantHealthyPeers := 4, 2 + if deviceFailure { + wantPeers, wantHealthyPeers = 3, 0 + } + + if summary.PeerCount != wantPeers || summary.HealthyPeers != wantHealthyPeers || summary.RouteMismatch { + t.Fatalf("unexpected observed counts/mismatch: %+v", summary) + } + + if summary.HealthCheck == nil || summary.HealthCheck.Healthy != full.HealthCheck.Healthy || + summary.HealthCheck.PeerCount != full.HealthCheck.PeerCount || summary.HealthCheck.Summary != full.HealthCheck.Summary { + t.Fatalf("health aggregates differ: %v vs %v", summary.HealthCheck, full.HealthCheck) + } + + assertSummaryHasNoDetails(t, summary) + }) + } +} + +func assertSummaryHasNoDetails(t *testing.T, summary *NodeStatusOverview) { + t.Helper() + + data, err := json.Marshal(summary) + if err != nil { + t.Fatal(err) + } + + var fields map[string]json.RawMessage + if err := json.Unmarshal(data, &fields); err != nil { + t.Fatal(err) + } + + for _, name := range []string{"peers", "routingTable", "bpfEntries", "peerMeasurements"} { + if _, exists := fields[name]; exists { + t.Fatalf("summary contains detail field %q", name) + } + } +} + +func TestSummaryBootstrapRecoveryAndLocalRouting(t *testing.T) { + h := blockedBootstrapHealthState() + h.transientErrors = []NodeError{{Type: "transport", Message: "failed"}, {Type: "expired", Message: "old", Timestamp: time.Now().Add(-2 * time.Minute)}} + mux := newHealthMux(h) + + for _, path := range []string{"/status/summary", "/status", "/status/json"} { + recorder := httptest.NewRecorder() + mux.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, path, nil)) + + if recorder.Code != http.StatusOK || recorder.Header().Get("Content-Type") != "application/json" { + t.Fatalf("%s: code=%d headers=%v", path, recorder.Code, recorder.Header()) + } + + var summary NodeStatusOverview + if err := json.Unmarshal(recorder.Body.Bytes(), &summary); err != nil { + t.Fatal(err) + } + + if summary.NodeInfo.Name != "node-a" || len(summary.NodeErrors) != 2 || summary.NodeErrors[1].Type != configPodCIDRGuard { + t.Fatalf("%s: lost bootstrap identity/errors: %+v", path, summary) + } + + var fields map[string]json.RawMessage + if err := json.Unmarshal(recorder.Body.Bytes(), &fields); err != nil { + t.Fatal(err) + } + + _, hasPeers := fields["peers"] + if hasPeers == (path == "/status/summary") { + t.Fatalf("%s: endpoint returned wrong representation", path) + } + } + + h.setCNIReady("cbr0", []string{"10.244.7.0/24"}) + + summary := h.getSummarySnapshot() + if len(summary.NodeErrors) != 1 || summary.NodeErrors[0].Type != "transport" { + t.Fatalf("guard did not recover: %+v", summary.NodeErrors) + } + + summary.NodeInfo.PodCIDRs[0] = "mutated" + if h.getSummarySnapshot().NodeInfo.PodCIDRs[0] == "mutated" { + t.Fatal("bootstrap CIDRs alias shared state") + } + + s := summaryRouteFixture() + s.state.nodeErrors = []NodeError{{Type: configPodCIDRGuard, Message: "obsolete guard"}} + h.setStatusServer(s) + + if got := h.getSummarySnapshot(); got.NodeInfo.Name != "local" || len(got.NodeErrors) != 1 || got.NodeErrors[0].Type != "transport" { + t.Fatalf("initialized summary lost recovery/transport state: %+v", got) + } +} + +func TestSummaryHealthyAggregate(t *testing.T) { + s := summaryRouteFixture() + + manager, err := healthcheck.NewManager("local", 0, nil) + if err != nil { + t.Fatal(err) + } + + s.state.healthCheckManager = manager + + summary := s.getNodeSummary() + if summary.HealthCheck == nil || !summary.HealthCheck.Healthy || summary.HealthCheck.PeerCount != 0 || + summary.HealthCheck.Summary != "all peers healthy" || summary.HealthCheck.CheckedAt.IsZero() { + t.Fatalf("lost healthy aggregate: %+v", summary.HealthCheck) + } +} + +func TestSummaryConcurrentBootstrapState(t *testing.T) { + h := blockedBootstrapHealthState() + + var wg sync.WaitGroup + for range 4 { + wg.Go(func() { + for range 20 { + h.beginManagedCNI("cbr0") + h.getSummarySnapshot() + h.setCNIReady("cbr0", []string{"10.244.7.0/24"}) + } + }) + } + + wg.Wait() +} + +func TestNodeSummaryToProto(t *testing.T) { + if nodeSummaryToProto(nil) != nil { + t.Fatal("nil summary converted") + } + + now := time.Now() + summary := &NodeStatusOverview{ + Timestamp: now, NodeInfo: NodeInfo{Name: "node", K8sReady: "Unknown"}, + PeerCount: 10, HealthyPeers: 4, RouteCount: 12, RouteMismatch: true, + FetchError: "unavailable", StatusSource: "error", LastPushTime: &now, + NodeErrors: []NodeError{{Type: "failure", Message: "failed"}}, + HealthCheck: &HealthCheckStatus{Healthy: false, Summary: "unhealthy"}, + } + encoded := nodeSummaryToProto(summary) + + data, err := proto.Marshal(encoded) + if err != nil { + t.Fatal(err) + } + + var decoded statusproto.NodeStatusOverview + if err := proto.Unmarshal(data, &decoded); err != nil { + t.Fatal(err) + } + + if !proto.Equal(encoded, &decoded) || decoded.PeerCount != 10 || decoded.HealthyPeers != 4 || decoded.RouteCount != 12 || + !decoded.RouteMismatch || decoded.FetchError != "unavailable" || decoded.NodeInfo.K8SReady != "Unknown" || + decoded.LastPushTimeUnixNs != now.UnixNano() || decoded.StatusSource != "error" || len(decoded.NodeErrors) != 1 { + t.Fatalf("summary conversion lost facts: %v", &decoded) + } +} + +func BenchmarkNodeSummaryCollection(b *testing.B) { + for _, full := range []bool{false, true} { + b.Run(map[bool]string{false: "summary", true: "full"}[full], func(b *testing.B) { + s := summaryRouteFixture() + s.bpfCollector = func() []BpfEntry { return nil } + + s.netlinkOps.(*fakeNetlinkOps).mainRoutes = map[int][]netlink.Route{ + netlink.FAMILY_V4: {summaryRoute("10.0.0.0/8", 2, 0, 0)}, + } + for i := range 2000 { + s.state.peers = append(s.state.peers, meshPeerInfo{ + Name: fmt.Sprintf("peer-%d", i), WireGuardPublicKey: fmt.Sprintf("key-%d", i), + TunnelProtocol: "GENEVE", InternalIPs: []string{"192.0.2.2"}, + PodCIDRs: []string{fmt.Sprintf("10.%d.%d.0/24", i/256, i%256)}, + }) + } + + b.ReportAllocs() + + for b.Loop() { + if full { + s.getNodeStatus() + } else { + s.getNodeSummary() + } + } + }) + } +} From 190d868980d7e4f8969a12d427901e4b19bb9100 Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Wed, 16 Sep 2026 21:39:42 +0000 Subject: [PATCH 2/2] refactor(net-node): reuse shared overview health semantics Use PeerHealthyForOverview during streamed summary collection and shared legacy projection/mismatch helpers in parity tests. Do not collect full snapshots in the production summary path. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- cmd/unbounded-net-node/status_summary.go | 13 ++----------- .../status_summary_routes_test.go | 7 +++---- cmd/unbounded-net-node/status_summary_test.go | 15 +++++---------- 3 files changed, 10 insertions(+), 25 deletions(-) diff --git a/cmd/unbounded-net-node/status_summary.go b/cmd/unbounded-net-node/status_summary.go index 5d088de91..4866afe85 100644 --- a/cmd/unbounded-net-node/status_summary.go +++ b/cmd/unbounded-net-node/status_summary.go @@ -11,6 +11,7 @@ import ( "k8s.io/klog/v2" "github.com/Azure/unbounded/internal/net/routeplan" + netstatus "github.com/Azure/unbounded/internal/net/status" statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" ) @@ -27,7 +28,7 @@ func (s *nodeStatusServer) getNodeSummary() *NodeStatusOverview { now := time.Now() facts := s.inspectNodePeers(func(peer WireGuardPeerStatus) { summary.PeerCount++ - if summaryPeerHealthy(peer, now) { + if netstatus.PeerHealthyForOverview(&peer, now) { summary.HealthyPeers++ } @@ -58,16 +59,6 @@ func (s *nodeStatusServer) getNodeSummary() *NodeStatusOverview { return summary } -// This is deliberately stricter than link-warning suppression: the controller -// counts only "up"/"Up" when enabled, and otherwise uses handshake freshness. -func summaryPeerHealthy(peer WireGuardPeerStatus, now time.Time) bool { - if peer.HealthCheck != nil && peer.HealthCheck.Enabled { - return peer.HealthCheck.Status == "up" || peer.HealthCheck.Status == "Up" - } - - return !peer.Tunnel.LastHandshake.IsZero() && now.Sub(peer.Tunnel.LastHandshake) < 3*time.Minute -} - func (h *nodeHealthState) getSummarySnapshot() *NodeStatusOverview { h.mu.RLock() srv := h.statusServer diff --git a/cmd/unbounded-net-node/status_summary_routes_test.go b/cmd/unbounded-net-node/status_summary_routes_test.go index 6bd1a7863..7d8e598b3 100644 --- a/cmd/unbounded-net-node/status_summary_routes_test.go +++ b/cmd/unbounded-net-node/status_summary_routes_test.go @@ -11,6 +11,7 @@ import ( "golang.org/x/sys/unix" "github.com/Azure/unbounded/internal/net/routeplan" + netstatus "github.com/Azure/unbounded/internal/net/status" ) func summaryRoute(destination string, index, table, distance int) netlink.Route { @@ -89,10 +90,8 @@ func TestRouteSummaryParity(t *testing.T) { wantMismatch := false for _, route := range full.RoutingTable.Routes { - for _, hop := range route.NextHops { - if (hop.Expected != nil && *hop.Expected) != (hop.Present != nil && *hop.Present) { - wantMismatch = true - } + if netstatus.RouteMismatchForOverview(route) { + wantMismatch = true } } diff --git a/cmd/unbounded-net-node/status_summary_test.go b/cmd/unbounded-net-node/status_summary_test.go index 75cecacfe..215b1c2c0 100644 --- a/cmd/unbounded-net-node/status_summary_test.go +++ b/cmd/unbounded-net-node/status_summary_test.go @@ -21,6 +21,7 @@ import ( "github.com/Azure/unbounded/internal/net/healthcheck" unboundednetnetlink "github.com/Azure/unbounded/internal/net/netlink" + netstatus "github.com/Azure/unbounded/internal/net/status" statusproto "github.com/Azure/unbounded/internal/net/status/proto" ) @@ -45,7 +46,7 @@ func TestSummaryPeerHealthy(t *testing.T) { } { t.Run(tc.name, func(t *testing.T) { peer := WireGuardPeerStatus{HealthCheck: tc.health, Tunnel: PeerTunnelStatus{LastHandshake: tc.handshake}} - if got := summaryPeerHealthy(peer, now); got != tc.want { + if got := netstatus.PeerHealthyForOverview(&peer, now); got != tc.want { t.Fatalf("healthy=%v, want %v", got, tc.want) } }) @@ -118,15 +119,9 @@ func TestNodeSummaryParityAndNoBPF(t *testing.T) { t.Fatalf("metadata/errors differ: summary=%+v full=%+v", summary, full) } - wantHealthy := 0 - - for _, peer := range full.Peers { - if summaryPeerHealthy(peer, time.Now()) { - wantHealthy++ - } - } - - if summary.PeerCount != len(full.Peers) || summary.HealthyPeers != wantHealthy || summary.RouteCount != len(full.RoutingTable.Routes) { + legacy := netstatus.OverviewFromStatus(full, time.Now()) + if summary.PeerCount != legacy.PeerCount || summary.HealthyPeers != legacy.HealthyPeers || + summary.RouteCount != legacy.RouteCount || summary.RouteMismatch != legacy.RouteMismatch { t.Fatalf("counts differ: summary=%+v full peers=%+v routes=%+v", summary, full.Peers, full.RoutingTable) }