fix: keep xDS watches whose response has to be withheld

changelog/v0.41.2/retain-unanswered-xds-watches.yaml

@@ -0,0 +1,35 @@
+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. 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.
+  - 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.

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"
 
@@ -29,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",
@@ -43,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
@@ -100,8 +126,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 +188,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,9 +226,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 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 {
+							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 answered watches, or all watches when retention is disabled.
 					watches.Delete(pi)
 				}
 			}
@@ -266,25 +332,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 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 sent := cache.respond(request, value, snapshot.GetResources(request.TypeUrl).Items, version); sent || !cache.retainUnansweredWatches {
+			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 +378,27 @@ 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 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) {
+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, 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 {
 				cache.log.Debugf("ADS mode: not responding to request: %v", err)
 			}
-			return
+			stats.RecordWithTags(context.TODO(), []tag.Mutator{
+				tag.Insert(KeyType, request.GetTypeUrl()),
+				tag.Insert(KeyReason, suppressedResponseReasonAdsNameMismatch),
+			}, MSuppressedResponses.M(1))
+			return false
 		}
 	}
 	if cache.log != nil {
@@ -325,6 +409,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 {

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))
+	})
+})