From 20b18fdda1f333dd5816c1074defbcf98bd915fa Mon Sep 17 00:00:00 2001 From: Valerian Roche Date: Wed, 24 Aug 2022 22:06:29 -0400 Subject: [PATCH] Rework Cache interface to isolate it from streamState and make it more uniform between sotw and delta --- pkg/cache/v3/cache.go | 26 +++++++- pkg/cache/v3/delta.go | 15 ++--- pkg/cache/v3/delta_test.go | 46 ++++++------- pkg/cache/v3/linear.go | 21 +++--- pkg/cache/v3/linear_test.go | 118 ++++++++++++++++----------------- pkg/cache/v3/mux.go | 6 +- pkg/cache/v3/simple.go | 27 ++++---- pkg/cache/v3/simple_test.go | 43 +++++++----- pkg/cache/v3/status.go | 19 +++++- pkg/server/delta/v3/server.go | 8 +-- pkg/server/sotw/v3/server.go | 24 ++++--- pkg/server/stream/v3/stream.go | 51 +------------- pkg/server/v3/delta_test.go | 37 +++++------ pkg/server/v3/server.go | 7 +- pkg/server/v3/server_test.go | 3 +- 15 files changed, 222 insertions(+), 229 deletions(-) diff --git a/pkg/cache/v3/cache.go b/pkg/cache/v3/cache.go index 35e963343b..be124cb484 100644 --- a/pkg/cache/v3/cache.go +++ b/pkg/cache/v3/cache.go @@ -26,7 +26,6 @@ import ( "google.golang.org/protobuf/types/known/durationpb" "github.com/envoyproxy/go-control-plane/pkg/cache/types" - "github.com/envoyproxy/go-control-plane/pkg/server/stream/v3" discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3" ) @@ -37,6 +36,27 @@ type Request = discovery.DiscoveryRequest // DeltaRequest is an alias for the delta discovery request type. type DeltaRequest = discovery.DeltaDiscoveryRequest +// ClientState provides additional data on the client knowledge for the type matching the request +// This allows proper implementation of stateful aspects of the protocol (e.g. returning only some updated resources) +// 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 ClientState interface { + // GetKnownResources returns the list of resources the clients has ACKed and their associated version. + // 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 ACKed + GetKnownResources() map[string]string + + // GetSubscribedResources returns the list of resources currently subscribed to by the client for the type. + // For delta it keeps track across requests + // For sotw it is a normalized view of the request resources + GetSubscribedResources() map[string]struct{} + + // IsWildcard returns whether the client has a wildcard watch. + // This considers subtilities related to the current migration of wildcard definition within the protocol. + IsWildcard() bool +} + // ConfigWatcher requests watches for configuration resources by a node, last // applied version identifier, and resource names hint. The watch should send // the responses when they are ready. The watch can be canceled by the @@ -50,7 +70,7 @@ type ConfigWatcher interface { // // Cancel is an optional function to release resources in the producer. If // provided, the consumer may call this function multiple times. - CreateWatch(*Request, stream.StreamState, chan Response) (cancel func()) + CreateWatch(*Request, ClientState, chan Response) (cancel func()) // CreateDeltaWatch returns a new open incremental xDS watch. // @@ -59,7 +79,7 @@ type ConfigWatcher interface { // // Cancel is an optional function to release resources in the producer. If // provided, the consumer may call this function multiple times. - CreateDeltaWatch(*DeltaRequest, stream.StreamState, chan DeltaResponse) (cancel func()) + CreateDeltaWatch(*DeltaRequest, ClientState, chan DeltaResponse) (cancel func()) } // ConfigFetcher fetches configuration resources from cache diff --git a/pkg/cache/v3/delta.go b/pkg/cache/v3/delta.go index deaeeb7ed1..c5adf9fb8b 100644 --- a/pkg/cache/v3/delta.go +++ b/pkg/cache/v3/delta.go @@ -18,7 +18,6 @@ import ( "context" "github.com/envoyproxy/go-control-plane/pkg/cache/types" - "github.com/envoyproxy/go-control-plane/pkg/server/stream/v3" ) // groups together resource-related arguments for the createDeltaResponse function @@ -28,7 +27,7 @@ type resourceContainer struct { systemVersion string } -func createDeltaResponse(ctx context.Context, req *DeltaRequest, state stream.StreamState, resources resourceContainer) *RawDeltaResponse { +func createDeltaResponse(ctx context.Context, req *DeltaRequest, state ClientState, resources resourceContainer) *RawDeltaResponse { // variables to build our response with var nextVersionMap map[string]string var filtered []types.Resource @@ -37,7 +36,7 @@ func createDeltaResponse(ctx context.Context, req *DeltaRequest, state stream.St // If we are handling a wildcard request, we want to respond with all resources switch { case state.IsWildcard(): - if len(state.GetResourceVersions()) == 0 { + if len(state.GetKnownResources()) == 0 { filtered = make([]types.Resource, 0, len(resources.resourceMap)) } nextVersionMap = make(map[string]string, len(resources.resourceMap)) @@ -46,7 +45,7 @@ func createDeltaResponse(ctx context.Context, req *DeltaRequest, state stream.St // we can just set it here to be used for comparison later version := resources.versionMap[name] nextVersionMap[name] = version - prevVersion, found := state.GetResourceVersions()[name] + prevVersion, found := state.GetKnownResources()[name] if !found || (prevVersion != version) { filtered = append(filtered, r) } @@ -54,17 +53,17 @@ func createDeltaResponse(ctx context.Context, req *DeltaRequest, state stream.St // Compute resources for removal // The resource version can be set to "" here to trigger a removal even if never returned before - for name := range state.GetResourceVersions() { + for name := range state.GetKnownResources() { if _, ok := resources.resourceMap[name]; !ok { toRemove = append(toRemove, name) } } default: - nextVersionMap = make(map[string]string, len(state.GetSubscribedResourceNames())) + nextVersionMap = make(map[string]string, len(state.GetSubscribedResources())) // state.GetResourceVersions() may include resources no longer subscribed // In the current code this gets silently cleaned when updating the version map - for name := range state.GetSubscribedResourceNames() { - prevVersion, found := state.GetResourceVersions()[name] + for name := range state.GetSubscribedResources() { + prevVersion, found := state.GetKnownResources()[name] if r, ok := resources.resourceMap[name]; ok { nextVersion := resources.versionMap[name] if prevVersion != nextVersion { diff --git a/pkg/cache/v3/delta_test.go b/pkg/cache/v3/delta_test.go index 4999cad603..e171abad69 100644 --- a/pkg/cache/v3/delta_test.go +++ b/pkg/cache/v3/delta_test.go @@ -3,12 +3,12 @@ package cache_test import ( "context" "fmt" - "reflect" "testing" "time" "github.com/google/go-cmp/cmp" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "google.golang.org/protobuf/testing/protocmp" core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" @@ -35,13 +35,14 @@ func TestSnapshotCacheDeltaWatch(t *testing.T) { // Make our initial request as a wildcard to get all resources and make sure the wildcard requesting works as intended for _, typ := range testTypes { watches[typ] = make(chan cache.DeltaResponse, 1) + state := stream.NewStreamState(true, nil) c.CreateDeltaWatch(&discovery.DeltaDiscoveryRequest{ Node: &core.Node{ Id: "node", }, TypeUrl: typ, ResourceNamesSubscribe: names[typ], - }, stream.NewStreamState(true, nil), watches[typ]) + }, &state, watches[typ]) } if err := c.SetSnapshot(context.Background(), key, fixture.snapshot()); err != nil { @@ -69,7 +70,7 @@ func TestSnapshotCacheDeltaWatch(t *testing.T) { watches[typ] = make(chan cache.DeltaResponse, 1) state := stream.NewStreamState(false, versionMap[typ]) for resource := range versionMap[typ] { - state.GetSubscribedResourceNames()[resource] = struct{}{} + state.GetSubscribedResources()[resource] = struct{}{} } c.CreateDeltaWatch(&discovery.DeltaDiscoveryRequest{ Node: &core.Node{ @@ -77,7 +78,7 @@ func TestSnapshotCacheDeltaWatch(t *testing.T) { }, TypeUrl: typ, ResourceNamesSubscribe: names[typ], - }, state, watches[typ]) + }, &state, watches[typ]) } if count := c.GetStatusInfo(key).GetNumDeltaWatches(); count != len(testTypes) { @@ -123,12 +124,10 @@ func TestDeltaRemoveResources(t *testing.T) { Id: "node", }, TypeUrl: typ, - }, *streams[typ], watches[typ]) + }, streams[typ], watches[typ]) } - if err := c.SetSnapshot(context.Background(), key, fixture.snapshot()); err != nil { - t.Fatal(err) - } + require.NoError(t, c.SetSnapshot(context.Background(), key, fixture.snapshot())) for _, typ := range testTypes { t.Run(typ, func(t *testing.T) { @@ -139,7 +138,7 @@ func TestDeltaRemoveResources(t *testing.T) { nextVersionMap := out.GetNextVersionMap() streams[typ].SetResourceVersions(nextVersionMap) case <-time.After(time.Second): - t.Fatal("failed to receive a snapshot response") + require.Fail(t, "failed to receive a snapshot response") } }) } @@ -152,20 +151,17 @@ func TestDeltaRemoveResources(t *testing.T) { Node: &core.Node{ Id: "node", }, - TypeUrl: typ, - }, *streams[typ], watches[typ]) + TypeUrl: typ, + ResponseNonce: "nonce", + }, streams[typ], watches[typ]) } - if count := c.GetStatusInfo(key).GetNumDeltaWatches(); count != len(testTypes) { - t.Errorf("watches should be created for the latest version, saw %d watches expected %d", count, len(testTypes)) - } + assert.Equal(t, len(testTypes), c.GetStatusInfo(key).GetNumDeltaWatches(), "watches should be created for the latest version") // set a partially versioned snapshot with no endpoints snapshot2 := fixture.snapshot() snapshot2.Resources[types.Endpoint] = cache.NewResources(fixture.version2, []types.Resource{}) - if err := c.SetSnapshot(context.Background(), key, snapshot2); err != nil { - t.Fatal(err) - } + require.NoError(t, c.SetSnapshot(context.Background(), key, snapshot2)) // validate response for endpoints select { @@ -176,11 +172,9 @@ func TestDeltaRemoveResources(t *testing.T) { nextVersionMap := out.GetNextVersionMap() // make sure the version maps are different since we no longer are tracking any endpoint resources - if reflect.DeepEqual(streams[testTypes[0]].GetResourceVersions(), nextVersionMap) { - t.Fatalf("versionMap for the endpoint resource type did not change, received: %v, instead of an empty map", nextVersionMap) - } + require.Equal(t, nextVersionMap, streams[testTypes[0]].GetKnownResources(), "versionMap for the endpoint resource type did not change") case <-time.After(time.Second): - t.Fatal("failed to receive snapshot response") + assert.Fail(t, "failed to receive snapshot response") } } @@ -203,13 +197,14 @@ func TestConcurrentSetDeltaWatch(t *testing.T) { t.Fatalf("snapshot failed: %s", err) } } else { + state := stream.NewStreamState(false, make(map[string]string)) cancel := c.CreateDeltaWatch(&discovery.DeltaDiscoveryRequest{ Node: &core.Node{ Id: id, }, TypeUrl: rsrc.EndpointType, ResourceNamesSubscribe: []string{clusterName}, - }, stream.NewStreamState(false, make(map[string]string)), responses) + }, &state, responses) defer cancel() } @@ -226,14 +221,14 @@ func TestSnapshotDeltaCacheWatchTimeout(t *testing.T) { // Create a non-buffered channel that will block sends. watchCh := make(chan cache.DeltaResponse) state := stream.NewStreamState(false, nil) - state.SetSubscribedResourceNames(map[string]struct{}{names[rsrc.EndpointType][0]: {}}) + state.SetSubscribedResources(map[string]struct{}{names[rsrc.EndpointType][0]: {}}) c.CreateDeltaWatch(&discovery.DeltaDiscoveryRequest{ Node: &core.Node{ Id: key, }, TypeUrl: rsrc.EndpointType, ResourceNamesSubscribe: names[rsrc.EndpointType], - }, state, watchCh) + }, &state, watchCh) // The first time we set the snapshot without consuming from the blocking channel, so this should time out. ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond) @@ -269,13 +264,14 @@ func TestSnapshotCacheDeltaWatchCancel(t *testing.T) { c := cache.NewSnapshotCache(true, group{}, logger{t: t}) for _, typ := range testTypes { responses := make(chan cache.DeltaResponse, 1) + state := stream.NewStreamState(false, make(map[string]string)) cancel := c.CreateDeltaWatch(&discovery.DeltaDiscoveryRequest{ Node: &core.Node{ Id: key, }, TypeUrl: typ, ResourceNamesSubscribe: names[typ], - }, stream.NewStreamState(false, make(map[string]string)), responses) + }, &state, responses) // Cancel the watch cancel() diff --git a/pkg/cache/v3/linear.go b/pkg/cache/v3/linear.go index ae55105d59..8b4b4b1652 100644 --- a/pkg/cache/v3/linear.go +++ b/pkg/cache/v3/linear.go @@ -24,7 +24,6 @@ import ( "github.com/envoyproxy/go-control-plane/pkg/cache/types" "github.com/envoyproxy/go-control-plane/pkg/log" - "github.com/envoyproxy/go-control-plane/pkg/server/stream/v3" ) type watches = map[chan Response]struct{} @@ -164,11 +163,11 @@ func (cache *LinearCache) notifyAll(modified map[string]struct{}) { } for id, watch := range cache.deltaWatches { - if !watch.StreamState.WatchesResources(modified) { + if !watch.WatchesResources(modified) { continue } - res := cache.respondDelta(watch.Request, watch.Response, watch.StreamState) + res := cache.respondDelta(watch.Request, watch.Response, watch.clientState) if res != nil { delete(cache.deltaWatches, id) } @@ -176,8 +175,8 @@ func (cache *LinearCache) notifyAll(modified map[string]struct{}) { } } -func (cache *LinearCache) respondDelta(request *DeltaRequest, value chan DeltaResponse, state stream.StreamState) *RawDeltaResponse { - resp := createDeltaResponse(context.Background(), request, state, resourceContainer{ +func (cache *LinearCache) respondDelta(request *DeltaRequest, value chan DeltaResponse, clientState ClientState) *RawDeltaResponse { + resp := createDeltaResponse(context.Background(), request, clientState, resourceContainer{ resourceMap: cache.resources, versionMap: cache.versionMap, systemVersion: cache.getVersion(), @@ -187,7 +186,7 @@ func (cache *LinearCache) respondDelta(request *DeltaRequest, value chan DeltaRe if len(resp.Resources) > 0 || len(resp.RemovedResources) > 0 { if cache.log != nil { cache.log.Debugf("[linear cache] node: %s, sending delta response with resources: %v removed resources %v wildcard: %t", - request.GetNode().GetId(), resp.Resources, resp.RemovedResources, state.IsWildcard()) + request.GetNode().GetId(), resp.Resources, resp.RemovedResources, clientState.IsWildcard()) } value <- resp return resp @@ -298,7 +297,7 @@ func (cache *LinearCache) GetResources() map[string]types.Resource { return resources } -func (cache *LinearCache) CreateWatch(request *Request, streamState stream.StreamState, value chan Response) func() { +func (cache *LinearCache) CreateWatch(request *Request, clientState ClientState, value chan Response) func() { if request.TypeUrl != cache.typeURL { value <- nil return nil @@ -371,7 +370,7 @@ func (cache *LinearCache) CreateWatch(request *Request, streamState stream.Strea } } -func (cache *LinearCache) CreateDeltaWatch(request *DeltaRequest, state stream.StreamState, value chan DeltaResponse) func() { +func (cache *LinearCache) CreateDeltaWatch(request *DeltaRequest, clientState ClientState, value chan DeltaResponse) func() { cache.mu.Lock() defer cache.mu.Unlock() @@ -388,7 +387,7 @@ func (cache *LinearCache) CreateDeltaWatch(request *DeltaRequest, state stream.S cache.log.Errorf("failed to update version map: %v", err) } } - response := cache.respondDelta(request, value, state) + response := cache.respondDelta(request, value, clientState) // if respondDelta returns nil this means that there is no change in any resource version // create a new watch accordingly @@ -396,10 +395,10 @@ func (cache *LinearCache) CreateDeltaWatch(request *DeltaRequest, state stream.S watchID := cache.nextDeltaWatchID() if cache.log != nil { cache.log.Infof("[linear cache] open delta watch ID:%d for %s Resources:%v, system version %q", watchID, - cache.typeURL, state.GetSubscribedResourceNames(), cache.getVersion()) + cache.typeURL, clientState.GetSubscribedResources(), cache.getVersion()) } - cache.deltaWatches[watchID] = DeltaResponseWatch{Request: request, Response: value, StreamState: state} + cache.deltaWatches[watchID] = DeltaResponseWatch{Request: request, Response: value, clientState: clientState} return cache.cancelDeltaWatch(watchID) } diff --git a/pkg/cache/v3/linear_test.go b/pkg/cache/v3/linear_test.go index 617d90366e..8f8db7fe04 100644 --- a/pkg/cache/v3/linear_test.go +++ b/pkg/cache/v3/linear_test.go @@ -191,19 +191,19 @@ func hashResource(t *testing.T, resource types.Resource) string { func createWildcardDeltaWatch(c *LinearCache, w chan DeltaResponse) { state := stream.NewStreamState(true, nil) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) resp := <-w state.SetResourceVersions(resp.GetNextVersionMap()) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) // Ensure the watch is set properly with cache values + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) // Ensure the watch is set properly with cache values } func TestLinearInitialResources(t *testing.T) { streamState := stream.NewStreamState(false, map[string]string{}) c := NewLinearCache(testType, WithInitialResources(map[string]types.Resource{"a": testResource("a"), "b": testResource("b")})) w := make(chan Response, 1) - c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType}, streamState, w) + c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType}, &streamState, w) verifyResponse(t, w, "0", 1) - c.CreateWatch(&Request{TypeUrl: testType}, streamState, w) + c.CreateWatch(&Request{TypeUrl: testType}, &streamState, w) verifyResponse(t, w, "0", 2) checkVersionMapNotSet(t, c) } @@ -217,7 +217,7 @@ func TestLinearCornerCases(t *testing.T) { } // create an incorrect type URL request w := make(chan Response, 1) - c.CreateWatch(&Request{TypeUrl: "test"}, streamState, w) + c.CreateWatch(&Request{TypeUrl: "test"}, &streamState, w) select { case r := <-w: if r != nil { @@ -234,12 +234,12 @@ func TestLinearBasic(t *testing.T) { // Create watches before a resource is ready w1 := make(chan Response, 1) - c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, streamState, w1) + c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, &streamState, w1) mustBlock(t, w1) checkVersionMapNotSet(t, c) w := make(chan Response, 1) - c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, streamState, w) + c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, &streamState, w) mustBlock(t, w) checkWatchCount(t, c, "a", 2) checkWatchCount(t, c, "b", 1) @@ -250,19 +250,19 @@ func TestLinearBasic(t *testing.T) { verifyResponse(t, w, "1", 1) // Request again, should get same response - c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, streamState, w) + c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, &streamState, w) checkWatchCount(t, c, "a", 0) verifyResponse(t, w, "1", 1) - c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, streamState, w) + c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, &streamState, w) checkWatchCount(t, c, "a", 0) verifyResponse(t, w, "1", 1) // Add another element and update the first, response should be different require.NoError(t, c.UpdateResource("b", testResource("b"))) require.NoError(t, c.UpdateResource("a", testResource("aa"))) - c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, streamState, w) + c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, &streamState, w) verifyResponse(t, w, "3", 1) - c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, streamState, w) + c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, &streamState, w) verifyResponse(t, w, "3", 2) // Ensure the version map was not created as we only ever used stow watches checkVersionMapNotSet(t, c) @@ -274,10 +274,10 @@ func TestLinearSetResources(t *testing.T) { // Create new resources w1 := make(chan Response, 1) - c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, streamState, w1) + c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, &streamState, w1) mustBlock(t, w1) w2 := make(chan Response, 1) - c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, streamState, w2) + c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, &streamState, w2) mustBlock(t, w2) c.SetResources(map[string]types.Resource{ "a": testResource("a"), @@ -287,9 +287,9 @@ func TestLinearSetResources(t *testing.T) { verifyResponse(t, w2, "1", 2) // the version was only incremented once for all resources // Add another element and update the first, response should be different - c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "1"}, streamState, w1) + c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "1"}, &streamState, w1) mustBlock(t, w1) - c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"}, streamState, w2) + c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"}, &streamState, w2) mustBlock(t, w2) c.SetResources(map[string]types.Resource{ "a": testResource("aa"), @@ -300,9 +300,9 @@ func TestLinearSetResources(t *testing.T) { verifyResponse(t, w2, "2", 3) // Delete resource - c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "2"}, streamState, w1) + c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "2"}, &streamState, w1) mustBlock(t, w1) - c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "2"}, streamState, w2) + c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "2"}, &streamState, w2) mustBlock(t, w2) c.SetResources(map[string]types.Resource{ "b": testResource("b"), @@ -334,14 +334,14 @@ func TestLinearVersionPrefix(t *testing.T) { c := NewLinearCache(testType, WithVersionPrefix("instance1-")) w := make(chan Response, 1) - c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, streamState, w) + c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, &streamState, w) verifyResponse(t, w, "instance1-0", 0) require.NoError(t, c.UpdateResource("a", testResource("a"))) - c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, streamState, w) + c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, &streamState, w) verifyResponse(t, w, "instance1-1", 1) - c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "instance1-1"}, streamState, w) + c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "instance1-1"}, &streamState, w) mustBlock(t, w) checkWatchCount(t, c, "a", 1) } @@ -350,17 +350,17 @@ func TestLinearDeletion(t *testing.T) { streamState := stream.NewStreamState(false, map[string]string{}) c := NewLinearCache(testType, WithInitialResources(map[string]types.Resource{"a": testResource("a"), "b": testResource("b")})) w := make(chan Response, 1) - c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, streamState, w) + c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"}, &streamState, w) mustBlock(t, w) checkWatchCount(t, c, "a", 1) require.NoError(t, c.DeleteResource("a")) verifyResponse(t, w, "1", 0) checkWatchCount(t, c, "a", 0) - c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, streamState, w) + c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, &streamState, w) verifyResponse(t, w, "1", 1) checkWatchCount(t, c, "b", 0) require.NoError(t, c.DeleteResource("b")) - c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"}, streamState, w) + c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"}, &streamState, w) verifyResponse(t, w, "2", 0) checkWatchCount(t, c, "b", 0) } @@ -369,10 +369,10 @@ func TestLinearWatchTwo(t *testing.T) { streamState := stream.NewStreamState(false, map[string]string{}) c := NewLinearCache(testType, WithInitialResources(map[string]types.Resource{"a": testResource("a"), "b": testResource("b")})) w := make(chan Response, 1) - c.CreateWatch(&Request{ResourceNames: []string{"a", "b"}, TypeUrl: testType, VersionInfo: "0"}, streamState, w) + c.CreateWatch(&Request{ResourceNames: []string{"a", "b"}, TypeUrl: testType, VersionInfo: "0"}, &streamState, w) mustBlock(t, w) w1 := make(chan Response, 1) - c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, streamState, w1) + c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"}, &streamState, w1) mustBlock(t, w1) require.NoError(t, c.UpdateResource("a", testResource("aa"))) // should only get the modified resource @@ -387,14 +387,14 @@ func TestLinearCancel(t *testing.T) { // cancel watch-all w := make(chan Response, 1) - cancel := c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"}, streamState, w) + cancel := c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"}, &streamState, w) mustBlock(t, w) checkWatchCount(t, c, "a", 1) cancel() checkWatchCount(t, c, "a", 0) // cancel watch for "a" - cancel = c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "1"}, streamState, w) + cancel = c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "1"}, &streamState, w) mustBlock(t, w) checkWatchCount(t, c, "a", 1) cancel() @@ -404,10 +404,10 @@ func TestLinearCancel(t *testing.T) { w2 := make(chan Response, 1) w3 := make(chan Response, 1) w4 := make(chan Response, 1) - cancel = c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "1"}, streamState, w) - cancel2 := c.CreateWatch(&Request{ResourceNames: []string{"b"}, TypeUrl: testType, VersionInfo: "1"}, streamState, w2) - cancel3 := c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"}, streamState, w3) - cancel4 := c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"}, streamState, w4) + cancel = c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "1"}, &streamState, w) + cancel2 := c.CreateWatch(&Request{ResourceNames: []string{"b"}, TypeUrl: testType, VersionInfo: "1"}, &streamState, w2) + cancel3 := c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"}, &streamState, w3) + cancel4 := c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"}, &streamState, w4) mustBlock(t, w) mustBlock(t, w2) mustBlock(t, w3) @@ -449,7 +449,7 @@ func TestLinearConcurrentSetWatch(t *testing.T) { ResourceNames: []string{id, id2}, VersionInfo: "0", TypeUrl: testType, - }, streamState, value) + }, &streamState, value) // wait until all updates apply verifyResponse(t, value, "", 1) } @@ -462,11 +462,11 @@ func TestLinearDeltaWildcard(t *testing.T) { c := NewLinearCache(testType) state1 := stream.NewStreamState(true, map[string]string{}) w1 := make(chan DeltaResponse, 1) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state1, w1) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state1, w1) mustBlockDelta(t, w1) state2 := stream.NewStreamState(true, map[string]string{}) w2 := make(chan DeltaResponse, 1) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state2, w2) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state2, w2) mustBlockDelta(t, w1) checkDeltaWatchCount(t, c, 2) @@ -491,16 +491,16 @@ func TestLinearDeltaExistingResources(t *testing.T) { assert.NoError(t, err) state := stream.NewStreamState(false, nil) - state.SetSubscribedResourceNames(map[string]struct{}{"b": {}, "c": {}}) // watching b and c - not interested in a + state.SetSubscribedResources(map[string]struct{}{"b": {}, "c": {}}) // watching b and c - not interested in a w := make(chan DeltaResponse, 1) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) checkDeltaWatchCount(t, c, 0) verifyDeltaResponse(t, w, []resourceInfo{{"b", hashB}}, []string{}) state = stream.NewStreamState(false, nil) - state.SetSubscribedResourceNames(map[string]struct{}{"a": {}, "b": {}}) + state.SetSubscribedResources(map[string]struct{}{"a": {}, "b": {}}) w = make(chan DeltaResponse, 1) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) checkDeltaWatchCount(t, c, 0) verifyDeltaResponse(t, w, []resourceInfo{{"b", hashB}, {"a", hashA}}, nil) } @@ -517,16 +517,16 @@ func TestLinearDeltaInitialResourcesVersionSet(t *testing.T) { assert.NoError(t, err) state := stream.NewStreamState(false, map[string]string{"b": hashB}) - state.SetSubscribedResourceNames(map[string]struct{}{"a": {}, "b": {}}) + state.SetSubscribedResources(map[string]struct{}{"a": {}, "b": {}}) w := make(chan DeltaResponse, 1) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) checkDeltaWatchCount(t, c, 0) verifyDeltaResponse(t, w, []resourceInfo{{"a", hashA}}, nil) // b is up to date and shouldn't be returned state = stream.NewStreamState(false, map[string]string{"a": hashA, "b": hashB}) - state.SetSubscribedResourceNames(map[string]struct{}{"a": {}, "b": {}}) + state.SetSubscribedResources(map[string]struct{}{"a": {}, "b": {}}) w = make(chan DeltaResponse, 1) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) mustBlockDelta(t, w) checkDeltaWatchCount(t, c, 1) b = &endpoint.ClusterLoadAssignment{ClusterName: "b", Endpoints: []*endpoint.LocalityLbEndpoints{{Priority: 10}}} // new version of b @@ -551,17 +551,17 @@ func TestLinearDeltaResourceUpdate(t *testing.T) { checkVersionMapNotSet(t, c) state := stream.NewStreamState(false, nil) - state.SetSubscribedResourceNames(map[string]struct{}{"a": {}, "b": {}}) + state.SetSubscribedResources(map[string]struct{}{"a": {}, "b": {}}) w := make(chan DeltaResponse, 1) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) checkDeltaWatchCount(t, c, 0) verifyDeltaResponse(t, w, []resourceInfo{{"b", hashB}, {"a", hashA}}, nil) checkVersionMapSet(t, c) state = stream.NewStreamState(false, map[string]string{"a": hashA, "b": hashB}) - state.SetSubscribedResourceNames(map[string]struct{}{"a": {}, "b": {}}) + state.SetSubscribedResources(map[string]struct{}{"a": {}, "b": {}}) w = make(chan DeltaResponse, 1) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) mustBlockDelta(t, w) checkDeltaWatchCount(t, c, 1) @@ -587,16 +587,16 @@ func TestLinearDeltaResourceDelete(t *testing.T) { assert.NoError(t, err) state := stream.NewStreamState(false, nil) - state.SetSubscribedResourceNames(map[string]struct{}{"a": {}, "b": {}}) + state.SetSubscribedResources(map[string]struct{}{"a": {}, "b": {}}) w := make(chan DeltaResponse, 1) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) checkDeltaWatchCount(t, c, 0) verifyDeltaResponse(t, w, []resourceInfo{{"b", hashB}, {"a", hashA}}, nil) state = stream.NewStreamState(false, map[string]string{"a": hashA, "b": hashB}) - state.SetSubscribedResourceNames(map[string]struct{}{"a": {}, "b": {}}) + state.SetSubscribedResources(map[string]struct{}{"a": {}, "b": {}}) w = make(chan DeltaResponse, 1) - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) mustBlockDelta(t, w) checkDeltaWatchCount(t, c, 1) @@ -612,13 +612,13 @@ func TestLinearDeltaMultiResourceUpdates(t *testing.T) { c := NewLinearCache(testType) state := stream.NewStreamState(false, nil) - state.SetSubscribedResourceNames(map[string]struct{}{"a": {}, "b": {}}) + state.SetSubscribedResources(map[string]struct{}{"a": {}, "b": {}}) w := make(chan DeltaResponse, 1) checkVersionMapNotSet(t, c) assert.Equal(t, 0, c.NumResources()) // Initial update - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) mustBlockDelta(t, w) checkDeltaWatchCount(t, c, 1) // The version map should now be created, even if empty @@ -636,7 +636,7 @@ func TestLinearDeltaMultiResourceUpdates(t *testing.T) { state.SetResourceVersions(resp.GetNextVersionMap()) // Multiple updates - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) mustBlockDelta(t, w) checkDeltaWatchCount(t, c, 1) a = &endpoint.ClusterLoadAssignment{ClusterName: "a", Endpoints: []*endpoint.LocalityLbEndpoints{ //resource update @@ -656,7 +656,7 @@ func TestLinearDeltaMultiResourceUpdates(t *testing.T) { state.SetResourceVersions(resp.GetNextVersionMap()) // Update/add/delete - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) mustBlockDelta(t, w) checkDeltaWatchCount(t, c, 1) a = &endpoint.ClusterLoadAssignment{ClusterName: "a", Endpoints: []*endpoint.LocalityLbEndpoints{ //resource update @@ -675,7 +675,7 @@ func TestLinearDeltaMultiResourceUpdates(t *testing.T) { state.SetResourceVersions(resp.GetNextVersionMap()) // Re-add previously deleted watched resource - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &state, w) mustBlockDelta(t, w) checkDeltaWatchCount(t, c, 1) b = &endpoint.ClusterLoadAssignment{ClusterName: "b", Endpoints: []*endpoint.LocalityLbEndpoints{}} // recreate watched resource @@ -738,7 +738,7 @@ func TestLinearMixedWatches(t *testing.T) { sotwState := stream.NewStreamState(false, nil) w := make(chan Response, 1) - c.CreateWatch(&Request{ResourceNames: []string{"a", "b"}, TypeUrl: testType, VersionInfo: c.getVersion()}, sotwState, w) + c.CreateWatch(&Request{ResourceNames: []string{"a", "b"}, TypeUrl: testType, VersionInfo: c.getVersion()}, &sotwState, w) mustBlock(t, w) checkVersionMapNotSet(t, c) @@ -752,16 +752,16 @@ func TestLinearMixedWatches(t *testing.T) { verifyResponse(t, w, c.getVersion(), 1) checkVersionMapNotSet(t, c) - c.CreateWatch(&Request{ResourceNames: []string{"a", "b"}, TypeUrl: testType, VersionInfo: c.getVersion()}, sotwState, w) + c.CreateWatch(&Request{ResourceNames: []string{"a", "b"}, TypeUrl: testType, VersionInfo: c.getVersion()}, &sotwState, w) mustBlock(t, w) checkVersionMapNotSet(t, c) deltaState := stream.NewStreamState(false, map[string]string{"a": hashA, "b": hashB}) - deltaState.SetSubscribedResourceNames(map[string]struct{}{"a": {}, "b": {}}) + deltaState.SetSubscribedResources(map[string]struct{}{"a": {}, "b": {}}) wd := make(chan DeltaResponse, 1) // Initial update - c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, deltaState, wd) + c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, &deltaState, wd) mustBlockDelta(t, wd) checkDeltaWatchCount(t, c, 1) checkVersionMapSet(t, c) diff --git a/pkg/cache/v3/mux.go b/pkg/cache/v3/mux.go index db5a65d0a7..5b5db4b489 100644 --- a/pkg/cache/v3/mux.go +++ b/pkg/cache/v3/mux.go @@ -17,8 +17,6 @@ package cache import ( "context" "errors" - - "github.com/envoyproxy/go-control-plane/pkg/server/stream/v3" ) // MuxCache multiplexes across several caches using a classification function. @@ -37,7 +35,7 @@ type MuxCache struct { var _ Cache = &MuxCache{} -func (mux *MuxCache) CreateWatch(request *Request, state stream.StreamState, value chan Response) func() { +func (mux *MuxCache) CreateWatch(request *Request, state ClientState, value chan Response) func() { key := mux.Classify(request) cache, exists := mux.Caches[key] if !exists { @@ -47,7 +45,7 @@ func (mux *MuxCache) CreateWatch(request *Request, state stream.StreamState, val return cache.CreateWatch(request, state, value) } -func (mux *MuxCache) CreateDeltaWatch(request *DeltaRequest, state stream.StreamState, value chan DeltaResponse) func() { +func (mux *MuxCache) CreateDeltaWatch(request *DeltaRequest, state ClientState, value chan DeltaResponse) func() { key := mux.ClassifyDelta(request) cache, exists := mux.Caches[key] if !exists { diff --git a/pkg/cache/v3/simple.go b/pkg/cache/v3/simple.go index 5fe36c5b9e..bb2c5a313a 100644 --- a/pkg/cache/v3/simple.go +++ b/pkg/cache/v3/simple.go @@ -23,7 +23,6 @@ import ( "github.com/envoyproxy/go-control-plane/pkg/cache/types" "github.com/envoyproxy/go-control-plane/pkg/log" - "github.com/envoyproxy/go-control-plane/pkg/server/stream/v3" ) // ResourceSnapshot is an abstract snapshot of a collection of resources that @@ -265,7 +264,7 @@ func (cache *snapshotCache) SetSnapshot(ctx context.Context, node string, snapsh snapshot, watch.Request, watch.Response, - watch.StreamState, + watch.clientState, ) if err != nil { return err @@ -322,7 +321,7 @@ func superset(names map[string]bool, resources map[string]types.ResourceWithTTL) } // CreateWatch returns a watch for an xDS request. -func (cache *snapshotCache) CreateWatch(request *Request, streamState stream.StreamState, value chan Response) func() { +func (cache *snapshotCache) CreateWatch(request *Request, clientState ClientState, value chan Response) func() { nodeID := cache.hash.ID(request.Node) cache.mu.Lock() @@ -347,7 +346,7 @@ func (cache *snapshotCache) CreateWatch(request *Request, streamState stream.Str } if exists { - knownResourceNames := streamState.GetKnownResourceNames(request.TypeUrl) + knownResourceNames := clientState.GetKnownResources() diff := []string{} for _, r := range request.ResourceNames { if _, ok := knownResourceNames[r]; !ok { @@ -461,7 +460,7 @@ func createResponse(ctx context.Context, request *Request, resources map[string] } // CreateDeltaWatch returns a watch for a delta xDS request which implements the Simple SnapshotCache. -func (cache *snapshotCache) CreateDeltaWatch(request *DeltaRequest, state stream.StreamState, value chan DeltaResponse) func() { +func (cache *snapshotCache) CreateDeltaWatch(request *DeltaRequest, clientState ClientState, value chan DeltaResponse) func() { nodeID := cache.hash.ID(request.Node) t := request.GetTypeUrl() @@ -490,7 +489,7 @@ func (cache *snapshotCache) CreateDeltaWatch(request *DeltaRequest, state stream if err != nil { cache.log.Errorf("failed to compute version for snapshot resources inline: %s", err) } - response, err := cache.respondDelta(context.Background(), snapshot, request, value, state) + response, err := cache.respondDelta(context.Background(), snapshot, request, value, clientState) if err != nil { cache.log.Errorf("failed to respond with delta response: %s", err) } @@ -502,12 +501,12 @@ func (cache *snapshotCache) CreateDeltaWatch(request *DeltaRequest, state stream watchID := cache.nextDeltaWatchID() if exists { - cache.log.Infof("open delta watch ID:%d for %s Resources:%v from nodeID: %q, version %q", watchID, t, state.GetSubscribedResourceNames(), nodeID, snapshot.GetVersion(t)) + cache.log.Infof("open delta watch ID:%d for %s Resources:%v from nodeID: %q, version %q", watchID, t, clientState.GetSubscribedResources(), nodeID, snapshot.GetVersion(t)) } else { - cache.log.Infof("open delta watch ID:%d for %s Resources:%v from nodeID: %q", watchID, t, state.GetSubscribedResourceNames(), nodeID) + cache.log.Infof("open delta watch ID:%d for %s Resources:%v from nodeID: %q", watchID, t, clientState.GetSubscribedResources(), nodeID) } - info.setDeltaResponseWatch(watchID, DeltaResponseWatch{Request: request, Response: value, StreamState: state}) + info.setDeltaResponseWatch(watchID, DeltaResponseWatch{Request: request, Response: value, clientState: clientState}) return cache.cancelDeltaWatch(nodeID, watchID) } @@ -515,20 +514,20 @@ func (cache *snapshotCache) CreateDeltaWatch(request *DeltaRequest, state stream } // Respond to a delta watch with the provided snapshot value. If the response is nil, there has been no state change. -func (cache *snapshotCache) respondDelta(ctx context.Context, snapshot ResourceSnapshot, request *DeltaRequest, value chan DeltaResponse, state stream.StreamState) (*RawDeltaResponse, error) { - resp := createDeltaResponse(ctx, request, state, resourceContainer{ +func (cache *snapshotCache) respondDelta(ctx context.Context, snapshot ResourceSnapshot, request *DeltaRequest, value chan DeltaResponse, clientState ClientState) (*RawDeltaResponse, error) { + resp := createDeltaResponse(ctx, request, clientState, resourceContainer{ resourceMap: snapshot.GetResources(request.TypeUrl), versionMap: snapshot.GetVersionMap(request.TypeUrl), systemVersion: snapshot.GetVersion(request.TypeUrl), }) // Only send a response if there were changes - // We want to respond immediately for the first wildcard request in a stream, even if the response is empty + // We want to respond immediately for the first request in a stream if it is wildcard, even if the response is empty // otherwise, envoy won't complete initialization - if len(resp.Resources) > 0 || len(resp.RemovedResources) > 0 || (state.IsWildcard() && state.IsFirst()) { + if len(resp.Resources) > 0 || len(resp.RemovedResources) > 0 || (clientState.IsWildcard() && request.ResponseNonce == "") { if cache.log != nil { cache.log.Debugf("node: %s, sending delta response with resources: %v removed resources %v wildcard: %t", - request.GetNode().GetId(), resp.Resources, resp.RemovedResources, state.IsWildcard()) + request.GetNode().GetId(), resp.Resources, resp.RemovedResources, clientState.IsWildcard()) } select { case value <- resp: diff --git a/pkg/cache/v3/simple_test.go b/pkg/cache/v3/simple_test.go index d5e1f92711..1f5a626239 100644 --- a/pkg/cache/v3/simple_test.go +++ b/pkg/cache/v3/simple_test.go @@ -125,7 +125,7 @@ func TestSnapshotCacheWithTTL(t *testing.T) { t.Run(typ, func(t *testing.T) { defer wg.Done() value := make(chan cache.Response, 1) - c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ]}, streamState, value) + c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ]}, &streamState, value) select { case out := <-value: if gotVersion, _ := out.GetVersion(); gotVersion != fixture.version { @@ -135,7 +135,9 @@ func TestSnapshotCacheWithTTL(t *testing.T) { t.Errorf("get resources %v, want %v", out.(*cache.RawResponse).Resources, snapshotWithTTL.GetResourcesAndTTL(typ)) } // Update streamState - streamState.SetKnownResourceNamesAsList(typ, out.GetRequest().GetResourceNames()) + for _, resource := range out.GetRequest().GetResourceNames() { + streamState.GetKnownResources()[resource] = fixture.version + } case <-time.After(2 * time.Second): t.Errorf("failed to receive snapshot response") } @@ -155,7 +157,7 @@ func TestSnapshotCacheWithTTL(t *testing.T) { for { value := make(chan cache.Response, 1) cancel := c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ], VersionInfo: fixture.version}, - streamState, value) + &streamState, value) select { case out := <-value: @@ -172,7 +174,9 @@ func TestSnapshotCacheWithTTL(t *testing.T) { updatesByType[typ]++ - streamState.SetKnownResourceNamesAsList(typ, out.GetRequest().ResourceNames) + for _, resource := range out.GetRequest().GetResourceNames() { + streamState.GetKnownResources()[resource] = fixture.version + } case <-end: cancel() return @@ -216,7 +220,7 @@ func TestSnapshotCache(t *testing.T) { value := make(chan cache.Response, 1) streamState := stream.NewStreamState(false, map[string]string{}) c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: rsrc.EndpointType, ResourceNames: []string{"none"}}, - streamState, value) + &streamState, value) select { case out := <-value: t.Errorf("watch for endpoints and mismatched names => got %v, want none", out) @@ -228,7 +232,7 @@ func TestSnapshotCache(t *testing.T) { value := make(chan cache.Response, 1) streamState := stream.NewStreamState(false, map[string]string{}) c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ]}, - streamState, value) + &streamState, value) select { case out := <-value: snapshot := fixture.snapshot() @@ -282,7 +286,7 @@ func TestSnapshotCacheWatch(t *testing.T) { streamState := stream.NewStreamState(false, map[string]string{}) for _, typ := range testTypes { watches[typ] = make(chan cache.Response, 1) - c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ]}, streamState, watches[typ]) + c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ]}, &streamState, watches[typ]) } if err := c.SetSnapshot(context.Background(), key, fixture.snapshot()); err != nil { t.Fatal(err) @@ -298,7 +302,9 @@ func TestSnapshotCacheWatch(t *testing.T) { if !reflect.DeepEqual(cache.IndexResourcesByName(out.(*cache.RawResponse).Resources), snapshot.GetResourcesAndTTL(typ)) { t.Errorf("get resources %v, want %v", out.(*cache.RawResponse).Resources, snapshot.GetResourcesAndTTL(typ)) } - streamState.SetKnownResourceNamesAsList(typ, out.GetRequest().GetResourceNames()) + for _, resource := range out.GetRequest().GetResourceNames() { + streamState.GetKnownResources()[resource] = fixture.version + } case <-time.After(time.Second): t.Fatal("failed to receive snapshot response") } @@ -309,7 +315,7 @@ func TestSnapshotCacheWatch(t *testing.T) { for _, typ := range testTypes { watches[typ] = make(chan cache.Response, 1) c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ], VersionInfo: fixture.version}, - streamState, watches[typ]) + &streamState, watches[typ]) } if count := c.GetStatusInfo(key).GetNumWatches(); count != len(testTypes) { t.Errorf("watches should be created for the latest version: %d", count) @@ -357,7 +363,7 @@ func TestConcurrentSetWatch(t *testing.T) { cancel := c.CreateWatch(&discovery.DiscoveryRequest{ Node: &core.Node{Id: id}, TypeUrl: rsrc.EndpointType, - }, streamState, value) + }, &streamState, value) defer cancel() } @@ -370,7 +376,7 @@ func TestSnapshotCacheWatchCancel(t *testing.T) { streamState := stream.NewStreamState(false, map[string]string{}) for _, typ := range testTypes { value := make(chan cache.Response, 1) - cancel := c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ]}, streamState, value) + cancel := c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ]}, &streamState, value) cancel() } // should be status info for the node @@ -396,7 +402,7 @@ func TestSnapshotCacheWatchTimeout(t *testing.T) { watchCh := make(chan cache.Response) streamState := stream.NewStreamState(false, map[string]string{}) c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: rsrc.EndpointType, ResourceNames: names[rsrc.EndpointType]}, - streamState, watchCh) + &streamState, watchCh) // The first time we set the snapshot without consuming from the blocking channel, so this should time out. ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond) @@ -450,8 +456,9 @@ func TestSnapshotCreateWatchWithResourcePreviouslyNotRequested(t *testing.T) { // Request resource with name=ClusterName go func() { + state := stream.NewStreamState(false, map[string]string{}) c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: rsrc.EndpointType, ResourceNames: []string{clusterName}}, - stream.NewStreamState(false, map[string]string{}), watch) + &state, watch) }() select { @@ -470,9 +477,9 @@ func TestSnapshotCreateWatchWithResourcePreviouslyNotRequested(t *testing.T) { // Request additional resource with name=clusterName2 for same version go func() { state := stream.NewStreamState(false, map[string]string{}) - state.SetKnownResourceNames(rsrc.EndpointType, map[string]struct{}{clusterName: {}}) + state.SetResourceVersions(map[string]string{clusterName: fixture.version}) c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: rsrc.EndpointType, VersionInfo: fixture.version, - ResourceNames: []string{clusterName, clusterName2}}, state, watch) + ResourceNames: []string{clusterName, clusterName2}}, &state, watch) }() select { @@ -489,9 +496,9 @@ func TestSnapshotCreateWatchWithResourcePreviouslyNotRequested(t *testing.T) { // Repeat request for with same version and make sure a watch is created state := stream.NewStreamState(false, map[string]string{}) - state.SetKnownResourceNames(rsrc.EndpointType, map[string]struct{}{clusterName: {}, clusterName2: {}}) + state.SetResourceVersions(map[string]string{clusterName: fixture.version, clusterName2: fixture.version}) if cancel := c.CreateWatch(&discovery.DiscoveryRequest{TypeUrl: rsrc.EndpointType, VersionInfo: fixture.version, - ResourceNames: []string{clusterName, clusterName2}}, state, watch); cancel == nil { + ResourceNames: []string{clusterName, clusterName2}}, &state, watch); cancel == nil { t.Fatal("Should create a watch") } else { cancel() @@ -617,7 +624,7 @@ func TestAvertPanicForWatchOnNonExistentSnapshot(t *testing.T) { } ss := stream.NewStreamState(false, map[string]string{"cluster": "abcdef"}) responder := make(chan cache.Response) - c.CreateWatch(req, ss, responder) + c.CreateWatch(req, &ss, responder) go func() { // Wait for at least one heartbeat to occur, then set snapshot. diff --git a/pkg/cache/v3/status.go b/pkg/cache/v3/status.go index 84db1f9821..43c6d109db 100644 --- a/pkg/cache/v3/status.go +++ b/pkg/cache/v3/status.go @@ -19,7 +19,6 @@ import ( "time" core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" - "github.com/envoyproxy/go-control-plane/pkg/server/stream/v3" ) // NodeHash computes string identifiers for Envoy nodes. @@ -99,7 +98,23 @@ type DeltaResponseWatch struct { Response chan DeltaResponse // VersionMap for the stream - StreamState stream.StreamState + clientState ClientState +} + +// WatchesResources returns whether at least one of the resource provided is currently watch by the stream +// It is currently only applicable to delta-xds +// If the request is wildcard, it will always return true +// Otherwise it will compare the provided resources to the list of resources currently subscribed +func (w *DeltaResponseWatch) WatchesResources(resourceNames map[string]struct{}) bool { + if w.clientState.IsWildcard() { + return true + } + for resourceName := range resourceNames { + if _, ok := w.clientState.GetSubscribedResources()[resourceName]; ok { + return true + } + } + return false } // newStatusInfo initializes a status info data structure. diff --git a/pkg/server/delta/v3/server.go b/pkg/server/delta/v3/server.go index 5f10266aba..e556e54a74 100644 --- a/pkg/server/delta/v3/server.go +++ b/pkg/server/delta/v3/server.go @@ -174,7 +174,7 @@ func (s *server) processDelta(str stream.DeltaStream, reqCh <-chan *discovery.De s.unsubscribe(req.GetResourceNamesUnsubscribe(), &watch.state) watch.responses = make(chan cache.DeltaResponse, 1) - watch.cancel = s.cache.CreateDeltaWatch(req, watch.state, watch.responses) + watch.cancel = s.cache.CreateDeltaWatch(req, &watch.state, watch.responses) watches.deltaWatches[typeURL] = watch go func() { @@ -216,7 +216,7 @@ func (s *server) DeltaStreamHandler(str stream.DeltaStream, typeURL string) erro // When we subscribe, we just want to make the cache know we are subscribing to a resource. // Even if the stream is wildcard, we keep the list of explicitly subscribed resources as the wildcard subscription can be discarded later on. func (s *server) subscribe(resources []string, streamState *stream.StreamState) { - sv := streamState.GetSubscribedResourceNames() + sv := streamState.GetSubscribedResources() for _, resource := range resources { if resource == "*" { streamState.SetWildcard(true) @@ -229,7 +229,7 @@ func (s *server) subscribe(resources []string, streamState *stream.StreamState) // Unsubscriptions remove resources from the stream's subscribed resource list. // If a client explicitly unsubscribes from a wildcard request, the stream is updated and now watches only subscribed resources. func (s *server) unsubscribe(resources []string, streamState *stream.StreamState) { - sv := streamState.GetSubscribedResourceNames() + sv := streamState.GetSubscribedResources() for _, resource := range resources { if resource == "*" { streamState.SetWildcard(false) @@ -244,7 +244,7 @@ func (s *server) unsubscribe(resources []string, streamState *stream.StreamState // To achieve that, we mark the resource as having been returned with an empty version. While creating the response, the cache will either: // * detect the version change, and return the resource (as an update) // * detect the resource deletion, and set it as removed in the response - streamState.GetResourceVersions()[resource] = "" + streamState.GetKnownResources()[resource] = "" } delete(sv, resource) } diff --git a/pkg/server/sotw/v3/server.go b/pkg/server/sotw/v3/server.go index 91681237c9..fcde7d89b2 100644 --- a/pkg/server/sotw/v3/server.go +++ b/pkg/server/sotw/v3/server.go @@ -69,7 +69,7 @@ type server struct { // regardless current snapshot version (even if it is not changed yet) type lastDiscoveryResponse struct { nonce string - resources map[string]struct{} + resources map[string]string } // process handles a bi-di stream request @@ -81,7 +81,7 @@ func (s *server) process(str stream.Stream, reqCh <-chan *discovery.DiscoveryReq // ignores stale nonces. nonce is only modified within send() function. var streamNonce int64 - streamState := stream.NewStreamState(false, map[string]string{}) + streamStates := map[string]stream.StreamState{} lastDiscoveryResponses := map[string]lastDiscoveryResponse{} // a collection of stack allocated watches per request type @@ -112,12 +112,17 @@ func (s *server) process(str stream.Stream, reqCh <-chan *discovery.DiscoveryReq streamNonce = streamNonce + 1 out.Nonce = strconv.FormatInt(streamNonce, 10) + version, err := resp.GetVersion() + if err != nil { + return "", err + } + lastResponse := lastDiscoveryResponse{ nonce: out.Nonce, - resources: make(map[string]struct{}), + resources: make(map[string]string), } for _, r := range resp.GetRequest().ResourceNames { - lastResponse.resources[r] = struct{}{} + lastResponse.resources[r] = version } lastDiscoveryResponses[resp.GetRequest().TypeUrl] = lastResponse @@ -183,14 +188,15 @@ func (s *server) process(str stream.Stream, reqCh <-chan *discovery.DiscoveryReq } } + typeURL := req.GetTypeUrl() + state := streamStates[typeURL] if lastResponse, ok := lastDiscoveryResponses[req.TypeUrl]; ok { if lastResponse.nonce == "" || lastResponse.nonce == nonce { // Let's record Resource names that a client has received. - streamState.SetKnownResourceNames(req.TypeUrl, lastResponse.resources) + state.SetResourceVersions(lastResponse.resources) } } - typeURL := req.GetTypeUrl() responder := make(chan cache.Response, 1) if w, ok := watches.responders[typeURL]; ok { // We've found a pre-existing watch, lets check and update if needed. @@ -199,7 +205,7 @@ func (s *server) process(str stream.Stream, reqCh <-chan *discovery.DiscoveryReq w.close() watches.addWatch(typeURL, &watch{ - cancel: s.cache.CreateWatch(req, streamState, responder), + cancel: s.cache.CreateWatch(req, &state, responder), response: responder, }) } @@ -207,11 +213,13 @@ func (s *server) process(str stream.Stream, reqCh <-chan *discovery.DiscoveryReq // No pre-existing watch exists, let's create one. // We need to precompute the watches first then open a watch in the cache. watches.addWatch(typeURL, &watch{ - cancel: s.cache.CreateWatch(req, streamState, responder), + cancel: s.cache.CreateWatch(req, &state, responder), response: responder, }) } + streamStates[typeURL] = state + // Recompute the dynamic select cases for this stream. watches.recompute(s.ctx, reqCh) default: diff --git a/pkg/server/stream/v3/stream.go b/pkg/server/stream/v3/stream.go index b5832b7d58..2147dc46e0 100644 --- a/pkg/server/stream/v3/stream.go +++ b/pkg/server/stream/v3/stream.go @@ -33,57 +33,30 @@ type StreamState struct { // nolint:golint,revive // ResourceVersions contains a hash of the resource as the value and the resource name as the key. // This field stores the last state sent to the client. resourceVersions map[string]string - - // knownResourceNames contains resource names that a client has received previously - knownResourceNames map[string]map[string]struct{} - - // indicates whether the object has been modified since its creation - first bool } // GetSubscribedResourceNames returns the list of resources currently explicitly subscribed to // If the request is set to wildcard it may be empty // Currently populated only when using delta-xds -func (s *StreamState) GetSubscribedResourceNames() map[string]struct{} { +func (s *StreamState) GetSubscribedResources() map[string]struct{} { return s.subscribedResourceNames } // SetSubscribedResourceNames is setting the list of resources currently explicitly subscribed to // It is decorrelated from the wildcard state of the stream // Currently used only when using delta-xds -func (s *StreamState) SetSubscribedResourceNames(subscribedResourceNames map[string]struct{}) { +func (s *StreamState) SetSubscribedResources(subscribedResourceNames map[string]struct{}) { s.subscribedResourceNames = subscribedResourceNames } -// WatchesResources returns whether at least one of the resource provided is currently watch by the stream -// It is currently only applicable to delta-xds -// If the request is wildcard, it will always return true -// Otherwise it will compare the provided resources to the list of resources currently subscribed -func (s *StreamState) WatchesResources(resourceNames map[string]struct{}) bool { - if s.IsWildcard() { - return true - } - for resourceName := range resourceNames { - if _, ok := s.subscribedResourceNames[resourceName]; ok { - return true - } - } - return false -} - -func (s *StreamState) GetResourceVersions() map[string]string { +func (s *StreamState) GetKnownResources() map[string]string { return s.resourceVersions } func (s *StreamState) SetResourceVersions(resourceVersions map[string]string) { - s.first = false s.resourceVersions = resourceVersions } -func (s *StreamState) IsFirst() bool { - return s.first -} - func (s *StreamState) SetWildcard(wildcard bool) { s.wildcard = wildcard } @@ -92,30 +65,12 @@ func (s *StreamState) IsWildcard() bool { return s.wildcard } -func (s *StreamState) SetKnownResourceNames(url string, names map[string]struct{}) { - s.knownResourceNames[url] = names -} - -func (s *StreamState) SetKnownResourceNamesAsList(url string, names []string) { - m := map[string]struct{}{} - for _, name := range names { - m[name] = struct{}{} - } - s.knownResourceNames[url] = m -} - -func (s *StreamState) GetKnownResourceNames(url string) map[string]struct{} { - return s.knownResourceNames[url] -} - // NewStreamState initializes a stream state. func NewStreamState(wildcard bool, initialResourceVersions map[string]string) StreamState { state := StreamState{ wildcard: wildcard, subscribedResourceNames: map[string]struct{}{}, resourceVersions: initialResourceVersions, - first: true, - knownResourceNames: map[string]map[string]struct{}{}, } if initialResourceVersions == nil { diff --git a/pkg/server/v3/delta_test.go b/pkg/server/v3/delta_test.go index 204599613c..66ddc5bd1a 100644 --- a/pkg/server/v3/delta_test.go +++ b/pkg/server/v3/delta_test.go @@ -15,12 +15,11 @@ import ( "github.com/envoyproxy/go-control-plane/pkg/cache/types" "github.com/envoyproxy/go-control-plane/pkg/cache/v3" 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/server/v3" "github.com/envoyproxy/go-control-plane/pkg/test/resource/v3" ) -func (config *mockConfigWatcher) CreateDeltaWatch(req *discovery.DeltaDiscoveryRequest, state stream.StreamState, out chan cache.DeltaResponse) func() { +func (config *mockConfigWatcher) CreateDeltaWatch(req *discovery.DeltaDiscoveryRequest, state cache.ClientState, out chan cache.DeltaResponse) func() { config.deltaCounts[req.TypeUrl] = config.deltaCounts[req.TypeUrl] + 1 // This is duplicated from pkg/cache/v3/delta.go as private there @@ -37,7 +36,7 @@ func (config *mockConfigWatcher) CreateDeltaWatch(req *discovery.DeltaDiscoveryR // If we are handling a wildcard request, we want to respond with all resources switch { case state.IsWildcard(): - if len(state.GetResourceVersions()) == 0 { + if len(state.GetKnownResources()) == 0 { filtered = make([]types.Resource, 0, len(resourceMap)) } nextVersionMap = make(map[string]string, len(resourceMap)) @@ -46,24 +45,24 @@ func (config *mockConfigWatcher) CreateDeltaWatch(req *discovery.DeltaDiscoveryR // we can just set it here to be used for comparison later version := versionMap[name] nextVersionMap[name] = version - prevVersion, found := state.GetResourceVersions()[name] + prevVersion, found := state.GetKnownResources()[name] if !found || (prevVersion != version) { filtered = append(filtered, r) } } // Compute resources for removal - for name := range state.GetResourceVersions() { + for name := range state.GetKnownResources() { if _, ok := resourceMap[name]; !ok { toRemove = append(toRemove, name) } } default: - nextVersionMap = make(map[string]string, len(state.GetSubscribedResourceNames())) + nextVersionMap = make(map[string]string, len(state.GetSubscribedResources())) // state.GetResourceVersions() may include resources no longer subscribed // In the current code this gets silently cleaned when updating the version map - for name := range state.GetSubscribedResourceNames() { - prevVersion, found := state.GetResourceVersions()[name] + for name := range state.GetSubscribedResources() { + prevVersion, found := state.GetKnownResources()[name] if r, ok := resourceMap[name]; ok { nextVersion := versionMap[name] if prevVersion != nextVersion { @@ -155,36 +154,36 @@ func makeMockDeltaStream(t *testing.T) *mockDeltaStream { func makeDeltaResources() map[string]map[string]types.Resource { return map[string]map[string]types.Resource{ - rsrc.EndpointType: map[string]types.Resource{ + rsrc.EndpointType: { endpoint.GetClusterName(): endpoint, }, - rsrc.ClusterType: map[string]types.Resource{ + rsrc.ClusterType: { cluster.Name: cluster, }, - rsrc.RouteType: map[string]types.Resource{ + rsrc.RouteType: { route.Name: route, }, - rsrc.ScopedRouteType: map[string]types.Resource{ + rsrc.ScopedRouteType: { scopedRoute.Name: scopedRoute, }, - rsrc.VirtualHostType: map[string]types.Resource{ + rsrc.VirtualHostType: { virtualHost.Name: virtualHost, }, - rsrc.ListenerType: map[string]types.Resource{ + rsrc.ListenerType: { httpListener.Name: httpListener, httpScopedListener.Name: httpScopedListener, }, - rsrc.SecretType: map[string]types.Resource{ + rsrc.SecretType: { secret.Name: secret, }, - rsrc.RuntimeType: map[string]types.Resource{ + rsrc.RuntimeType: { runtime.Name: runtime, }, - rsrc.ExtensionConfigType: map[string]types.Resource{ + rsrc.ExtensionConfigType: { extensionConfig.Name: extensionConfig, }, // Pass-through type (types without explicit handling) - opaqueType: map[string]types.Resource{ + opaqueType: { "opaque": opaque, }, } @@ -460,7 +459,7 @@ func TestDeltaCallbackError(t *testing.T) { func TestDeltaWildcardSubscriptions(t *testing.T) { config := makeMockConfigWatcher() config.deltaResources = map[string]map[string]types.Resource{ - rsrc.EndpointType: map[string]types.Resource{ + rsrc.EndpointType: { "endpoints0": resource.MakeEndpoint("endpoints0", 1234), "endpoints1": resource.MakeEndpoint("endpoints1", 1234), "endpoints2": resource.MakeEndpoint("endpoints2", 1234), diff --git a/pkg/server/v3/server.go b/pkg/server/v3/server.go index ed0b0cb6be..56c683a2de 100644 --- a/pkg/server/v3/server.go +++ b/pkg/server/v3/server.go @@ -29,7 +29,6 @@ import ( core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" clusterservice "github.com/envoyproxy/go-control-plane/envoy/service/cluster/v3" discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3" - discoverygrpc "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3" endpointservice "github.com/envoyproxy/go-control-plane/envoy/service/endpoint/v3" extensionconfigservice "github.com/envoyproxy/go-control-plane/envoy/service/extension/v3" listenerservice "github.com/envoyproxy/go-control-plane/envoy/service/listener/v3" @@ -48,7 +47,7 @@ type Server interface { routeservice.ScopedRoutesDiscoveryServiceServer routeservice.VirtualHostDiscoveryServiceServer listenerservice.ListenerDiscoveryServiceServer - discoverygrpc.AggregatedDiscoveryServiceServer + discovery.AggregatedDiscoveryServiceServer secretservice.SecretDiscoveryServiceServer runtimeservice.RuntimeDiscoveryServiceServer extensionconfigservice.ExtensionConfigDiscoveryServiceServer @@ -184,7 +183,7 @@ func (s *server) StreamHandler(stream stream.Stream, typeURL string) error { return s.sotw.StreamHandler(stream, typeURL) } -func (s *server) StreamAggregatedResources(stream discoverygrpc.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error { +func (s *server) StreamAggregatedResources(stream discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error { return s.StreamHandler(stream, resource.AnyType) } @@ -297,7 +296,7 @@ func (s *server) DeltaStreamHandler(stream stream.DeltaStream, typeURL string) e return s.delta.DeltaStreamHandler(stream, typeURL) } -func (s *server) DeltaAggregatedResources(stream discoverygrpc.AggregatedDiscoveryService_DeltaAggregatedResourcesServer) error { +func (s *server) DeltaAggregatedResources(stream discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesServer) error { return s.DeltaStreamHandler(stream, resource.AnyType) } diff --git a/pkg/server/v3/server_test.go b/pkg/server/v3/server_test.go index 02078bc3ad..8668e46296 100644 --- a/pkg/server/v3/server_test.go +++ b/pkg/server/v3/server_test.go @@ -32,7 +32,6 @@ import ( "github.com/envoyproxy/go-control-plane/pkg/cache/types" "github.com/envoyproxy/go-control-plane/pkg/cache/v3" 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/server/v3" "github.com/envoyproxy/go-control-plane/pkg/test/resource/v3" ) @@ -48,7 +47,7 @@ type mockConfigWatcher struct { mu *sync.RWMutex } -func (config *mockConfigWatcher) CreateWatch(req *discovery.DiscoveryRequest, state stream.StreamState, out chan cache.Response) func() { +func (config *mockConfigWatcher) CreateWatch(req *discovery.DiscoveryRequest, state cache.ClientState, out chan cache.Response) func() { config.counts[req.TypeUrl] = config.counts[req.TypeUrl] + 1 if len(config.responses[req.TypeUrl]) > 0 { out <- config.responses[req.TypeUrl][0]