diff --git a/cmd/unbounded-net-controller/detail_lifecycle.go b/cmd/unbounded-net-controller/detail_lifecycle.go index ce2f49b9f..95bf5d1ba 100644 --- a/cmd/unbounded-net-controller/detail_lifecycle.go +++ b/cmd/unbounded-net-controller/detail_lifecycle.go @@ -107,6 +107,10 @@ func (h *healthState) startDetailRequests(ctx context.Context, nodeInformer cach h.detailRequests = manager h.detailMu.Unlock() + if h.statusCache != nil { + h.statusCache.ObserveLegacyDetails(manager) + } + go func() { <-manager.done diff --git a/cmd/unbounded-net-controller/node_detail_observer.go b/cmd/unbounded-net-controller/node_detail_observer.go new file mode 100644 index 000000000..29fbee53f --- /dev/null +++ b/cmd/unbounded-net-controller/node_detail_observer.go @@ -0,0 +1,33 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import "k8s.io/klog/v2" + +// ObserveLegacyDetails bridges accepted legacy publications into the detail API +// while preserving existing full-cache and callback behavior during migration. +// It does not replay existing entries or refresh TTL on reads/source changes. +func (c *NodeStatusCache) ObserveLegacyDetails(manager *nodeDetailRequests) { + c.mu.Lock() + defer c.mu.Unlock() + + c.legacyObserver = manager +} + +// The cache lock preserves publication order against concurrent full/delta +// updates. The observer resolves informer identity but performs no network I/O. +func (c *NodeStatusCache) observeLegacyLocked(nodeName string, entry *CachedNodeStatus) { + if c.legacyObserver == nil { + return + } + + status := *entry.Status + if status.NodeInfo.Name == "" { + status.NodeInfo.Name = nodeName + } + + if err := c.legacyObserver.ObserveLegacy(nodeName, &status, entry.Revision, entry.peerIdentity, nil); err != nil { + klog.Errorf("Observing legacy node %q details failed: %v", nodeName, err) + } +} diff --git a/cmd/unbounded-net-controller/node_detail_observer_test.go b/cmd/unbounded-net-controller/node_detail_observer_test.go new file mode 100644 index 000000000..3230b2e29 --- /dev/null +++ b/cmd/unbounded-net-controller/node_detail_observer_test.go @@ -0,0 +1,167 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func TestLegacyObserverBridgeAuthenticatedPublishAndCachedDetails(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var pulls atomic.Int32 + + manager := testDetailRequests(t, nodeDetailRequestHooks{ + Pull: func(context.Context, string) (*NodeStatusResponse, error) { + pulls.Add(1) + + return nil, errors.New("node HTTP endpoint is unreachable") + }, + }) + health := &healthState{ + statusCache: NewNodeStatusCache(), detailRequests: manager, + nodeTokenVerifier: fakeServiceAccountTokenVerifier{}, + } + health.isLeader.Store(true) + health.statusCache.ObserveLegacyDetails(manager) + + fullCallbacks := 0 + + health.statusCache.SetOnChange(func(_ string, status *NodeStatusResponse) { + fullCallbacks++ + + if len(status.Peers) == 0 { + t.Error("bridge replaced the existing full callback payload") + } + }) + + issuer := testTokenIssuer(t) + token := testNodeToken(t, issuer) + mux := http.NewServeMux() + registerPushHandlers(mux, health, nil, make(chan struct{}, maxConcurrentNodeWS), issuer) + registerStatusHandlers(mux, health, false, nil, nil, nil) + + fullBody := `{"mode":"full","nodeName":"node-a","status":{"nodeInfo":{"name":"node-a"},"peers":[{"name":"peer-1"}]}}` + + publish := func(body, bearer string) int { + request := httptest.NewRequest(http.MethodPost, "/status/push", strings.NewReader(body)) + request.Header.Set("Authorization", "Bearer "+bearer) + + response := httptest.NewRecorder() + mux.ServeHTTP(response, request) + + return response.Code + } + if code := publish(fullBody, "invalid"); code != http.StatusUnauthorized { + t.Fatalf("unauthenticated publication returned %d", code) + } + + assertNodeDetailEntries(t, manager.cache, 0) + + if code := publish(fullBody, token); code != http.StatusOK { + t.Fatalf("full publication returned %d", code) + } + + response, first := serveDetailRequest(t, mux, http.MethodPost, "/status/node/node-a/details", "{}") + if response.Code != http.StatusOK || first.State != statusv1alpha1.NodeDetailComplete || + first.Details == nil || first.Details.Status.Peers[0].Name != "peer-1" || pulls.Load() != 0 { + t.Fatal("cached authenticated details required a direct HTTP pull") + } + + time.Sleep(time.Second) + + deltaBody := `{"mode":"delta","nodeName":"node-a","baseRevision":1,"delta":{"peers":[{"name":"peer-2"},{"name":"peer-3"}]}}` + if code := publish(deltaBody, token); code != http.StatusOK { + t.Fatalf("delta publication returned %d", code) + } + + current := manager.Request("node-a", false) + if current.RequestID != first.RequestID || current.Details == nil || len(current.Details.Status.Peers) != 2 || + !current.Details.ExpiresAt.Equal(first.Details.ExpiresAt.Add(time.Second)) || pulls.Load() != 0 { + t.Fatal("delta did not refresh the existing cache association") + } + + legacy, ok := health.statusCache.Get("node-a") + if !ok || legacy.Overview != nil || len(legacy.Status.Peers) != 2 || fullCallbacks != 2 { + t.Fatal("bridge changed existing full-cache or callback behavior") + } + + health.statusCache.UpdateSource("node-a", "ws") + + if _, err := health.statusCache.StoreOverview("summary-node", statusv1alpha1.NodeStatusOverview{ + NodeInfo: NodeInfo{Name: "summary-node"}, + }, "push"); err != nil { + t.Fatal(err) + } + + after, _ := manager.cache.Get("node-a") + if after.ExpiresAt != current.Details.ExpiresAt || fullCallbacks != 3 { + t.Fatal("source/summary change renewed the detail TTL or changed callbacks") + } + + fresh := manager.Request("node-a", true) + + if code := publish(fullBody, token); code != http.StatusOK { + t.Fatalf("next full publication returned %d", code) + } + + pending := manager.Result("node-a", fresh.RequestID) + if pending.State != statusv1alpha1.NodeDetailPending || pending.RequestID != fresh.RequestID || pending.Deadline != fresh.Deadline { + t.Fatal("legacy publication reset or completed a pending correlated refresh") + } + + time.Sleep(manager.cache.ttl) + synctest.Wait() + assertNodeDetailEntries(t, manager.cache, 0) + + legacy, ok = health.statusCache.Get("node-a") + + if !ok || len(legacy.Status.Peers) != 1 || legacy.Overview != nil { + t.Fatal("bridge prematurely removed the legacy full base") + } + }) +} + +func TestLegacyObserverBridgeFailurePreservesOldStore(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + manager := testDetailRequests(t, nodeDetailRequestHooks{}) + cache := NewNodeStatusCache() + cache.ObserveLegacyDetails(manager) + + status := NodeStatusResponse{Peers: []WireGuardPeerStatus{{Name: "peer"}}} + cache.StoreFull("node", status, "ws") + + details := manager.Request("node", false) + + if details.Details == nil || details.Details.Status.NodeInfo.Name != "node" || + status.NodeInfo.Name != "" || cache.entries["node"].Status.NodeInfo.Name != "" { + t.Fatal("bridge failed to normalize only its own shallow snapshot") + } + + manager.Close() + + if revision := cache.StoreFull("node", status, "ws"); revision != 2 { + t.Fatal("observer failure changed old cache revision behavior") + } + + if len(cache.entries["node"].Status.Peers) != 1 { + t.Fatal("observer failure discarded the old full payload") + } + + cache.ObserveLegacyDetails(nil) + + if revision := cache.StoreFull("node", status, "ws"); revision != 3 { + t.Fatal("detaching the bridge changed legacy storage") + } + }) +} diff --git a/cmd/unbounded-net-controller/node_status.go b/cmd/unbounded-net-controller/node_status.go index 3dbb37b88..a1708718f 100644 --- a/cmd/unbounded-net-controller/node_status.go +++ b/cmd/unbounded-net-controller/node_status.go @@ -32,6 +32,7 @@ type NodeStatusCache struct { entries map[string]*CachedNodeStatus onChange func(nodeName string, status *NodeStatusResponse) onOverviewChange func(nodeName string, overview statusv1alpha1.NodeStatusOverview) + legacyObserver *nodeDetailRequests } // NewNodeStatusCache creates an empty NodeStatusCache. @@ -67,6 +68,8 @@ func (c *NodeStatusCache) StoreFull(nodeName string, status NodeStatusResponse, Source: source, Revision: revision, } + c.observeLegacyLocked(nodeName, c.entries[nodeName]) + fn := c.onChange statusPtr := c.entries[nodeName].Status c.mu.Unlock() @@ -368,6 +371,8 @@ func (c *NodeStatusCache) commitParsedDelta(nodeName string, previous *CachedNod Revision: revision, peerIdentity: peerIdentity, } + c.observeLegacyLocked(nodeName, c.entries[nodeName]) + fn := c.onChange mergedPtr := c.entries[nodeName].Status c.mu.Unlock()