From 8f6cce5c5edb9180ab4c37a1426ac8e253a51039 Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Thu, 17 Sep 2026 14:03:28 +0000 Subject: [PATCH] net node: collect lightweight summaries with exact diagnostic parity Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- cmd/unbounded-net-node/status_proto.go | 28 ++ .../status_publishing_bootstrap_test.go | 45 +++ cmd/unbounded-net-node/status_server.go | 213 +++++++----- cmd/unbounded-net-node/status_summary.go | 110 ++++++ .../status_summary_routes.go | 123 +++++++ .../status_summary_routes_test.go | 116 +++++++ cmd/unbounded-net-node/status_summary_test.go | 319 ++++++++++++++++++ 7 files changed, 873 insertions(+), 81 deletions(-) create mode 100644 cmd/unbounded-net-node/status_summary.go create mode 100644 cmd/unbounded-net-node/status_summary_routes.go create mode 100644 cmd/unbounded-net-node/status_summary_routes_test.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..7d2992bc3 100644 --- a/cmd/unbounded-net-node/status_proto.go +++ b/cmd/unbounded-net-node/status_proto.go @@ -18,6 +18,34 @@ 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, + RouteMismatchCount: int32(summary.RouteMismatchCount), + UnhealthyPeerLinks: int32(summary.UnhealthyPeerLinks), + UsesIpip: summary.UsesIPIP, + } + 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_publishing_bootstrap_test.go b/cmd/unbounded-net-node/status_publishing_bootstrap_test.go index b2320b137..975622cf1 100644 --- a/cmd/unbounded-net-node/status_publishing_bootstrap_test.go +++ b/cmd/unbounded-net-node/status_publishing_bootstrap_test.go @@ -6,6 +6,7 @@ package main import ( "compress/gzip" "context" + "encoding/json" "io" "net/http" "net/http/httptest" @@ -17,6 +18,7 @@ import ( "github.com/coder/websocket" "google.golang.org/protobuf/proto" + configpkg "github.com/Azure/unbounded/internal/net/config" statusproto "github.com/Azure/unbounded/internal/net/status/proto" ) @@ -302,6 +304,7 @@ func TestStatusPublishersHonorDisabledTogglesAndJoin(t *testing.T) { cfg := &config{ NodeName: "node-a", + StatusDetailMode: configpkg.DefaultStatusDetailMode, StatusPushEnabled: false, StatusPushURL: server.URL, StatusPushInterval: time.Millisecond, @@ -311,8 +314,50 @@ func TestStatusPublishersHonorDisabledTogglesAndJoin(t *testing.T) { } health := blockedBootstrapHealthState() + status := summaryRouteFixture() + + var bpfCollections atomic.Int32 + + status.bpfCollector = func() []BpfEntry { + bpfCollections.Add(1) + + return []BpfEntry{{}} + } + health.setStatusServer(status) startStatusPublishers(context.Background(), cfg, health) + local := httptest.NewServer(newHealthMux(health)) + defer local.Close() + + for _, path := range []string{"/status/summary", "/status/json"} { + request, err := http.NewRequestWithContext(t.Context(), http.MethodGet, local.URL+path, nil) + if err != nil { + t.Fatal(err) + } + + response, err := local.Client().Do(request) + if err != nil { + t.Fatal(err) + } + + var snapshot NodeStatusResponse + + err = json.NewDecoder(response.Body).Decode(&snapshot) + _ = response.Body.Close() + + if err != nil || response.StatusCode != http.StatusOK || snapshot.NodeInfo.Name != "local" { + t.Fatalf("disabled publishers prevented local diagnostics: code=%d err=%v", response.StatusCode, err) + } + + if path == "/status/summary" { + if len(snapshot.BpfEntries) != 0 || bpfCollections.Load() != 0 { + t.Fatal("local summary collected BPF details") + } + } else if len(snapshot.BpfEntries) != 1 || bpfCollections.Load() != 1 { + t.Fatal("explicit HTTP pull did not collect full details") + } + } + done := make(chan struct{}) go func() { diff --git a/cmd/unbounded-net-node/status_server.go b/cmd/unbounded-net-node/status_server.go index 08a453828..d05ccd005 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 { @@ -2373,6 +2381,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 @@ -2471,8 +2491,16 @@ func (s *nodeStatusServer) startRouteChangeWatcher(ctx context.Context) { }() } -// getNodeStatus collects all status information about this node -func (s *nodeStatusServer) getNodeStatus() *NodeStatusResponse { +type nodeStatusFacts struct { + Timestamp time.Time + NodeInfo NodeInfo + NodeErrors []NodeError + HealthCheck *HealthCheckStatus +} + +// inspectNodePeers visits peers without retaining an outbound peer array. +// The visitor must not acquire state.mu: non-WireGuard peers are visited under it. +func (s *nodeStatusServer) inspectNodePeers(visit func(WireGuardPeerStatus)) *nodeStatusFacts { // Snapshot state under the lock - copy all fields we need, then release. // Expensive operations (WireGuard GetDevice, collectRoutingTable) happen outside the lock. lockStart := time.Now() @@ -2480,7 +2508,7 @@ func (s *nodeStatusServer) getNodeStatus() *NodeStatusResponse { s.state.mu.Lock() lockWait := time.Since(lockStart) - status := &NodeStatusResponse{ + status := &nodeStatusFacts{ Timestamp: time.Now(), NodeInfo: NodeInfo{ Name: s.cfg.NodeName, @@ -2624,7 +2652,7 @@ func (s *nodeStatusServer) getNodeStatus() *NodeStatusResponse { // 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) @@ -2690,7 +2718,7 @@ func (s *nodeStatusServer) getNodeStatus() *NodeStatusResponse { } } - status.Peers = append(status.Peers, peer) + visit(peer) } } } @@ -2701,7 +2729,7 @@ func (s *nodeStatusServer) getNodeStatus() *NodeStatusResponse { 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{ @@ -2745,7 +2773,8 @@ func (s *nodeStatusServer) getNodeStatus() *NodeStatusResponse { } } - status.Peers = append(status.Peers, peer) + visit(peer) + addedPeerNames[gw.gatewayName] = true } } @@ -2809,7 +2838,7 @@ func (s *nodeStatusServer) getNodeStatus() *NodeStatusResponse { } } - status.Peers = append(status.Peers, peer) + visit(peer) } for _, gp := range s.state.gatewayPeers { @@ -2864,10 +2893,35 @@ func (s *nodeStatusServer) getNodeStatus() *NodeStatusResponse { } } - status.Peers = append(status.Peers, peer) + visit(peer) } s.state.mu.Unlock() + expensiveDuration := time.Since(expensiveStart) + + totalDuration := time.Since(lockStart) + if totalDuration > 2*time.Second { + klog.Warningf("inspectNodePeers() slow: total=%v (lock_wait=%v, snapshot=%v, expensive=%v)", + totalDuration, lockWait, snapshotDuration-lockWait, expensiveDuration) + } else { + klog.V(4).Infof("inspectNodePeers() timing: total=%v (lock_wait=%v, snapshot=%v, expensive=%v)", + totalDuration, lockWait, snapshotDuration-lockWait, expensiveDuration) + } + + return status +} + +// getNodeStatus collects all status information about this node. +func (s *nodeStatusServer) getNodeStatus() *NodeStatusResponse { + status := &NodeStatusResponse{} + facts := s.inspectNodePeers(func(peer WireGuardPeerStatus) { + status.Peers = append(status.Peers, peer) + }) + status.Timestamp = facts.Timestamp + status.NodeInfo = facts.NodeInfo + status.NodeErrors = facts.NodeErrors + status.HealthCheck = facts.HealthCheck + sortStatusPeers(status.Peers) // Collect routing table from kernel via netlink @@ -2891,17 +2945,10 @@ func (s *nodeStatusServer) getNodeStatus() *NodeStatusResponse { } // Collect BPF trie entries. - status.BpfEntries = s.collectBpfEntries() - - expensiveDuration := time.Since(expensiveStart) - - totalDuration := time.Since(lockStart) - if totalDuration > 2*time.Second { - klog.Warningf("getNodeStatus() slow: total=%v (lock_wait=%v, snapshot=%v, expensive=%v)", - totalDuration, lockWait, snapshotDuration-lockWait, expensiveDuration) + if s.bpfCollector != nil { + status.BpfEntries = s.bpfCollector() } else { - klog.V(4).Infof("getNodeStatus() timing: total=%v (lock_wait=%v, snapshot=%v, expensive=%v)", - totalDuration, lockWait, snapshotDuration-lockWait, expensiveDuration) + status.BpfEntries = s.collectBpfEntries() } return status @@ -2925,26 +2972,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) { @@ -3039,6 +3092,37 @@ func (s *nodeStatusServer) collectRoutingTableFromKernel() RoutingTableInfo { s.routingTableCacheMu.RUnlock() + s.inspectKernelRoutes(func(family, destination string, table int, hops []observedNextHop) { + entry := RouteEntry{Destination: destination, Family: family, Table: table} + for _, hop := range hops { + entry.NextHops = append(entry.NextHops, NextHop{ + Gateway: hop.gateway, Device: hop.device, Distance: hop.distance, MTU: hop.mtu, + RouteTypes: []RouteType{{Type: "kernel", Attributes: []string{"fib"}}}, + }) + } + + info.Routes = append(info.Routes, entry) + }) + + s.routingTableCacheMu.Lock() + s.routingTableCache = info + s.routingTableCachedAt = time.Now() + s.routingTableCacheMu.Unlock() + s.routingTableDirty.Store(false) + + return info +} + +type observedNextHop struct { + gateway string + device string + distance int + mtu int +} + +// inspectKernelRoutes shares filtering and deduplication without constructing +// status routes, annotations, or a full-detail routing cache. +func (s *nodeStatusServer) inspectKernelRoutes(visit func(family, destination string, table int, hops []observedNextHop)) { // Build a set of managed route prefixes from the route manager so we can // include routes on non-tunnel interfaces (e.g. eth0 with tunnelProtocol: None). managedPrefixes := make(map[string]bool) @@ -3049,7 +3133,7 @@ func (s *nodeStatusServer) collectRoutingTableFromKernel() RoutingTableInfo { } } - collect := func(family int, familyLabel string) []RouteEntry { + collect := func(family int, familyLabel string) { // Collect routes from the main table and, if configured, our dedicated table. // RouteList(nil, family) only returns routes from the main table, so we // explicitly request routes from our dedicated table via RouteListFiltered. @@ -3105,7 +3189,7 @@ func (s *nodeStatusServer) collectRoutingTableFromKernel() RoutingTableInfo { type destEntry struct { destination string table int - nexthops map[nhKey]NextHop + nexthops map[nhKey]observedNextHop nhOrder []nhKey } @@ -3177,7 +3261,7 @@ func (s *nodeStatusServer) collectRoutingTableFromKernel() RoutingTableInfo { de, exists := destMap[mapKey] if !exists { - de = &destEntry{destination: prefix, table: table, nexthops: make(map[nhKey]NextHop)} + de = &destEntry{destination: prefix, table: table, nexthops: make(map[nhKey]observedNextHop)} destMap[mapKey] = de destOrder = append(destOrder, mapKey) } @@ -3185,12 +3269,11 @@ func (s *nodeStatusServer) collectRoutingTableFromKernel() RoutingTableInfo { for _, wh := range wgHops { nk := nhKey{gateway: wh.gwStr, device: wh.devName} if _, nhExists := de.nexthops[nk]; !nhExists { - nh := NextHop{ - Gateway: wh.gwStr, Device: wh.devName, Distance: r.Priority, - RouteTypes: []RouteType{{Type: "kernel", Attributes: []string{"fib"}}}, + nh := observedNextHop{ + gateway: wh.gwStr, device: wh.devName, distance: r.Priority, } if r.MTU > 0 { - nh.MTU = r.MTU + nh.mtu = r.MTU } de.nexthops[nk] = nh @@ -3234,7 +3317,7 @@ func (s *nodeStatusServer) collectRoutingTableFromKernel() RoutingTableInfo { de, exists := destMap[mapKey] if !exists { - de = &destEntry{destination: prefix, table: table, nexthops: make(map[nhKey]NextHop)} + de = &destEntry{destination: prefix, table: table, nexthops: make(map[nhKey]observedNextHop)} destMap[mapKey] = de destOrder = append(destOrder, mapKey) } @@ -3246,17 +3329,11 @@ func (s *nodeStatusServer) collectRoutingTableFromKernel() RoutingTableInfo { nk := nhKey{gateway: gwStr, device: devName} if _, nhExists := de.nexthops[nk]; !nhExists { - nh := NextHop{ - Gateway: gwStr, - Device: devName, - Distance: r.Priority, - RouteTypes: []RouteType{{ - Type: "kernel", - Attributes: []string{"fib"}, - }}, + nh := observedNextHop{ + gateway: gwStr, device: devName, distance: r.Priority, } if r.MTU > 0 { - nh.MTU = r.MTU + nh.mtu = r.MTU } de.nexthops[nk] = nh @@ -3264,46 +3341,20 @@ func (s *nodeStatusServer) collectRoutingTableFromKernel() RoutingTableInfo { } } - result := make([]RouteEntry, 0, len(destOrder)) for _, mapKey := range destOrder { de := destMap[mapKey] - nhs := make([]NextHop, 0, len(de.nhOrder)) + nhs := make([]observedNextHop, 0, len(de.nhOrder)) for _, nk := range de.nhOrder { nhs = append(nhs, de.nexthops[nk]) } - result = append(result, RouteEntry{ - Destination: de.destination, - Family: familyLabel, - Table: de.table, - NextHops: nhs, - }) + visit(familyLabel, de.destination, de.table, nhs) } - - return result - } - - v4Routes := collect(netlink.FAMILY_V4, "IPv4") - v6Routes := collect(netlink.FAMILY_V6, "IPv6") - - if v4Routes == nil { - v4Routes = []RouteEntry{} } - if v6Routes == nil { - v6Routes = []RouteEntry{} - } - - info.Routes = append(v4Routes, v6Routes...) - - s.routingTableCacheMu.Lock() - s.routingTableCache = info - s.routingTableCachedAt = time.Now() - s.routingTableCacheMu.Unlock() - s.routingTableDirty.Store(false) - - return info + collect(netlink.FAMILY_V4, "IPv4") + collect(netlink.FAMILY_V6, "IPv6") } // isManagedTunnelInterface returns true for the interfaces created by the diff --git a/cmd/unbounded-net-node/status_summary.go b/cmd/unbounded-net-node/status_summary.go new file mode 100644 index 000000000..7f999b7e9 --- /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" + netstatus "github.com/Azure/unbounded/internal/net/status" + 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 netstatus.PeerHealthyForOverview(&peer, now) { + summary.HealthyPeers++ + } + + if !netstatus.PeerLinkHealthyForDiagnostics(&peer, now) { + summary.UnhealthyPeerLinks++ + } + + if peer.Tunnel.Protocol == "IPIP" { + summary.UsesIPIP = true + } + + 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.RouteMismatchCount = s.collectRouteSummary(routePeers, facts.NodeInfo.SiteName) + summary.RouteMismatch = summary.RouteMismatchCount > 0 + + 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 +} + +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_routes.go b/cmd/unbounded-net-node/status_summary_routes.go new file mode 100644 index 000000000..dcf22d69b --- /dev/null +++ b/cmd/unbounded-net-node/status_summary_routes.go @@ -0,0 +1,123 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "strings" + + "k8s.io/klog/v2" + + "github.com/Azure/unbounded/internal/net/routeplan" +) + +type routeSummaryFamily struct { + expected map[routeKey]expectedRoute + lowest map[string]int + destinations map[string]bool + hasUnbounded bool +} + +func newRouteSummaryFamily(plan []routeplan.ExpectedRoute) *routeSummaryFamily { + family := &routeSummaryFamily{ + expected: make(map[routeKey]expectedRoute), + lowest: make(map[string]int), + destinations: make(map[string]bool), + } + + for _, route := range plan { + distance := effectiveRouteDistance(route.Distance) + key := routeKey{destination: route.Destination, gateway: route.Gateway, device: route.Device, distance: distance, weight: route.Weight} + + family.expected[key] = expectedRoute{ + destination: route.Destination, + nextHop: NextHop{Gateway: route.Gateway, Device: route.Device, Distance: distance, Weight: route.Weight}, + } + if previous, ok := family.lowest[route.Destination]; !ok || distance < previous { + family.lowest[route.Destination] = distance + } + } + + return family +} + +// collectRouteSummary preserves annotation counts, including synthetic missing +// routes and the per-family unbounded0 suppression of missing tunnel hops. +// It never creates status route arrays or fills the full-detail route cache. +func (s *nodeStatusServer) collectRouteSummary(peers []routeplan.Peer, localSite string) (count, mismatchCount int) { + defer func() { + if r := recover(); r != nil { + klog.Warningf("route summary recovered from panic: %v", r) + } + }() + + actx := buildAnnotationContext(s.siteInformer, s.sliceInformer, s.gatewayPoolInformer, s.sitePeeringInformer) + for i := range peers { + peers[i].SitePeered = sitesAreDirectlyPeered(strings.TrimSpace(localSite), peers[i].SiteName, actx.directSitePeerings) + } + + ipv4, ipv6 := routeplan.BuildExpectedWireGuardRoutes(peers, actx.routeNodes, routeplan.InterfaceNames{ + WireGuardPrefix: s.cfg.WireGuardInterfacePrefix, + Geneve: s.cfg.GeneveInterfaceName, + VXLAN: s.cfg.VXLANInterfaceName, + IPIP: s.cfg.IPIPInterfaceName, + }) + families := map[string]*routeSummaryFamily{ + "IPv4": newRouteSummaryFamily(ipv4), + "IPv6": newRouteSummaryFamily(ipv6), + } + + s.inspectKernelRoutes(func(familyName, destination string, _ int, hops []observedNextHop) { + family := families[familyName] + count++ + family.destinations[destination] = true + normalized, _ := normalizeRouteDestination(destination) + + for _, observed := range hops { + if observed.device == unbounded0DeviceName { + family.hasUnbounded = true + } + + if !isPeerRoutingInterface(s.cfg, observed.device) { + continue + } + + hop := NextHop{Gateway: observed.gateway, Device: observed.device, Distance: observed.distance} + + key, matched := findExpectedWireGuardMatch(family.expected, normalized, &hop) + if matched { + delete(family.expected, key) + } else { + // Kernel inspection only emits "kernel" route types, so the + // connected/local host-route exception cannot apply here. + mismatchCount++ + } + } + }) + + // The legacy collector does not annotate an entirely empty kernel result. + if count == 0 { + return count, mismatchCount + } + + for _, family := range families { + for _, expected := range family.expected { + if effectiveRouteDistance(expected.nextHop.Distance) > family.lowest[expected.destination] { + continue + } + + if family.hasUnbounded { + continue + } + + mismatchCount++ + + if !family.destinations[expected.destination] { + family.destinations[expected.destination] = true + count++ + } + } + } + + return count, mismatchCount +} diff --git a/cmd/unbounded-net-node/status_summary_routes_test.go b/cmd/unbounded-net-node/status_summary_routes_test.go new file mode 100644 index 000000000..b0ec7b7bf --- /dev/null +++ b/cmd/unbounded-net-node/status_summary_routes_test.go @@ -0,0 +1,116 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "net" + "testing" + + "github.com/vishvananda/netlink" + "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 { + _, prefix, _ := net.ParseCIDR(destination) + + return netlink.Route{Dst: prefix, LinkIndex: index, Table: table, Priority: distance, Protocol: unix.RTPROT_BOOT} +} + +func summaryRouteFixture() *nodeStatusServer { + return &nodeStatusServer{ + cfg: &config{ + NodeName: "local", WireGuardInterfacePrefix: "wg", WireGuardPort: 51820, + GeneveInterfaceName: "gn0", VXLANInterfaceName: "vx0", IPIPInterfaceName: "ip0", + }, + state: &wireGuardState{routeTableID: 100}, + netlinkOps: &fakeNetlinkOps{ + links: map[int]netlink.Link{ + 1: &fakeLink{attrs: netlink.LinkAttrs{Index: 1, Name: "wg51820"}}, + 2: &fakeLink{attrs: netlink.LinkAttrs{Index: 2, Name: unbounded0DeviceName}}, + 3: &fakeLink{attrs: netlink.LinkAttrs{Index: 3, Name: "eth0"}}, + 4: &fakeLink{attrs: netlink.LinkAttrs{Index: 4, Name: "gn0"}}, + }, + }, + } +} + +func TestRouteSummaryParity(t *testing.T) { + peer := WireGuardPeerStatus{ + Name: "peer", PeerType: "site", SiteName: "local", + PodCIDRGateways: []string{"10.42.1.1", "fd00:1::1"}, + Tunnel: PeerTunnelStatus{Interface: "wg51820", AllowedIPs: []string{"10.42.1.0/24", "fd00:1::/64"}}, + } + planPeer := routeplan.Peer{ + Name: peer.Name, PeerType: peer.PeerType, SiteName: peer.SiteName, + Interface: peer.Tunnel.Interface, AllowedIPs: peer.Tunnel.AllowedIPs, PodCIDRGateways: peer.PodCIDRGateways, + } + + for _, tc := range []struct { + name string + v4 []netlink.Route + v6 []netlink.Route + table []netlink.Route + withoutPeer bool + exactMismatchCount int + }{ + {name: "empty kernel does not synthesize"}, + {name: "missing expected", exactMismatchCount: 5, v4: []netlink.Route{summaryRoute("10.9.0.0/24", 1, 0, 0)}}, + {name: "matched", v4: []netlink.Route{summaryRoute("10.42.1.0/24", 1, 0, 0)}}, + {name: "unexpected without peers", withoutPeer: true, v4: []netlink.Route{summaryRoute("10.9.0.0/24", 1, 0, 0)}}, + {name: "unbounded suppresses only its family", exactMismatchCount: 2, v4: []netlink.Route{summaryRoute("10.42.0.0/16", 2, 0, 0)}}, + {name: "unbounded both families", v4: []netlink.Route{summaryRoute("10.42.0.0/16", 2, 0, 0)}, v6: []netlink.Route{summaryRoute("fd00::/48", 2, 0, 0)}}, + {name: "duplicate prefix distinct tables", v4: []netlink.Route{summaryRoute("10.42.1.0/24", 1, 0, 0)}, table: []netlink.Route{summaryRoute("10.42.1.0/24", 1, 100, 0)}}, + {name: "duplicate next hop", v4: []netlink.Route{summaryRoute("10.42.1.0/24", 1, 0, 0), summaryRoute("10.42.1.0/24", 1, 0, 100)}}, + {name: "wrong distance", v4: []netlink.Route{summaryRoute("10.42.1.0/24", 1, 0, 500)}}, + {name: "unmanaged ignored", v4: []netlink.Route{summaryRoute("10.42.1.0/24", 3, 0, 0)}}, + {name: "multipath", v4: []netlink.Route{{ + Dst: summaryRoute("10.42.1.0/24", 1, 0, 0).Dst, + MultiPath: []*netlink.NexthopInfo{{LinkIndex: 1}, {LinkIndex: 3}, {LinkIndex: 4}}, + }}}, + {name: "two unexpected tunnel hops", withoutPeer: true, exactMismatchCount: 2, v4: []netlink.Route{{ + Dst: summaryRoute("10.99.0.0/24", 1, 0, 0).Dst, + MultiPath: []*netlink.NexthopInfo{{LinkIndex: 1}, {LinkIndex: 4}}, + }}}, + } { + t.Run(tc.name, func(t *testing.T) { + s := summaryRouteFixture() + ops := s.netlinkOps.(*fakeNetlinkOps) + ops.mainRoutes = map[int][]netlink.Route{netlink.FAMILY_V4: tc.v4, netlink.FAMILY_V6: tc.v6} + ops.tableRoutes = map[int]map[int][]netlink.Route{netlink.FAMILY_V4: {100: tc.table}} + full := &NodeStatusResponse{NodeInfo: NodeInfo{SiteName: "local"}, RoutingTable: s.collectRoutingTableFromKernel()} + + var peers []routeplan.Peer + + if !tc.withoutPeer { + full.Peers = []WireGuardPeerStatus{peer} + peers = []routeplan.Peer{planPeer} + } + + annotateNodeRoutes(full, s.cfg, nil, nil, nil, nil) + + wantMismatch := false + + for _, route := range full.RoutingTable.Routes { + if netstatus.RouteMismatchForOverview(route) { + wantMismatch = true + } + } + + count, mismatchCount := s.collectRouteSummary(peers, "local") + + mismatch := mismatchCount > 0 + if count != len(full.RoutingTable.Routes) || mismatch != wantMismatch { + t.Fatalf("summary=(%d,%v), legacy=(%d,%v): %+v", count, mismatch, len(full.RoutingTable.Routes), wantMismatch, full.RoutingTable.Routes) + } + + wantMismatchCount := netstatus.RouteMismatchCount(full.RoutingTable.Routes) + if mismatchCount != wantMismatchCount || (tc.exactMismatchCount > 0 && mismatchCount != tc.exactMismatchCount) { + t.Fatalf("mismatched hops=%d, legacy=%d, explicit=%d: %+v", mismatchCount, wantMismatchCount, tc.exactMismatchCount, full.RoutingTable.Routes) + } + }) + } +} 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..1c0562aca --- /dev/null +++ b/cmd/unbounded-net-node/status_summary_test.go @@ -0,0 +1,319 @@ +// 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" + netstatus "github.com/Azure/unbounded/internal/net/status" + 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 := netstatus.PeerHealthyForOverview(&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) + } + + legacy := netstatus.OverviewFromStatus(full, time.Now()) + if summary.PeerCount != legacy.PeerCount || summary.HealthyPeers != legacy.HealthyPeers || + summary.RouteCount != legacy.RouteCount || summary.RouteMismatch != legacy.RouteMismatch || + summary.RouteMismatchCount != legacy.RouteMismatchCount || summary.UnhealthyPeerLinks != legacy.UnhealthyPeerLinks || + summary.UsesIPIP != legacy.UsesIPIP { + 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, + RouteMismatchCount: 3, UnhealthyPeerLinks: 2, UsesIPIP: 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.RouteMismatchCount != 3 || decoded.UnhealthyPeerLinks != 2 || !decoded.UsesIpip || + !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() + } + } + }) + } +}