diff --git a/controller/model/link_manager.go b/controller/model/link_manager.go index 59464be20..a766be38e 100644 --- a/controller/model/link_manager.go +++ b/controller/model/link_manager.go @@ -29,7 +29,6 @@ import ( "github.com/openziti/ziti/v2/controller/storage/boltz" "github.com/openziti/ziti/v2/controller/storage/objectz" cmap "github.com/orcaman/concurrent-map/v2" - "go.etcd.io/bbolt" ) // linkLog is the logger for the controller's link manager. Its channel name is @@ -290,42 +289,10 @@ func (self *LinkManager) All() []*Link { return self.linkTable.all() } -// RouterDeleted drops the deleted router's link index, unless the id has since come back. -// -// A router id can be reused: fabric router ids are the enrollment certificate's common name, so re-adding -// a router from the same cert reuses its id. Nothing orders this call against that, since store commit -// handlers run after bolt has released the writer lock, so it can execute long after the id was recreated, -// connected, and reported links. Dropping a live router's index is invisible in the link table and surfaces -// only when a per-router query fails to find links it should have. -// -// Re-reading the store closes that rather than narrowing it, because the read happens while the index map's -// shard is held. Indexing a link takes the same shard, so an entry for a live router cannot appear without -// its create having committed first, which this read would then see. Anything indexed while the store says -// the id is absent belongs to the deleted router. +// RouterDeleted drops the deleted router's link index. A router id is only ever added to the index, so +// without this every router that has ever had a link is held for the controller's lifetime. func (self *LinkManager) RouterDeleted(routerId string) { - self.linkTable.dropRouterIndexIf(routerId, func() bool { - return !self.routerExists(routerId) - }) -} - -// routerExists reports whether a router with this id is in the store. A nil env, which only happens in -// tests that do not build one, reports absent so cleanup still runs. -func (self *LinkManager) routerExists(routerId string) bool { - if self.env == nil { - return false - } - found := false - if err := self.env.GetDb().View(func(tx *bbolt.Tx) error { - found = self.env.GetStores().Router.IsEntityPresent(tx, routerId) - return nil - }); err != nil { - // Unreadable is not evidence the router is gone, and keeping an index costs only its entry. - linkLog.Warn("could not check whether router still exists, keeping its link index", - "routerId", routerId, - "error", err) - return true - } - return found + self.linkTable.dropRouterIndex(routerId) } // LinksForRouter returns the links with routerId at either end. Per-router work must use this rather than @@ -462,12 +429,9 @@ func (lt *linkTable) indexFor(routerId string) *concurrency.LockedSet[string] { }) } -// dropRouterIndexIf forgets a router's link index when shouldDrop agrees. shouldDrop runs with the index -// map's shard held, so it must not index a link, and callers relying on that exclusion should say so. -func (lt *linkTable) dropRouterIndexIf(routerId string, shouldDrop func() bool) { - lt.byRouter.RemoveCb(routerId, func(_ string, _ *concurrency.LockedSet[string], _ bool) bool { - return shouldDrop() - }) +// dropRouterIndex forgets a router's link index. +func (lt *linkTable) dropRouterIndex(routerId string) { + lt.byRouter.Remove(routerId) } // linkIdsForRouter returns the ids indexed under the given router. Ids the table no longer holds are diff --git a/controller/network/canary_gossip.go b/controller/network/canary_gossip.go index 55e2b08c5..90fae80db 100644 --- a/controller/network/canary_gossip.go +++ b/controller/network/canary_gossip.go @@ -42,7 +42,7 @@ type CanaryValue = gossip_pb.CanaryGossipValue // tombstones and no anti-entropy — they are purely fire-and-forget probes. // Called once from initGossip, which owns the store this registers against. func (network *Network) initCanaryGossip() { - listener := &canaryGossipListener{network: network} + listener := &canaryGossipListener{network: network, canaries: map[string]*CanaryValue{}} canaryType := gossip.Register[*CanaryValue](network.GossipStore, gossip.StateTypeConfig[*CanaryValue]{ Name: CanaryGossipStoreType, @@ -55,6 +55,13 @@ func (network *Network) initCanaryGossip() { network.CanaryGossipType = canaryType network.canaryListener = listener + + // Deliberately not registered as a gossip type: those are swept on an epoch change, and a canary is + // what carries the epoch that detects one. It still has to be cleaned up when a router is deleted. + network.onRouterDeleted(CanaryGossipStoreType, func(routerId string) { + canaryType.DropOwner(routerId) + listener.remove(routerId) + }) } // GetCanaryForRouter returns the latest canary sequence number this controller @@ -72,15 +79,44 @@ func (network *Network) GetCanaryValueForRouter(routerId string) (*CanaryValue, // canaryGossipListener tracks the latest canary state per router and detects // epoch changes to trigger cleanup of old link gossip entries. type canaryGossipListener struct { - network *Network - canaries sync.Map // routerId (string) -> CanaryValue + network *Network + + // A plain map under a mutex rather than a sync.Map, so forget can decide and delete atomically. A + // sync.Map offers no compare-and-delete, which would leave the reused-id check racing the delete. + lock sync.RWMutex + canaries map[string]*CanaryValue +} + +// swap records a router's canary, returning the one it replaced. +func (l *canaryGossipListener) swap(routerId string, value *CanaryValue) (*CanaryValue, bool) { + l.lock.Lock() + defer l.lock.Unlock() + prev, loaded := l.canaries[routerId] + if l.canaries == nil { + l.canaries = map[string]*CanaryValue{} + } + l.canaries[routerId] = value + return prev, loaded +} + +func (l *canaryGossipListener) load(routerId string) (*CanaryValue, bool) { + l.lock.RLock() + defer l.lock.RUnlock() + v, ok := l.canaries[routerId] + return v, ok +} + +func (l *canaryGossipListener) remove(routerId string) { + l.lock.Lock() + defer l.lock.Unlock() + delete(l.canaries, routerId) } func (l *canaryGossipListener) EntryChanged(key string, value *CanaryValue, _ uint64, _ string, _ bool, _ gossip.ChangeOrigin) { - prev, loaded := l.canaries.Swap(key, value) + prev, loaded := l.swap(key, value) if loaded && len(value.Epoch) > 0 { - prevVal := prev.(*CanaryValue) + prevVal := prev if len(prevVal.Epoch) > 0 && bytes.Compare(value.Epoch, prevVal.Epoch) > 0 { // New epoch is newer — the router restarted. Clean up old-epoch // entries for this router across all gossip store types. @@ -95,7 +131,7 @@ func (l *canaryGossipListener) EntryChanged(key string, value *CanaryValue, _ ui } func (l *canaryGossipListener) EntryRemoved(key string, _ string, _ uint64, _ gossip.ChangeOrigin) { - l.canaries.Delete(key) + l.remove(key) } // checkEpoch compares the given epoch against the stored epoch for a router. @@ -112,9 +148,8 @@ func (l *canaryGossipListener) checkEpoch(routerId string, epoch []byte) { return } - prev, loaded := l.canaries.Load(routerId) + prevVal, loaded := l.load(routerId) if loaded { - prevVal := prev.(*CanaryValue) if len(prevVal.Epoch) > 0 && bytes.Compare(epoch, prevVal.Epoch) > 0 { canaryLog.Info("router epoch changed, cleaning up old-epoch gossip entries", "routerId", routerId, @@ -133,15 +168,15 @@ func (l *canaryGossipListener) checkEpoch(routerId string, epoch []byte) { } func (l *canaryGossipListener) getSeq(routerId string) (uint64, bool) { - if v, ok := l.canaries.Load(routerId); ok { - return v.(*CanaryValue).Seq, true + if v, ok := l.load(routerId); ok { + return v.Seq, true } return 0, false } func (l *canaryGossipListener) getValue(routerId string) (*CanaryValue, bool) { - if v, ok := l.canaries.Load(routerId); ok { - return v.(*CanaryValue), true + if v, ok := l.load(routerId); ok { + return v, true } return nil, false } diff --git a/controller/network/link_gossip.go b/controller/network/link_gossip.go index 05eac4f1b..3bf790cbd 100644 --- a/controller/network/link_gossip.go +++ b/controller/network/link_gossip.go @@ -85,6 +85,7 @@ func (network *Network) initLinkGossip() { network.LinkGossipType = linkType network.RegisterGossipType(LinkGossipStoreType, linkType) + network.onRouterDeleted(LinkGossipStoreType, linkType.DropOwner) } // NotifyLinkViaGossip is called when an old router reports a link. Only the diff --git a/controller/network/link_metrics_gossip.go b/controller/network/link_metrics_gossip.go index ffbe3fb90..1844501f6 100644 --- a/controller/network/link_metrics_gossip.go +++ b/controller/network/link_metrics_gossip.go @@ -58,6 +58,7 @@ func (network *Network) initLinkMetricsGossip() { network.LinkMetricsType = metricsType network.RegisterGossipType(LinkMetricsGossipStoreType, metricsType) + network.onRouterDeleted(LinkMetricsGossipStoreType, metricsType.DropOwner) } // applyLinkMetric sets the source or destination latency of the given link from a diff --git a/controller/network/network.go b/controller/network/network.go index 5bf992538..a68888ae7 100644 --- a/controller/network/network.go +++ b/controller/network/network.go @@ -126,10 +126,13 @@ type Network struct { LinkGossipType *gossip.StateType[*ctrl_pb.RouterLinks_RouterLink] LinkMetricsType *gossip.StateType[*ctrl_pb.LinkMetrics] CanaryGossipType *gossip.StateType[*CanaryValue] - gossipTypes map[string]gossip.StateTypeInfo // non-generic lookup by store type - canaryListener *canaryGossipListener - isHA bool - leaderCheck func() bool + // gossipTypes drives the epoch sweep only. Router-deleted cleanup is a separate lifecycle with a + // different membership, since canaries carry the epoch that detects a change and so must survive one. + gossipTypes map[string]gossip.StateTypeInfo // non-generic lookup by store type + routerDeletedCleanups []routerDeletedCleanup + canaryListener *canaryGossipListener + isHA bool + leaderCheck func() bool serviceEventMetrics servermetrics.UsageRegistry serviceDialSuccessCounter servermetrics.IntervalCounter @@ -226,6 +229,9 @@ func NewNetwork(config Config, env model.Env) (*Network, error) { env.GetManagers().Router.Store.AddEntityIdListener(network.HandleRouterDelete, boltz.EntityDeletedAsync) + linkManager := env.GetManagers().Link + network.onRouterDeleted("linkIndex", linkManager.RouterDeleted) + network.AddCapability("ziti.fabric") network.showOptions() network.relayControllerMetrics() @@ -288,13 +294,8 @@ func (network *Network) IsLeader() bool { func (self *Network) HandleRouterDelete(id string) { self.routerDeleted(id) - self.Link.RouterDeleted(id) self.RouterMessaging.RouterDeleted(id) - // Drop the router's gossip-store owner data so memory doesn't grow - // unboundedly when routers are added/removed over time (e.g., autoscaling). - // Live entries are tombstoned and broadcast; the ownerData is compacted out - // of the gossip store by the reaper once tombstones age out. - self.dropGossipOwner(id) + self.runRouterDeletedCleanups(id) } func (self *Network) decodeSyncSnapshotCommand(_ int32, data []byte) (command.Command, error) { @@ -689,9 +690,41 @@ func (network *Network) GossipStoreTypes() []string { // dropGossipOwner tombstones every registered store type's entries for the owner, // so a removed router's gossip state is reclaimed across all stores. -func (network *Network) dropGossipOwner(id string) { - for _, t := range network.gossipTypes { - t.DropOwner(id) +// routerDeletedCleanup is one subsystem's per-router state teardown, named so a test can check every +// per-router store is covered. +type routerDeletedCleanup struct { + name string + cleanup func(routerId string) +} + +// onRouterDeleted registers cleanup for state keyed by router id. Every store holding per-router state +// must register, or it leaks on delete: nothing else enumerates them, and the leak is invisible until a +// memory profile is read. +// +// This assumes a deleted router id does not come back. Router ids are generated at create, apart from the +// deprecated path that creates a fabric router outside enrollment with a caller-supplied id, and cleanup +// is not ordered against a create of the same id. See the deprecation note on that path. +func (network *Network) onRouterDeleted(name string, cleanup func(routerId string)) { + network.routerDeletedCleanups = append(network.routerDeletedCleanups, routerDeletedCleanup{ + name: name, + cleanup: cleanup, + }) +} + +// routerDeletedCleanupNames returns the registered cleanups, for tests that check every per-router store +// is covered. +func (network *Network) routerDeletedCleanupNames() []string { + names := make([]string, 0, len(network.routerDeletedCleanups)) + for _, c := range network.routerDeletedCleanups { + names = append(names, c.name) + } + return names +} + +// runRouterDeletedCleanups runs every registered cleanup for a deleted router. +func (network *Network) runRouterDeletedCleanups(id string) { + for _, c := range network.routerDeletedCleanups { + c.cleanup(id) } } diff --git a/controller/network/router_connect_test.go b/controller/network/router_connect_test.go index 285931ba4..d1ac8b158 100644 --- a/controller/network/router_connect_test.go +++ b/controller/network/router_connect_test.go @@ -621,34 +621,6 @@ func TestRouterDelete_DropsTheLinkIndex(t *testing.T) { require.Len(t, network.Link.LinksForRouter(peer.Id), 1, "the other endpoint's index must be untouched") } -// TestRouterDelete_KeepsTheIndexOfAReusedId covers the delete cleanup running late against an id that has -// come back. Fabric router ids are the enrollment certificate's common name, so re-adding a router from the -// same cert reuses its id, and nothing orders the cleanup against that: store commit handlers run after -// bolt has released the writer lock. Dropping the live router's index here is invisible in the link table -// and shows up only when a per-router query misses links it should have. -// -// The late callback is invoked directly rather than by pausing the real one, since the guard is a store read -// taken while the index map's shard is held, so the observable property is the same either way. -func TestRouterDelete_KeepsTheIndexOfAReusedId(t *testing.T) { - _, network, addr := newConnectTestNetwork(t) - - peer := newPersistedRouter(t, network, addr, "r1") - - original := newPersistedRouter(t, network, addr, "r0") - require.NoError(t, network.Router.Delete(original.Id, change.New())) - - // The same id comes back and its links are reported, all before the earlier delete's cleanup runs. - reused := newPersistedRouter(t, network, addr, "r0") - network.Link.Add(model.NewTestLink("l0", reused, peer)) - require.Len(t, network.Link.LinksForRouter(reused.Id), 1) - - network.Link.RouterDeleted(original.Id) - - require.Len(t, network.Link.LinksForRouter(reused.Id), 1, - "a delete that lands after the id was recreated must not drop the live router's index") - require.Len(t, network.Link.LinksForRouter(peer.Id), 1) -} - // TestRouterReportedLink_RepairsDestDisplacedMidReport is the same interleave with a stale destination // rather than an absent one. The report resolves the destination under the source's connect stripe, not the // destination's, so that router can be replaced before the report lands. A link left pointing at the diff --git a/controller/network/router_deleted_test.go b/controller/network/router_deleted_test.go new file mode 100644 index 000000000..2c6ceca72 --- /dev/null +++ b/controller/network/router_deleted_test.go @@ -0,0 +1,82 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package network + +import ( + "testing" + + "github.com/openziti/ziti/v2/controller/change" + "github.com/stretchr/testify/require" +) + +// TestRouterDeletedCleanups_CoverEveryPerRouterStore is the guard on the defect that let canary state leak: +// a per-router store was added, wired into the epoch sweep's registry only, and nothing cleaned it up when a +// router was deleted. Adding a store without registering its cleanup must fail here rather than leak until +// someone reads a memory profile. +func TestRouterDeletedCleanups_CoverEveryPerRouterStore(t *testing.T) { + _, network, _ := newConnectTestNetwork(t) + + registered := map[string]bool{} + for _, name := range network.routerDeletedCleanupNames() { + registered[name] = true + } + + // Every gossip store keyed by owner holds per-router state, whether or not it takes part in the epoch + // sweep. Canaries deliberately sit out the sweep, since they carry the epoch that detects a change, and + // that exclusion is exactly what hid their missing delete cleanup. + for _, storeType := range []string{LinkGossipStoreType, LinkMetricsGossipStoreType, CanaryGossipStoreType} { + require.True(t, registered[storeType], + "gossip store %q holds per-router state and must register router-deleted cleanup", storeType) + } + + require.True(t, registered["linkIndex"], "the link index is keyed by router id and must be cleaned up") +} + +// TestRouterDeleted_DropsCanaryStateOfAGoneId: a deleted router must not leave canary state behind, which +// is what grew without bound under router churn. Canary state is tracked outside the gossip store as well, +// so the cleanup has to drop both. +func TestRouterDeleted_DropsCanaryStateOfAGoneId(t *testing.T) { + _, network, addr := newConnectTestNetwork(t) + + gone := newPersistedRouter(t, network, addr, "r0") + network.canaryListener.EntryChanged(gone.Id, &CanaryValue{Seq: 3}, 1, gone.Id, false, 0) + _, found := network.GetCanaryForRouter(gone.Id) + require.True(t, found) + + require.NoError(t, network.Router.Delete(gone.Id, change.New())) + network.runRouterDeletedCleanups(gone.Id) + + _, found = network.GetCanaryForRouter(gone.Id) + require.False(t, found, "a deleted router's canary state must not be left behind") +} + +// TestRouterDeleted_DropsGossipOwnerOfAGoneId: a deleted router must have its gossip owner data dropped, +// which is what keeps the store from growing with every router ever deleted. +func TestRouterDeleted_DropsGossipOwnerOfAGoneId(t *testing.T) { + _, network, addr := newConnectTestNetwork(t) + + gone := newPersistedRouter(t, network, addr, "r0") + require.NoError(t, network.CanaryGossipType.Set("r0", gone.Id, &CanaryValue{Seq: 1})) + _, _, ok := network.CanaryGossipType.GetForOwner(gone.Id, "r0") + require.True(t, ok) + + require.NoError(t, network.Router.Delete(gone.Id, change.New())) + network.runRouterDeletedCleanups(gone.Id) + + _, _, ok = network.CanaryGossipType.GetForOwner(gone.Id, "r0") + require.False(t, ok, "a deleted router's gossip entries must not be left behind") +}