From 0e9a99b375212cb58f2d654575a82d737e035951 Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Thu, 17 Sep 2026 14:00:37 +0000 Subject: [PATCH] net controller: transport summaries and correlated diagnostics over WebSocket and HTTP Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- .../cluster_status.go | 24 ++ .../cluster_status_cache.go | 34 +- .../detail_dispatch.go | 101 ++++++ .../detail_dispatch_test.go | 206 +++++++++++ .../detail_lifecycle.go | 1 + .../detail_requests.go | 22 +- .../detail_responses.go | 80 ++++ .../detail_transport_test.go | 342 ++++++++++++++++++ cmd/unbounded-net-controller/health_state.go | 41 +-- cmd/unbounded-net-controller/node_overview.go | 61 ++++ .../node_overview_test.go | 198 ++++++++++ cmd/unbounded-net-controller/node_status.go | 35 +- cmd/unbounded-net-controller/server.go | 210 ++++++++--- .../status_overview_proto.go | 38 ++ cmd/unbounded-net-controller/status_proto.go | 92 ++++- cmd/unbounded-net-controller/status_types.go | 124 +++---- .../ws_source_ownership_test.go | 35 ++ 17 files changed, 1478 insertions(+), 166 deletions(-) create mode 100644 cmd/unbounded-net-controller/detail_dispatch.go create mode 100644 cmd/unbounded-net-controller/detail_dispatch_test.go create mode 100644 cmd/unbounded-net-controller/detail_responses.go create mode 100644 cmd/unbounded-net-controller/detail_transport_test.go create mode 100644 cmd/unbounded-net-controller/node_overview.go create mode 100644 cmd/unbounded-net-controller/node_overview_test.go create mode 100644 cmd/unbounded-net-controller/status_overview_proto.go create mode 100644 cmd/unbounded-net-controller/ws_source_ownership_test.go diff --git a/cmd/unbounded-net-controller/cluster_status.go b/cmd/unbounded-net-controller/cluster_status.go index 6185da8c8..56beaff1d 100644 --- a/cmd/unbounded-net-controller/cluster_status.go +++ b/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)) } diff --git a/cmd/unbounded-net-controller/cluster_status_cache.go b/cmd/unbounded-net-controller/cluster_status_cache.go index 6a9b5d13e..33437a4a8 100644 --- a/cmd/unbounded-net-controller/cluster_status_cache.go +++ b/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. diff --git a/cmd/unbounded-net-controller/detail_dispatch.go b/cmd/unbounded-net-controller/detail_dispatch.go new file mode 100644 index 000000000..659ff45e2 --- /dev/null +++ b/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) + } + }) +} diff --git a/cmd/unbounded-net-controller/detail_dispatch_test.go b/cmd/unbounded-net-controller/detail_dispatch_test.go new file mode 100644 index 000000000..8c75514f6 --- /dev/null +++ b/cmd/unbounded-net-controller/detail_dispatch_test.go @@ -0,0 +1,206 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func TestDetailDispatchUsesOnlyCurrentCapableConnection(t *testing.T) { + health := &healthState{} + + var canceled, sent atomic.Int32 + + cancel := func() { canceled.Add(1) } + old := health.registerNodeWS("node", cancel) + current := health.registerNodeWS("node", cancel) + health.unregisterNodeWS("node", old) + + if canceled.Load() != 1 { + t.Fatal("replaced connection was not canceled") + } + + command := statusv1alpha1.DetailRequest{RequestID: "request", Deadline: time.Now().Add(time.Minute)} + if ok, err := health.dispatchNodeDetail(t.Context(), "node", command); ok || err != nil { + t.Fatal("legacy connection was treated as command-capable") + } + + health.setNodeWSDetailSender("node", old, func(context.Context, statusv1alpha1.DetailRequest) error { + t.Error("an obsolete connection sent a command") + return nil + }) + health.setNodeWSDetailSender("node", current, func(_ context.Context, got statusv1alpha1.DetailRequest) error { + if got != command { + t.Error("command identity or deadline changed") + } + + sent.Add(1) + + return nil + }) + + if ok, err := health.dispatchNodeDetail(t.Context(), "node", command); !ok || err != nil || sent.Load() != 1 { + t.Fatalf("current connection did not receive command: %v %v", ok, err) + } + + health.unregisterNodeWS("node", current) + + if ok, _ := health.dispatchNodeDetail(t.Context(), "node", command); ok { + t.Fatal("closed connection remained usable") + } +} + +func TestDetailDispatchDisconnectAndReconnectPreserveDeadline(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + health := &healthState{} + commands := make(chan statusv1alpha1.DetailRequest, 3) + sender := func(_ context.Context, command statusv1alpha1.DetailRequest) error { + select { + case commands <- command: + return nil + default: + return errors.New("unexpected extra dispatch") + } + } + + var pulls atomic.Int32 + + manager := testDetailRequests(t, nodeDetailRequestHooks{ + Dispatch: health.dispatchNodeDetail, + Pull: func(context.Context, string) (*NodeStatusResponse, error) { + pulls.Add(1) + return nil, errors.New("unreachable") + }, + }) + health.detailRequests = manager + connection := health.registerNodeWS("node", func() {}) + health.setNodeWSDetailSender("node", connection, sender) + + request := manager.Request("node", true) + + synctest.Wait() + + if len(commands) != 1 { + t.Fatal("expected one WebSocket command") + } + + first := <-commands + if pulls.Load() != 0 || first.RequestID != request.RequestID { + t.Fatal("active WebSocket did not take priority") + } + + time.Sleep(time.Second) + health.unregisterNodeWS("node", connection) + synctest.Wait() + + pending, ok := manager.Pending("node") + if !ok || pulls.Load() != 1 || pending != first { + t.Fatal("disconnect failed to pull and retain the original POST command") + } + + reconnected := health.registerNodeWS("node", func() {}) + health.setNodeWSDetailSender("node", reconnected, sender) + synctest.Wait() + + if len(commands) != 1 { + t.Fatal("expected one command on reconnect") + } + + if next := <-commands; next != first { + t.Fatal("reconnect reset request identity or deadline") + } + + health.setNodeWSDetailSender("node", reconnected, sender) + synctest.Wait() + + if len(commands) != 0 { + t.Fatal("a routine capability update dispatched another collection") + } + + if err := manager.Complete("node", first.RequestID, testDetailStatus()); err != nil { + t.Fatal(err) + } + + health.unregisterNodeWS("node", reconnected) + synctest.Wait() + + if pulls.Load() != 1 { + t.Fatal("a completed request restarted after disconnect") + } + }) +} + +func TestDetailDispatchRetryCoalescesAndCancels(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var active, peak atomic.Int32 + + manager := testDetailRequests(t, nodeDetailRequestHooks{ + Dispatch: func(ctx context.Context, _ string, _ statusv1alpha1.DetailRequest) (bool, error) { + n := active.Add(1) + if n > peak.Load() { + peak.Store(n) + } + + defer active.Add(-1) + + <-ctx.Done() + + return false, ctx.Err() + }, + }) + manager.Request("node", true) + synctest.Wait() + + for range 10 { + manager.Retry("node") + } + + synctest.Wait() + manager.Close() + + if peak.Load() != 1 || active.Load() != 0 { + t.Fatal("retry created concurrent dispatches or shutdown left one running") + } + }) +} + +func TestDetailDispatchWriteTimeoutFallsBackWithinOriginalDeadline(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + health := &healthState{} + connection := health.registerNodeWS("node", func() {}) + health.setNodeWSDetailSender("node", connection, func(ctx context.Context, _ statusv1alpha1.DetailRequest) error { + <-ctx.Done() + return ctx.Err() + }) + + var pulls atomic.Int32 + + manager := testDetailRequests(t, nodeDetailRequestHooks{ + Dispatch: health.dispatchNodeDetail, + Pull: func(context.Context, string) (*NodeStatusResponse, error) { + pulls.Add(1) + return testDetailStatus(), nil + }, + }) + manager.timeout = 20 * time.Second + health.detailRequests = manager + request := manager.Request("node", true) + + synctest.Wait() + time.Sleep(5 * time.Second) + synctest.Wait() + + result := manager.Result("node", request.RequestID) + if pulls.Load() != 1 || result.State != statusv1alpha1.NodeDetailComplete || !result.Deadline.Equal(request.Deadline) { + t.Fatalf("failed WebSocket write did not fall back within the original deadline: %+v", result) + } + }) +} diff --git a/cmd/unbounded-net-controller/detail_lifecycle.go b/cmd/unbounded-net-controller/detail_lifecycle.go index 3da73cdb8..ce2f49b9f 100644 --- a/cmd/unbounded-net-controller/detail_lifecycle.go +++ b/cmd/unbounded-net-controller/detail_lifecycle.go @@ -50,6 +50,7 @@ func (h *healthState) startDetailRequests(ctx context.Context, nodeInformer cach lister := corev1listers.NewNodeLister(nodeInformer.GetIndexer()) manager, err := newNodeDetailRequests(ctx, detailCache, h.statusDetailRequestTimeout, nodeDetailRequestHooks{ + Dispatch: h.dispatchNodeDetail, Resolve: func(name string) (types.UID, error) { node, err := lister.Get(name) if err != nil { diff --git a/cmd/unbounded-net-controller/detail_requests.go b/cmd/unbounded-net-controller/detail_requests.go index e2cc5bcac..0477b9a81 100644 --- a/cmd/unbounded-net-controller/detail_requests.go +++ b/cmd/unbounded-net-controller/detail_requests.go @@ -26,14 +26,17 @@ type nodeDetailRequestHooks struct { // A request owns metadata and cancellation only, never a result payload. type nodeDetailRequest struct { - nodeName string - uid types.UID - command statusv1alpha1.DetailRequest - state statusv1alpha1.NodeDetailState - message string - wakeAt time.Time - poll bool - cancel context.CancelFunc + nodeName string + uid types.UID + command statusv1alpha1.DetailRequest + state statusv1alpha1.NodeDetailState + message string + wakeAt time.Time + poll bool + cancel context.CancelFunc + ctx context.Context + dispatching bool + retry bool } type nodeDetailRequests struct { @@ -118,10 +121,11 @@ func (m *nodeDetailRequests) Request(nodeName string, forceRefresh bool) statusv } ctx, cancel := context.WithDeadline(m.ctx, request.command.Deadline) request.cancel = cancel + request.ctx = ctx m.requests[request.command.RequestID] = request m.active[nodeName] = request m.notify() - m.workers.Go(func() { m.dispatch(ctx, nodeName, request.command) }) + m.startDispatchLocked(request) return m.resultLocked(request) } diff --git a/cmd/unbounded-net-controller/detail_responses.go b/cmd/unbounded-net-controller/detail_responses.go new file mode 100644 index 000000000..9b4e3a8a7 --- /dev/null +++ b/cmd/unbounded-net-controller/detail_responses.go @@ -0,0 +1,80 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "errors" + "time" + + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +// Fail records a correlated collection failure without deleting an older, +// still-valid snapshot that a viewer may display during a failed refresh. +func (m *nodeDetailRequests) Fail(nodeName, requestID, reason string) error { + m.mu.Lock() + defer m.mu.Unlock() + + m.expireLocked(time.Now()) + + request := m.requests[requestID] + if reason == "" || m.ctx.Err() != nil || m.closed || request == nil || request.nodeName != nodeName { + return errors.New("detail failure does not match an available request") + } + + if uid, err := m.hooks.Resolve(nodeName); err != nil || uid != request.uid { + m.invalidateLocked(request) + return errors.New("detail request node was deleted or replaced") + } + + if request.state == statusv1alpha1.NodeDetailComplete || + (request.state == statusv1alpha1.NodeDetailUnavailable && request.message == reason) { + return nil + } + + if request.state != statusv1alpha1.NodeDetailPending { + return errors.New("detail request is no longer pending") + } + + request.cancel() + request.state = statusv1alpha1.NodeDetailUnavailable + request.message = reason + request.poll = false + request.wakeAt = time.Now().Add(m.timeout) + delete(m.active, nodeName) + m.notify() + + return nil +} + +func handleNodeDetailResponse(health *healthState, nodeName, requestID string, status *NodeStatusResponse, failure string) NodeStatusPushAck { + ack := NodeStatusPushAck{Status: "error", DetailRequestID: requestID, SummarySupported: true} + + manager := health.getDetailRequests() + if manager == nil || requestID == "" { + ack.Reason = "detail request leader or request identity is unavailable" + return ack + } + + if failure != "" && status != nil { + ack.Reason = "detail response cannot contain both data and a collection error" + return ack + } + + var err error + if failure != "" { + err = manager.Fail(nodeName, requestID, failure) + } else { + err = manager.Complete(nodeName, requestID, status) + } + + if err != nil { + ack.Reason = err.Error() + return ack + } + + ack.Status = "ok" + + return ack +} diff --git a/cmd/unbounded-net-controller/detail_transport_test.go b/cmd/unbounded-net-controller/detail_transport_test.go new file mode 100644 index 000000000..205dddbf6 --- /dev/null +++ b/cmd/unbounded-net-controller/detail_transport_test.go @@ -0,0 +1,342 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "testing/synctest" + "time" + + "github.com/coder/websocket" + "google.golang.org/protobuf/proto" + + statuspkg "github.com/Azure/unbounded/internal/net/status" + statusproto "github.com/Azure/unbounded/internal/net/status/proto" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func encodeDetailTransportMessage(t *testing.T, binary bool, requestID string, failure ...string) []byte { + t.Helper() + + message := &statusproto.NodeStatusMessage{ + Type: statusv1alpha1.NodeStatusSummaryType, NodeName: "node-a", SupportsDetails: true, + Summary: &statusproto.NodeStatusOverview{NodeInfo: &statusproto.NodeInfo{Name: "node-a"}, PeerCount: 1}, + } + jsonMessage := NodeStatusWSMessage{Type: message.Type, NodeName: message.NodeName, SupportsDetails: true} + overview := protoToNodeOverview(message.Summary) + jsonMessage.Summary = &overview + + if requestID != "" { + message.Type = statusv1alpha1.NodeStatusDetailsType + message.Summary = nil + message.DetailRequestId = requestID + message.Status = &statusproto.NodeStatusFull{ + NodeInfo: &statusproto.NodeInfo{Name: "node-a"}, + Peers: []*statusproto.PeerStatus{{Name: "peer"}}, + } + full := protoToNodeStatus(message.Status) + jsonMessage.Type = message.Type + jsonMessage.Summary = nil + jsonMessage.Status = &full + jsonMessage.DetailRequestID = requestID + } + + if len(failure) > 0 { + message.DetailError = failure[0] + message.Status = nil + jsonMessage.DetailError = failure[0] + jsonMessage.Status = nil + } + + var ( + data []byte + err error + ) + if binary { + data, err = proto.Marshal(message) + } else { + data, err = json.Marshal(jsonMessage) + } + + if err != nil { + t.Fatal(err) + } + + return data +} + +func decodeDetailTransportAck(t *testing.T, binary, ws bool, data []byte) *NodeStatusPushAck { + t.Helper() + + if binary { + var ack statusproto.NodeStatusAck + if err := proto.Unmarshal(data, &ack); err != nil { + t.Fatal(err) + } + + return statuspkg.NodeStatusAckFromProto(&ack) + } + + var ack NodeStatusPushAck + if ws { + var envelope struct { + Data NodeStatusPushAck `json:"data"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + t.Fatal(err) + } + + ack = envelope.Data + } else if err := json.Unmarshal(data, &ack); err != nil { + t.Fatal(err) + } + + return &ack +} + +func awaitDetailTransport(t *testing.T, ready func() bool) { + t.Helper() + + timeout := time.NewTimer(3 * time.Second) + defer timeout.Stop() + + tick := time.NewTicker(time.Millisecond) + defer tick.Stop() + + for !ready() { + select { + case <-timeout.C: + t.Fatal("detail transport did not become ready") + case <-tick.C: + } + } +} + +func TestDetailWebSocketCommandAndResponse(t *testing.T) { + for _, binary := range []bool{false, true} { + t.Run(map[bool]string{false: "json", true: "protobuf"}[binary], func(t *testing.T) { + health := newJSONIdentityHealth() + health.registerAggregatedAPIServer = false + manager := testDetailRequests(t, nodeDetailRequestHooks{ + Dispatch: health.dispatchNodeDetail, + Pull: func(context.Context, string) (*NodeStatusResponse, error) { + t.Error("active WebSocket request unexpectedly used HTTP") + return nil, nil + }, + }) + health.detailRequests = manager + issuer := testTokenIssuer(t) + mux := http.NewServeMux() + registerPushHandlers(mux, health, nil, make(chan struct{}, 1), issuer) + + server := httptest.NewServer(mux) + defer server.Close() + + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + conn, _, err := websocket.Dial(ctx, server.URL+"/status/nodews", &websocket.DialOptions{ + HTTPHeader: http.Header{"Authorization": {"Bearer " + testNodeToken(t, issuer)}}, + }) + if err != nil { + t.Fatal(err) + } + defer conn.CloseNow() + + frameType := websocket.MessageText + if binary { + frameType = websocket.MessageBinary + } + + send := func(id string) { + t.Helper() + + if err := conn.Write(ctx, frameType, encodeDetailTransportMessage(t, binary, id)); err != nil { + t.Fatal(err) + } + } + read := func() *NodeStatusPushAck { + t.Helper() + + _, data, err := conn.Read(ctx) + if err != nil { + t.Fatal(err) + } + + return decodeDetailTransportAck(t, binary, true, data) + } + + send("") + + publication := read() + + awaitDetailTransport(t, func() bool { + health.nodeWSMu.Lock() + defer health.nodeWSMu.Unlock() + + return health.nodeWSRegistry["node-a"] != nil && health.nodeWSRegistry["node-a"].send != nil + }) + + request := manager.Request("node-a", true) + + command := read() + if command.IsPublicationAck() || command.DetailRequest == nil || + command.DetailRequest.RequestID != request.RequestID || !command.SummarySupported { + t.Fatalf("wire command was not distinct from an ordinary ACK: %+v", command) + } + + send(request.RequestID) + + ack := read() + if ack.Status != "ok" || ack.DetailRequestID != request.RequestID || ack.Revision != 0 || ack.IsPublicationAck() { + t.Fatalf("invalid correlated ACK: %+v", ack) + } + + result := manager.Result("node-a", request.RequestID) + if result.Details == nil || len(result.Details.Status.Peers) != 1 { + t.Fatal("one-shot details did not complete the request") + } + + cached, _ := health.statusCache.Get("node-a") + if cached.Revision != publication.Revision || cached.Status.Peers != nil { + t.Fatal("one-shot reply became the routine delta base") + } + + expires := result.Details.ExpiresAt + + send(request.RequestID) + + if duplicate := read(); duplicate.Status != "ok" { + t.Fatal("duplicate detail reply was not idempotent") + } + + if !manager.Result("node-a", request.RequestID).Details.ExpiresAt.Equal(expires) { + t.Fatal("duplicate detail reply renewed TTL") + } + + refresh := manager.Request("node-a", true) + if command := read(); command.DetailRequest == nil || command.DetailRequest.RequestID != refresh.RequestID { + t.Fatal("refresh command was not delivered") + } + + for range 2 { + if err := conn.Write(ctx, frameType, encodeDetailTransportMessage(t, binary, refresh.RequestID, "collection failed")); err != nil { + t.Fatal(err) + } + + if receipt := read(); receipt.Status != "ok" || receipt.DetailRequestID != refresh.RequestID { + t.Fatal("collection error receipt was not idempotently acknowledged") + } + } + + failed := manager.Result("node-a", refresh.RequestID) + if failed.State != statusv1alpha1.NodeDetailUnavailable || failed.Error != "collection failed" || failed.Details != nil { + t.Fatal("collection error did not become an explicit failed request") + } + }) + } +} + +func TestDetailHTTPPollingCommandAndResponse(t *testing.T) { + for _, binary := range []bool{false, true} { + t.Run(map[bool]string{false: "json", true: "protobuf"}[binary], func(t *testing.T) { + health := newJSONIdentityHealth() + health.registerAggregatedAPIServer = false + manager := testDetailRequests(t, nodeDetailRequestHooks{}) + health.detailRequests = manager + issuer := testTokenIssuer(t) + token := testNodeToken(t, issuer) + mux := http.NewServeMux() + registerPushHandlers(mux, health, nil, make(chan struct{}, 1), issuer) + + request := manager.Request("node-a", true) + + awaitDetailTransport(t, func() bool { _, ok := manager.Pending("node-a"); return ok }) + + post := func(id string, failure ...string) *NodeStatusPushAck { + t.Helper() + r := httptest.NewRequest(http.MethodPost, "/status/push", bytes.NewReader(encodeDetailTransportMessage(t, binary, id, failure...))) + r.Header.Set("Authorization", "Bearer "+token) + + if binary { + r.Header.Set("Content-Type", "application/x-protobuf") + } + + w := httptest.NewRecorder() + mux.ServeHTTP(w, r) + + if w.Code != http.StatusOK { + t.Fatalf("POST failed: %d %s", w.Code, w.Body.String()) + } + + return decodeDetailTransportAck(t, binary, false, w.Body.Bytes()) + } + + publication := post("") + if !publication.SummarySupported || !publication.IsPublicationAck() || + publication.DetailRequest == nil || publication.DetailRequest.RequestID != request.RequestID { + t.Fatalf("POST ACK lost capabilities or polling command: %+v", publication) + } + + detailAck := post(request.RequestID) + if detailAck.Status != "ok" || detailAck.DetailRequestID != request.RequestID || detailAck.IsPublicationAck() || detailAck.DetailRequest != nil { + t.Fatalf("POST detail ACK corrupted publication/polling state: %+v", detailAck) + } + + cached, _ := health.statusCache.Get("node-a") + if cached.Revision != publication.Revision || manager.Result("node-a", request.RequestID).Details == nil { + t.Fatal("POST detail response changed routine state or failed to complete") + } + + refresh := manager.Request("node-a", true) + for range 2 { + receipt := post(refresh.RequestID, "response exceeds the transport frame limit") + if receipt.Status != "ok" || receipt.DetailRequestID != refresh.RequestID { + t.Fatal("POST collection failure receipt was not acknowledged") + } + } + + if failed := manager.Result("node-a", refresh.RequestID); failed.State != statusv1alpha1.NodeDetailUnavailable || failed.Error == "" { + t.Fatal("POST collection failure did not terminate the request") + } + }) + } +} + +func TestDetailFailureKeepsPreviousValidSnapshot(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + manager := testDetailRequests(t, nodeDetailRequestHooks{}) + + first := manager.Request("node", true) + if err := manager.Complete("node", first.RequestID, testDetailStatus()); err != nil { + t.Fatal(err) + } + + before, _ := manager.cache.Get("node") + + refresh := manager.Request("node", true) + if err := manager.Fail("other", refresh.RequestID, "bad"); err == nil { + t.Fatal("failure for a different node was accepted") + } + + if err := manager.Fail("node", refresh.RequestID, "response exceeds the transport frame limit"); err != nil { + t.Fatal(err) + } + + result := manager.Result("node", refresh.RequestID) + if result.State != statusv1alpha1.NodeDetailUnavailable || result.Error == "" || result.Details != nil { + t.Fatal("collection failure was not explicit") + } + + after, ok := manager.cache.Get("node") + if !ok || after.Status != before.Status || !after.ExpiresAt.Equal(before.ExpiresAt) { + t.Fatal("failed refresh destroyed or renewed the old valid snapshot") + } + }) +} diff --git a/cmd/unbounded-net-controller/health_state.go b/cmd/unbounded-net-controller/health_state.go index 1f09d89fd..9b20ca3ed 100644 --- a/cmd/unbounded-net-controller/health_state.go +++ b/cmd/unbounded-net-controller/health_state.go @@ -85,11 +85,11 @@ type healthState struct { // kubeProxyMonitor checks the local kube-proxy health endpoint. kubeProxyMonitor *kubeProxyMonitor - // nodeWSRegistry tracks the active WS cancel function per node name. + // nodeWSRegistry tracks the active authenticated connection per node name. // When a node reconnects, the previous connection is canceled to avoid // duplicate connections consuming resources. nodeWSMu sync.Mutex - nodeWSRegistry map[string]context.CancelFunc + nodeWSRegistry map[string]*nodeWSConnection } const defaultMaxPullConcurrency = 20 @@ -97,45 +97,44 @@ const defaultMaxPullConcurrency = 20 // registerNodeWS registers a WS connection for a node. If an existing // connection is registered for the same node, its context is canceled // to force it to close (preventing duplicate connections). -func (h *healthState) registerNodeWS(nodeName string, cancel context.CancelFunc) { +func (h *healthState) registerNodeWS(nodeName string, cancel context.CancelFunc) *nodeWSConnection { if nodeName == "" { - return + return nil } h.nodeWSMu.Lock() defer h.nodeWSMu.Unlock() if h.nodeWSRegistry == nil { - h.nodeWSRegistry = make(map[string]context.CancelFunc) + h.nodeWSRegistry = make(map[string]*nodeWSConnection) } if prev, ok := h.nodeWSRegistry[nodeName]; ok { - prev() // cancel the old connection + prev.cancel() } - h.nodeWSRegistry[nodeName] = cancel + connection := &nodeWSConnection{cancel: cancel} + h.nodeWSRegistry[nodeName] = connection + + return connection } -// unregisterNodeWS removes a node's WS registration. Only removes if the -// cancel function matches (to avoid unregistering a newer connection). -func (h *healthState) unregisterNodeWS(nodeName string, cancel context.CancelFunc) { - if nodeName == "" { +// unregisterNodeWS cannot remove a newer connection with the same node identity. +func (h *healthState) unregisterNodeWS(nodeName string, connection *nodeWSConnection) { + if connection == nil { return } h.nodeWSMu.Lock() - defer h.nodeWSMu.Unlock() - if h.nodeWSRegistry == nil { - return + removed := h.nodeWSRegistry[nodeName] == connection + if removed { + delete(h.nodeWSRegistry, nodeName) } - // Only remove if it's still our registration (not replaced by a newer connection) - if existing, ok := h.nodeWSRegistry[nodeName]; ok { - // Compare by pointer identity -- Go func values aren't comparable, - // but context.CancelFunc from the same WithCancel call is the same pointer. - if fmt.Sprintf("%p", existing) == fmt.Sprintf("%p", cancel) { - delete(h.nodeWSRegistry, nodeName) - } + h.nodeWSMu.Unlock() + + if removed { + h.retryNodeDetails(nodeName) } } diff --git a/cmd/unbounded-net-controller/node_overview.go b/cmd/unbounded-net-controller/node_overview.go new file mode 100644 index 000000000..fe5f79225 --- /dev/null +++ b/cmd/unbounded-net-controller/node_overview.go @@ -0,0 +1,61 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "fmt" + "time" + + statuspkg "github.com/Azure/unbounded/internal/net/status" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +// StoreOverview replaces routine wire state without retaining diagnostic arrays. +func (c *NodeStatusCache) StoreOverview(nodeName string, overview statusv1alpha1.NodeStatusOverview, source string) (uint64, error) { + if nodeName == "" || (overview.NodeInfo.Name != "" && overview.NodeInfo.Name != nodeName) { + return 0, fmt.Errorf("summary identity does not match node %q", nodeName) + } + + if overview.PeerCount < 0 || overview.HealthyPeers < 0 || + overview.HealthyPeers > overview.PeerCount || overview.RouteCount < 0 { + return 0, fmt.Errorf("summary contains invalid observed counts") + } + + overview.NodeInfo.Name = nodeName + + if source == "" { + source = "push" + } + + overview.StatusSource = source + metadata := statuspkg.OverviewMetadata(overview) + + c.mu.Lock() + + revision := uint64(1) + if previous := c.entries[nodeName]; previous != nil { + revision = previous.Revision + 1 + } + + c.entries[nodeName] = &CachedNodeStatus{ + Status: &metadata, Overview: &overview, Source: source, + Revision: revision, ReceivedAt: time.Now(), + } + fn := c.onOverviewChange + c.mu.Unlock() + + if fn != nil { + fn(nodeName, overview) + } + + return revision, nil +} + +// SetOnOverviewChange registers the summary-only cache mutation callback. +func (c *NodeStatusCache) SetOnOverviewChange(fn func(string, statusv1alpha1.NodeStatusOverview)) { + c.mu.Lock() + defer c.mu.Unlock() + + c.onOverviewChange = fn +} diff --git a/cmd/unbounded-net-controller/node_overview_test.go b/cmd/unbounded-net-controller/node_overview_test.go new file mode 100644 index 000000000..7b5734891 --- /dev/null +++ b/cmd/unbounded-net-controller/node_overview_test.go @@ -0,0 +1,198 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "bytes" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func TestNodeOverviewCacheReplacesOnlyRoutineState(t *testing.T) { + cache := NewNodeStatusCache() + cache.StoreFull("node", NodeStatusResponse{Peers: []WireGuardPeerStatus{{Name: "peer"}}}, "push") + + var notified statusv1alpha1.NodeStatusOverview + + cache.SetOnOverviewChange(func(name string, overview statusv1alpha1.NodeStatusOverview) { + if name != "node" || cache.Len() != 1 { + t.Error("notification has wrong identity or ran under the cache lock") + } + + notified = overview + }) + overview := statusv1alpha1.NodeStatusOverview{ + PeerCount: 20, HealthyPeers: 17, RouteCount: 30, RouteMismatch: true, + NodeErrors: []NodeError{{Type: "cni", Message: "bootstrap blocked"}}, + } + + revision, err := cache.StoreOverview("node", overview, "ws") + if err != nil || revision != 2 { + t.Fatalf("store: revision=%d error=%v", revision, err) + } + + cached, ok := cache.Get("node") + if !ok || cached.Overview == nil || cached.Overview.PeerCount != 20 || cached.peerIdentity != nil { + t.Fatalf("unexpected overview wire state: %+v", cached) + } + + if cached.Status.Peers != nil || cached.Status.RoutingTable.Routes != nil || cached.Status.BpfEntries != nil { + t.Fatal("summary retained diagnostic arrays") + } + + cached.Overview.PeerCount = 99 + + unchanged, _ := cache.Get("node") + if unchanged.Overview.PeerCount != 20 { + t.Fatal("changing the returned overview mutated the cache") + } + + if notified.NodeInfo.Name != "node" || notified.StatusSource != "ws" || len(cached.Status.NodeErrors) != 1 { + t.Fatal("notification or metadata lost identity, source, or errors") + } + + rev, resync, err := cache.ApplyDelta("node", revision, map[string]json.RawMessage{}, "push") + if err != nil || !resync || rev != revision { + t.Fatalf("legacy delta must not apply to a summary: %d %v %v", rev, resync, err) + } + + snapshot := cache.GetAll() + cache.UpdateSource("node", "apiserver-ws") + + if snapshot["node"].Source != "ws" || notified.StatusSource != "apiserver-ws" { + t.Fatal("source change mutated an older snapshot or lost its notification") + } + + if next := cache.StoreFull("node", NodeStatusResponse{Peers: []WireGuardPeerStatus{{Name: "legacy"}}}, "push"); next != 3 { + t.Fatalf("legacy full resync revision=%d", next) + } + + legacy, _ := cache.Get("node") + if legacy.Overview != nil || len(legacy.Status.Peers) != 1 { + t.Fatal("explicit legacy full resync did not replace summary state") + } +} + +func TestNodeOverviewCacheRejectsInvalidFacts(t *testing.T) { + for _, overview := range []statusv1alpha1.NodeStatusOverview{ + {NodeInfo: NodeInfo{Name: "different-node"}}, + {PeerCount: -1}, + {HealthyPeers: -1}, + {PeerCount: 1, HealthyPeers: 2}, + {RouteCount: -1}, + } { + cache := NewNodeStatusCache() + if _, err := cache.StoreOverview("node", overview, "ws"); err == nil || cache.Len() != 0 { + t.Fatalf("invalid summary accepted: %+v", overview) + } + } + + if _, err := NewNodeStatusCache().StoreOverview("", statusv1alpha1.NodeStatusOverview{}, ""); err == nil { + t.Fatal("empty node identity accepted") + } +} + +func TestClusterOverviewPreservesCountsAndEnrichment(t *testing.T) { + c := NewClusterStatusCache(&healthState{}) + c.status = &ClusterStatusResponse{ + Nodes: []*NodeStatusResponse{{NodeInfo: NodeInfo{Name: "node", K8sReady: "Ready", ProviderID: "provider"}}}, + } + c.nodeIndex["node"] = 0 + overview := statusv1alpha1.NodeStatusOverview{ + NodeInfo: NodeInfo{Name: "node", SiteName: "site", WireGuard: &WireGuardStatusInfo{Interface: "wg0"}}, + StatusSource: "ws", PeerCount: 20, HealthyPeers: 17, RouteCount: 30, RouteMismatch: true, + } + c.PatchOverview("node", overview) + snapshot := c.Get() + + row := buildClusterSummary(snapshot).NodeSummaries[0] + if row.PeerCount != 20 || row.HealthyPeers != 17 || row.RouteCount != 30 || !row.RouteMismatch || + row.K8sReady != "Ready" || row.CniStatus != "Route mismatch" || row.SiteName != "site" { + t.Fatalf("summary lost observed facts or enriched fields: %+v", row) + } + + if snapshot.Nodes[0].NodeInfo.ProviderID != "provider" { + t.Fatal("controller enrichment was lost") + } + + problems := collectClusterProblems(snapshot) + if len(problems) != 1 || len(problems[0].Errors) != 2 { + t.Fatalf("summary health/mismatch problems were hidden: %+v", problems) + } + + overview.PeerCount = 25 + overview.NodeErrors = []NodeError{{Type: "cni", Message: "blocked"}} + c.PatchOverview("node", overview) + + if buildClusterSummary(snapshot).NodeSummaries[0] != row { + t.Fatal("patching changed a previously returned snapshot") + } + + nextRow := buildClusterSummary(c.Get()).NodeSummaries[0] + if nextRow.PeerCount != 25 || nextRow.FirstError != "blocked" || nextRow.CniTone != "danger" { + t.Fatalf("summary update lost errors or counts: %+v", nextRow) + } + + c.PatchNode("node", NodeStatusResponse{NodeInfo: overview.NodeInfo, Peers: []WireGuardPeerStatus{{}}}) + + if legacy := buildClusterSummary(c.Get()).NodeSummaries[0]; legacy.PeerCount != 1 { + t.Fatal("legacy update retained stale explicit summary counts") + } +} + +func TestClusterOverviewWireIgnoresDiagnosticArrays(t *testing.T) { + node := &NodeStatusResponse{NodeInfo: NodeInfo{Name: "node"}} + status := &ClusterStatusResponse{ + Nodes: []*NodeStatusResponse{node}, + NodeOverviews: map[string]*statusv1alpha1.NodeStatusOverview{ + "node": {PeerCount: 5, HealthyPeers: 4, RouteCount: 9}, + }, + } + + before, err := json.Marshal(buildClusterSummary(status)) + if err != nil { + t.Fatal(err) + } + + node.Peers = make([]WireGuardPeerStatus, 10000) + node.RoutingTable.Routes = make([]RouteEntry, 10000) + node.BpfEntries = make([]BpfEntry, 10000) + + after, err := json.Marshal(buildClusterSummary(status)) + if err != nil { + t.Fatal(err) + } + + if !bytes.Equal(before, after) { + t.Fatal("overview wire size or facts depend on diagnostic arrays") + } + + for _, field := range []string{`"peers":`, `"routingTable":`, `"bpfEntries":`, `"NodeOverviews":`} { + if strings.Contains(string(after), field) { + t.Fatalf("overview exposed %s", field) + } + } +} + +func TestClusterOverviewConcurrentSnapshots(t *testing.T) { + c := NewClusterStatusCache(&healthState{}) + c.status = &ClusterStatusResponse{} + + var wg sync.WaitGroup + for range 4 { + wg.Go(func() { + for range 100 { + c.PatchOverview("node", statusv1alpha1.NodeStatusOverview{NodeInfo: NodeInfo{Name: "node"}, Timestamp: time.Now()}) + buildClusterSummary(c.Get()) + } + }) + } + + wg.Wait() +} diff --git a/cmd/unbounded-net-controller/node_status.go b/cmd/unbounded-net-controller/node_status.go index 7fb9b09b1..3dbb37b88 100644 --- a/cmd/unbounded-net-controller/node_status.go +++ b/cmd/unbounded-net-controller/node_status.go @@ -12,6 +12,7 @@ import ( "time" statusproto "github.com/Azure/unbounded/internal/net/status/proto" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" ) // CachedNodeStatus stores a node's pushed status with timestamp and revision. @@ -20,15 +21,17 @@ type CachedNodeStatus struct { ReceivedAt time.Time Source string Revision uint64 + Overview *statusv1alpha1.NodeStatusOverview peerIdentity *peerIdentityDigest } // NodeStatusCache is a thread-safe cache of node status data pushed from node agents. type NodeStatusCache struct { - mu sync.RWMutex - entries map[string]*CachedNodeStatus - onChange func(nodeName string, status *NodeStatusResponse) + mu sync.RWMutex + entries map[string]*CachedNodeStatus + onChange func(nodeName string, status *NodeStatusResponse) + onOverviewChange func(nodeName string, overview statusv1alpha1.NodeStatusOverview) } // NewNodeStatusCache creates an empty NodeStatusCache. @@ -230,7 +233,7 @@ func (c *NodeStatusCache) applyParsedDelta(nodeName string, baseRevision uint64, return 0, true, nil } - if (pd.peerMeasurements != nil && baseRevision == 0) || (baseRevision != 0 && entry.Revision != baseRevision) { + if entry.Overview != nil || (pd.peerMeasurements != nil && baseRevision == 0) || (baseRevision != 0 && entry.Revision != baseRevision) { rev := entry.Revision c.mu.RUnlock() @@ -399,6 +402,11 @@ func (c *NodeStatusCache) Get(nodeName string) (*CachedNodeStatus, bool) { copy := *entry copy.Status = &statusCopy + if entry.Overview != nil { + overviewCopy := *entry.Overview + copy.Overview = &overviewCopy + } + return ©, true } @@ -428,6 +436,12 @@ func (c *NodeStatusCache) Delete(nodeName string) { // UpdateSource updates the cached status source for a node without changing // the cached payload or ReceivedAt timestamp. func (c *NodeStatusCache) UpdateSource(nodeName, source string) bool { + return c.UpdateSourceIf(nodeName, "", source) +} + +// UpdateSourceIf changes the source only if the expected transport still owns it. +// An empty expected source preserves the unconditional UpdateSource behavior. +func (c *NodeStatusCache) UpdateSourceIf(nodeName, expectedSource, source string) bool { if source == "" { return false } @@ -435,7 +449,7 @@ func (c *NodeStatusCache) UpdateSource(nodeName, source string) bool { c.mu.Lock() entry, ok := c.entries[nodeName] - if !ok { + if !ok || (expectedSource != "" && entry.Source != expectedSource) { c.mu.Unlock() return false } @@ -445,12 +459,19 @@ func (c *NodeStatusCache) UpdateSource(nodeName, source string) bool { return true } - entry.Source = source + updated := *entry + updated.Source = source + c.entries[nodeName] = &updated fn := c.onChange + overviewFn := c.onOverviewChange statusCopy := entry.Status c.mu.Unlock() - if fn != nil { + if entry.Overview != nil && overviewFn != nil { + overview := *entry.Overview + overview.StatusSource = source + overviewFn(nodeName, overview) + } else if fn != nil { fn(nodeName, statusCopy) } diff --git a/cmd/unbounded-net-controller/server.go b/cmd/unbounded-net-controller/server.go index 6bd4551db..dace866f1 100644 --- a/cmd/unbounded-net-controller/server.go +++ b/cmd/unbounded-net-controller/server.go @@ -27,6 +27,7 @@ import ( "github.com/Azure/unbounded/internal/net/html" "github.com/Azure/unbounded/internal/net/metrics" statusproto "github.com/Azure/unbounded/internal/net/status/proto" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" webhookpkg "github.com/Azure/unbounded/internal/net/webhook" ) @@ -58,6 +59,9 @@ type nodeStatusWSIdentity struct { Status *struct { NodeInfo nodeStatusIdentityInfo `json:"nodeInfo"` } `json:"status"` + Summary *struct { + NodeInfo nodeStatusIdentityInfo `json:"nodeInfo"` + } `json:"summary"` Delta map[string]json.RawMessage `json:"delta"` } @@ -76,6 +80,10 @@ func extractNodeNameFromWSMessage(data []byte) (string, error) { nodeNames = append(nodeNames, identity.Status.NodeInfo.Name) } + if identity.Summary != nil { + nodeNames = append(nodeNames, identity.Summary.NodeInfo.Name) + } + // Match ApplyDelta's case-sensitive map lookup, not struct field matching. if raw, ok := identity.Delta["nodeInfo"]; ok { var info nodeStatusIdentityInfo @@ -96,10 +104,10 @@ func rejectDuplicateStatusIdentityFields(data []byte, object string) error { return nil } - fields := []string{"nodeName", "nodeInfo", "status", "delta"} + fields := []string{"nodeName", "nodeInfo", "status", "delta", "summary"} switch object { - case "status", "delta": + case "status", "delta", "summary": fields = []string{"nodeInfo"} case "nodeInfo": fields = []string{"name"} @@ -147,7 +155,7 @@ func rejectDuplicateStatusIdentityFields(data []byte, object string) error { return fmt.Errorf("invalid status identity value: %w", err) } - if matched == "status" || matched == "delta" || matched == "nodeInfo" { + if matched == "status" || matched == "delta" || matched == "nodeInfo" || matched == "summary" { if err := rejectDuplicateStatusIdentityFields(value, matched); err != nil { return err } @@ -233,6 +241,11 @@ func startServer(ctx context.Context, healthPort int, requireDashboardAuth bool, clusterStatusCache.MarkDirty() broadcaster.Notify() }) + health.statusCache.SetOnOverviewChange(func(nodeName string, overview statusv1alpha1.NodeStatusOverview) { + clusterStatusCache.PatchOverview(nodeName, overview) + clusterStatusCache.MarkDirty() + broadcaster.Notify() + }) // Node WebSocket connection semaphore. wsSemaphore := make(chan struct{}, maxConcurrentNodeWS) @@ -550,17 +563,19 @@ func registerPushHandlers(mux *http.ServeMux, health *healthState, webhookServer return } + if ack.IsPublicationAck() && ack.Status == "ok" { + if manager := health.getDetailRequests(); manager != nil { + if command, ok := manager.Pending(authorizedNodeName(r)); ok { + ack.DetailRequest = &command + } + } + } + isProto := isProtobufContentType(r) if isProto { w.Header().Set("Content-Type", "application/x-protobuf") - pbAck := &statusproto.NodeStatusAck{ - Status: ack.Status, - Revision: ack.Revision, - Reason: ack.Reason, - } - - data, marshalErr := proto.Marshal(pbAck) + data, marshalErr := marshalProtoAck("node_status_ack", ack) if marshalErr != nil { klog.V(4).Infof("status push proto ack marshal failed: %v", marshalErr) http.Error(w, "internal error", http.StatusInternalServerError) @@ -648,37 +663,57 @@ func registerPushHandlers(mux *http.ServeMux, health *healthState, webhookServer } }() - send := func(frameType websocket.MessageType, ackMsgType string, ack NodeStatusPushAck) { - if frameType == websocket.MessageBinary { - payload, marshalErr := marshalProtoAck(ackMsgType, ack) - if marshalErr != nil { - klog.V(4).Infof("Node WebSocket proto ack marshal failed (source=%s, node=%s): %v", source, nodeNameForLog(), marshalErr) - return - } + wsCtx, wsCancel := context.WithCancel(r.Context()) + defer wsCancel() - if writeErr := conn.Write(r.Context(), websocket.MessageBinary, payload); writeErr != nil { - klog.V(4).Infof("Node WebSocket ack write failed (source=%s, node=%s): %v", source, nodeNameForLog(), writeErr) - } + var registration *nodeWSConnection + defer func() { + health.markNodeWSStale(lastWSNodeName, registration, source) + health.unregisterNodeWS(lastWSNodeName, registration) + }() - return + writeGate := make(chan struct{}, 1) + + sendContext := func(ctx context.Context, frameType websocket.MessageType, ackMsgType string, ack NodeStatusPushAck) error { + select { + case writeGate <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + case <-wsCtx.Done(): + return wsCtx.Err() + } + + defer func() { <-writeGate }() + + var ( + payload []byte + marshalErr error + ) + if frameType == websocket.MessageBinary { + payload, marshalErr = marshalProtoAck(ackMsgType, ack) + } else { + payload, marshalErr = json.Marshal(map[string]interface{}{"type": ackMsgType, "data": ack}) } - payload, marshalErr := json.Marshal(map[string]interface{}{"type": ackMsgType, "data": ack}) if marshalErr != nil { - klog.V(4).Infof("Node WebSocket ack marshal failed (source=%s, node=%s): %v", source, nodeNameForLog(), marshalErr) - return + return marshalErr } - if writeErr := conn.Write(r.Context(), websocket.MessageText, payload); writeErr != nil { - klog.V(4).Infof("Node WebSocket ack write failed (source=%s, node=%s): %v", source, nodeNameForLog(), writeErr) + return conn.Write(ctx, frameType, payload) + } + send := func(frameType websocket.MessageType, ackMsgType string, ack NodeStatusPushAck) { + if err := sendContext(wsCtx, frameType, ackMsgType, ack); err != nil { + klog.V(4).Infof("Node WebSocket ack failed (source=%s): %v", source, err) + wsCancel() } } - - wsCtx, wsCancel := context.WithCancel(r.Context()) - defer wsCancel() - defer func() { - health.unregisterNodeWS(lastWSNodeName, wsCancel) - }() + enableDetails := func(nodeName string, frameType websocket.MessageType) { + health.setNodeWSDetailSender(nodeName, registration, func(ctx context.Context, command statusv1alpha1.DetailRequest) error { + return sendContext(ctx, frameType, "node_status_ack", NodeStatusPushAck{ + Status: statusv1alpha1.DetailRequestStatus, DetailRequest: &command, SummarySupported: true, + }) + }) + } recvCh := make(chan wsFrame) errCh := make(chan error, 1) @@ -763,7 +798,7 @@ func registerPushHandlers(mux *http.ServeMux, health *healthState, webhookServer return case readErr := <-errCh: if lastWSNodeName != "" { - health.statusCache.UpdateSource(lastWSNodeName, "stale-cache") + health.markNodeWSStale(lastWSNodeName, registration, source) } // Log graceful close frames and expected disconnections at // V(4) to reduce noise during rolling restarts. @@ -786,7 +821,7 @@ func registerPushHandlers(mux *http.ServeMux, health *healthState, webhookServer case frame, ok := <-recvCh: if !ok { if lastWSNodeName != "" { - health.statusCache.UpdateSource(lastWSNodeName, "stale-cache") + health.markNodeWSStale(lastWSNodeName, registration, source) } return @@ -819,7 +854,7 @@ func registerPushHandlers(mux *http.ServeMux, health *healthState, webhookServer if nodeName != "" { if lastWSNodeName == "" { // First message identifies the node -- register and evict old connections. - health.registerNodeWS(nodeName, wsCancel) + registration = health.registerNodeWS(nodeName, wsCancel) } lastWSNodeName = nodeName @@ -827,6 +862,10 @@ func registerPushHandlers(mux *http.ServeMux, health *healthState, webhookServer ackType, ack := handleProtoWSMessage(health, decoded, source) send(websocket.MessageBinary, ackType, ack) + + if ack.Status == "ok" && decoded.message.SupportsDetails { + enableDetails(nodeName, websocket.MessageBinary) + } } else { nodeName, identityErr := extractNodeNameFromWSMessage(frame.data) if identityErr != nil || nodeName == "" { @@ -854,7 +893,7 @@ func registerPushHandlers(mux *http.ServeMux, health *healthState, webhookServer if nodeName != "" { if lastWSNodeName == "" { - health.registerNodeWS(nodeName, wsCancel) + registration = health.registerNodeWS(nodeName, wsCancel) } lastWSNodeName = nodeName @@ -862,6 +901,13 @@ func registerPushHandlers(mux *http.ServeMux, health *healthState, webhookServer ackType, ack := handleNodeStatusWSMessageWithSource(health, frame.data, source) send(websocket.MessageText, ackType, ack) + + var capability struct { + SupportsDetails bool `json:"supportsDetails"` + } + if err := json.Unmarshal(frame.data, &capability); err == nil && ack.Status == "ok" && capability.SupportsDetails { + enableDetails(nodeName, websocket.MessageText) + } } case <-keepaliveCh: // Skip ping if we received a message recently @@ -905,7 +951,7 @@ func registerPushHandlers(mux *http.ServeMux, health *healthState, webhookServer klog.V(2).Infof("Node WebSocket keepalive closing connection after reaching failure threshold (source=%s, node=%s, failures=%d, threshold=%d)", source, nodeNameLog, keepaliveFailures, health.statusWSKeepaliveFailureCount) if lastWSNodeName != "" { - health.statusCache.UpdateSource(lastWSNodeName, "stale-cache") + health.markNodeWSStale(lastWSNodeName, registration, source) } if closeErr := conn.Close(websocket.StatusGoingAway, "keepalive failure threshold reached"); closeErr != nil { @@ -1174,7 +1220,9 @@ func handleStatusPushBody(health *healthState, r *http.Request, bodyBytes []byte return handleStatusPushRequestWithSource(health, bodyBytes, source) } -func handleStatusPushRequestWithSource(health *healthState, bodyBytes []byte, source string) (NodeStatusPushAck, int, error) { +func handleStatusPushRequestWithSource(health *healthState, bodyBytes []byte, source string) (ack NodeStatusPushAck, code int, err error) { + defer func() { ack.SummarySupported = true }() + if _, err := extractNodeNameFromWSMessage(bodyBytes); err != nil { return NodeStatusPushAck{}, http.StatusBadRequest, err } @@ -1184,7 +1232,35 @@ func handleStatusPushRequestWithSource(health *healthState, bodyBytes []byte, so return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("invalid request body: %v", err) } - ack := NodeStatusPushAck{Status: "ok"} + ack = NodeStatusPushAck{Status: "ok"} + + if envelope.Type == statusv1alpha1.NodeStatusSummaryType { + if envelope.Mode != "" && envelope.Mode != "summary" { + return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("conflicting status mode and type") + } + + envelope.Mode = "summary" + } + + if envelope.Type == statusv1alpha1.NodeStatusDetailsType { + if envelope.Mode != "" && envelope.Mode != "details" { + return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("conflicting status mode and type") + } + + envelope.Mode = "details" + } + + if envelope.DetailError != "" && envelope.Mode != "details" { + return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("collection error requires a detail response") + } + + if envelope.Summary != nil && envelope.Mode != "summary" { + return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("overview requires summary mode") + } + + if envelope.Mode == "summary" && envelope.Type != "" && envelope.Type != statusv1alpha1.NodeStatusSummaryType { + return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("conflicting status mode and type") + } if envelope.Mode == "" { var nodeStatus NodeStatusResponse @@ -1207,11 +1283,32 @@ func handleStatusPushRequestWithSource(health *healthState, bodyBytes []byte, so nodeName = envelope.Status.NodeInfo.Name } + if envelope.Summary != nil && envelope.Summary.NodeInfo.Name != "" { + nodeName = envelope.Summary.NodeInfo.Name + } + if nodeName == "" { return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("nodeName is required") } switch envelope.Mode { + case "details": + if envelope.Delta != nil { + return NodeStatusPushAck{Status: "error", DetailRequestID: envelope.DetailRequestID, Reason: "details cannot include a delta"}, http.StatusOK, nil + } + + return handleNodeDetailResponse(health, nodeName, envelope.DetailRequestID, envelope.Status, envelope.DetailError), http.StatusOK, nil + case "summary": + if envelope.Summary == nil || envelope.Status != nil || envelope.Delta != nil || envelope.DetailRequestID != "" { + return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("summary must contain only overview data") + } + + ack.Revision, err = health.statusCache.StoreOverview(nodeName, *envelope.Summary, source) + if err != nil { + return NodeStatusPushAck{}, http.StatusBadRequest, err + } + + return ack, http.StatusOK, nil case "full": if envelope.Status == nil { return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("status is required for full mode") @@ -1244,7 +1341,7 @@ func handleStatusPushRequestWithSource(health *healthState, bodyBytes []byte, so return ack, http.StatusOK, nil default: - return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("mode must be full or delta") + return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("unsupported status mode %q", envelope.Mode) } } @@ -1252,7 +1349,9 @@ func handleNodeStatusWSMessage(health *healthState, data []byte) (string, NodeSt return handleNodeStatusWSMessageWithSource(health, data, "ws") } -func handleNodeStatusWSMessageWithSource(health *healthState, data []byte, source string) (string, NodeStatusPushAck) { +func handleNodeStatusWSMessageWithSource(health *healthState, data []byte, source string) (ackType string, ack NodeStatusPushAck) { + defer func() { ack.SummarySupported = true }() + if _, err := extractNodeNameFromWSMessage(data); err != nil { return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: err.Error()} } @@ -1262,16 +1361,45 @@ func handleNodeStatusWSMessageWithSource(health *healthState, data []byte, sourc return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "invalid message"} } + if message.Summary != nil && message.Type != statusv1alpha1.NodeStatusSummaryType { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "overview requires summary message type"} + } + + if message.DetailError != "" && message.Type != statusv1alpha1.NodeStatusDetailsType { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "collection error requires a detail response"} + } + nodeName := message.NodeName if message.Status != nil && message.Status.NodeInfo.Name != "" { nodeName = message.Status.NodeInfo.Name } + if message.Summary != nil && message.Summary.NodeInfo.Name != "" { + nodeName = message.Summary.NodeInfo.Name + } + if nodeName == "" { return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "nodeName is required"} } switch message.Type { + case statusv1alpha1.NodeStatusDetailsType: + if message.Delta != nil { + return "node_status_ack", NodeStatusPushAck{Status: "error", DetailRequestID: message.DetailRequestID, Reason: "details cannot include a delta"} + } + + return "node_status_ack", handleNodeDetailResponse(health, nodeName, message.DetailRequestID, message.Status, message.DetailError) + case statusv1alpha1.NodeStatusSummaryType: + if message.Summary == nil || message.Status != nil || message.Delta != nil || message.DetailRequestID != "" { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "summary must contain only overview data"} + } + + revision, err := health.statusCache.StoreOverview(nodeName, *message.Summary, source) + if err != nil { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: err.Error()} + } + + return "node_status_ack", NodeStatusPushAck{Status: "ok", Revision: revision} case "node_status_full": if message.Status == nil { return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "full message missing status"} diff --git a/cmd/unbounded-net-controller/status_overview_proto.go b/cmd/unbounded-net-controller/status_overview_proto.go new file mode 100644 index 000000000..5255dfe91 --- /dev/null +++ b/cmd/unbounded-net-controller/status_overview_proto.go @@ -0,0 +1,38 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "time" + + statusproto "github.com/Azure/unbounded/internal/net/status/proto" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func protoToNodeOverview(msg *statusproto.NodeStatusOverview) statusv1alpha1.NodeStatusOverview { + overview := statusv1alpha1.NodeStatusOverview{ + HealthCheck: protoToHealthCheckStatus(msg.HealthCheck), + NodeErrors: protoToNodeErrors(msg.NodeErrors), + FetchError: msg.FetchError, StatusSource: msg.StatusSource, + NodePodInfo: protoToNodePodInfo(msg.NodePodInfo), + PeerCount: int(msg.PeerCount), HealthyPeers: int(msg.HealthyPeers), + RouteCount: int(msg.RouteCount), RouteMismatch: msg.RouteMismatch, + RouteMismatchCount: int(msg.RouteMismatchCount), + UnhealthyPeerLinks: int(msg.UnhealthyPeerLinks), UsesIPIP: msg.UsesIpip, + } + if msg.NodeInfo != nil { + overview.NodeInfo = protoToNodeInfo(msg.NodeInfo) + } + + if msg.TimestampUnixNs != 0 { + overview.Timestamp = time.Unix(0, msg.TimestampUnixNs) + } + + if msg.LastPushTimeUnixNs != 0 { + lastPush := time.Unix(0, msg.LastPushTimeUnixNs) + overview.LastPushTime = &lastPush + } + + return overview +} diff --git a/cmd/unbounded-net-controller/status_proto.go b/cmd/unbounded-net-controller/status_proto.go index cb10196b6..635ebc611 100644 --- a/cmd/unbounded-net-controller/status_proto.go +++ b/cmd/unbounded-net-controller/status_proto.go @@ -9,6 +9,7 @@ import ( "google.golang.org/protobuf/proto" + statuspkg "github.com/Azure/unbounded/internal/net/status" statusproto "github.com/Azure/unbounded/internal/net/status/proto" statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" ) @@ -468,20 +469,58 @@ func validatedProtoNodeName(msg *statusproto.NodeStatusMessage) (string, error) nodeNames = append(nodeNames, msg.Delta.NodeInfo.Name) } + if msg.Summary != nil && msg.Summary.NodeInfo != nil { + nodeNames = append(nodeNames, msg.Summary.NodeInfo.Name) + } + return validatedNodeNames(nodeNames) } // handleProtoWSMessage applies the same decoded message used for authorization. -func handleProtoWSMessage(health *healthState, decoded *decodedProtoWSMessage, source string) (string, NodeStatusPushAck) { +func handleProtoWSMessage(health *healthState, decoded *decodedProtoWSMessage, source string) (ackType string, ack NodeStatusPushAck) { + defer func() { ack.SummarySupported = true }() + msg := &decoded.message nodeName := decoded.nodeName + if msg.DetailError != "" && msg.Type != statusv1alpha1.NodeStatusDetailsType { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "collection error requires a detail response"} + } + + if msg.Summary != nil && msg.Type != statusv1alpha1.NodeStatusSummaryType { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "overview requires summary message type"} + } + if msg.Delta.GetPeerMeasurements() != nil && (msg.Type != "node_status_delta" || msg.Status != nil) { peerMeasurementUpdatesTotal.WithLabelValues("error").Inc() return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "full status conflicts with measurements"} } switch msg.Type { + case statusv1alpha1.NodeStatusDetailsType: + if msg.Delta != nil { + return "node_status_ack", NodeStatusPushAck{Status: "error", DetailRequestID: msg.DetailRequestId, Reason: "details cannot include a delta"} + } + + var status *NodeStatusResponse + + if msg.Status != nil { + full := protoToNodeStatus(msg.Status) + status = &full + } + + return "node_status_ack", handleNodeDetailResponse(health, nodeName, msg.DetailRequestId, status, msg.DetailError) + case statusv1alpha1.NodeStatusSummaryType: + if msg.Summary == nil || msg.Status != nil || msg.Delta != nil || msg.DetailRequestId != "" { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "summary must contain only overview data"} + } + + revision, err := health.statusCache.StoreOverview(nodeName, protoToNodeOverview(msg.Summary), source) + if err != nil { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: err.Error()} + } + + return "node_status_ack", NodeStatusPushAck{Status: "ok", Revision: revision} case "node_status_full": if msg.Status == nil { return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "full message missing status"} @@ -518,7 +557,9 @@ func handleProtoWSMessage(health *healthState, decoded *decodedProtoWSMessage, s } // handleProtoPushRequest processes an HTTP push request with protobuf body. -func handleProtoPushRequest(health *healthState, bodyBytes []byte, source string) (NodeStatusPushAck, int, error) { +func handleProtoPushRequest(health *healthState, bodyBytes []byte, source string) (ack NodeStatusPushAck, code int, err error) { + defer func() { ack.SummarySupported = true }() + var msg statusproto.NodeStatusMessage if err := proto.Unmarshal(bodyBytes, &msg); err != nil { return NodeStatusPushAck{}, 400, fmt.Errorf("invalid protobuf body: %v", err) @@ -533,13 +574,46 @@ func handleProtoPushRequest(health *healthState, bodyBytes []byte, source string return NodeStatusPushAck{}, 400, fmt.Errorf("nodeName is required") } - ack := NodeStatusPushAck{Status: "ok"} + ack = NodeStatusPushAck{Status: "ok"} + + if msg.DetailError != "" && msg.Type != statusv1alpha1.NodeStatusDetailsType { + return NodeStatusPushAck{}, 400, fmt.Errorf("collection error requires a detail response") + } + + if msg.Summary != nil && msg.Type != statusv1alpha1.NodeStatusSummaryType { + return NodeStatusPushAck{}, 400, fmt.Errorf("overview requires summary message type") + } + if msg.Delta.GetPeerMeasurements() != nil && (msg.Type != "node_status_delta" || msg.Status != nil) { peerMeasurementUpdatesTotal.WithLabelValues("error").Inc() return NodeStatusPushAck{Status: "resync_required", Reason: "full status conflicts with measurements"}, 429, nil } switch msg.Type { + case statusv1alpha1.NodeStatusDetailsType: + if msg.Delta != nil { + return NodeStatusPushAck{Status: "error", DetailRequestID: msg.DetailRequestId, Reason: "details cannot include a delta"}, 200, nil + } + + var status *NodeStatusResponse + + if msg.Status != nil { + full := protoToNodeStatus(msg.Status) + status = &full + } + + return handleNodeDetailResponse(health, nodeName, msg.DetailRequestId, status, msg.DetailError), 200, nil + case statusv1alpha1.NodeStatusSummaryType: + if msg.Summary == nil || msg.Status != nil || msg.Delta != nil || msg.DetailRequestId != "" { + return NodeStatusPushAck{}, 400, fmt.Errorf("summary must contain only overview data") + } + + ack.Revision, err = health.statusCache.StoreOverview(nodeName, protoToNodeOverview(msg.Summary), source) + if err != nil { + return NodeStatusPushAck{}, 400, err + } + + return ack, 200, nil case "node_status_full": if msg.Status == nil { return NodeStatusPushAck{}, 400, fmt.Errorf("status is required for full mode") @@ -573,18 +647,14 @@ func handleProtoPushRequest(health *healthState, bodyBytes []byte, source string return ack, 200, nil default: - return NodeStatusPushAck{}, 400, fmt.Errorf("type must be node_status_full or node_status_delta") + return NodeStatusPushAck{}, 400, fmt.Errorf("unsupported status message type %q", msg.Type) } } // marshalProtoAck serializes a NodeStatusPushAck into a protobuf NodeStatusAck. func marshalProtoAck(ackType string, ack NodeStatusPushAck) ([]byte, error) { - pbAck := &statusproto.NodeStatusAck{ - PeerMeasurements: true, - Status: ack.Status, - Revision: ack.Revision, - Reason: ack.Reason, - } + ack.PeerMeasurements = true + ack.SummarySupported = true - return proto.Marshal(pbAck) + return proto.Marshal(statuspkg.NodeStatusAckToProto(&ack)) } diff --git a/cmd/unbounded-net-controller/status_types.go b/cmd/unbounded-net-controller/status_types.go index 98b72c1bd..db908722c 100644 --- a/cmd/unbounded-net-controller/status_types.go +++ b/cmd/unbounded-net-controller/status_types.go @@ -9,27 +9,29 @@ import ( "sort" "time" + statuspkg "github.com/Azure/unbounded/internal/net/status" statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" ) // ClusterStatusResponse is the top-level status response for the cluster. type ClusterStatusResponse struct { - Seq uint64 `json:"seq"` - Timestamp time.Time `json:"timestamp"` - NodeCount int `json:"nodeCount"` - SiteCount int `json:"siteCount"` - AzureTenantID string `json:"azureTenantId,omitempty"` - LeaderInfo *LeaderInfo `json:"leaderInfo,omitempty"` - BuildInfo *BuildInfo `json:"buildInfo,omitempty"` - Nodes []*NodeStatusResponse `json:"nodes"` - Sites []SiteStatus `json:"sites"` - GatewayPools []GatewayPoolStatus `json:"gatewayPools"` - Peerings []PeeringStatus `json:"peerings"` - Errors []string `json:"errors,omitempty"` - Warnings []string `json:"warnings,omitempty"` - Problems []StatusProblem `json:"problems"` - ConnectivityMatrix map[string]*SiteMatrix `json:"connectivityMatrix,omitempty"` - PullEnabled bool `json:"pullEnabled"` + Seq uint64 `json:"seq"` + Timestamp time.Time `json:"timestamp"` + NodeCount int `json:"nodeCount"` + SiteCount int `json:"siteCount"` + AzureTenantID string `json:"azureTenantId,omitempty"` + LeaderInfo *LeaderInfo `json:"leaderInfo,omitempty"` + BuildInfo *BuildInfo `json:"buildInfo,omitempty"` + Nodes []*NodeStatusResponse `json:"nodes"` + Sites []SiteStatus `json:"sites"` + GatewayPools []GatewayPoolStatus `json:"gatewayPools"` + Peerings []PeeringStatus `json:"peerings"` + Errors []string `json:"errors,omitempty"` + Warnings []string `json:"warnings,omitempty"` + Problems []StatusProblem `json:"problems"` + ConnectivityMatrix map[string]*SiteMatrix `json:"connectivityMatrix,omitempty"` + PullEnabled bool `json:"pullEnabled"` + NodeOverviews map[string]*statusv1alpha1.NodeStatusOverview `json:"-"` } // ClusterStatusDelta is a WebSocket delta update. @@ -77,28 +79,23 @@ type NodeStatusResponse = statusv1alpha1.NodeStatusResponse // NodeStatusPushEnvelope carries a push status update from a node. type NodeStatusPushEnvelope struct { - Mode string `json:"mode,omitempty"` - NodeName string `json:"nodeName,omitempty"` - BaseRevision uint64 `json:"baseRevision,omitempty"` - Status *NodeStatusResponse `json:"status,omitempty"` - Delta map[string]json.RawMessage `json:"delta,omitempty"` + Type string `json:"type,omitempty"` + Mode string `json:"mode,omitempty"` + NodeName string `json:"nodeName,omitempty"` + BaseRevision uint64 `json:"baseRevision,omitempty"` + Status *NodeStatusResponse `json:"status,omitempty"` + Delta map[string]json.RawMessage `json:"delta,omitempty"` + Summary *statusv1alpha1.NodeStatusOverview `json:"summary,omitempty"` + DetailRequestID string `json:"detailRequestId,omitempty"` + DetailError string `json:"detailError,omitempty"` + SupportsDetails bool `json:"supportsDetails,omitempty"` } // NodeStatusPushAck is the acknowledgment returned for push updates. -type NodeStatusPushAck struct { - Status string `json:"status"` - Revision uint64 `json:"revision,omitempty"` - Reason string `json:"reason,omitempty"` -} +type NodeStatusPushAck = statusv1alpha1.NodeStatusAck // NodeStatusWSMessage is the status message format used over WebSockets. -type NodeStatusWSMessage struct { - Type string `json:"type"` - NodeName string `json:"nodeName,omitempty"` - BaseRevision uint64 `json:"baseRevision,omitempty"` - Status *NodeStatusResponse `json:"status,omitempty"` - Delta map[string]json.RawMessage `json:"delta,omitempty"` -} +type NodeStatusWSMessage = statusv1alpha1.NodeStatusMessage // NodePodInfo aliases the shared node pod status schema. type NodePodInfo = statusv1alpha1.NodePodInfo @@ -186,23 +183,32 @@ type NodeSummary struct { } // buildClusterSummary extracts a ClusterSummary from a full ClusterStatusResponse. -// This is O(N) in nodes with simple field reads -- no route annotation work. +// Only legacy payloads require scanning peers and route next hops. func buildClusterSummary(status *ClusterStatusResponse) *ClusterSummary { summaries := make([]NodeSummary, 0, len(status.Nodes)) now := time.Now() for i := range status.Nodes { node := status.Nodes[i] + + overview := status.NodeOverviews[node.NodeInfo.Name] + if overview == nil { + projected := statuspkg.OverviewFromStatus(node, now) + overview = &projected + } + ns := NodeSummary{ - Name: node.NodeInfo.Name, - SiteName: node.NodeInfo.SiteName, - IsGateway: node.NodeInfo.IsGateway, - K8sReady: node.NodeInfo.K8sReady, - StatusSource: node.StatusSource, - PeerCount: len(node.Peers), - RouteCount: len(node.RoutingTable.Routes), - FetchError: node.FetchError, - ErrorCount: len(node.NodeErrors), + Name: node.NodeInfo.Name, + SiteName: node.NodeInfo.SiteName, + IsGateway: node.NodeInfo.IsGateway, + K8sReady: node.NodeInfo.K8sReady, + StatusSource: node.StatusSource, + PeerCount: overview.PeerCount, + HealthyPeers: overview.HealthyPeers, + RouteCount: overview.RouteCount, + RouteMismatch: overview.RouteMismatch, + FetchError: node.FetchError, + ErrorCount: len(node.NodeErrors), } // Include first error message so the frontend can show it inline @@ -211,38 +217,6 @@ func buildClusterSummary(status *ClusterStatusResponse) *ClusterSummary { ns.FirstError = node.NodeErrors[0].Message } - // Count healthy peers - for j := range node.Peers { - peer := &node.Peers[j] - if peer.HealthCheck != nil && peer.HealthCheck.Enabled { - if peer.HealthCheck.Status == "up" || peer.HealthCheck.Status == "Up" { - ns.HealthyPeers++ - } - } else { - // Fall back to handshake freshness - if !peer.Tunnel.LastHandshake.IsZero() && now.Sub(peer.Tunnel.LastHandshake) < 3*time.Minute { - ns.HealthyPeers++ - } - } - } - - // Route mismatch check - for _, route := range node.RoutingTable.Routes { - for _, hop := range route.NextHops { - expected := hop.Expected != nil && *hop.Expected - - present := hop.Present != nil && *hop.Present - if expected != present { - ns.RouteMismatch = true - break - } - } - - if ns.RouteMismatch { - break - } - } - // Derive CNI status and tone ns.CniStatus, ns.CniTone = deriveCniStatusAndTone(node, ns.RouteMismatch) summaries = append(summaries, ns) diff --git a/cmd/unbounded-net-controller/ws_source_ownership_test.go b/cmd/unbounded-net-controller/ws_source_ownership_test.go new file mode 100644 index 000000000..ea1e1900a --- /dev/null +++ b/cmd/unbounded-net-controller/ws_source_ownership_test.go @@ -0,0 +1,35 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import "testing" + +func TestWebSocketTeardownPreservesNewerStatusSource(t *testing.T) { + health := &healthState{statusCache: NewNodeStatusCache()} + status := NodeStatusResponse{NodeInfo: NodeInfo{Name: "node"}} + health.statusCache.StoreFull("node", status, "ws") + old := health.registerNodeWS("node", func() {}) + current := health.registerNodeWS("node", func() {}) + + health.markNodeWSStale("node", old, "ws") + + if cached, _ := health.statusCache.Get("node"); cached.Source != "ws" { + t.Fatal("old connection teardown marked the replacement stale") + } + + health.statusCache.StoreFull("node", status, "push") + health.markNodeWSStale("node", current, "ws") + + if cached, _ := health.statusCache.Get("node"); cached.Source != "push" { + t.Fatal("WebSocket teardown overwrote a newer HTTP publication") + } + + health.statusCache.StoreFull("node", status, "ws") + before := health.statusCache.GetAll() + health.markNodeWSStale("node", current, "ws") + + if cached, _ := health.statusCache.Get("node"); cached.Source != "stale-cache" || before["node"].Source != "ws" { + t.Fatal("current teardown lost its stale signal or mutated an old snapshot") + } +}