From cd14adce61d6dbe0c71f0424a4a645a068f4b946 Mon Sep 17 00:00:00 2001 From: "David L. Chandler" Date: Thu, 3 Sep 2026 11:38:38 -0400 Subject: [PATCH 1/5] fix: keep xDS watches whose response has to be withheld In ADS mode the snapshot cache deliberately does not answer a request whose resource names do not cover the snapshot's resources of that type: sending it would hand the client resources it has not subscribed to. The withheld response was then treated as a completed request by both call sites, and the watch was lost: - SetSnapshot called respond() and deleted the watch unconditionally, so a parked watch was discarded even though nothing had been sent to it. - CreateWatch called respond() for a request whose version was stale and returned a cancel function that only closes the channel, registering no watch at all. The comment above respond() explains the intent as "it is expected that envoy makes another request", but a client only issues a new request after it receives a response: Envoy re-subscribes EDS once it has applied a CDS update, and otherwise waits. So when the mismatch is caused by a resource the client will not subscribe to soon -- a cluster it has just rejected, or one it has not applied yet -- no new request arrives, the watch is gone, and every later snapshot has nothing to deliver to. The client keeps serving its last configuration until it reconnects, with no error surfaced anywhere: gateway proxies observed holding removed endpoints indefinitely after control-plane churn while readiness and endpoint health stayed green. This makes respond() report whether it sent a response, and fixes both paths: - SetSnapshot discards the watch only when a response was sent, and keeps it otherwise, so the next snapshot the request can accept is delivered to it. - CreateWatch falls through to opening a watch when the response is withheld, instead of returning a request that nothing will ever answer. Nothing changes when a response is sent, when the requested version is already up to date, or for wildcard requests, which are never withheld. PR #586 fixed the SetSnapshot path only. The request path is reachable in the same window -- a proxy that (re)subscribes while the snapshot is misaligned, which is exactly what happens during a rejected cluster update or a reconnect mid-churn -- so both are closed here. For reference, go-control-plane's equivalent rework (bf9b60b56, "[Snapshot Cache] Simplify Snapshot cache responses") also closed only the publish path and left the request path behind a TODO. Supersedes #586. Tests: five specs in pkg/api/v1/control-plane/cache covering retention on both paths, delivery once the snapshot realigns, immediate answers, an up-to-date request still parking, and wildcard behaviour unchanged. Signed-off-by: David L. Chandler --- .../retain-unanswered-xds-watches.yaml | 15 ++ pkg/api/v1/control-plane/cache/simple.go | 66 ++++++--- .../cache/watch_retention_test.go | 132 ++++++++++++++++++ 3 files changed, 192 insertions(+), 21 deletions(-) create mode 100644 changelog/v0.41.2/retain-unanswered-xds-watches.yaml create mode 100644 pkg/api/v1/control-plane/cache/watch_retention_test.go diff --git a/changelog/v0.41.2/retain-unanswered-xds-watches.yaml b/changelog/v0.41.2/retain-unanswered-xds-watches.yaml new file mode 100644 index 000000000..bbe82ce04 --- /dev/null +++ b/changelog/v0.41.2/retain-unanswered-xds-watches.yaml @@ -0,0 +1,15 @@ +changelog: + - type: FIX + issueLink: https://github.com/solo-io/solo-kit/pull/586 + resolvesIssue: false + description: >- + Keep an xDS watch open when its response has to be withheld. In ADS mode + the snapshot cache does not answer a request whose resource names do not + cover the snapshot's resources of that type, which happens while a client + has not yet subscribed to a resource the snapshot already carries. Both + call sites treated the withheld response as a completed request, so the + watch was discarded on a snapshot update and never registered for a new + request. Because a client only issues a new request after receiving a + response, nothing re-armed delivery, and it stayed on its previous + configuration until it reconnected. The watch is now retained on both + paths and answered by the first snapshot it can accept. diff --git a/pkg/api/v1/control-plane/cache/simple.go b/pkg/api/v1/control-plane/cache/simple.go index 529d62905..1580f41be 100644 --- a/pkg/api/v1/control-plane/cache/simple.go +++ b/pkg/api/v1/control-plane/cache/simple.go @@ -168,9 +168,17 @@ func (cache *snapshotCache) SetSnapshot(node string, snapshot Snapshot) { // empty resource - intended behavior is for the resources to be cleared if resources != nil { // snapshot has been initialized and exists - cache.respond(watch.Request, watch.Response, resources, version) - - // discard the watch + if !cache.respond(watch.Request, watch.Response, resources, version) { + // The response was withheld (see respond): keep the watch so a + // later snapshot that this request can accept is delivered to it. + if cache.log != nil { + cache.log.Debugf("keeping open watch priority %d and index %d :%v; version %q could not be sent", + pi.Index, pi.Priority, watch.Request.ResourceNames, version) + } + return + } + + // discard the watch: the client creates a new one when it acks watches.Delete(pi) } } @@ -266,25 +274,32 @@ func (cache *snapshotCache) CreateWatch(request Request) (chan Response, func()) } version := snapshot.GetResources(request.TypeUrl).Version - // if the requested version is up-to-date or missing a response, leave an open watch - if !exists || request.VersionInfo == version { - info.mu.Lock() - // check SetSnapshot() for responses on the watches map - priority := info.watches.Add(ResponseWatch{Request: request, Response: value}) - info.mu.Unlock() - if cache.log != nil { - cache.log.Debugf("open watch Priority Index %d and Element Index %d for %s%v from nodeID %q, version %q", - priority.Priority, priority.Index, request.TypeUrl, request.ResourceNames, nodeID, request.VersionInfo) + // If the snapshot is newer than the requested version, try to answer the + // request immediately. The response can still be withheld in ADS mode (see + // respond), in which case the request has to become an open watch: the + // client is waiting for this response and does not issue a new request + // until it receives one, so dropping the request here would leave the + // client on its current configuration until it reconnects. + if exists && request.VersionInfo != version { + if cache.respond(request, value, snapshot.GetResources(request.TypeUrl).Items, version) { + return value, func() { + close(value) + } } - return value, cache.cancelWatch(nodeID, priority) } - // otherwise, the watch may be responded immediately - cache.respond(request, value, snapshot.GetResources(request.TypeUrl).Items, version) - - return value, func() { - close(value) + // Leave an open watch: the requested version is up to date, there is no + // snapshot for this node yet, or the response above was withheld. + info.mu.Lock() + // check SetSnapshot() for responses on the watches map + priority := info.watches.Add(ResponseWatch{Request: request, Response: value}) + info.mu.Unlock() + if cache.log != nil { + cache.log.Debugf("open watch Priority Index %d and Element Index %d for %s%v from nodeID %q, version %q", + priority.Priority, priority.Index, request.TypeUrl, request.ResourceNames, nodeID, request.VersionInfo) } + + return value, cache.cancelWatch(nodeID, priority) } // cancellation function for cleaning stale watches @@ -305,16 +320,23 @@ func (cache *snapshotCache) cancelWatch(nodeID string, watchIndex PriorityIndex) } // Respond to a watch with the snapshot value. The value channel should have capacity not to block. +// +// It reports whether a response was sent. In ADS mode a request whose resource +// names do not cover the snapshot cannot be answered yet, and callers must keep +// the watch open instead of treating the request as served: nothing else will +// answer it, because the client is waiting for this response and only issues a +// new request once it receives one. +// // TODO(kuat) do not respond always, see issue https://github.com/envoyproxy/go-control-plane/issues/46 -func (cache *snapshotCache) respond(request Request, value chan Response, resources map[string]Resource, version string) { +func (cache *snapshotCache) respond(request Request, value chan Response, resources map[string]Resource, version string) bool { // for ADS, the request names must match the snapshot names - // if they do not, then the watch is never responded, and it is expected that envoy makes another request + // if they do not, the response is withheld and false is returned, so the caller keeps the watch open if len(request.ResourceNames) != 0 && cache.ads { if err := Superset(nameSet(request.ResourceNames), resources); err != nil { if cache.log != nil { cache.log.Debugf("ADS mode: not responding to request: %v", err) } - return + return false } } if cache.log != nil { @@ -325,6 +347,8 @@ func (cache *snapshotCache) respond(request Request, value chan Response, resour tag.Insert(KeyType, request.GetTypeUrl()), }, MResponses.M(1)) value <- createResponse(request, resources, version) + + return true } func createResponse(request Request, resources map[string]Resource, version string) Response { diff --git a/pkg/api/v1/control-plane/cache/watch_retention_test.go b/pkg/api/v1/control-plane/cache/watch_retention_test.go new file mode 100644 index 000000000..3a0d185f6 --- /dev/null +++ b/pkg/api/v1/control-plane/cache/watch_retention_test.go @@ -0,0 +1,132 @@ +package cache_test + +import ( + "time" + + envoy_config_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/cache" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/resource" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/types" +) + +// In ADS mode a response is withheld when the request's resource names do not +// cover the snapshot's resources of that type, which happens whenever the +// snapshot knows about a resource the client has not subscribed to yet (a +// cluster it has not applied, or one it rejected). The request must stay an +// open watch through that window: the client is waiting for a response and +// only issues a new request once it gets one, so a dropped watch leaves it on +// its current configuration until it reconnects. +var _ = Describe("Snapshot cache watch retention", func() { + const retentionNode = "watch-retention-node" + + var snapshotCache cache.SnapshotCache + + BeforeEach(func() { + snapshotCache = cache.NewSnapshotCache(cache.CacheSettings{Ads: true, Hash: TestIDHash{}}) + }) + + // endpointSnapshot builds a snapshot whose EDS resources are assignments + // for the named clusters. + endpointSnapshot := func(version string, clusters ...string) cache.Snapshot { + resources := make([]cache.Resource, 0, len(clusters)) + for _, name := range clusters { + resources = append(resources, resource.NewEnvoyResource(makeEndpoint(name))) + } + return cache.NewEasyGenericSnapshot(version, resources) + } + + endpointRequest := func(version string, names ...string) cache.Request { + return cache.Request{ + Node: &envoy_config_core_v3.Node{Id: retentionNode}, + TypeUrl: types.EndpointTypeV3, + ResourceNames: names, + VersionInfo: version, + } + } + + openWatches := func() int { + return snapshotCache.GetStatusInfo(retentionNode).GetNumWatches() + } + + It("keeps an open watch whose response a new snapshot cannot carry, and answers it once one can", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("1", "cluster-a")) + + initial, _ := snapshotCache.CreateWatch(endpointRequest("", "cluster-a")) + Eventually(initial, time.Second).Should(Receive()) + + // The client acks and re-requests: the watch is parked. + parked, cancel := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + defer cancel() + Expect(openWatches()).To(Equal(1)) + + // A snapshot that also carries an assignment for a cluster this client + // has not subscribed to cannot be sent to that watch. + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a", "cluster-b")) + Consistently(parked, 200*time.Millisecond, 20*time.Millisecond).ShouldNot(Receive()) + Expect(openWatches()).To(Equal(1), "the watch must survive a snapshot it cannot be sent") + + // Once the snapshot is one this request can accept, the retained watch + // receives it. + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("3", "cluster-a")) + var response cache.Response + Eventually(parked, time.Second).Should(Receive(&response)) + Expect(response.Version).To(Equal("3")) + Expect(openWatches()).To(Equal(0), "an answered watch is discarded") + }) + + It("opens a watch for a request the current snapshot cannot answer, and answers it once one can", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a", "cluster-b")) + + // The request asks for a newer version than it holds, so the cache + // tries to answer it immediately -- and cannot, because the snapshot + // carries cluster-b as well. + pending, cancel := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + defer cancel() + Consistently(pending, 200*time.Millisecond, 20*time.Millisecond).ShouldNot(Receive()) + Expect(openWatches()).To(Equal(1), "a request that could not be answered must become an open watch") + + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("3", "cluster-a")) + var response cache.Response + Eventually(pending, time.Second).Should(Receive(&response)) + Expect(response.Version).To(Equal("3")) + }) + + It("still answers a request the snapshot can carry immediately", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a")) + + answered, _ := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + var response cache.Response + Eventually(answered, time.Second).Should(Receive(&response)) + Expect(response.Version).To(Equal("2")) + Expect(openWatches()).To(Equal(0), "an immediately answered request leaves no watch behind") + }) + + It("still parks a request that is already up to date", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("1", "cluster-a")) + + parked, cancel := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + defer cancel() + Consistently(parked, 200*time.Millisecond, 20*time.Millisecond).ShouldNot(Receive()) + Expect(openWatches()).To(Equal(1)) + + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a")) + Eventually(parked, time.Second).Should(Receive()) + }) + + It("is unaffected for wildcard requests, which are never withheld", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("1", "cluster-a")) + + initial, _ := snapshotCache.CreateWatch(endpointRequest("")) + Eventually(initial, time.Second).Should(Receive()) + + parked, cancel := snapshotCache.CreateWatch(endpointRequest("1")) + defer cancel() + Expect(openWatches()).To(Equal(1)) + + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a", "cluster-b")) + Eventually(parked, time.Second).Should(Receive()) + Expect(openWatches()).To(Equal(0)) + }) +}) From 5c94c4914502e671ee84668dacbf52279f0dd3d7 Mon Sep 17 00:00:00 2001 From: "David L. Chandler" Date: Tue, 15 Sep 2026 17:50:10 -0400 Subject: [PATCH 2/5] configurability via SOLOKIT_RETAIN_UNANSWERED_XDS_WATCHES Signed-off-by: David L. Chandler --- .../retain-unanswered-xds-watches.yaml | 8 +- pkg/api/v1/control-plane/cache/simple.go | 61 +++++++++++--- .../cache/watch_retention_env_test.go | 82 +++++++++++++++++++ .../cache/watch_retention_test.go | 29 +++++++ 4 files changed, 166 insertions(+), 14 deletions(-) create mode 100644 pkg/api/v1/control-plane/cache/watch_retention_env_test.go diff --git a/changelog/v0.41.2/retain-unanswered-xds-watches.yaml b/changelog/v0.41.2/retain-unanswered-xds-watches.yaml index bbe82ce04..b147096ff 100644 --- a/changelog/v0.41.2/retain-unanswered-xds-watches.yaml +++ b/changelog/v0.41.2/retain-unanswered-xds-watches.yaml @@ -11,5 +11,11 @@ changelog: watch was discarded on a snapshot update and never registered for a new request. Because a client only issues a new request after receiving a response, nothing re-armed delivery, and it stayed on its previous - configuration until it reconnected. The watch is now retained on both + configuration until it reconnected. When enabled, the watch is retained on both paths and answered by the first snapshot it can accept. + Set SOLOKIT_RETAIN_UNANSWERED_XDS_WATCHES to "off" to restore the previous + behavior on both paths, or "on" to force retention. "default", unset, and + empty values follow the release default (currently off), which may change + in a future release; explicit "on" and "off" remain pinned. Other values + log a warning and use the release default. The setting is read when each + snapshot cache is created. diff --git a/pkg/api/v1/control-plane/cache/simple.go b/pkg/api/v1/control-plane/cache/simple.go index 1580f41be..e2e8c7275 100644 --- a/pkg/api/v1/control-plane/cache/simple.go +++ b/pkg/api/v1/control-plane/cache/simple.go @@ -18,6 +18,9 @@ import ( "context" "errors" "fmt" + "os" + + "github.com/solo-io/go-utils/contextutils" "sync" "time" @@ -100,8 +103,38 @@ type CacheSettings struct { PrioritySet map[int][]string } +// RetainUnansweredXDSWatchesEnv controls retention of withheld ADS watches. +// "on" and "off" pin the behavior; "default", unset, and empty follow the +// release default. Invalid values warn and also follow the release default. +const RetainUnansweredXDSWatchesEnv = "SOLOKIT_RETAIN_UNANSWERED_XDS_WATCHES" + +// Change this default independently of the explicit on/off overrides. +const retainUnansweredXDSWatchesDefault = false + +func retainUnansweredXDSWatches(logger log.Logger) bool { + value := os.Getenv(RetainUnansweredXDSWatchesEnv) + switch value { + case "on": + return true + case "off": + return false + case "", "default": + return retainUnansweredXDSWatchesDefault + default: + // The cache logger predates warning support. Use it when available, + // otherwise use the standard application logger without changing its interface. + warningLogger, ok := logger.(interface{ Warnf(string, ...interface{}) }) + if !ok { + warningLogger = contextutils.LoggerFrom(context.Background()) + } + warningLogger.Warnf("invalid %s value %q: expected default, on, or off; using default (%t)", RetainUnansweredXDSWatchesEnv, value, retainUnansweredXDSWatchesDefault) + return retainUnansweredXDSWatchesDefault + } +} + type snapshotCache struct { - log log.Logger + retainUnansweredWatches bool + log log.Logger // ads flag to hold responses until all resources are named ads bool @@ -132,14 +165,16 @@ type snapshotCache struct { // is OK. // // Logger is optional. +// RetainUnansweredXDSWatchesEnv is read once when each cache is created. func NewSnapshotCache(settings CacheSettings) SnapshotCache { return &snapshotCache{ - log: settings.Logger, - ads: settings.Ads, - snapshots: make(map[string]Snapshot), - status: make(map[string]*statusInfo), - hash: settings.Hash, - prioritySet: settings.PrioritySet, + retainUnansweredWatches: retainUnansweredXDSWatches(settings.Logger), + log: settings.Logger, + ads: settings.Ads, + snapshots: make(map[string]Snapshot), + status: make(map[string]*statusInfo), + hash: settings.Hash, + prioritySet: settings.PrioritySet, } } @@ -168,7 +203,7 @@ func (cache *snapshotCache) SetSnapshot(node string, snapshot Snapshot) { // empty resource - intended behavior is for the resources to be cleared if resources != nil { // snapshot has been initialized and exists - if !cache.respond(watch.Request, watch.Response, resources, version) { + if sent := cache.respond(watch.Request, watch.Response, resources, version); !sent && cache.retainUnansweredWatches { // The response was withheld (see respond): keep the watch so a // later snapshot that this request can accept is delivered to it. if cache.log != nil { @@ -178,7 +213,7 @@ func (cache *snapshotCache) SetSnapshot(node string, snapshot Snapshot) { return } - // discard the watch: the client creates a new one when it acks + // Discard answered watches, or all watches when retention is disabled. watches.Delete(pi) } } @@ -276,12 +311,12 @@ func (cache *snapshotCache) CreateWatch(request Request) (chan Response, func()) // If the snapshot is newer than the requested version, try to answer the // request immediately. The response can still be withheld in ADS mode (see - // respond), in which case the request has to become an open watch: the + // respond), in which case retention opens a watch: the // client is waiting for this response and does not issue a new request // until it receives one, so dropping the request here would leave the // client on its current configuration until it reconnects. if exists && request.VersionInfo != version { - if cache.respond(request, value, snapshot.GetResources(request.TypeUrl).Items, version) { + if sent := cache.respond(request, value, snapshot.GetResources(request.TypeUrl).Items, version); sent || !cache.retainUnansweredWatches { return value, func() { close(value) } @@ -323,14 +358,14 @@ func (cache *snapshotCache) cancelWatch(nodeID string, watchIndex PriorityIndex) // // It reports whether a response was sent. In ADS mode a request whose resource // names do not cover the snapshot cannot be answered yet, and callers must keep -// the watch open instead of treating the request as served: nothing else will +// the watch open when retention is enabled: nothing else will // answer it, because the client is waiting for this response and only issues a // new request once it receives one. // // TODO(kuat) do not respond always, see issue https://github.com/envoyproxy/go-control-plane/issues/46 func (cache *snapshotCache) respond(request Request, value chan Response, resources map[string]Resource, version string) bool { // for ADS, the request names must match the snapshot names - // if they do not, the response is withheld and false is returned, so the caller keeps the watch open + // if they do not, return false so callers can retain the watch when enabled if len(request.ResourceNames) != 0 && cache.ads { if err := Superset(nameSet(request.ResourceNames), resources); err != nil { if cache.log != nil { diff --git a/pkg/api/v1/control-plane/cache/watch_retention_env_test.go b/pkg/api/v1/control-plane/cache/watch_retention_env_test.go new file mode 100644 index 000000000..7be20b3f1 --- /dev/null +++ b/pkg/api/v1/control-plane/cache/watch_retention_env_test.go @@ -0,0 +1,82 @@ +package cache + +import ( + "fmt" + "os" + "strings" + "testing" +) + +type warningRecorder struct{ warnings []string } + +func (*warningRecorder) Debugf(string, ...interface{}) {} +func (*warningRecorder) Infof(string, ...interface{}) {} +func (*warningRecorder) Errorf(string, ...interface{}) {} +func (l *warningRecorder) Warnf(format string, args ...interface{}) { + l.warnings = append(l.warnings, fmt.Sprintf(format, args...)) +} + +func TestWatchRetentionEnvironment(t *testing.T) { + for _, tc := range []struct { + value string + enabled bool + warning bool + }{ + {"", retainUnansweredXDSWatchesDefault, false}, + {"default", retainUnansweredXDSWatchesDefault, false}, + {"on", true, false}, + {"off", false, false}, + {"invalid", retainUnansweredXDSWatchesDefault, true}, + {"ON", retainUnansweredXDSWatchesDefault, true}, + {" ", retainUnansweredXDSWatchesDefault, true}, + } { + t.Run(fmt.Sprintf("value=%q", tc.value), func(t *testing.T) { + t.Setenv(RetainUnansweredXDSWatchesEnv, tc.value) + logger := &warningRecorder{} + c := NewSnapshotCache(CacheSettings{Logger: logger}).(*snapshotCache) + if c.retainUnansweredWatches != tc.enabled { + t.Fatalf("retention = %t, want %t", c.retainUnansweredWatches, tc.enabled) + } + wantWarnings := 0 + if tc.warning { + wantWarnings = 1 + } + if len(logger.warnings) != wantWarnings { + t.Fatalf("warnings = %v, want %d", logger.warnings, wantWarnings) + } + if tc.warning && (!strings.Contains(logger.warnings[0], RetainUnansweredXDSWatchesEnv) || !strings.Contains(logger.warnings[0], fmt.Sprintf("%q", tc.value))) { + t.Fatalf("warning does not identify invalid setting: %s", logger.warnings[0]) + } + }) + } +} + +func TestWatchRetentionEnvironmentIsReadAtConstruction(t *testing.T) { + t.Setenv(RetainUnansweredXDSWatchesEnv, "on") + enabled := NewSnapshotCache(CacheSettings{}).(*snapshotCache) + t.Setenv(RetainUnansweredXDSWatchesEnv, "off") + disabled := NewSnapshotCache(CacheSettings{}).(*snapshotCache) + if !enabled.retainUnansweredWatches || disabled.retainUnansweredWatches { + t.Fatal("each cache must keep the setting read at construction") + } +} + +func TestWatchRetentionInvalidEnvironmentWithoutLogger(t *testing.T) { + t.Setenv(RetainUnansweredXDSWatchesEnv, "invalid") + c := NewSnapshotCache(CacheSettings{}).(*snapshotCache) + if c.retainUnansweredWatches != retainUnansweredXDSWatchesDefault { + t.Fatal("invalid setting must use the release default") + } +} + +func TestWatchRetentionUnsetEnvironment(t *testing.T) { + t.Setenv(RetainUnansweredXDSWatchesEnv, "") + if err := os.Unsetenv(RetainUnansweredXDSWatchesEnv); err != nil { + t.Fatal(err) + } + logger := &warningRecorder{} + c := NewSnapshotCache(CacheSettings{Logger: logger}).(*snapshotCache) + if c.retainUnansweredWatches != retainUnansweredXDSWatchesDefault || len(logger.warnings) != 0 { + t.Fatal("unset setting must use the release default without warning") + } +} diff --git a/pkg/api/v1/control-plane/cache/watch_retention_test.go b/pkg/api/v1/control-plane/cache/watch_retention_test.go index 3a0d185f6..f713a2f62 100644 --- a/pkg/api/v1/control-plane/cache/watch_retention_test.go +++ b/pkg/api/v1/control-plane/cache/watch_retention_test.go @@ -24,6 +24,7 @@ var _ = Describe("Snapshot cache watch retention", func() { var snapshotCache cache.SnapshotCache BeforeEach(func() { + GinkgoT().Setenv(cache.RetainUnansweredXDSWatchesEnv, "on") snapshotCache = cache.NewSnapshotCache(cache.CacheSettings{Ads: true, Hash: TestIDHash{}}) }) @@ -129,4 +130,32 @@ var _ = Describe("Snapshot cache watch retention", func() { Eventually(parked, time.Second).Should(Receive()) Expect(openWatches()).To(Equal(0)) }) + Context("with retention disabled", func() { + BeforeEach(func() { + GinkgoT().Setenv(cache.RetainUnansweredXDSWatchesEnv, "off") + snapshotCache = cache.NewSnapshotCache(cache.CacheSettings{Ads: true, Hash: TestIDHash{}}) + }) + + It("discards a parked watch even when the snapshot response is withheld", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("1", "cluster-a")) + pending, cancel := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + defer cancel() + Expect(openWatches()).To(Equal(1)) + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a", "cluster-b")) + Expect(openWatches()).To(Equal(0)) + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("3", "cluster-a")) + Expect(pending).ShouldNot(Receive()) + }) + + It("does not register a stale request whose immediate response is withheld", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a", "cluster-b")) + pending, cancel := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + Expect(openWatches()).To(Equal(0)) + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("3", "cluster-a")) + Expect(pending).ShouldNot(Receive()) + cancel() + Expect(pending).To(BeClosed()) + }) + }) + }) From f557ad2cfe4b8dcc82d21e969fd9cdc563e91cfb Mon Sep 17 00:00:00 2001 From: "David L. Chandler" Date: Thu, 17 Sep 2026 01:53:05 -0400 Subject: [PATCH 3/5] observability borrowed from #587 Signed-off-by: David L. Chandler --- .../retain-unanswered-xds-watches.yaml | 14 +++ pkg/api/v1/control-plane/cache/simple.go | 33 ++++- .../cache/suppressed_responses_test.go | 117 ++++++++++++++++++ 3 files changed, 161 insertions(+), 3 deletions(-) create mode 100644 pkg/api/v1/control-plane/cache/suppressed_responses_test.go diff --git a/changelog/v0.41.2/retain-unanswered-xds-watches.yaml b/changelog/v0.41.2/retain-unanswered-xds-watches.yaml index b147096ff..2f363e719 100644 --- a/changelog/v0.41.2/retain-unanswered-xds-watches.yaml +++ b/changelog/v0.41.2/retain-unanswered-xds-watches.yaml @@ -19,3 +19,17 @@ changelog: in a future release; explicit "on" and "off" remain pinned. Other values log a warning and use the release default. The setting is read when each snapshot cache is created. + - type: NEW_FEATURE + issueLink: https://github.com/solo-io/solo-kit/pull/587 + resolvesIssue: false + description: >- + Add the xds/suppressed_responses measure, counting the times the snapshot + cache intentionally withholds an xDS response, tagged with the resource + type and a reason. The only reason today is ads_name_mismatch: the + request's resource names do not cover the snapshot's resources of that + type. The measure is recorded whether or not watch retention is enabled, + because nothing else reports that the condition happened -- with retention + off the watch is silently discarded, and with it on the watch survives but + still goes unanswered until a compatible snapshot arrives. It is therefore + the signal for how often this occurs in a running fleet, and the basis for + deciding the release default of SOLOKIT_RETAIN_UNANSWERED_XDS_WATCHES. diff --git a/pkg/api/v1/control-plane/cache/simple.go b/pkg/api/v1/control-plane/cache/simple.go index e2e8c7275..5b6a4b738 100644 --- a/pkg/api/v1/control-plane/cache/simple.go +++ b/pkg/api/v1/control-plane/cache/simple.go @@ -32,9 +32,11 @@ import ( ) var ( - MResponses = stats.Int64("xds/responses", "The responses for envoy", "1") + MResponses = stats.Int64("xds/responses", "The responses for envoy", "1") + MSuppressedResponses = stats.Int64("xds/suppressed_responses", "The responses intentionally suppressed by cache logic", "1") - KeyType = tag.MustNewKey("type") + KeyType = tag.MustNewKey("type") + KeyReason = tag.MustNewKey("reason") ResponsesView = &view.View{ Name: "xds/responses", @@ -46,14 +48,35 @@ var ( }, } + // SuppressedResponsesView counts responses the cache withholds. It is + // recorded whether or not watch retention is enabled, because it is the only + // signal that says how often the condition fires at all: with retention off + // a withheld response silently discards the watch, and with it on the watch + // is kept but still goes unanswered until a compatible snapshot arrives. + // Either way nothing else in the cache reports that it happened. + SuppressedResponsesView = &view.View{ + Name: "xds/suppressed_responses", + Measure: MSuppressedResponses, + Description: "The times the cache intentionally withheld an xDS response", + Aggregation: view.Count(), + TagKeys: []tag.Key{ + KeyType, + KeyReason, + }, + } + VersionUpToDateError = errors.New("skip fetch: version up to date") // Compile-time assertion _ SnapshotCache = new(snapshotCache) ) +// suppressedResponseReasonAdsNameMismatch is recorded when ADS withholds a +// response because the request does not name every resource in the snapshot. +const suppressedResponseReasonAdsNameMismatch = "ads_name_mismatch" + func init() { - view.Register(ResponsesView) + view.Register(ResponsesView, SuppressedResponsesView) } // SnapshotCache is a snapshot-based cache that maintains a single versioned @@ -371,6 +394,10 @@ func (cache *snapshotCache) respond(request Request, value chan Response, resour if cache.log != nil { cache.log.Debugf("ADS mode: not responding to request: %v", err) } + stats.RecordWithTags(context.TODO(), []tag.Mutator{ + tag.Insert(KeyType, request.GetTypeUrl()), + tag.Insert(KeyReason, suppressedResponseReasonAdsNameMismatch), + }, MSuppressedResponses.M(1)) return false } } diff --git a/pkg/api/v1/control-plane/cache/suppressed_responses_test.go b/pkg/api/v1/control-plane/cache/suppressed_responses_test.go new file mode 100644 index 000000000..437d9c992 --- /dev/null +++ b/pkg/api/v1/control-plane/cache/suppressed_responses_test.go @@ -0,0 +1,117 @@ +package cache_test + +import ( + envoy_config_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/cache" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/resource" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/types" + "go.opencensus.io/stats/view" +) + +// A withheld response is otherwise invisible. With retention off the watch is +// silently discarded; with it on the watch survives but still goes unanswered +// until a compatible snapshot arrives. Neither case logs above debug or moves +// any other counter, so this measure is the only thing that says how often the +// condition fires -- which is what makes the retention default decidable from +// field data rather than from argument. +var _ = Describe("Suppressed response metric", func() { + + const suppressedNode = "suppressed-response-node" + + // suppressedCount totals the rows of xds/suppressed_responses. The view is + // process-wide and other specs record into it, so a test reads the delta + // across the action rather than an absolute. + suppressedCount := func() int64 { + rows, err := view.RetrieveData(cache.SuppressedResponsesView.Name) + Expect(err).NotTo(HaveOccurred()) + var total int64 + for _, row := range rows { + if data, ok := row.Data.(*view.CountData); ok { + total += data.Value + } + } + return total + } + + // withholdOneResponse asks for "a" while the snapshot also holds "b", which + // is the ADS name mismatch that makes the cache withhold. + withholdOneResponse := func(snapshotCache cache.SnapshotCache) func() { + snapshotCache.SetSnapshot(suppressedNode, &TestSnapshot{ + Endpoints: cache.NewResources("v1", []cache.Resource{ + resource.NewEnvoyResource(makeEndpoint("a")), + resource.NewEnvoyResource(makeEndpoint("b")), + }), + }) + _, cancel := snapshotCache.CreateWatch(cache.Request{ + TypeUrl: types.EndpointTypeV3, + ResourceNames: []string{"a"}, + VersionInfo: "v0", + Node: &envoy_config_core_v3.Node{Id: suppressedNode}, + }) + return cancel + } + + DescribeTable("counts a withheld response whether or not watches are retained", + func(retention string) { + GinkgoT().Setenv(cache.RetainUnansweredXDSWatchesEnv, retention) + snapshotCache := cache.NewSnapshotCache(cache.CacheSettings{Ads: true, Hash: TestIDHash{}}) + + before := suppressedCount() + cancel := withholdOneResponse(snapshotCache) + defer cancel() + + Expect(suppressedCount()).To(BeNumerically(">", before), + "a withheld response must be counted regardless of the retention setting") + }, + Entry("retention on", "on"), + Entry("retention off", "off"), + ) + + It("does not count a response that was sent", func() { + GinkgoT().Setenv(cache.RetainUnansweredXDSWatchesEnv, "on") + snapshotCache := cache.NewSnapshotCache(cache.CacheSettings{Ads: true, Hash: TestIDHash{}}) + snapshotCache.SetSnapshot(suppressedNode, &TestSnapshot{ + Endpoints: cache.NewResources("v1", []cache.Resource{ + resource.NewEnvoyResource(makeEndpoint("a")), + }), + }) + + before := suppressedCount() + // The request now names every resource in the snapshot, so it is answered. + _, cancel := snapshotCache.CreateWatch(cache.Request{ + TypeUrl: types.EndpointTypeV3, + ResourceNames: []string{"a"}, + VersionInfo: "v0", + Node: &envoy_config_core_v3.Node{Id: suppressedNode}, + }) + defer cancel() + + Expect(suppressedCount()).To(Equal(before)) + }) + + It("tags the reason so the cause is distinguishable", func() { + GinkgoT().Setenv(cache.RetainUnansweredXDSWatchesEnv, "on") + snapshotCache := cache.NewSnapshotCache(cache.CacheSettings{Ads: true, Hash: TestIDHash{}}) + cancel := withholdOneResponse(snapshotCache) + defer cancel() + + rows, err := view.RetrieveData(cache.SuppressedResponsesView.Name) + Expect(err).NotTo(HaveOccurred()) + + var reasons, typeURLs []string + for _, row := range rows { + for _, t := range row.Tags { + switch t.Key.Name() { + case "reason": + reasons = append(reasons, t.Value) + case "type": + typeURLs = append(typeURLs, t.Value) + } + } + } + Expect(reasons).To(ContainElement("ads_name_mismatch")) + Expect(typeURLs).To(ContainElement(types.EndpointTypeV3)) + }) +}) From 922b48c410f172ecd12668728530cf64db61c030 Mon Sep 17 00:00:00 2001 From: "David L. Chandler" Date: Thu, 17 Sep 2026 11:05:30 -0400 Subject: [PATCH 4/5] lint Signed-off-by: David L. Chandler --- pkg/api/v1/control-plane/cache/simple.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/api/v1/control-plane/cache/simple.go b/pkg/api/v1/control-plane/cache/simple.go index 5b6a4b738..6833393a2 100644 --- a/pkg/api/v1/control-plane/cache/simple.go +++ b/pkg/api/v1/control-plane/cache/simple.go @@ -20,10 +20,11 @@ import ( "fmt" "os" - "github.com/solo-io/go-utils/contextutils" "sync" "time" + "github.com/solo-io/go-utils/contextutils" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/log" "go.opencensus.io/stats" From d74a0353abef037d458b59ee6ed51e5981bd486b Mon Sep 17 00:00:00 2001 From: "David L. Chandler" Date: Thu, 17 Sep 2026 11:07:39 -0400 Subject: [PATCH 5/5] EXP_FIX Signed-off-by: David L. Chandler --- changelog/v0.41.2/retain-unanswered-xds-watches.yaml | 4 ++-- pkg/api/v1/control-plane/cache/simple.go | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/changelog/v0.41.2/retain-unanswered-xds-watches.yaml b/changelog/v0.41.2/retain-unanswered-xds-watches.yaml index 2f363e719..01c4d6423 100644 --- a/changelog/v0.41.2/retain-unanswered-xds-watches.yaml +++ b/changelog/v0.41.2/retain-unanswered-xds-watches.yaml @@ -13,7 +13,7 @@ changelog: response, nothing re-armed delivery, and it stayed on its previous configuration until it reconnected. When enabled, the watch is retained on both paths and answered by the first snapshot it can accept. - Set SOLOKIT_RETAIN_UNANSWERED_XDS_WATCHES to "off" to restore the previous + Set EXP_FIX to "off" to restore the previous behavior on both paths, or "on" to force retention. "default", unset, and empty values follow the release default (currently off), which may change in a future release; explicit "on" and "off" remain pinned. Other values @@ -32,4 +32,4 @@ changelog: off the watch is silently discarded, and with it on the watch survives but still goes unanswered until a compatible snapshot arrives. It is therefore the signal for how often this occurs in a running fleet, and the basis for - deciding the release default of SOLOKIT_RETAIN_UNANSWERED_XDS_WATCHES. + deciding the release default of EXP_FIX. diff --git a/pkg/api/v1/control-plane/cache/simple.go b/pkg/api/v1/control-plane/cache/simple.go index 6833393a2..154fb811c 100644 --- a/pkg/api/v1/control-plane/cache/simple.go +++ b/pkg/api/v1/control-plane/cache/simple.go @@ -130,7 +130,7 @@ type CacheSettings struct { // RetainUnansweredXDSWatchesEnv controls retention of withheld ADS watches. // "on" and "off" pin the behavior; "default", unset, and empty follow the // release default. Invalid values warn and also follow the release default. -const RetainUnansweredXDSWatchesEnv = "SOLOKIT_RETAIN_UNANSWERED_XDS_WATCHES" +const RetainUnansweredXDSWatchesEnv = "EXP_FIX" // Change this default independently of the explicit on/off overrides. const retainUnansweredXDSWatchesDefault = false