[Sotw][Linear cache] Ensure watches are properly considering subscription changes to not miss new resources (#1301)

This PR leverages the new Subscription interface passed into the CreateWatch method to properly track the resources returned as well as the evolution of the watch subscription. This allows a proper handling of resource re-subscription as well as semantics needed for the new wildcard mode

Signed-off-by: Valerian Roche <valerian.roche@datadoghq.com>

Author: valerian-roche

pkg/cache/v3/cache.go

@@ -40,15 +40,17 @@ type DeltaRequest = discovery.DeltaDiscoveryRequest
 // Though the methods may return mutable parts of the state for performance reasons,
 // the cache is expected to consider this state as immutable and thread safe between a watch creation and its cancellation.
 type Subscription interface {
-	// ReturnedResources returns a list of resources that were sent to the client and their associated versions.
+	// ReturnedResources returns the list of resources the client currently knows and their associated versions.
 	// The versions are:
 	//  - delta protocol: version of the specific resource set in the response.
 	//  - sotw protocol: version of the global response when the resource was last sent.
+	// The returned map must not be altered by the Cache.
 	ReturnedResources() map[string]string
 
 	// SubscribedResources returns the list of resources currently subscribed to by the client for the type.
 	// For delta it keeps track of subscription updates across requests
 	// For sotw it is a normalized view of the last request resources
+	// The returned map must not be altered by the Cache.
 	SubscribedResources() map[string]struct{}
 
 	// IsWildcard returns whether the client has a wildcard watch.
@@ -118,6 +120,11 @@ type Response interface {
 	// Get the version in the Response.
 	GetVersion() (string, error)
 
+	// GetReturnedResources returns the map of resources and their versions returned in the subscription.
+	// It may include more resources than directly set in the response to consider the full state of the client.
+	// The caller is expected to provide this unchanged to the next call to CreateWatch as part of the subscription.
+	GetReturnedResources() map[string]string
+
 	// Get the context provided during response creation.
 	GetContext() context.Context
 }
@@ -155,6 +162,12 @@ type RawResponse struct {
 	// Resources to be included in the response.
 	Resources []types.ResourceWithTTL
 
+	// ReturnedResources tracks the resources returned for the subscription and the version when it was last returned,
+	// including previously returned ones when using non-full state resources.
+	// It allows the cache to know what the client knows. The server will transparently forward this
+	// across requests, and the cache is responsible for its interpretation.
+	ReturnedResources map[string]string
+
 	// Whether this is a heartbeat response. For xDS versions that support TTL, this
 	// will be converted into a response that doesn't contain the actual resource protobuf.
 	// This allows for more lightweight updates that server only to update the TTL timer.
@@ -207,6 +220,12 @@ type PassthroughResponse struct {
 	DiscoveryResponse *discovery.DiscoveryResponse
 
 	ctx context.Context
+
+	// ReturnedResources tracks the resources returned for the subscription and the version when it was last returned,
+	// including previously returned ones when using non-full state resources.
+	// It allows the cache to know what the client knows. The server will transparently forward this
+	// across requests, and the cache is responsible for its interpretation.
+	ReturnedResources map[string]string
 }
 
 // DeltaPassthroughResponse is a pre constructed xDS response that need not go through marshaling transformations.
@@ -264,6 +283,10 @@ func (r *RawResponse) GetDiscoveryResponse() (*discovery.DiscoveryResponse, erro
 	return marshaledResponse.(*discovery.DiscoveryResponse), nil
 }
 
+func (r *RawResponse) GetReturnedResources() map[string]string {
+	return r.ReturnedResources
+}
+
 // GetDeltaDiscoveryResponse performs the marshaling the first time its called and uses the cached response subsequently.
 // We can do this because the marshaled response does not change across the calls.
 // This caching behavior is important in high throughput scenarios because grpc marshaling has a cost and it drives the cpu utilization under load.
@@ -367,6 +390,10 @@ func (r *PassthroughResponse) GetDiscoveryResponse() (*discovery.DiscoveryRespon
 	return r.DiscoveryResponse, nil
 }
 
+func (r *PassthroughResponse) GetReturnedResources() map[string]string {
+	return r.ReturnedResources
+}
+
 // GetDeltaDiscoveryResponse returns the final passthrough Delta Discovery Response.
 func (r *DeltaPassthroughResponse) GetDeltaDiscoveryResponse() (*discovery.DeltaDiscoveryResponse, error) {
 	return r.DeltaDiscoveryResponse, nil

pkg/cache/v3/delta_test.go

@@ -17,6 +17,7 @@ import (
 	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
 	"github.com/envoyproxy/go-control-plane/pkg/cache/v3"
+	"github.com/envoyproxy/go-control-plane/pkg/log"
 	rsrc "github.com/envoyproxy/go-control-plane/pkg/resource/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/server/stream/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/test/resource/v3"
@@ -31,7 +32,7 @@ func assertResourceMapEqual(t *testing.T, want, got map[string]types.Resource) {
 }
 
 func TestSnapshotCacheDeltaWatch(t *testing.T) {
-	c := cache.NewSnapshotCache(false, group{}, logger{t: t})
+	c := cache.NewSnapshotCache(false, group{}, log.NewTestLogger(t))
 	watches := make(map[string]chan cache.DeltaResponse)
 	subscriptions := make(map[string]stream.Subscription)
 
@@ -113,7 +114,7 @@ func TestSnapshotCacheDeltaWatch(t *testing.T) {
 }
 
 func TestDeltaRemoveResources(t *testing.T) {
-	c := cache.NewSnapshotCache(false, group{}, logger{t: t})
+	c := cache.NewSnapshotCache(false, group{}, log.NewTestLogger(t))
 	watches := make(map[string]chan cache.DeltaResponse)
 	subscriptions := make(map[string]*stream.Subscription)
 
@@ -188,7 +189,7 @@ func TestDeltaRemoveResources(t *testing.T) {
 }
 
 func TestConcurrentSetDeltaWatch(t *testing.T) {
-	c := cache.NewSnapshotCache(false, group{}, logger{t: t})
+	c := cache.NewSnapshotCache(false, group{}, log.NewTestLogger(t))
 	for i := 0; i < 50; i++ {
 		version := fmt.Sprintf("v%d", i)
 		func(i int) {
@@ -222,7 +223,7 @@ func TestConcurrentSetDeltaWatch(t *testing.T) {
 type testKey struct{}
 
 func TestSnapshotDeltaCacheWatchTimeout(t *testing.T) {
-	c := cache.NewSnapshotCache(true, group{}, logger{t: t})
+	c := cache.NewSnapshotCache(true, group{}, log.NewTestLogger(t))
 
 	// Create a non-buffered channel that will block sends.
 	watchCh := make(chan cache.DeltaResponse)
@@ -267,7 +268,7 @@ func TestSnapshotDeltaCacheWatchTimeout(t *testing.T) {
 }
 
 func TestSnapshotCacheDeltaWatchCancel(t *testing.T) {
-	c := cache.NewSnapshotCache(true, group{}, logger{t: t})
+	c := cache.NewSnapshotCache(true, group{}, log.NewTestLogger(t))
 	for _, typ := range testTypes {
 		responses := make(chan cache.DeltaResponse, 1)
 		cancel, err := c.CreateDeltaWatch(&discovery.DeltaDiscoveryRequest{