From 43186d60513cb3051c5d2da64d1d33639f18410f Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Thu, 17 Sep 2026 14:04:54 +0000 Subject: [PATCH 1/2] net node: publish summaries and answer one-shot diagnostics immediately Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- cmd/unbounded-net-node/main.go | 4 +- cmd/unbounded-net-node/status_ack.go | 60 ++- cmd/unbounded-net-node/status_details.go | 264 ++++++++++++ cmd/unbounded-net-node/status_details_http.go | 80 ++++ .../status_details_http_test.go | 287 +++++++++++++ cmd/unbounded-net-node/status_details_test.go | 178 ++++++++ .../status_details_ws_test.go | 262 ++++++++++++ cmd/unbounded-net-node/status_publication.go | 77 ++++ .../status_publication_test.go | 205 +++++++++ cmd/unbounded-net-node/status_server.go | 397 ++++++++++++------ cmd/unbounded-net-node/status_server_test.go | 2 +- .../status_websocket_auth_test.go | 35 +- 12 files changed, 1702 insertions(+), 149 deletions(-) create mode 100644 cmd/unbounded-net-node/status_details.go create mode 100644 cmd/unbounded-net-node/status_details_http.go create mode 100644 cmd/unbounded-net-node/status_details_http_test.go create mode 100644 cmd/unbounded-net-node/status_details_test.go create mode 100644 cmd/unbounded-net-node/status_details_ws_test.go create mode 100644 cmd/unbounded-net-node/status_publication.go create mode 100644 cmd/unbounded-net-node/status_publication_test.go diff --git a/cmd/unbounded-net-node/main.go b/cmd/unbounded-net-node/main.go index 34b0cd46a..c2f836343 100644 --- a/cmd/unbounded-net-node/main.go +++ b/cmd/unbounded-net-node/main.go @@ -93,7 +93,7 @@ type config struct { StatusPushInterval time.Duration // Interval between status pushes to controller StatusPushAPIServerInterval time.Duration // Interval between status pushes via aggregated API server StatusPushDelta bool // Whether periodic HTTP pushes use deltas - StatusDetailMode string // Startup-loaded; publication wiring follows separately. + StatusDetailMode string // Startup-loaded routine publication mode. StatusWSEnabled bool // Whether websocket push is enabled StatusWSURL string // Controller websocket URL for status push StatusWSAPIServerMode string // API server fallback mode: never, fallback, preferred (alias for fallback) @@ -330,7 +330,7 @@ then annotates the node with the public key.`, flags.DurationVar(&cfg.StatusPushInterval, "status-push-interval", 60*time.Second, "Interval between status pushes to controller") flags.DurationVar(&cfg.StatusPushAPIServerInterval, "status-push-apiserver-interval", 60*time.Second, "Interval between status pushes via aggregated API server") flags.BoolVar(&cfg.StatusPushDelta, "status-push-delta", true, "Enable delta mode for periodic HTTP status push") - flags.StringVar(&cfg.StatusDetailMode, "status-detail-mode", configpkg.DefaultStatusDetailMode, "Routine status detail mode: summary or full (preparatory; publication behavior unchanged)") + flags.StringVar(&cfg.StatusDetailMode, "status-detail-mode", configpkg.DefaultStatusDetailMode, "Routine status detail mode: summary or full") flags.BoolVar(&cfg.StatusWSEnabled, "status-ws-enabled", true, "Enable websocket status push to controller") flags.StringVar(&cfg.StatusWSURL, "status-ws-url", "", "Controller websocket URL for status push (default: ws://service/status/nodews)") flags.StringVar(&cfg.StatusWSAPIServerMode, "status-ws-apiserver-mode", statusWSAPIServerModeFallback, "API server fallback mode: never, fallback, preferred (alias for fallback); direct controller endpoints are tried first") diff --git a/cmd/unbounded-net-node/status_ack.go b/cmd/unbounded-net-node/status_ack.go index b81a85770..d457a4ce4 100644 --- a/cmd/unbounded-net-node/status_ack.go +++ b/cmd/unbounded-net-node/status_ack.go @@ -5,11 +5,14 @@ package main import ( "encoding/json" + "fmt" "sync/atomic" "google.golang.org/protobuf/proto" + netstatus "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" ) // statusAckState is created fresh for every connection. One outstanding message @@ -19,31 +22,54 @@ type statusAckState struct { resync atomic.Bool pending atomic.Bool compact atomic.Bool + summary atomic.Bool } -func (s *statusAckState) accept(data []byte) bool { +func decodeNodeStatusAck(data []byte) (*statusv1alpha1.NodeStatusAck, error) { var ack statusproto.NodeStatusAck - if err := proto.Unmarshal(data, &ack); err != nil { - var envelope struct { - Type string `json:"type"` - Data nodeStatusPushAck `json:"data"` - } - if err := json.Unmarshal(data, &envelope); err != nil { - return false - } + if err := proto.Unmarshal(data, &ack); err == nil && ack.Status != "" { + return netstatus.NodeStatusAckFromProto(&ack), nil + } + + var jsonAck statusv1alpha1.NodeStatusAck + if err := json.Unmarshal(data, &jsonAck); err == nil && jsonAck.Status != "" { + return &jsonAck, nil + } + + var envelope struct { + Type string `json:"type"` + Data statusv1alpha1.NodeStatusAck `json:"data"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, err + } - switch envelope.Type { - case "node_status_ack": - ack.Status = "ok" - case "node_status_resync": - ack.Status = "resync_required" - default: - return false + switch envelope.Type { + case "node_status_ack": + if envelope.Data.Status == "" { + envelope.Data.Status = "ok" } + case "node_status_resync": + envelope.Data.Status = "resync_required" + default: + return nil, fmt.Errorf("unrecognized status acknowledgment") + } - ack.Revision = envelope.Data.Revision + return &envelope.Data, nil +} + +func (s *statusAckState) accept(data []byte) bool { + ack, err := decodeNodeStatusAck(data) + return err == nil && s.acceptAck(ack) +} + +func (s *statusAckState) acceptAck(ack *statusv1alpha1.NodeStatusAck) bool { + if !ack.IsPublicationAck() { + return false } + s.summary.Store(ack.SummarySupported) + switch ack.Status { case "ok": s.compact.Store(ack.PeerMeasurements && ack.Revision > 0) diff --git a/cmd/unbounded-net-node/status_details.go b/cmd/unbounded-net-node/status_details.go new file mode 100644 index 000000000..0223a9047 --- /dev/null +++ b/cmd/unbounded-net-node/status_details.go @@ -0,0 +1,264 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "fmt" + "strings" + "sync" + "time" + + "google.golang.org/protobuf/proto" + "k8s.io/klog/v2" + + netstatus "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" +) + +const ( + nodeDetailFrameLimit = 2 * 1024 * 1024 + nodeDetailRetryInterval = time.Second +) + +type nodeDetailReply struct { + request statusv1alpha1.DetailRequest + payload []byte + sending bool + done bool + retryAt time.Time +} + +// One state spans both publishers and reconnects. Successful ACKs retain only +// request identity/deadline markers; no routine publisher owns detail snapshots. +type nodeDetailState struct { + mu sync.Mutex + replies map[string]*nodeDetailReply + wsWake chan struct{} + httpWake chan struct{} +} + +func (h *nodeHealthState) detailState() *nodeDetailState { + h.mu.Lock() + defer h.mu.Unlock() + + if h.details == nil { + h.details = &nodeDetailState{ + replies: make(map[string]*nodeDetailReply), + wsWake: make(chan struct{}, 1), httpWake: make(chan struct{}, 1), + } + } + + return h.details +} + +func (s *nodeDetailState) wake() { + for _, ch := range []chan struct{}{s.wsWake, s.httpWake} { + select { + case ch <- struct{}{}: + default: + } + } +} + +func (s *nodeDetailState) expireLocked(now time.Time) { + for id, reply := range s.replies { + if !reply.request.Deadline.After(now) { + delete(s.replies, id) + } + } +} + +func (s *nodeDetailState) enqueue(request *statusv1alpha1.DetailRequest, now time.Time) error { + if err := netstatus.ValidateDetailRequest(request, now); err != nil { + return err + } + + s.mu.Lock() + s.expireLocked(now) + + if _, exists := s.replies[request.RequestID]; !exists { + s.replies[request.RequestID] = &nodeDetailReply{request: *request} + time.AfterFunc(time.Until(request.Deadline), func() { + s.mu.Lock() + defer s.mu.Unlock() + + s.expireLocked(time.Now()) + }) + } + s.mu.Unlock() + s.wake() + + return nil +} + +func (s *nodeDetailState) receive(ack *statusv1alpha1.NodeStatusAck) { + if ack == nil { + return + } + + if ack.DetailRequestID != "" && ack.Status != "ok" { + klog.V(2).Infof("Detail reply %q not acknowledged: status=%s reason=%s", ack.DetailRequestID, ack.Status, ack.Reason) + } + + s.acknowledge(ack) + + if ack.DetailRequest != nil { + if err := s.enqueue(ack.DetailRequest, time.Now()); err != nil { + klog.V(2).Infof("Ignoring invalid detail command: %v", err) + } + } +} + +func (s *nodeDetailState) clear() { + s.mu.Lock() + defer s.mu.Unlock() + + clear(s.replies) +} + +func (s *nodeDetailState) acknowledge(ack *statusv1alpha1.NodeStatusAck) { + if ack == nil || ack.DetailRequestID == "" || ack.Status != "ok" { + return + } + + s.mu.Lock() + defer s.mu.Unlock() + + if reply := s.replies[ack.DetailRequestID]; reply != nil { + reply.payload = nil + reply.done = true + reply.sending = false + reply.retryAt = time.Time{} + } +} + +type nodeDetailDelivery struct { + id string + deadline time.Time + payload []byte +} + +func (s *nodeDetailState) take(nodeName string, collect func() *NodeStatusResponse, now time.Time) *nodeDetailDelivery { + s.mu.Lock() + s.expireLocked(now) + + var ( + selected *nodeDetailReply + payload []byte + ) + + for _, reply := range s.replies { + if !reply.done && !reply.sending && !now.Before(reply.retryAt) { + selected = reply + selected.sending = true + payload = selected.payload + + break + } + } + s.mu.Unlock() + + if selected == nil { + return nil + } + + if payload == nil { + payload = collectDetailPayload(nodeName, selected.request.RequestID, collect) + } + + s.mu.Lock() + defer s.mu.Unlock() + + if s.replies[selected.request.RequestID] != selected { + return nil + } + + if !selected.request.Deadline.After(time.Now()) { + delete(s.replies, selected.request.RequestID) + return nil + } + + if selected.done { + return nil + } + + selected.payload = payload + + return &nodeDetailDelivery{id: selected.request.RequestID, deadline: selected.request.Deadline, payload: payload} +} + +func (s *nodeDetailState) finish(id string) { + s.mu.Lock() + defer s.mu.Unlock() + + if reply := s.replies[id]; reply != nil && !reply.done { + reply.sending = false + reply.retryAt = time.Now().Add(nodeDetailRetryInterval) + } +} + +func detailErrorPayload(nodeName, requestID, message string) []byte { + payload, err := proto.Marshal(&statusproto.NodeStatusMessage{ + Type: statusv1alpha1.NodeStatusDetailsType, NodeName: nodeName, DetailRequestId: requestID, + DetailError: strings.ToValidUTF8(message, "?"), SupportsDetails: true, + }) + if err != nil { + klog.Errorf("Failed to encode correlated detail failure: %v", err) + return nil + } + + return payload +} + +func (s *nodeDetailState) failDelivery(nodeName string, delivery *nodeDetailDelivery, message string) []byte { + payload := detailErrorPayload(nodeName, delivery.id, message) + + s.mu.Lock() + if reply := s.replies[delivery.id]; reply != nil && !reply.done { + reply.payload = payload + } + s.mu.Unlock() + + delivery.payload = payload + + return payload +} + +func (s *nodeDetailState) wsPayload(nodeName string, delivery *nodeDetailDelivery) []byte { + if len(delivery.payload) <= nodeDetailFrameLimit { + return delivery.payload + } + + return s.failDelivery(nodeName, delivery, "detail response exceeds 2 MiB WebSocket frame limit") +} + +func collectDetailPayload(nodeName, requestID string, collect func() *NodeStatusResponse) (payload []byte) { + defer func() { + if failure := recover(); failure != nil { + payload = detailErrorPayload(nodeName, requestID, fmt.Sprintf("detail collection failed: %v", failure)) + } + }() + + full := collect() + if full == nil { + return detailErrorPayload(nodeName, requestID, "detail collection returned no snapshot") + } + + if full.FetchError != "" { + return detailErrorPayload(nodeName, requestID, full.FetchError) + } + + message := &statusproto.NodeStatusMessage{ + Type: statusv1alpha1.NodeStatusDetailsType, NodeName: nodeName, DetailRequestId: requestID, + Status: nodeStatusToProto(full), SupportsDetails: true, + } + + payload, err := proto.Marshal(message) + if err != nil { + return detailErrorPayload(nodeName, requestID, fmt.Sprintf("detail encoding failed: %v", err)) + } + + return payload +} diff --git a/cmd/unbounded-net-node/status_details_http.go b/cmd/unbounded-net-node/status_details_http.go new file mode 100644 index 000000000..c72f4b730 --- /dev/null +++ b/cmd/unbounded-net-node/status_details_http.go @@ -0,0 +1,80 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "bytes" + "compress/gzip" + "context" + "errors" + "time" +) + +const nodeDetailHTTPBodyLimit = 1024 * 1024 + +// Detail commands and retry wakeups share the existing one-in-flight HTTP loop. +// The normal publication ticker remains independent of diagnostic traffic. +func statusPushEvents(ctx context.Context, interval time.Duration, detailWake <-chan struct{}) <-chan bool { + events := make(chan bool) + + go func() { + ticker := time.NewTicker(interval) + retry := time.NewTicker(nodeDetailRetryInterval) + + defer ticker.Stop() + defer retry.Stop() + + for { + detailOnly := false + + select { + case <-ctx.Done(): + return + case <-ticker.C: + case <-detailWake: + detailOnly = true + case <-retry.C: + detailOnly = true + } + + select { + case events <- detailOnly: + case <-ctx.Done(): + return + } + } + }() + + return events +} + +func gzipStatusPayload(data []byte) ([]byte, error) { + var body bytes.Buffer + + writer, err := gzip.NewWriterLevel(&body, gzip.BestSpeed) + if err != nil { + return nil, err + } + + if _, err := writer.Write(data); err != nil { + return nil, errors.Join(err, writer.Close()) + } + + if err := writer.Close(); err != nil { + return nil, err + } + + return body.Bytes(), nil +} + +func (s *nodeDetailState) httpBody(nodeName string, delivery *nodeDetailDelivery, data []byte) ([]byte, error) { + body, err := gzipStatusPayload(data) + if err != nil || delivery == nil || len(body) <= nodeDetailHTTPBodyLimit { + return body, err + } + + payload := s.failDelivery(nodeName, delivery, "detail response exceeds 1 MiB compressed HTTP body limit") + + return gzipStatusPayload(payload) +} diff --git a/cmd/unbounded-net-node/status_details_http_test.go b/cmd/unbounded-net-node/status_details_http_test.go new file mode 100644 index 000000000..a7b4190fe --- /dev/null +++ b/cmd/unbounded-net-node/status_details_http_test.go @@ -0,0 +1,287 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "bytes" + "compress/gzip" + "context" + "crypto/rand" + "encoding/base64" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "google.golang.org/protobuf/proto" + + statusproto "github.com/Azure/unbounded/internal/net/status/proto" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func decodeTestHTTPStatus(t *testing.T, body io.Reader) *statusproto.NodeStatusMessage { + t.Helper() + + reader, err := gzip.NewReader(body) + if err != nil { + t.Error(err) + return nil + } + + data, err := io.ReadAll(reader) + _ = reader.Close() + + if err != nil { + t.Error(err) + return nil + } + + var msg statusproto.NodeStatusMessage + if err := proto.Unmarshal(data, &msg); err != nil { + t.Error(err) + return nil + } + + return &msg +} + +func TestHTTPDetailsWakeImmediatelyAndPreserveBase(t *testing.T) { + t.Setenv("UNBOUNDED_NET_CONTROLLER_SERVICE_HOST", "") + t.Setenv("KUBERNETES_SERVICE_HOST", "") + + for _, mode := range []string{"summary", "full"} { + for _, fallback := range []bool{false, true} { + t.Run(mode+map[bool]string{true: "/fallback", false: "/direct"}[fallback], func(t *testing.T) { + messages := make(chan *statusproto.NodeStatusMessage, 8) + + var detailCount atomic.Int32 + + deadline := time.Now().Add(time.Minute) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if fallback && r.URL.Path == "/direct" { + http.Error(w, "direct unavailable", http.StatusServiceUnavailable) + return + } + + if r.Header.Get("Content-Type") != "application/x-protobuf" || r.Header.Get("Content-Encoding") != "gzip" { + t.Error("detail path changed status HTTP encoding") + } + + msg := decodeTestHTTPStatus(t, r.Body) + if msg == nil { + return + } + + select { + case messages <- msg: + case <-r.Context().Done(): + return + } + + ack := statusv1alpha1.NodeStatusAck{ + Status: "ok", Revision: 7, SummarySupported: true, + DetailRequest: &statusv1alpha1.DetailRequest{RequestID: "request", Deadline: deadline}, + } + + if msg.Type == statusv1alpha1.NodeStatusDetailsType { + detailCount.Add(1) + + ack.DetailRequestID, ack.Revision = "request", 99 + } + // Shared JSON ACKs must carry the same commands as protobuf. + if err := json.NewEncoder(w).Encode(ack); err != nil { + t.Error(err) + } + })) + defer server.Close() + + cfg := &config{ + NodeName: "node-a", StatusDetailMode: mode, StatusPushEnabled: true, StatusPushDelta: true, + StatusPushURL: server.URL + "/direct", StatusPushInterval: time.Second, + StatusPushAPIServerInterval: time.Hour, StatusWSAPIServerMode: statusWSAPIServerModeFallback, + StatusWSAPIServerURL: "ws" + strings.TrimPrefix(server.URL, "http") + "/apis/status/nodews", + } + h := blockedBootstrapHealthState() + ctx, cancel := context.WithCancel(t.Context()) + startStatusPublishers(ctx, cfg, h) + + defer func() { cancel(); h.stopStatusPublishers() }() + + select { + case msg := <-messages: + if !msg.SupportsDetails || msg.Type == statusv1alpha1.NodeStatusDetailsType { + t.Fatalf("bad initial publication: %v", msg) + } + case <-time.After(3 * time.Second): + t.Fatal("no initial publication") + } + + select { + case msg := <-messages: + if msg.Type != statusv1alpha1.NodeStatusDetailsType || msg.DetailRequestId != "request" || msg.Status == nil || msg.BaseRevision != 0 { + t.Fatalf("bad immediate detail response: %v", msg) + } + case <-time.After(400 * time.Millisecond): + t.Fatal("detail response waited for routine or API fallback interval") + } + + if !fallback { + select { + case msg := <-messages: + if msg.Type == statusv1alpha1.NodeStatusDetailsType || msg.BaseRevision != 7 { + t.Fatalf("detail ACK changed publication base or duplicate recollected: %v", msg) + } + case <-time.After(2 * time.Second): + t.Fatal("no subsequent publication") + } + } + + waitForStatusCondition(t, func() bool { + state := h.detailState() + state.mu.Lock() + defer state.mu.Unlock() + + reply := state.replies["request"] + + return reply != nil && reply.done && reply.payload == nil + }) + + if detailCount.Load() != 1 { + t.Fatalf("duplicate command generated %d replies", detailCount.Load()) + } + }) + } + } +} + +func TestHTTPDetailRetryDoesNotRecollectOrWaitForRoutineTick(t *testing.T) { + h := blockedBootstrapHealthState() + state := h.detailState() + + req := &statusv1alpha1.DetailRequest{RequestID: "retry", Deadline: time.Now().Add(time.Minute)} + if err := state.enqueue(req, time.Now()); err != nil { + t.Fatal(err) + } + // Seed the immutable reply as though a previous channel disconnected. + first := state.take("node-a", h.getStatusSnapshot, time.Now()) + state.finish(first.id) + + requests := make(chan *statusproto.NodeStatusMessage, 4) + + var count atomic.Int32 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + msg := decodeTestHTTPStatus(t, r.Body) + requests <- msg + + if count.Add(1) == 1 { + http.Error(w, "temporary failure", http.StatusServiceUnavailable) + return + } + + payload, err := proto.Marshal(&statusproto.NodeStatusAck{Status: "ok", DetailRequestId: "retry", Revision: 999}) + if err != nil { + t.Error(err) + return + } + + _, _ = w.Write(payload) + })) + defer server.Close() + + cfg := &config{ + NodeName: "node-a", StatusDetailMode: "summary", StatusPushEnabled: true, + StatusPushURL: server.URL, StatusPushInterval: time.Hour, StatusWSAPIServerMode: statusWSAPIServerModeNever, + } + ctx, cancel := context.WithCancel(t.Context()) + startStatusPublishers(ctx, cfg, h) + + defer func() { cancel(); h.stopStatusPublishers() }() + + var original statusproto.NodeStatusMessage + if err := proto.Unmarshal(first.payload, &original); err != nil { + t.Fatal(err) + } + + for range 2 { + select { + case msg := <-requests: + if !proto.Equal(&original, msg) { + t.Fatal("HTTP retry recollected or changed the response") + } + case <-time.After(4 * time.Second): + t.Fatal("HTTP detail retry waited for routine publication") + } + } +} + +func TestHTTPDetailDoesNotApplyWebSocketDecodedLimit(t *testing.T) { + message := strings.Repeat("x", nodeDetailFrameLimit+1024) + + payload := collectDetailPayload("node", "large", func() *NodeStatusResponse { + return &NodeStatusResponse{NodeErrors: []NodeError{{Message: message}}} + }) + if len(payload) <= nodeDetailFrameLimit { + t.Fatal("fixture must exceed the WebSocket frame limit") + } + + state := (&nodeHealthState{}).detailState() + + body, err := state.httpBody("node", &nodeDetailDelivery{id: "large", payload: payload}, payload) + if err != nil { + t.Fatal(err) + } + + result := decodeTestHTTPStatus(t, bytes.NewReader(body)) + if len(body) > nodeDetailHTTPBodyLimit || result == nil || result.Status == nil || result.DetailError != "" || + len(result.Status.NodeErrors) != 1 || result.Status.NodeErrors[0].Message != message { + t.Fatal("HTTP incorrectly applied a decoded-body/WebSocket size cap") + } +} + +func TestHTTPDetailCompressedLimitProducesRetriableError(t *testing.T) { + random := make([]byte, nodeDetailHTTPBodyLimit+64*1024) + if _, err := rand.Read(random); err != nil { + t.Fatal(err) + } + + state := (&nodeHealthState{}).detailState() + + req := &statusv1alpha1.DetailRequest{RequestID: "large", Deadline: time.Now().Add(time.Minute)} + if err := state.enqueue(req, time.Now()); err != nil { + t.Fatal(err) + } + + collections := 0 + + delivery := state.take("node", func() *NodeStatusResponse { + collections++ + return &NodeStatusResponse{NodeErrors: []NodeError{{Message: base64.StdEncoding.EncodeToString(random)}}} + }, time.Now()) + if delivery == nil { + t.Fatal("no detail delivery") + } + + body, err := state.httpBody("node", delivery, delivery.payload) + if err != nil { + t.Fatal(err) + } + + msg := decodeTestHTTPStatus(t, bytes.NewReader(body)) + if msg == nil || msg.Status != nil || !strings.Contains(msg.DetailError, "1 MiB") || msg.DetailRequestId != "large" { + t.Fatalf("oversized detail did not produce correlated failure: %v", msg) + } + + state.finish(delivery.id) + + retry := state.take("node", func() *NodeStatusResponse { t.Fatal("oversized retry recollected"); return nil }, time.Now().Add(2*time.Second)) + if retry == nil || !bytes.Equal(retry.payload, delivery.payload) || collections != 1 { + t.Fatal("oversized payload retained or changed on retry") + } +} diff --git a/cmd/unbounded-net-node/status_details_test.go b/cmd/unbounded-net-node/status_details_test.go new file mode 100644 index 000000000..6f5e0b828 --- /dev/null +++ b/cmd/unbounded-net-node/status_details_test.go @@ -0,0 +1,178 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "bytes" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "google.golang.org/protobuf/proto" + + statusproto "github.com/Azure/unbounded/internal/net/status/proto" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func TestDetailStateCoalescesRetriesAndReleases(t *testing.T) { + h := blockedBootstrapHealthState() + state := h.detailState() + now := time.Now() + + req := &statusv1alpha1.DetailRequest{RequestID: "one", Deadline: now.Add(time.Minute)} + if err := state.enqueue(req, now); err != nil { + t.Fatal(err) + } + + originalDeadline := req.Deadline + + req.Deadline = req.Deadline.Add(time.Hour) + if err := state.enqueue(req, now); err != nil { + t.Fatal(err) + } + + count := 0 + collect := func() *NodeStatusResponse { count++; return h.getStatusSnapshot() } + + first := state.take("node-a", collect, now) + if first == nil || !first.deadline.Equal(originalDeadline) || count != 1 { + t.Fatalf("lost command deadline/collection: %+v count=%d", first, count) + } + + if state.take("node-a", collect, now) != nil { + t.Fatal("same request sent concurrently") + } + + state.finish(first.id) + + retry := state.take("node-a", collect, now.Add(2*time.Second)) + if retry == nil || !bytes.Equal(first.payload, retry.payload) || count != 1 { + t.Fatal("retry recollected or changed the snapshot") + } + + state.acknowledge(&statusv1alpha1.NodeStatusAck{Status: "ok", DetailRequestID: first.id}) + state.finish(first.id) + + if state.replies[first.id].payload != nil || !state.replies[first.id].done { + t.Fatal("ACK retained heavy payload") + } + + if err := state.enqueue(req, now); err != nil { + t.Fatal(err) + } + + if state.take("node-a", collect, now.Add(3*time.Second)) != nil || count != 1 { + t.Fatal("delayed duplicate recollected an acknowledged request") + } + + state.take("node-a", collect, originalDeadline) + + if len(state.replies) != 0 { + t.Fatal("deadline did not remove idempotency marker") + } +} + +func TestDetailStateExpiryAndConcurrentClaims(t *testing.T) { + state := (&nodeHealthState{}).detailState() + + now := time.Now() + if state.enqueue(&statusv1alpha1.DetailRequest{RequestID: "old", Deadline: now}, now) == nil { + t.Fatal("expired command accepted") + } + + req := &statusv1alpha1.DetailRequest{RequestID: "one", Deadline: now.Add(time.Minute)} + + var ( + count atomic.Int32 + wg sync.WaitGroup + ) + for range 8 { + wg.Go(func() { + if err := state.enqueue(req, now); err != nil { + t.Error(err) + } + + state.take("node", func() *NodeStatusResponse { + count.Add(1) + return &NodeStatusResponse{} + }, now) + }) + } + + wg.Wait() + + if count.Load() != 1 { + t.Fatalf("collected %d duplicate snapshots", count.Load()) + } + + state.take("node", nil, req.Deadline) + + if len(state.replies) != 0 { + t.Fatal("expired unacknowledged detail retained") + } +} + +func TestDetailDeadlineExpiresDuringCollectionAndDisconnect(t *testing.T) { + state := (&nodeHealthState{}).detailState() + + req := &statusv1alpha1.DetailRequest{RequestID: "slow", Deadline: time.Now().Add(20 * time.Millisecond)} + if err := state.enqueue(req, time.Now()); err != nil { + t.Fatal(err) + } + + delivery := state.take("node", func() *NodeStatusResponse { + <-time.After(time.Until(req.Deadline) + 10*time.Millisecond) + return &NodeStatusResponse{} + }, time.Now()) + if delivery != nil { + t.Fatal("expired collection was delivered") + } + + req = &statusv1alpha1.DetailRequest{RequestID: "disconnected", Deadline: time.Now().Add(20 * time.Millisecond)} + if err := state.enqueue(req, time.Now()); err != nil { + t.Fatal(err) + } + + if state.take("node", func() *NodeStatusResponse { return &NodeStatusResponse{} }, time.Now()) == nil { + t.Fatal("missing unacknowledged reply") + } + + waitForStatusCondition(t, func() bool { + state.mu.Lock() + defer state.mu.Unlock() + + return len(state.replies) == 0 + }) +} + +func TestDetailPayloadErrorsAreCorrelated(t *testing.T) { + for _, tc := range []struct { + name string + collect func() *NodeStatusResponse + }{ + {"nil", func() *NodeStatusResponse { return nil }}, + {"panic", func() *NodeStatusResponse { panic("failed syscall") }}, + {"fetch error", func() *NodeStatusResponse { return &NodeStatusResponse{FetchError: "unavailable"} }}, + {"oversized", func() *NodeStatusResponse { + return &NodeStatusResponse{NodeErrors: []NodeError{{Message: strings.Repeat("x", nodeDetailFrameLimit)}}} + }}, + } { + t.Run(tc.name, func(t *testing.T) { + payload := collectDetailPayload("node", "request", tc.collect) + payload = (&nodeHealthState{}).detailState().wsPayload("node", &nodeDetailDelivery{id: "request", payload: payload}) + + var message statusproto.NodeStatusMessage + if err := proto.Unmarshal(payload, &message); err != nil { + t.Fatal(err) + } + + if message.Type != statusv1alpha1.NodeStatusDetailsType || message.NodeName != "node" || + message.DetailRequestId != "request" || message.DetailError == "" || message.Status != nil || len(payload) >= nodeDetailFrameLimit { + t.Fatalf("invalid correlated failure: %v", &message) + } + }) + } +} diff --git a/cmd/unbounded-net-node/status_details_ws_test.go b/cmd/unbounded-net-node/status_details_ws_test.go new file mode 100644 index 000000000..5774dc296 --- /dev/null +++ b/cmd/unbounded-net-node/status_details_ws_test.go @@ -0,0 +1,262 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" + "google.golang.org/protobuf/proto" + + statusproto "github.com/Azure/unbounded/internal/net/status/proto" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func sendTestStatusAck(ctx context.Context, conn *websocket.Conn, ack *statusproto.NodeStatusAck) error { + payload, err := proto.Marshal(ack) + if err != nil { + return err + } + + return conn.Write(ctx, websocket.MessageBinary, payload) +} + +func TestWebSocketDetailsWakeWhilePublicationPending(t *testing.T) { + for _, tc := range []struct { + name string + mode string + oversized bool + }{ + {name: "summary", mode: "summary"}, + {name: "full", mode: "full"}, + {name: "oversized", mode: "summary", oversized: true}, + } { + t.Run(tc.name, func(t *testing.T) { + messages := make(chan *statusproto.NodeStatusMessage, 8) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer func() { _ = conn.Close(websocket.StatusNormalClosure, "done") }() + + for { + _, data, err := conn.Read(r.Context()) + if err != nil { + return + } + + var msg statusproto.NodeStatusMessage + if err := proto.Unmarshal(data, &msg); err != nil { + t.Error(err) + return + } + + select { + case messages <- &msg: + case <-r.Context().Done(): + return + } + + if msg.Type == statusv1alpha1.NodeStatusDetailsType { + if err := sendTestStatusAck(r.Context(), conn, &statusproto.NodeStatusAck{ + Status: "ok", DetailRequestId: "request", Revision: 99, + }); err != nil { + return + } + } else { + if err := sendTestStatusAck(r.Context(), conn, &statusproto.NodeStatusAck{ + Status: statusv1alpha1.DetailRequestStatus, + DetailRequest: &statusproto.DetailRequest{RequestId: "expired", DeadlineUnixNs: time.Now().Add(-time.Second).UnixNano()}, + }); err != nil { + return + } + } + // No publication ACK is sent. This command (and its delayed + // duplicate after the detail ACK) cannot release that pending ACK. + if err := sendTestStatusAck(r.Context(), conn, &statusproto.NodeStatusAck{ + Status: statusv1alpha1.DetailRequestStatus, + DetailRequest: &statusproto.DetailRequest{RequestId: "request", DeadlineUnixNs: time.Now().Add(time.Minute).UnixNano()}, + }); err != nil { + return + } + } + })) + defer server.Close() + + s := summaryRouteFixture() + s.cfg.NodeName = "node-a" + + var collections atomic.Int32 + + s.bpfCollector = func() []BpfEntry { + collections.Add(1) + + if tc.oversized { + return []BpfEntry{{CIDR: strings.Repeat("x", nodeDetailFrameLimit)}} + } + + return []BpfEntry{{CIDR: "10.0.0.0/8"}} + } + h := blockedBootstrapHealthState() + h.setStatusServer(s) + + cfg := &config{ + NodeName: "node-a", StatusDetailMode: tc.mode, StatusWSEnabled: true, + StatusWSURL: "ws" + strings.TrimPrefix(server.URL, "http"), StatusWSAPIServerMode: statusWSAPIServerModeNever, + CriticalDeltaEvery: time.Hour, StatsDeltaEvery: time.Hour, FullSyncEvery: time.Hour, + } + ctx, cancel := context.WithCancel(t.Context()) + startStatusPublishers(ctx, cfg, h) + + defer func() { cancel(); h.stopStatusPublishers() }() + + for i := range 2 { + select { + case msg := <-messages: + if !msg.SupportsDetails { + t.Fatal("missing responder capability") + } + + if i == 1 && (msg.Type != statusv1alpha1.NodeStatusDetailsType || msg.DetailRequestId != "request" || + msg.BaseRevision != 0) { + t.Fatalf("invalid immediate detail reply: %v", msg) + } + + if i == 1 { + if tc.oversized { + if msg.Status != nil || !strings.Contains(msg.DetailError, "WebSocket frame limit") { + t.Fatalf("oversized websocket details were not rejected: %v", msg) + } + } else if msg.Status == nil || len(msg.Status.BpfEntries) != 1 || msg.DetailError != "" { + t.Fatalf("detail response lost payload: %v", msg) + } + } + case <-time.After(3 * time.Second): + t.Fatal("detail command waited for a periodic publication tick") + } + } + + waitForStatusCondition(t, func() bool { + state := h.detailState() + state.mu.Lock() + defer state.mu.Unlock() + + reply := state.replies["request"] + + return reply != nil && reply.done && reply.payload == nil + }) + + select { + case msg := <-messages: + t.Fatalf("duplicate or expired command produced another reply: %v", msg) + case <-time.After(100 * time.Millisecond): + } + + want := int32(1) + if tc.mode == "full" { + want++ + } + + if collections.Load() != want { + t.Fatalf("collected %d full snapshots, want %d", collections.Load(), want) + } + }) + } +} + +func TestWebSocketDetailRetrySurvivesReconnect(t *testing.T) { + deliveries := make(chan []byte, 4) + + var connections atomic.Int32 + + deadline := time.Now().Add(time.Minute) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + + number := connections.Add(1) + + defer func() { _ = conn.Close(websocket.StatusGoingAway, "reconnect") }() + + for { + _, data, err := conn.Read(r.Context()) + if err != nil { + return + } + + var msg statusproto.NodeStatusMessage + if err := proto.Unmarshal(data, &msg); err != nil { + t.Error(err) + return + } + + if msg.Type == statusv1alpha1.NodeStatusDetailsType { + deliveries <- data + + if number == 1 { + return + } + + if err := sendTestStatusAck(r.Context(), conn, &statusproto.NodeStatusAck{Status: "ok", DetailRequestId: "retry"}); err != nil { + return + } + } else if err := sendTestStatusAck(r.Context(), conn, &statusproto.NodeStatusAck{ + Status: "ok", Revision: 1, SummarySupported: true, + DetailRequest: &statusproto.DetailRequest{RequestId: "retry", DeadlineUnixNs: deadline.UnixNano()}, + }); err != nil { + return + } + } + })) + defer server.Close() + + h := blockedBootstrapHealthState() + s := summaryRouteFixture() + s.cfg.NodeName = "node-a" + + var collections atomic.Int32 + + s.bpfCollector = func() []BpfEntry { collections.Add(1); return nil } + h.setStatusServer(s) + + cfg := &config{ + NodeName: "node-a", StatusDetailMode: "summary", StatusWSEnabled: true, + StatusWSURL: "ws" + strings.TrimPrefix(server.URL, "http"), StatusWSAPIServerMode: statusWSAPIServerModeNever, + CriticalDeltaEvery: time.Hour, StatsDeltaEvery: time.Hour, FullSyncEvery: time.Hour, + } + ctx, cancel := context.WithCancel(t.Context()) + startStatusPublishers(ctx, cfg, h) + + defer func() { cancel(); h.stopStatusPublishers() }() + + var first []byte + + for range 2 { + select { + case data := <-deliveries: + if first != nil && string(first) != string(data) { + t.Fatal("reconnect changed the collected reply") + } + + first = data + case <-time.After(4 * time.Second): + t.Fatal("outstanding detail was not retried after reconnect") + } + } + + if collections.Load() != 1 { + t.Fatal("reconnect recollected details") + } +} diff --git a/cmd/unbounded-net-node/status_publication.go b/cmd/unbounded-net-node/status_publication.go new file mode 100644 index 000000000..4a35fda7b --- /dev/null +++ b/cmd/unbounded-net-node/status_publication.go @@ -0,0 +1,77 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "reflect" + "time" + + statusproto "github.com/Azure/unbounded/internal/net/status/proto" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +const nodeErrorSummaryUnsupported = "status-summary-unsupported" + +func collectPublication(health *nodeHealthState, cfg *config, previous *NodeStatusResponse, force bool, revision uint64) (*statusproto.NodeStatusMessage, *NodeStatusResponse) { + if cfg.StatusDetailMode == "summary" { + summary := health.getSummarySnapshot() + + return &statusproto.NodeStatusMessage{ + Type: statusv1alpha1.NodeStatusSummaryType, NodeName: summary.NodeInfo.Name, + BaseRevision: revision, Summary: nodeSummaryToProto(summary), + SupportsDetails: true, + }, nil + } + + full := health.getStatusSnapshot() + + msg := &statusproto.NodeStatusMessage{Type: "node_status_full", NodeName: full.NodeInfo.Name, SupportsDetails: true} + if cfg.StatusPushDelta && !force { + msg.Delta = typedStatusDelta(previous, full, false, true) + if msg.Delta != nil { + msg.Type, msg.BaseRevision = "node_status_delta", revision + } + } + + if msg.Delta == nil { + msg.Status = nodeStatusToProto(full) + } + + return msg, full +} + +func publicationNodeErrors(errors []NodeError) []NodeError { + result := make([]NodeError, 0, len(errors)) + for _, err := range errors { + switch err.Type { + case nodeErrorTypeDirectPush, nodeErrorTypeDirectWebSocket, nodeErrorTypeFallbackPush, nodeErrorTypeFallbackWS: + continue + default: + result = append(result, err) + } + } + + return result +} + +func equalPublicationSummaries(a, b *NodeStatusOverview) bool { + if a == nil || b == nil { + return a == b + } + + normalize := func(summary *NodeStatusOverview) NodeStatusOverview { + result := *summary + result.Timestamp = time.Time{} + + if summary.HealthCheck != nil { + health := *summary.HealthCheck + health.CheckedAt = time.Time{} + result.HealthCheck = &health + } + + return result + } + + return reflect.DeepEqual(normalize(a), normalize(b)) +} diff --git a/cmd/unbounded-net-node/status_publication_test.go b/cmd/unbounded-net-node/status_publication_test.go new file mode 100644 index 000000000..4747d1a8c --- /dev/null +++ b/cmd/unbounded-net-node/status_publication_test.go @@ -0,0 +1,205 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "compress/gzip" + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/coder/websocket" + "google.golang.org/protobuf/proto" + + netstatus "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 TestSummaryPublicationNeverCollectsDetails(t *testing.T) { + s := summaryRouteFixture() + s.bpfCollector = func() []BpfEntry { t.Fatal("routine summary collected BPF"); return nil } + h := blockedBootstrapHealthState() + h.setStatusServer(s) + + cfg := &config{StatusDetailMode: "summary", StatusPushDelta: true} + for _, force := range []bool{true, false} { + msg, base := collectPublication(h, cfg, &NodeStatusResponse{Peers: make([]WireGuardPeerStatus, 100)}, force, 42) + if base != nil || msg.Status != nil || msg.Delta != nil || msg.Type != statusv1alpha1.NodeStatusSummaryType || msg.Summary == nil { + t.Fatalf("summary retained details: %v base=%v", msg, base) + } + + if len(msg.Summary.NodeErrors) == 0 || msg.BaseRevision != 42 { + t.Fatal("summary lost guard/revision") + } + } +} + +func TestDetailACKDoesNotReleasePublication(t *testing.T) { + state := &statusAckState{} + state.revision.Store(7) + state.pending.Store(true) + + for _, ack := range []*statusv1alpha1.NodeStatusAck{ + {Status: statusv1alpha1.DetailRequestStatus, DetailRequest: &statusv1alpha1.DetailRequest{RequestID: "r", Deadline: time.Now().Add(time.Minute)}}, + {Status: "ok", DetailRequestID: "r", Revision: 99, SummarySupported: true}, + } { + data, err := proto.Marshal(netstatus.NodeStatusAckToProto(ack)) + if err != nil { + t.Fatal(err) + } + + if state.accept(data) || !state.pending.Load() || state.revision.Load() != 7 || state.summary.Load() { + t.Fatal("detail traffic changed publication ACK state") + } + + data, err = json.Marshal(ack) + if err != nil { + t.Fatal(err) + } + + if state.accept(data) || !state.pending.Load() || state.revision.Load() != 7 || state.summary.Load() { + t.Fatal("JSON detail traffic changed publication ACK state") + } + } +} + +func TestRoutineSummaryPublishers(t *testing.T) { + for _, transport := range []string{"HTTP", "WS"} { + for _, supported := range []bool{true, false} { + t.Run(transport+map[bool]string{true: "/supported", false: "/unsupported"}[supported], func(t *testing.T) { + messages := make(chan *statusproto.NodeStatusMessage, 32) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handle := func(data []byte) []byte { + var msg statusproto.NodeStatusMessage + if err := proto.Unmarshal(data, &msg); err != nil { + t.Error(err) + } + + select { + case messages <- &msg: + case <-r.Context().Done(): + } + + ack, _ := proto.Marshal(&statusproto.NodeStatusAck{Status: "resync_required", Revision: 3, SummarySupported: supported}) + + return ack + } + + if transport == "WS" { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer func() { _ = conn.Close(websocket.StatusNormalClosure, "done") }() + + for { + _, data, err := conn.Read(r.Context()) + if err != nil { + return + } + + if err := conn.Write(r.Context(), websocket.MessageBinary, handle(data)); err != nil { + return + } + } + } + + reader, err := gzip.NewReader(r.Body) + if err != nil { + t.Error(err) + return + } + + data, err := io.ReadAll(reader) + _ = reader.Close() + + if err != nil { + t.Error(err) + return + } + + _, _ = w.Write(handle(data)) + })) + defer server.Close() + + cfg := &config{ + NodeName: "node-a", StatusDetailMode: "summary", StatusPushEnabled: transport == "HTTP", + StatusWSEnabled: transport == "WS", StatusPushURL: server.URL, StatusWSURL: "ws" + strings.TrimPrefix(server.URL, "http"), + StatusPushInterval: 5 * time.Millisecond, StatusPushDelta: true, StatusWSAPIServerMode: statusWSAPIServerModeNever, + CriticalDeltaEvery: 5 * time.Millisecond, StatsDeltaEvery: 7 * time.Millisecond, FullSyncEvery: 9 * time.Millisecond, + } + h := blockedBootstrapHealthState() + ctx, cancel := context.WithCancel(t.Context()) + startStatusPublishers(ctx, cfg, h) + + defer func() { cancel(); h.stopStatusPublishers() }() + + want := 3 + if !supported { + want = 1 + } + + for range want { + select { + case msg := <-messages: + if msg.Type != statusv1alpha1.NodeStatusSummaryType || msg.Summary == nil || msg.Status != nil || msg.Delta != nil { + t.Fatalf("unexpected routine message: %v", msg) + } + + if len(msg.Summary.NodeErrors) == 0 || msg.Summary.NodeInfo.Name != "node-a" { + t.Fatal("lost bootstrap guard/identity") + } + case <-time.After(3 * time.Second): + t.Fatal("no summary received") + } + } + + if supported { + h.setCNIReady("cbr0", []string{"10.244.7.0/24"}) + + recovered := false + timeout := time.After(3 * time.Second) + + for !recovered { + select { + case msg := <-messages: + if msg.Summary == nil || msg.Status != nil || msg.Delta != nil { + t.Fatalf("recovery published details: %v", msg) + } + + recovered = true + + for _, nodeError := range msg.Summary.NodeErrors { + if nodeError.Type == configPodCIDRGuard { + recovered = false + } + } + case <-timeout: + t.Fatal("summary did not publish CNI recovery") + } + } + } + + if !supported { + waitForStatusCondition(t, func() bool { + for _, err := range h.getSummarySnapshot().NodeErrors { + if err.Type == nodeErrorSummaryUnsupported { + return true + } + } + + return false + }) + } + }) + } + } +} diff --git a/cmd/unbounded-net-node/status_server.go b/cmd/unbounded-net-node/status_server.go index d05ccd005..9412fa2c9 100644 --- a/cmd/unbounded-net-node/status_server.go +++ b/cmd/unbounded-net-node/status_server.go @@ -5,7 +5,6 @@ package main import ( "bytes" - "compress/gzip" "context" "crypto/tls" "crypto/x509" @@ -35,6 +34,7 @@ import ( "github.com/Azure/unbounded/internal/net/metrics" unboundednetnetlink "github.com/Azure/unbounded/internal/net/netlink" statusproto "github.com/Azure/unbounded/internal/net/status/proto" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" ) const routingTableRefreshBackstop = 30 * time.Second @@ -60,6 +60,7 @@ type nodeHealthState struct { statusTransportWg *sync.WaitGroup statusTransportCancel context.CancelFunc statusTransportStop sync.Once + details *nodeDetailState mu sync.RWMutex } @@ -99,6 +100,14 @@ func (h *nodeHealthState) stopStatusPublishers() { if wg != nil { wg.Wait() } + + h.mu.RLock() + details := h.details + h.mu.RUnlock() + + if details != nil { + details.clear() + } }) } @@ -551,14 +560,6 @@ func startHealthServer(port int, healthState *nodeHealthState) { } } -// nodeStatusPushAck is the JSON acknowledgment returned by the controller for push updates. -// Kept for backward-compatible JSON fallback parsing during protobuf rollout. -type nodeStatusPushAck struct { - Status string `json:"status"` - Revision uint64 `json:"revision,omitempty"` - Reason string `json:"reason,omitempty"` -} - const ( statusWSAPIServerModeNever = "never" statusWSAPIServerModeFallback = "fallback" @@ -1055,6 +1056,7 @@ func runStatusWebSocketPusher( dialHTTPClient *http.Client, hmacMgr *hmacTokenManager, ) { + details := healthState.detailState() // Reuse the push client's TLS trust setup so wss://KUBERNETES_SERVICE_HOST // can validate the cluster CA in fallback/preferred API server modes. // Keep timeout disabled for long-lived websocket connections. @@ -1272,14 +1274,15 @@ func runStatusWebSocketPusher( } var ( - conn *websocket.Conn - wsURL string - directErr error - fallbackErr error - directTried bool - fallbackTried bool - successes []dialResult - initialStatus *NodeStatusResponse + conn *websocket.Conn + wsURL string + directErr error + fallbackErr error + directTried bool + fallbackTried bool + successes []dialResult + initialStatus *NodeStatusResponse + initialSummary *NodeStatusOverview ) if recovered != nil { @@ -1287,6 +1290,7 @@ func runStatusWebSocketPusher( connCtx, connCancel = recovered.ctx, recovered.cancel initialStatus = recovered.status + initialSummary = recovered.summary successes = append(successes, dialResult{url: directWSURL, isDirect: true, conn: recovered.conn}) recovered = nil } @@ -1379,6 +1383,7 @@ func runStatusWebSocketPusher( var ( lastSentStatus *NodeStatusResponse + lastSentSummary *NodeStatusOverview lastCriticalSnapshot *NodeStatusResponse acks statusAckState lastAckTimeNs atomic.Int64 @@ -1387,7 +1392,7 @@ func runStatusWebSocketPusher( lastAckTimeNs.Store(time.Now().UnixNano()) - if initialStatus != nil { + if initialStatus != nil || initialSummary != nil { acks.pending.Store(true) lastWriteTime = time.Now() @@ -1405,13 +1410,60 @@ func runStatusWebSocketPusher( return } - if acks.accept(data) { + ack, err := decodeNodeStatusAck(data) + if err != nil { + continue + } + + details.receive(ack) + + if ack.IsPublicationAck() && cfg.StatusDetailMode == "summary" && !ack.SummarySupported { + appendNodeError(healthState, nodeErrorSummaryUnsupported, "controller does not advertise summary support; full publication is disabled in summary mode") + return + } + + if acks.acceptAck(ack) { lastAckTimeNs.Store(time.Now().UnixNano()) + + clearNodeErrorsByTypes(healthState, nodeErrorSummaryUnsupported) } } }() + sendSummary := func(onlyChanged bool) error { + summary := healthState.getSummarySnapshot() + + summary.NodeErrors = publicationNodeErrors(summary.NodeErrors) + if onlyChanged && equalPublicationSummaries(lastSentSummary, summary) { + return nil + } + + payload, err := marshalStatusWebSocketSummary(summary, acks.revision.Load()) + if err != nil { + return err + } + + acks.resync.Store(false) + acks.pending.Store(true) + + lastWriteTime = time.Now() + + if err := conn.Write(connCtx, websocket.MessageBinary, payload); err != nil { + return err + } + + lastSentSummary = summary + + clearNodeErrorsByTypes(healthState, nodeErrorTypeDirectPush, nodeErrorTypeDirectWebSocket, nodeErrorTypeFallbackPush, nodeErrorTypeFallbackWS) + + return nil + } + sendFull := func() error { + if cfg.StatusDetailMode == "summary" { + return sendSummary(false) + } + status, payload, err := marshalStatusWebSocketFull(healthState) if err != nil { return err @@ -1436,11 +1488,15 @@ func runStatusWebSocketPusher( var initialSendErr error - if initialStatus != nil { - // The recovery candidate already sent this full snapshot. Preserve its + if initialStatus != nil || initialSummary != nil { + // The recovery candidate already sent its initial snapshot. Preserve its // delta base and let the reader consume its queued ACK without resending. lastSentStatus = initialStatus - lastCriticalSnapshot = stripPeerStats(initialStatus) + lastSentSummary = initialSummary + + if initialStatus != nil { + lastCriticalSnapshot = stripPeerStats(initialStatus) + } clearNodeErrorsByTypes(healthState, nodeErrorTypeDirectPush, nodeErrorTypeDirectWebSocket, nodeErrorTypeFallbackPush, nodeErrorTypeFallbackWS) } else { @@ -1536,6 +1592,20 @@ func runStatusWebSocketPusher( } fallbackCloseTicker := time.NewTicker(500 * time.Millisecond) + sendDetails := func() error { + delivery := details.take(cfg.NodeName, healthState.getStatusSnapshot, time.Now()) + if delivery == nil { + return nil + } + defer details.finish(delivery.id) + + detailCtx, cancel := context.WithDeadline(ctx, delivery.deadline) + defer cancel() + + return conn.Write(detailCtx, websocket.MessageBinary, details.wsPayload(cfg.NodeName, delivery)) + } + + details.wake() loop: for { @@ -1544,11 +1614,24 @@ func runStatusWebSocketPusher( break loop case <-readCtx.Done(): break loop + case <-details.wsWake: + if err := sendDetails(); err != nil { + klog.V(2).Infof("Status websocket: detail reply write failed: %v", err) + break loop + } case <-criticalTicker.C: if acks.pending.Load() { continue } + if cfg.StatusDetailMode == "summary" { + if err := sendSummary(!acks.resync.Load()); err != nil { + break loop + } + + continue + } + if acks.resync.Load() || lastSentStatus == nil { if err := sendFull(); err != nil { klog.V(2).Infof("Status websocket: resync full send failed: %v", err) @@ -1573,10 +1656,11 @@ func runStatusWebSocketPusher( } message := &statusproto.NodeStatusMessage{ - Type: "node_status_delta", - NodeName: current.NodeInfo.Name, - BaseRevision: acks.revision.Load(), - Delta: delta, + Type: "node_status_delta", + NodeName: current.NodeInfo.Name, + BaseRevision: acks.revision.Load(), + Delta: delta, + SupportsDetails: true, } payload, err := proto.Marshal(message) @@ -1600,6 +1684,14 @@ func runStatusWebSocketPusher( continue } + if cfg.StatusDetailMode == "summary" { + if err := sendSummary(false); err != nil { + break loop + } + + continue + } + if acks.resync.Load() || lastSentStatus == nil { if err := sendFull(); err != nil { klog.V(2).Infof("Status websocket: stats resync failed: %v", err) @@ -1624,10 +1716,11 @@ func runStatusWebSocketPusher( } wsMsg := &statusproto.NodeStatusMessage{ - Type: "node_status_delta", - NodeName: current.NodeInfo.Name, - BaseRevision: acks.revision.Load(), - Delta: delta, + Type: "node_status_delta", + NodeName: current.NodeInfo.Name, + BaseRevision: acks.revision.Load(), + Delta: delta, + SupportsDetails: true, } payload, err := proto.Marshal(wsMsg) @@ -1686,7 +1779,7 @@ func runStatusWebSocketPusher( keepaliveFailures = 0 } case <-directRecoveryCh: - recovered = tryDirectRecoveryProbe(ctx, healthState, dialHTTPClient, getToken, hmacMgr.invalidate, directWSURL, cfg.NodeName) + recovered = tryDirectRecoveryProbe(ctx, healthState, dialHTTPClient, getToken, hmacMgr.invalidate, directWSURL, cfg.NodeName, cfg.StatusDetailMode) if recovered != nil { klog.V(2).Info("Status websocket: promoting initialized direct connection from API server fallback") break loop @@ -1697,6 +1790,10 @@ func runStatusWebSocketPusher( directRecoveryTimer.Reset(directRecoveryBackoff) } case <-fallbackCloseTicker.C: + if err := sendDetails(); err != nil { + break loop + } + if acks.pending.Load() && time.Since(lastWriteTime) > 30*time.Second { klog.V(2).Info("Status websocket: status acknowledgment timed out") break loop @@ -1705,7 +1802,7 @@ func runStatusWebSocketPusher( if wsURL == fallbackWSURL && closeFallbackWS != nil && closeFallbackWS.Load() { closeFallbackWS.Store(false) - recovered = tryDirectRecoveryProbe(ctx, healthState, dialHTTPClient, getToken, hmacMgr.invalidate, directWSURL, cfg.NodeName) + recovered = tryDirectRecoveryProbe(ctx, healthState, dialHTTPClient, getToken, hmacMgr.invalidate, directWSURL, cfg.NodeName, cfg.StatusDetailMode) if recovered != nil { klog.V(2).Info("Status websocket: promoting initialized direct connection after HTTP recovery") @@ -1746,6 +1843,8 @@ func runStatusWebSocketPusher( if wsMode != nil { wsMode.Store(statusWSModeNone) } + + details.wake() // Send a graceful WebSocket close frame. Use StatusNormalClosure // for clean shutdown and StatusGoingAway for reconnect scenarios. // conn.Close has its own 5s timeout for the close handshake. @@ -1760,46 +1859,50 @@ func runStatusWebSocketPusher( _ = conn.Close(closeCode, closeReason) //nolint:errcheck connCancel() // tear down the detached connection context after graceful close + + if cfg.StatusDetailMode == "summary" && !acks.summary.Load() { + if wsURL == directWSURL { + nextDirectAttemptAt = time.Now().Add(5 * time.Second) + } else { + nextFallbackAttemptAt = time.Now().Add(5 * time.Second) + } + } + klog.V(4).Info("Status websocket disconnected") } } func marshalStatusWebSocketFull(healthState *nodeHealthState) (*NodeStatusResponse, []byte, error) { status := healthState.getStatusSnapshot() - if len(status.NodeErrors) > 0 { - // Publish a clean snapshot, but retain local transport errors until the - // write succeeds and the connection is selected for publishing. - filtered := make([]NodeError, 0, len(status.NodeErrors)) - for _, nodeError := range status.NodeErrors { - switch nodeError.Type { - case nodeErrorTypeDirectPush, nodeErrorTypeDirectWebSocket, nodeErrorTypeFallbackPush, nodeErrorTypeFallbackWS: - continue - default: - filtered = append(filtered, nodeError) - } - } - - status.NodeErrors = filtered - } + status.NodeErrors = publicationNodeErrors(status.NodeErrors) payload, err := proto.Marshal(&statusproto.NodeStatusMessage{ - Type: "node_status_full", - NodeName: status.NodeInfo.Name, - Status: nodeStatusToProto(status), + Type: "node_status_full", + NodeName: status.NodeInfo.Name, + Status: nodeStatusToProto(status), + SupportsDetails: true, }) return status, payload, err } +func marshalStatusWebSocketSummary(summary *NodeStatusOverview, revision uint64) ([]byte, error) { + return proto.Marshal(&statusproto.NodeStatusMessage{ + Type: statusv1alpha1.NodeStatusSummaryType, NodeName: summary.NodeInfo.Name, + BaseRevision: revision, Summary: nodeSummaryToProto(summary), SupportsDetails: true, + }) +} + type initializedStatusWebSocket struct { - conn *websocket.Conn - ctx context.Context - cancel context.CancelFunc - status *NodeStatusResponse + conn *websocket.Conn + ctx context.Context + cancel context.CancelFunc + status *NodeStatusResponse + summary *NodeStatusOverview } // tryDirectRecoveryProbe prepares the connection that will replace fallback. -// Like normal initialization, success means the full write completed, not that +// Like normal initialization, success means the initial write completed, not that // the controller ACKed it. The publisher owns the returned connection and ACK. func tryDirectRecoveryProbe( ctx context.Context, @@ -1809,6 +1912,7 @@ func tryDirectRecoveryProbe( invalidateToken func(), directWSURL string, nodeName string, + detailMode string, ) *initializedStatusWebSocket { if directWSURL != "" { headers := http.Header{} @@ -1830,7 +1934,20 @@ func tryDirectRecoveryProbe( } if err == nil { - status, payload, sendErr := marshalStatusWebSocketFull(healthState) + var ( + status *NodeStatusResponse + summary *NodeStatusOverview + payload []byte + sendErr error + ) + if detailMode == "summary" { + summary = healthState.getSummarySnapshot() + summary.NodeErrors = publicationNodeErrors(summary.NodeErrors) + payload, sendErr = marshalStatusWebSocketSummary(summary, 0) + } else { + status, payload, sendErr = marshalStatusWebSocketFull(healthState) + } + if sendErr == nil { sendErr = conn.Write(probeCtx, websocket.MessageBinary, payload) } @@ -1840,7 +1957,7 @@ func tryDirectRecoveryProbe( clearNodeErrorsByTypes(healthState, nodeErrorTypeDirectPush, nodeErrorTypeDirectWebSocket) - return &initializedStatusWebSocket{conn: conn, ctx: connCtx, cancel: connCancel, status: status} + return &initializedStatusWebSocket{conn: conn, ctx: connCtx, cancel: connCancel, status: status, summary: summary} } if closeErr := conn.CloseNow(); closeErr != nil { @@ -2002,14 +2119,14 @@ func startStatusPusher( ) defer requests.Wait() - ticker := time.NewTicker(cfg.StatusPushInterval) - defer ticker.Stop() + details := healthState.detailState() + events := statusPushEvents(ctx, cfg.StatusPushInterval, details.httpWake) for { select { case <-ctx.Done(): return - case <-ticker.C: + case detailOnly := <-events: currentWSMode := statusWSModeNone if wsMode != nil { currentWSMode = wsMode.Load() @@ -2042,10 +2159,21 @@ func startStatusPusher( continue } + if detailOnly && currentWSMode != statusWSModeNone { + continue + } + // Collect status and prepare the request body synchronously. collectStart := time.Now() - nodeStatus := healthState.getStatusSnapshot() - collectDuration := time.Since(collectStart) + + var delivery *nodeDetailDelivery + if currentWSMode == statusWSModeNone { + delivery = details.take(cfg.NodeName, healthState.getStatusSnapshot, time.Now()) + } + + if detailOnly && delivery == nil { + continue + } pushStateMu.Lock() currentForceFull := forceFullPush @@ -2053,68 +2181,51 @@ func startStatusPusher( previousStatus := lastSentStatus pushStateMu.Unlock() - mode := "full" + var ( + nodeStatus *NodeStatusResponse + data []byte + mode string + ) + if delivery != nil { + data, mode = delivery.payload, statusv1alpha1.NodeStatusDetailsType + } else { + var protoMsg *statusproto.NodeStatusMessage - protoMsg := &statusproto.NodeStatusMessage{ - Type: "node_status_full", - NodeName: nodeStatus.NodeInfo.Name, - } - if cfg.StatusPushDelta && !currentForceFull { - delta := typedStatusDelta(previousStatus, nodeStatus, false, true) - if delta != nil { - mode = "delta" - protoMsg.Type = "node_status_delta" - protoMsg.BaseRevision = currentRevision - protoMsg.Status = nil - protoMsg.Delta = delta - } - } + protoMsg, nodeStatus = collectPublication(healthState, cfg, previousStatus, currentForceFull, currentRevision) - if protoMsg.Delta == nil { - protoMsg.Status = nodeStatusToProto(nodeStatus) - } + var err error - marshalStart := time.Now() + data, err = proto.Marshal(protoMsg) + if err != nil { + klog.V(3).Infof("Status push: failed to marshal protobuf status: %v", err) + continue + } - data, err := proto.Marshal(protoMsg) - if err != nil { - klog.V(3).Infof("Status push: failed to marshal protobuf status: %v", err) - continue + mode = protoMsg.Type } - // Gzip-compress the protobuf body to reduce bandwidth - var compressed bytes.Buffer + collectDuration := time.Since(collectStart) + marshalStart := time.Now() - gz, err := gzip.NewWriterLevel(&compressed, gzip.BestSpeed) + body, err := details.httpBody(cfg.NodeName, delivery, data) if err != nil { - klog.V(3).Infof("Status push: failed to init gzip writer: %v", err) - continue - } - - if _, err := gz.Write(data); err != nil { - _ = gz.Close() //nolint:errcheck + if delivery != nil { + details.finish(delivery.id) + } klog.V(3).Infof("Status push: failed to gzip status: %v", err) continue } - if err := gz.Close(); err != nil { - klog.V(3).Infof("Status push: failed to finalize gzip: %v", err) - continue - } - prepareDuration := time.Since(marshalStart) if collectDuration > 2*time.Second { - klog.Warningf("Status push: getNodeStatus() took %v (marshal+gzip: %v, body: %d bytes, mode=%s)", collectDuration, prepareDuration, compressed.Len(), mode) + klog.Warningf("Status push: getNodeStatus() took %v (marshal+gzip: %v, body: %d bytes, mode=%s)", collectDuration, prepareDuration, len(body), mode) } else { - klog.V(4).Infof("Status push: collected in %v, prepared in %v (%d bytes, mode=%s)", collectDuration, prepareDuration, compressed.Len(), mode) + klog.V(4).Infof("Status push: collected in %v, prepared in %v (%d bytes, mode=%s)", collectDuration, prepareDuration, len(body), mode) } - // Copy the compressed data so the goroutine owns it - body := compressed.Bytes() - // Send HTTP POST in background so slow network doesn't block the ticker. // The ticker loop stays responsive and can fire the next push on time. pushInFlight.Store(true) @@ -2122,7 +2233,14 @@ func startStatusPusher( go func(mode string, statusCopy *NodeStatusResponse) { defer requests.Done() - defer pushInFlight.Store(false) + defer func() { + if delivery != nil { + details.finish(delivery.id) + } + + pushInFlight.Store(false) + details.wake() + }() postStart := time.Now() @@ -2147,6 +2265,10 @@ func startStatusPusher( return false } + if delivery != nil { + return true + } + interval := cfg.StatusPushAPIServerInterval if interval <= 0 { interval = 30 * time.Second @@ -2162,8 +2284,16 @@ func startStatusPusher( postTo := func(targetURL, targetLabel string) (bool, bool) { pushStart := time.Now() + requestCtx := ctx - req, err := http.NewRequestWithContext(ctx, http.MethodPost, targetURL, bytes.NewReader(body)) + if delivery != nil { + var cancel context.CancelFunc + + requestCtx, cancel = context.WithDeadline(ctx, delivery.deadline) + defer cancel() + } + + req, err := http.NewRequestWithContext(requestCtx, http.MethodPost, targetURL, bytes.NewReader(body)) if err != nil { klog.V(2).Infof("Status push: failed to create %s request: %v", targetLabel, err) @@ -2231,22 +2361,18 @@ func startStatusPusher( defer func() { _ = resp.Body.Close() }() //nolint:errcheck - var ack statusproto.NodeStatusAck + ack := &statusv1alpha1.NodeStatusAck{} - respBody, readErr := io.ReadAll(resp.Body) + respBody, readErr := io.ReadAll(io.LimitReader(resp.Body, 64*1024+1)) if readErr != nil { klog.V(4).Infof("Status push: failed to read %s response body: %v", targetLabel, readErr) - } else if protoErr := proto.Unmarshal(respBody, &ack); protoErr != nil { - // Fallback: try JSON for backward compatibility during rollout. - var jsonAck nodeStatusPushAck - if json.Unmarshal(respBody, &jsonAck) == nil { - ack.Revision = jsonAck.Revision - ack.Status = jsonAck.Status - ack.Reason = jsonAck.Reason - } + } else if decoded, err := decodeNodeStatusAck(respBody); err == nil && len(respBody) <= 64*1024 { + ack = decoded } - if resp.StatusCode == http.StatusTooManyRequests { + if resp.StatusCode == http.StatusTooManyRequests && delivery == nil && + ack.DetailRequestID == "" && ack.Status != statusv1alpha1.DetailRequestStatus { + details.receive(ack) pushStateMu.Lock() forceFullPush = true lastAckRevision = ack.Revision @@ -2295,19 +2421,44 @@ func startStatusPusher( return false, false } - pushStateMu.Lock() - if ack.Revision > 0 { - lastAckRevision = ack.Revision + if delivery != nil && (ack.DetailRequestID != delivery.id || ack.Status != "ok") { + appendNodeError(healthState, "status-details", "controller did not acknowledge the correlated detail response") + return false, false + } + + details.receive(ack) + + legacyEmptyACK := ack.Status == "" && ack.DetailRequestID == "" && ack.DetailRequest == nil + if delivery == nil && cfg.StatusDetailMode == "summary" && + (ack.IsPublicationAck() || legacyEmptyACK) && !ack.SummarySupported { + appendNodeError(healthState, nodeErrorSummaryUnsupported, "controller does not advertise summary support; full publication is disabled in summary mode") + return false, true } - lastSentStatus = statusCopy - forceFullPush = false + if delivery == nil && ack.IsPublicationAck() && ack.SummarySupported { + clearNodeErrorsByTypes(healthState, nodeErrorSummaryUnsupported) + } else if delivery != nil { + clearNodeErrorsByTypes(healthState, "status-details") + } + + pushStateMu.Lock() + + if delivery == nil && (ack.IsPublicationAck() || legacyEmptyACK) { + if ack.Revision > 0 { + lastAckRevision = ack.Revision + } + + lastSentStatus = statusCopy + forceFullPush = ack.Status == "resync_required" + } pushStateMu.Unlock() clearNodeErrorsByTypes(healthState, nodeErrorTypeDirectPush, nodeErrorTypeFallbackPush) switch targetLabel { case "apiserver": - lastAPIServerPushUnix.Store(time.Now().UnixNano()) + if delivery == nil { + lastAPIServerPushUnix.Store(time.Now().UnixNano()) + } case "direct": pushStateMu.Lock() directPushDownSince = time.Time{} diff --git a/cmd/unbounded-net-node/status_server_test.go b/cmd/unbounded-net-node/status_server_test.go index d36ee64bd..0f151352a 100644 --- a/cmd/unbounded-net-node/status_server_test.go +++ b/cmd/unbounded-net-node/status_server_test.go @@ -819,7 +819,7 @@ func TestTryDirectRecoveryProbeClearsNodeErrors(t *testing.T) { transientErrors: []NodeError{{Type: "directWebsocket", Message: "node node-a direct websocket probe failed: dial tcp timeout"}}, } - candidate := tryDirectRecoveryProbe(t.Context(), health, &http.Client{Timeout: 5 * time.Second}, func() string { return "" }, nil, wsURL, "node-a") + candidate := tryDirectRecoveryProbe(t.Context(), health, &http.Client{Timeout: 5 * time.Second}, func() string { return "" }, nil, wsURL, "node-a", "full") if candidate == nil { t.Fatalf("expected direct recovery probe to succeed") } diff --git a/cmd/unbounded-net-node/status_websocket_auth_test.go b/cmd/unbounded-net-node/status_websocket_auth_test.go index 3e6033258..83c5079fb 100644 --- a/cmd/unbounded-net-node/status_websocket_auth_test.go +++ b/cmd/unbounded-net-node/status_websocket_auth_test.go @@ -252,8 +252,10 @@ func TestWebSocketEstablishedShutdownDoesNotReconnect(t *testing.T) { } func TestWebSocketRecoveryPromotesInitializedConnection(t *testing.T) { - for _, trigger := range []string{"timer", "HTTP recovery"} { - t.Run(trigger, func(t *testing.T) { + for _, testCase := range []string{"timer/full", "HTTP recovery/full", "timer/summary", "HTTP recovery/summary"} { + t.Run(testCase, func(t *testing.T) { + trigger, publicationMode, _ := strings.Cut(testCase, "/") + var ( directCalls, fallbackCalls, failedWrites atomic.Int32 directFrames, fallbackFrames atomic.Int32 @@ -261,6 +263,7 @@ func TestWebSocketRecoveryPromotesInitializedConnection(t *testing.T) { closeFallback atomic.Bool initialAckSent atomic.Bool mode atomic.Int32 + detailCollections atomic.Int32 ) failWrites.Store(true) @@ -298,6 +301,11 @@ func TestWebSocketRecoveryPromotesInitializedConnection(t *testing.T) { return } + if !message.SupportsDetails || (publicationMode == "summary" && (message.Summary == nil || message.Status != nil || message.Delta != nil)) { + t.Error("recovery changed publication mode or lost detail capability") + return + } + var revision int32 if direct { revision = directFrames.Add(1) @@ -309,7 +317,7 @@ func TestWebSocketRecoveryPromotesInitializedConnection(t *testing.T) { revision = fallbackFrames.Add(1) } - ack, err := proto.Marshal(&statusproto.NodeStatusAck{Status: "ok", Revision: uint64(revision)}) + ack, err := proto.Marshal(&statusproto.NodeStatusAck{Status: "ok", Revision: uint64(revision), SummarySupported: true}) if err != nil { t.Error(err) return @@ -349,7 +357,7 @@ func TestWebSocketRecoveryPromotesInitializedConnection(t *testing.T) { t.Setenv("UNBOUNDED_NET_CONTROLLER_SERVICE_PORT", port) cfg := &config{ - NodeName: "node-a", StatusWSEnabled: true, + NodeName: "node-a", StatusWSEnabled: true, StatusDetailMode: publicationMode, StatusWSAPIServerMode: statusWSAPIServerModeFallback, StatusWSAPIServerURL: "wss" + strings.TrimPrefix(server.URL, "https") + "/apis/status/nodews", StatusWSAPIServerStartupDelay: 25 * time.Millisecond, @@ -362,11 +370,22 @@ func TestWebSocketRecoveryPromotesInitializedConnection(t *testing.T) { manager := &hmacTokenManager{token: "valid-token", issuedAt: time.Now(), expiresAt: time.Now().Add(time.Hour)} ctx, cancel := context.WithCancel(t.Context()) done := make(chan struct{}) + health := blockedBootstrapHealthState() + + if publicationMode == "summary" { + statusServer := summaryRouteFixture() + statusServer.cfg.NodeName = "node-a" + statusServer.bpfCollector = func() []BpfEntry { + detailCollections.Add(1) + return nil + } + health.setStatusServer(statusServer) + } go func() { defer close(done) - runStatusWebSocketPusher(ctx, cfg, blockedBootstrapHealthState(), &connected, &mode, + runStatusWebSocketPusher(ctx, cfg, health, &connected, &mode, nil, nil, &closeFallback, client, manager) }() @@ -402,6 +421,10 @@ func TestWebSocketRecoveryPromotesInitializedConnection(t *testing.T) { t.Fatalf("recovery must promote its connection without redial: direct=%d fallback=%d fallbackClosed=%v", directCalls.Load(), fallbackCalls.Load(), fallbackClosed.Load()) } + + if detailCollections.Load() != 0 { + t.Fatal("summary recovery collected full diagnostics") + } }) } } @@ -681,7 +704,7 @@ func TestDirectRecoveryUnauthorizedInvalidatesToken(t *testing.T) { invalidated := false if tryDirectRecoveryProbe(t.Context(), &nodeHealthState{}, server.Client(), func() string { return "token" }, - func() { invalidated = true }, "ws"+strings.TrimPrefix(server.URL, "http"), "node-a") != nil || !invalidated { + func() { invalidated = true }, "ws"+strings.TrimPrefix(server.URL, "http"), "node-a", "full") != nil || !invalidated { t.Fatal("unauthorized recovery probe did not invalidate the credential") } } From af35a3d08078350c02d5278d23f4044d1540871a Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Tue, 22 Sep 2026 17:08:00 +0000 Subject: [PATCH 2/2] fix(net): keep detail collection off publishers Run correlated detail collection through one lifecycle-bound worker so slow kernel inspection cannot stall routine HTTP or WebSocket publications. Filter local transport diagnostics consistently from summary publications. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- cmd/unbounded-net-node/status_details.go | 177 +++++++++++++++--- .../status_details_http_test.go | 20 +- cmd/unbounded-net-node/status_details_test.go | 160 +++++++++++++--- cmd/unbounded-net-node/status_publication.go | 9 +- .../status_publication_test.go | 177 ++++++++++++++++++ cmd/unbounded-net-node/status_server.go | 14 +- 6 files changed, 478 insertions(+), 79 deletions(-) diff --git a/cmd/unbounded-net-node/status_details.go b/cmd/unbounded-net-node/status_details.go index 0223a9047..c1daa4ea9 100644 --- a/cmd/unbounded-net-node/status_details.go +++ b/cmd/unbounded-net-node/status_details.go @@ -4,6 +4,7 @@ package main import ( + "context" "fmt" "strings" "sync" @@ -23,20 +24,25 @@ const ( ) type nodeDetailReply struct { - request statusv1alpha1.DetailRequest - payload []byte - sending bool - done bool - retryAt time.Time + request statusv1alpha1.DetailRequest + payload []byte + sending bool + collecting bool + ready bool + done bool + retryAt time.Time } // One state spans both publishers and reconnects. Successful ACKs retain only // request identity/deadline markers; no routine publisher owns detail snapshots. type nodeDetailState struct { - mu sync.Mutex - replies map[string]*nodeDetailReply - wsWake chan struct{} - httpWake chan struct{} + mu sync.Mutex + replies map[string]*nodeDetailReply + wsWake chan struct{} + httpWake chan struct{} + collectWake chan struct{} + workerCancel context.CancelFunc + workerDone chan struct{} } func (h *nodeHealthState) detailState() *nodeDetailState { @@ -47,12 +53,61 @@ func (h *nodeHealthState) detailState() *nodeDetailState { h.details = &nodeDetailState{ replies: make(map[string]*nodeDetailReply), wsWake: make(chan struct{}, 1), httpWake: make(chan struct{}, 1), + collectWake: make(chan struct{}, 1), } } return h.details } +func (s *nodeDetailState) signalCollection() { + select { + case s.collectWake <- struct{}{}: + default: + } +} + +func (s *nodeDetailState) start(ctx context.Context, nodeName string, collect func() *NodeStatusResponse) <-chan struct{} { + s.mu.Lock() + if s.workerDone != nil { + done := s.workerDone + s.mu.Unlock() + + return done + } + + workerCtx, cancel := context.WithCancel(ctx) + done := make(chan struct{}) + s.workerCancel = cancel + s.workerDone = done + s.mu.Unlock() + + // One worker bounds a slow kernel-backed collection to one goroutine while + // routine HTTP and WebSocket publishers remain responsive. + go func() { + defer close(done) + + s.run(workerCtx, nodeName, collect) + }() + + s.signalCollection() + + return done +} + +func (s *nodeDetailState) stop() { + s.mu.Lock() + cancel := s.workerCancel + clear(s.replies) + s.mu.Unlock() + + if cancel != nil { + cancel() + } + + s.signalCollection() +} + func (s *nodeDetailState) wake() { for _, ch := range []chan struct{}{s.wsWake, s.httpWake} { select { @@ -62,6 +117,79 @@ func (s *nodeDetailState) wake() { } } +type nodeDetailCollection struct { + id string + deadline time.Time +} + +func (s *nodeDetailState) nextCollection(now time.Time) *nodeDetailCollection { + s.mu.Lock() + defer s.mu.Unlock() + + s.expireLocked(now) + + for _, reply := range s.replies { + if !reply.done && !reply.collecting && !reply.ready { + reply.collecting = true + + return &nodeDetailCollection{id: reply.request.RequestID, deadline: reply.request.Deadline} + } + } + + return nil +} + +func (s *nodeDetailState) completeCollection(collection *nodeDetailCollection, payload []byte, now time.Time) { + s.mu.Lock() + + reply := s.replies[collection.id] + if reply == nil || reply.request.Deadline != collection.deadline { + s.mu.Unlock() + + return + } + + reply.collecting = false + if !reply.request.Deadline.After(now) { + delete(s.replies, collection.id) + s.mu.Unlock() + + return + } + + if reply.done { + s.mu.Unlock() + + return + } + + reply.payload = payload + reply.ready = true + s.mu.Unlock() + s.wake() +} + +func (s *nodeDetailState) run(ctx context.Context, nodeName string, collect func() *NodeStatusResponse) { + for { + collection := s.nextCollection(time.Now()) + if collection == nil { + select { + case <-ctx.Done(): + return + case <-s.collectWake: + continue + } + } + + payload := collectDetailPayload(nodeName, collection.id, collect) + s.completeCollection(collection, payload, time.Now()) + + if ctx.Err() != nil { + return + } + } +} + func (s *nodeDetailState) expireLocked(now time.Time) { for id, reply := range s.replies { if !reply.request.Deadline.After(now) { @@ -88,7 +216,7 @@ func (s *nodeDetailState) enqueue(request *statusv1alpha1.DetailRequest, now tim }) } s.mu.Unlock() - s.wake() + s.signalCollection() return nil } @@ -111,13 +239,6 @@ func (s *nodeDetailState) receive(ack *statusv1alpha1.NodeStatusAck) { } } -func (s *nodeDetailState) clear() { - s.mu.Lock() - defer s.mu.Unlock() - - clear(s.replies) -} - func (s *nodeDetailState) acknowledge(ack *statusv1alpha1.NodeStatusAck) { if ack == nil || ack.DetailRequestID == "" || ack.Status != "ok" { return @@ -130,6 +251,8 @@ func (s *nodeDetailState) acknowledge(ack *statusv1alpha1.NodeStatusAck) { reply.payload = nil reply.done = true reply.sending = false + reply.collecting = false + reply.ready = false reply.retryAt = time.Time{} } } @@ -140,20 +263,16 @@ type nodeDetailDelivery struct { payload []byte } -func (s *nodeDetailState) take(nodeName string, collect func() *NodeStatusResponse, now time.Time) *nodeDetailDelivery { +func (s *nodeDetailState) take(now time.Time) *nodeDetailDelivery { s.mu.Lock() s.expireLocked(now) - var ( - selected *nodeDetailReply - payload []byte - ) + var selected *nodeDetailReply for _, reply := range s.replies { - if !reply.done && !reply.sending && !now.Before(reply.retryAt) { + if !reply.done && !reply.sending && reply.ready && !now.Before(reply.retryAt) { selected = reply selected.sending = true - payload = selected.payload break } @@ -161,11 +280,9 @@ func (s *nodeDetailState) take(nodeName string, collect func() *NodeStatusRespon s.mu.Unlock() if selected == nil { - return nil - } + s.signalCollection() - if payload == nil { - payload = collectDetailPayload(nodeName, selected.request.RequestID, collect) + return nil } s.mu.Lock() @@ -184,9 +301,7 @@ func (s *nodeDetailState) take(nodeName string, collect func() *NodeStatusRespon return nil } - selected.payload = payload - - return &nodeDetailDelivery{id: selected.request.RequestID, deadline: selected.request.Deadline, payload: payload} + return &nodeDetailDelivery{id: selected.request.RequestID, deadline: selected.request.Deadline, payload: selected.payload} } func (s *nodeDetailState) finish(id string) { diff --git a/cmd/unbounded-net-node/status_details_http_test.go b/cmd/unbounded-net-node/status_details_http_test.go index a7b4190fe..995cd4182 100644 --- a/cmd/unbounded-net-node/status_details_http_test.go +++ b/cmd/unbounded-net-node/status_details_http_test.go @@ -169,7 +169,8 @@ func TestHTTPDetailRetryDoesNotRecollectOrWaitForRoutineTick(t *testing.T) { t.Fatal(err) } // Seed the immutable reply as though a previous channel disconnected. - first := state.take("node-a", h.getStatusSnapshot, time.Now()) + startTestDetailWorker(t, state, "node-a", h.getStatusSnapshot) + first := waitForDetailDelivery(t, state) state.finish(first.id) requests := make(chan *statusproto.NodeStatusMessage, 4) @@ -258,15 +259,14 @@ func TestHTTPDetailCompressedLimitProducesRetriableError(t *testing.T) { t.Fatal(err) } - collections := 0 + var collections atomic.Int32 + + startTestDetailWorker(t, state, "node", func() *NodeStatusResponse { + collections.Add(1) - delivery := state.take("node", func() *NodeStatusResponse { - collections++ return &NodeStatusResponse{NodeErrors: []NodeError{{Message: base64.StdEncoding.EncodeToString(random)}}} - }, time.Now()) - if delivery == nil { - t.Fatal("no detail delivery") - } + }) + delivery := waitForDetailDelivery(t, state) body, err := state.httpBody("node", delivery, delivery.payload) if err != nil { @@ -280,8 +280,8 @@ func TestHTTPDetailCompressedLimitProducesRetriableError(t *testing.T) { state.finish(delivery.id) - retry := state.take("node", func() *NodeStatusResponse { t.Fatal("oversized retry recollected"); return nil }, time.Now().Add(2*time.Second)) - if retry == nil || !bytes.Equal(retry.payload, delivery.payload) || collections != 1 { + retry := state.take(time.Now().Add(2 * time.Second)) + if retry == nil || !bytes.Equal(retry.payload, delivery.payload) || collections.Load() != 1 { t.Fatal("oversized payload retained or changed on retry") } } diff --git a/cmd/unbounded-net-node/status_details_test.go b/cmd/unbounded-net-node/status_details_test.go index 6f5e0b828..dc7b7e6c8 100644 --- a/cmd/unbounded-net-node/status_details_test.go +++ b/cmd/unbounded-net-node/status_details_test.go @@ -17,6 +17,33 @@ import ( statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" ) +func startTestDetailWorker(t *testing.T, state *nodeDetailState, nodeName string, collect func() *NodeStatusResponse) <-chan struct{} { + t.Helper() + + done := state.start(t.Context(), nodeName, collect) + t.Cleanup(state.stop) + + return done +} + +func waitForDetailDelivery(t *testing.T, state *nodeDetailState) *nodeDetailDelivery { + t.Helper() + + deadline := time.After(time.Second) + + for { + if delivery := state.take(time.Now()); delivery != nil { + return delivery + } + + select { + case <-state.wsWake: + case <-deadline: + t.Fatal("timed out waiting for detail collection") + } + } +} + func TestDetailStateCoalescesRetriesAndReleases(t *testing.T) { h := blockedBootstrapHealthState() state := h.detailState() @@ -34,22 +61,27 @@ func TestDetailStateCoalescesRetriesAndReleases(t *testing.T) { t.Fatal(err) } - count := 0 - collect := func() *NodeStatusResponse { count++; return h.getStatusSnapshot() } + var count atomic.Int32 + + startTestDetailWorker(t, state, "node-a", func() *NodeStatusResponse { + count.Add(1) - first := state.take("node-a", collect, now) - if first == nil || !first.deadline.Equal(originalDeadline) || count != 1 { - t.Fatalf("lost command deadline/collection: %+v count=%d", first, count) + return h.getStatusSnapshot() + }) + + first := waitForDetailDelivery(t, state) + if !first.deadline.Equal(originalDeadline) || count.Load() != 1 { + t.Fatalf("lost command deadline/collection: %+v count=%d", first, count.Load()) } - if state.take("node-a", collect, now) != nil { + if state.take(now) != nil { t.Fatal("same request sent concurrently") } state.finish(first.id) - retry := state.take("node-a", collect, now.Add(2*time.Second)) - if retry == nil || !bytes.Equal(first.payload, retry.payload) || count != 1 { + retry := state.take(now.Add(2 * time.Second)) + if retry == nil || !bytes.Equal(first.payload, retry.payload) || count.Load() != 1 { t.Fatal("retry recollected or changed the snapshot") } @@ -64,11 +96,11 @@ func TestDetailStateCoalescesRetriesAndReleases(t *testing.T) { t.Fatal(err) } - if state.take("node-a", collect, now.Add(3*time.Second)) != nil || count != 1 { + if state.take(now.Add(3*time.Second)) != nil || count.Load() != 1 { t.Fatal("delayed duplicate recollected an acknowledged request") } - state.take("node-a", collect, originalDeadline) + state.take(originalDeadline) if len(state.replies) != 0 { t.Fatal("deadline did not remove idempotency marker") @@ -86,57 +118,127 @@ func TestDetailStateExpiryAndConcurrentClaims(t *testing.T) { req := &statusv1alpha1.DetailRequest{RequestID: "one", Deadline: now.Add(time.Minute)} var ( - count atomic.Int32 - wg sync.WaitGroup + collections atomic.Int32 + wg sync.WaitGroup ) + + startTestDetailWorker(t, state, "node", func() *NodeStatusResponse { + collections.Add(1) + + return &NodeStatusResponse{} + }) + for range 8 { wg.Go(func() { if err := state.enqueue(req, now); err != nil { t.Error(err) } + }) + } + + wg.Wait() - state.take("node", func() *NodeStatusResponse { - count.Add(1) - return &NodeStatusResponse{} - }, now) + waitForStatusCondition(t, func() bool { + state.mu.Lock() + defer state.mu.Unlock() + + reply := state.replies[req.RequestID] + + return reply != nil && reply.ready + }) + + var deliveries atomic.Int32 + + for range 8 { + wg.Go(func() { + if state.take(time.Now()) != nil { + deliveries.Add(1) + } }) } wg.Wait() - if count.Load() != 1 { - t.Fatalf("collected %d duplicate snapshots", count.Load()) + if collections.Load() != 1 || deliveries.Load() != 1 { + t.Fatalf("collections=%d deliveries=%d, want 1 each", collections.Load(), deliveries.Load()) } - state.take("node", nil, req.Deadline) + state.take(req.Deadline) if len(state.replies) != 0 { t.Fatal("expired unacknowledged detail retained") } } -func TestDetailDeadlineExpiresDuringCollectionAndDisconnect(t *testing.T) { +func TestDetailCollectionDoesNotBlockPublisherOrShutdown(t *testing.T) { state := (&nodeHealthState{}).detailState() + started := make(chan struct{}) + release := make(chan struct{}) + done := startTestDetailWorker(t, state, "node", func() *NodeStatusResponse { + close(started) + <-release + + return &NodeStatusResponse{} + }) - req := &statusv1alpha1.DetailRequest{RequestID: "slow", Deadline: time.Now().Add(20 * time.Millisecond)} + req := &statusv1alpha1.DetailRequest{RequestID: "slow", Deadline: time.Now().Add(30 * time.Millisecond)} if err := state.enqueue(req, time.Now()); err != nil { t.Fatal(err) } - delivery := state.take("node", func() *NodeStatusResponse { - <-time.After(time.Until(req.Deadline) + 10*time.Millisecond) - return &NodeStatusResponse{} - }, time.Now()) - if delivery != nil { - t.Fatal("expired collection was delivered") + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("detail collection did not start") + } + + takeStarted := time.Now() + if delivery := state.take(time.Now()); delivery != nil { + t.Fatal("unfinished collection was delivered") } - req = &statusv1alpha1.DetailRequest{RequestID: "disconnected", Deadline: time.Now().Add(20 * time.Millisecond)} + if elapsed := time.Since(takeStarted); elapsed > 50*time.Millisecond { + t.Fatalf("publisher blocked on detail collection for %v", elapsed) + } + + time.Sleep(time.Until(req.Deadline) + 10*time.Millisecond) + state.take(time.Now()) + + state.mu.Lock() + remaining := len(state.replies) + state.mu.Unlock() + + if remaining != 0 { + t.Fatal("expired collection retained its request") + } + + stopStarted := time.Now() + + state.stop() + + if elapsed := time.Since(stopStarted); elapsed > 50*time.Millisecond { + t.Fatalf("shutdown blocked on detail collection for %v", elapsed) + } + + close(release) + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("detail worker did not stop after collection returned") + } +} + +func TestDetailDisconnectExpiresUnacknowledgedReply(t *testing.T) { + state := (&nodeHealthState{}).detailState() + startTestDetailWorker(t, state, "node", func() *NodeStatusResponse { return &NodeStatusResponse{} }) + + req := &statusv1alpha1.DetailRequest{RequestID: "disconnected", Deadline: time.Now().Add(20 * time.Millisecond)} if err := state.enqueue(req, time.Now()); err != nil { t.Fatal(err) } - if state.take("node", func() *NodeStatusResponse { return &NodeStatusResponse{} }, time.Now()) == nil { + if waitForDetailDelivery(t, state) == nil { t.Fatal("missing unacknowledged reply") } diff --git a/cmd/unbounded-net-node/status_publication.go b/cmd/unbounded-net-node/status_publication.go index 4a35fda7b..2a992f19d 100644 --- a/cmd/unbounded-net-node/status_publication.go +++ b/cmd/unbounded-net-node/status_publication.go @@ -15,7 +15,7 @@ const nodeErrorSummaryUnsupported = "status-summary-unsupported" func collectPublication(health *nodeHealthState, cfg *config, previous *NodeStatusResponse, force bool, revision uint64) (*statusproto.NodeStatusMessage, *NodeStatusResponse) { if cfg.StatusDetailMode == "summary" { - summary := health.getSummarySnapshot() + summary := publicationSummary(health) return &statusproto.NodeStatusMessage{ Type: statusv1alpha1.NodeStatusSummaryType, NodeName: summary.NodeInfo.Name, @@ -41,6 +41,13 @@ func collectPublication(health *nodeHealthState, cfg *config, previous *NodeStat return msg, full } +func publicationSummary(health *nodeHealthState) *NodeStatusOverview { + summary := health.getSummarySnapshot() + summary.NodeErrors = publicationNodeErrors(summary.NodeErrors) + + return summary +} + func publicationNodeErrors(errors []NodeError) []NodeError { result := make([]NodeError, 0, len(errors)) for _, err := range errors { diff --git a/cmd/unbounded-net-node/status_publication_test.go b/cmd/unbounded-net-node/status_publication_test.go index 4747d1a8c..762c307bf 100644 --- a/cmd/unbounded-net-node/status_publication_test.go +++ b/cmd/unbounded-net-node/status_publication_test.go @@ -11,6 +11,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" "testing" "time" @@ -41,6 +42,47 @@ func TestSummaryPublicationNeverCollectsDetails(t *testing.T) { } } +func TestSummaryPublicationFiltersTransportErrorsAcrossHTTPAndWebSocket(t *testing.T) { + h := blockedBootstrapHealthState() + h.transientErrors = []NodeError{ + {Type: nodeErrorTypeDirectPush, Message: "direct push unavailable"}, + {Type: nodeErrorTypeDirectWebSocket, Message: "direct websocket unavailable"}, + {Type: "cni", Message: "CNI reconciliation failed"}, + } + + httpMessage, _ := collectPublication(h, &config{StatusDetailMode: "summary"}, nil, false, 42) + + wsPayload, err := marshalStatusWebSocketSummary(publicationSummary(h), 42) + if err != nil { + t.Fatal(err) + } + + var wsMessage statusproto.NodeStatusMessage + if err := proto.Unmarshal(wsPayload, &wsMessage); err != nil { + t.Fatal(err) + } + + httpMessage.Summary.TimestampUnixNs = 0 + + wsMessage.Summary.TimestampUnixNs = 0 + if !proto.Equal(httpMessage.Summary, wsMessage.Summary) { + t.Fatalf("HTTP and WebSocket summaries differ: HTTP=%v WS=%v", httpMessage.Summary, wsMessage.Summary) + } + + errorTypes := make(map[string]bool) + for _, nodeError := range httpMessage.Summary.NodeErrors { + errorTypes[nodeError.Type] = true + } + + if errorTypes[nodeErrorTypeDirectPush] || errorTypes[nodeErrorTypeDirectWebSocket] { + t.Fatalf("summary retained local transport diagnostics: %v", httpMessage.Summary.NodeErrors) + } + + if !errorTypes["cni"] || !errorTypes[configPodCIDRGuard] { + t.Fatalf("summary lost actual node errors: %v", httpMessage.Summary.NodeErrors) + } +} + func TestDetailACKDoesNotReleasePublication(t *testing.T) { state := &statusAckState{} state.revision.Store(7) @@ -70,6 +112,141 @@ func TestDetailACKDoesNotReleasePublication(t *testing.T) { } } +func TestSlowDetailCollectionDoesNotBlockRoutinePublishers(t *testing.T) { + for _, transport := range []string{"HTTP", "WS"} { + t.Run(transport, func(t *testing.T) { + messages := make(chan *statusproto.NodeStatusMessage, 8) + collectionStarted := make(chan struct{}) + releaseCollection := make(chan struct{}) + + var requestOnce sync.Once + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handle := func(data []byte) []byte { + var msg statusproto.NodeStatusMessage + if err := proto.Unmarshal(data, &msg); err != nil { + t.Error(err) + } + + messages <- &msg + + ack := &statusproto.NodeStatusAck{Status: "ok", Revision: 1, SummarySupported: true} + + requestOnce.Do(func() { + ack.DetailRequest = &statusproto.DetailRequest{ + RequestId: "slow", + DeadlineUnixNs: time.Now().Add(time.Minute).UnixNano(), + } + }) + + payload, err := proto.Marshal(ack) + if err != nil { + t.Error(err) + } + + return payload + } + + if transport == "WS" { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer func() { _ = conn.Close(websocket.StatusNormalClosure, "done") }() + + for { + _, data, err := conn.Read(r.Context()) + if err != nil { + return + } + + if err := conn.Write(r.Context(), websocket.MessageBinary, handle(data)); err != nil { + return + } + } + } + + reader, err := gzip.NewReader(r.Body) + if err != nil { + t.Error(err) + + return + } + + data, err := io.ReadAll(reader) + _ = reader.Close() + + if err != nil { + t.Error(err) + + return + } + + var ack statusproto.NodeStatusAck + if err := proto.Unmarshal(handle(data), &ack); err != nil { + t.Error(err) + + return + } + + if err := json.NewEncoder(w).Encode(netstatus.NodeStatusAckFromProto(&ack)); err != nil { + t.Error(err) + } + })) + defer server.Close() + + statusServer := summaryRouteFixture() + statusServer.bpfCollector = func() []BpfEntry { + close(collectionStarted) + <-releaseCollection + + return nil + } + + health := blockedBootstrapHealthState() + health.setStatusServer(statusServer) + + cfg := &config{ + NodeName: "node-a", StatusDetailMode: "summary", StatusPushEnabled: transport == "HTTP", + StatusWSEnabled: transport == "WS", StatusPushURL: server.URL, StatusWSURL: "ws" + strings.TrimPrefix(server.URL, "http"), + StatusPushInterval: 10 * time.Millisecond, StatusWSAPIServerMode: statusWSAPIServerModeNever, + CriticalDeltaEvery: 10 * time.Millisecond, StatsDeltaEvery: time.Hour, FullSyncEvery: time.Hour, + } + ctx, cancel := context.WithCancel(t.Context()) + startStatusPublishers(ctx, cfg, health) + + defer func() { + close(releaseCollection) + cancel() + health.stopStatusPublishers() + }() + + select { + case <-messages: + case <-time.After(3 * time.Second): + t.Fatal("no initial summary publication") + } + + select { + case <-collectionStarted: + case <-time.After(time.Second): + t.Fatal("detail collection did not start") + } + + health.setCNIReady("cbr0", []string{"10.244.7.0/24"}) + + select { + case msg := <-messages: + if msg.Type != statusv1alpha1.NodeStatusSummaryType || msg.Summary == nil { + t.Fatalf("routine publication changed while detail collection was blocked: %v", msg) + } + case <-time.After(500 * time.Millisecond): + t.Fatal("slow detail collection blocked routine publication") + } + }) + } +} + func TestRoutineSummaryPublishers(t *testing.T) { for _, transport := range []string{"HTTP", "WS"} { for _, supported := range []bool{true, false} { diff --git a/cmd/unbounded-net-node/status_server.go b/cmd/unbounded-net-node/status_server.go index 9412fa2c9..2cc9f6c26 100644 --- a/cmd/unbounded-net-node/status_server.go +++ b/cmd/unbounded-net-node/status_server.go @@ -106,7 +106,7 @@ func (h *nodeHealthState) stopStatusPublishers() { h.mu.RUnlock() if details != nil { - details.clear() + details.stop() } }) } @@ -587,6 +587,7 @@ func startStatusPublishers(ctx context.Context, cfg *config, healthState *nodeHe wg := &sync.WaitGroup{} wg.Add(2) healthState.setStatusTransportLifecycle(wg, cancel) + healthState.detailState().start(publisherCtx, cfg.NodeName, healthState.getStatusSnapshot) go func() { defer wg.Done() @@ -1431,9 +1432,7 @@ func runStatusWebSocketPusher( }() sendSummary := func(onlyChanged bool) error { - summary := healthState.getSummarySnapshot() - - summary.NodeErrors = publicationNodeErrors(summary.NodeErrors) + summary := publicationSummary(healthState) if onlyChanged && equalPublicationSummaries(lastSentSummary, summary) { return nil } @@ -1593,7 +1592,7 @@ func runStatusWebSocketPusher( fallbackCloseTicker := time.NewTicker(500 * time.Millisecond) sendDetails := func() error { - delivery := details.take(cfg.NodeName, healthState.getStatusSnapshot, time.Now()) + delivery := details.take(time.Now()) if delivery == nil { return nil } @@ -1941,8 +1940,7 @@ func tryDirectRecoveryProbe( sendErr error ) if detailMode == "summary" { - summary = healthState.getSummarySnapshot() - summary.NodeErrors = publicationNodeErrors(summary.NodeErrors) + summary = publicationSummary(healthState) payload, sendErr = marshalStatusWebSocketSummary(summary, 0) } else { status, payload, sendErr = marshalStatusWebSocketFull(healthState) @@ -2168,7 +2166,7 @@ func startStatusPusher( var delivery *nodeDetailDelivery if currentWSMode == statusWSModeNone { - delivery = details.take(cfg.NodeName, healthState.getStatusSnapshot, time.Now()) + delivery = details.take(time.Now()) } if detailOnly && delivery == nil {