deactivate legacy wildcard support

Author: zhiyanfoo

pkg/cache/v3/delta_test.go

@@ -19,6 +19,7 @@ import (
 	"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/config"
 	"github.com/envoyproxy/go-control-plane/pkg/server/stream/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/test/resource/v3"
 )
@@ -39,7 +40,7 @@ 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)
-		subscriptions[typ] = stream.NewDeltaSubscription(nil, nil, nil)
+		subscriptions[typ] = stream.NewDeltaSubscription(nil, nil, nil, config.NewOpts(), typ)
 		_, err := c.CreateDeltaWatch(&discovery.DeltaDiscoveryRequest{
 			Node: &core.Node{
 				Id: "node",
@@ -120,7 +121,7 @@ func TestDeltaRemoveResources(t *testing.T) {
 
 	for _, typ := range testTypes {
 		watches[typ] = make(chan cache.DeltaResponse, 1)
-		sub := stream.NewDeltaSubscription(nil, nil, nil)
+		sub := stream.NewDeltaSubscription(nil, nil, nil, config.NewOpts(), typ)
 		subscriptions[typ] = &sub
 		// We don't specify any resource name subscriptions here because we want to make sure we test wildcard
 		// functionality. This means we should receive all resources back without requesting a subscription by name.
@@ -210,7 +211,7 @@ func TestConcurrentSetDeltaWatch(t *testing.T) {
 						},
 						TypeUrl:                rsrc.EndpointType,
 						ResourceNamesSubscribe: []string{clusterName},
-					}, stream.NewDeltaSubscription([]string{clusterName}, nil, nil), responses)
+					}, stream.NewDeltaSubscription([]string{clusterName}, nil, nil, config.NewOpts(), rsrc.EndpointType), responses)
 
 					require.NoError(t, err)
 					defer cancel()
@@ -227,7 +228,7 @@ func TestSnapshotDeltaCacheWatchTimeout(t *testing.T) {
 
 	// Create a non-buffered channel that will block sends.
 	watchCh := make(chan cache.DeltaResponse)
-	sub := stream.NewDeltaSubscription(names[rsrc.EndpointType], nil, nil)
+	sub := stream.NewDeltaSubscription(names[rsrc.EndpointType], nil, nil, config.NewOpts(), rsrc.EndpointType)
 	_, err := c.CreateDeltaWatch(&discovery.DeltaDiscoveryRequest{
 		Node: &core.Node{
 			Id: key,
@@ -277,7 +278,7 @@ func TestSnapshotCacheDeltaWatchCancel(t *testing.T) {
 			},
 			TypeUrl:                typ,
 			ResourceNamesSubscribe: names[typ],
-		}, stream.NewDeltaSubscription(names[typ], nil, nil), responses)
+		}, stream.NewDeltaSubscription(names[typ], nil, nil, config.NewOpts(), typ), responses)
 		require.NoError(t, err)
 
 		// Cancel the watch

pkg/cache/v3/linear_test.go

@@ -32,6 +32,7 @@ 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/resource/v3"
+	"github.com/envoyproxy/go-control-plane/pkg/server/config"
 	"github.com/envoyproxy/go-control-plane/pkg/server/stream/v3"
 )
 
@@ -223,7 +224,7 @@ func hashResource(t *testing.T, resource types.Resource) string {
 
 func createWildcardDeltaWatch(t *testing.T, initialReq bool, c *LinearCache, w chan DeltaResponse) {
 	t.Helper()
-	sub := stream.NewDeltaSubscription(nil, nil, nil)
+	sub := stream.NewDeltaSubscription(nil, nil, nil, config.NewOpts(), testType)
 	req := &DeltaRequest{TypeUrl: testType}
 	if !initialReq {
 		req.ResponseNonce = "1"
@@ -237,7 +238,7 @@ func createWildcardDeltaWatch(t *testing.T, initialReq bool, c *LinearCache, w c
 }
 
 func subFromRequest(req *Request) stream.Subscription {
-	return stream.NewSotwSubscription(req.GetResourceNames())
+	return stream.NewSotwSubscription(req.GetResourceNames(), config.NewOpts(), req.GetTypeUrl())
 }
 
 // This method represents the expected behavior of client and servers regarding the request and the subscription.
@@ -250,7 +251,7 @@ func updateFromSotwResponse(resp Response, sub *stream.Subscription, req *Reques
 }
 
 func subFromDeltaRequest(req *DeltaRequest) stream.Subscription {
-	return stream.NewDeltaSubscription(req.GetResourceNamesSubscribe(), req.GetResourceNamesUnsubscribe(), req.GetInitialResourceVersions())
+	return stream.NewDeltaSubscription(req.GetResourceNamesSubscribe(), req.GetResourceNamesUnsubscribe(), req.GetInitialResourceVersions(), config.NewOpts(), req.GetTypeUrl())
 }
 
 func TestLinearInitialResources(t *testing.T) {

pkg/cache/v3/simple_test.go

@@ -36,6 +36,7 @@ import (
 	"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/config"
 	"github.com/envoyproxy/go-control-plane/pkg/server/stream/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/test/resource/v3"
 )
@@ -54,7 +55,7 @@ func (group) ID(node *core.Node) string {
 }
 
 func subFromRequest(req *cache.Request) stream.Subscription {
-	return stream.NewSotwSubscription(req.GetResourceNames())
+	return stream.NewSotwSubscription(req.GetResourceNames(), config.NewOpts(), req.GetTypeUrl())
 }
 
 // This method represents the expected behavior of client and servers regarding the request and the subscription.
@@ -601,7 +602,7 @@ func TestAvertPanicForWatchOnNonExistentSnapshot(t *testing.T) {
 		ResourceNames: []string{"rtds"},
 		TypeUrl:       rsrc.RuntimeType,
 	}
-	ss := stream.NewSotwSubscription([]string{"rtds"})
+	ss := stream.NewSotwSubscription([]string{"rtds"}, config.NewOpts(), rsrc.RuntimeType)
 	ss.SetReturnedResources(map[string]string{"cluster": "abcdef"})
 	responder := make(chan cache.Response)
 	_, err := c.CreateWatch(req, ss, responder)

pkg/server/config/config.go

@@ -9,6 +9,12 @@ type Opts struct {
 	Ordered bool
 
 	Logger log.Logger
+
+	// If true, deactivate legacy wildcard mode for all resource types
+	legacyWildcardDeactivated bool
+
+	// Deactivate legacy wildcard mode for specific resource types
+	legacyWildcardDeactivatedTypes map[string]struct{}
 }
 
 func NewOpts() Opts {
@@ -18,6 +24,16 @@ func NewOpts() Opts {
 	}
 }
 
+// LegacyWildcardDeactivated returns whether legacy wildcard mode is deactivated for all resource types
+func (o Opts) LegacyWildcardDeactivated() bool {
+	return o.legacyWildcardDeactivated
+}
+
+// LegacyWildcardDeactivatedTypes returns the set of resource types for which legacy wildcard mode is deactivated
+func (o Opts) LegacyWildcardDeactivatedTypes() map[string]struct{} {
+	return o.legacyWildcardDeactivatedTypes
+}
+
 // Each xDS implementation should implement their own functional opts.
 // It is recommended that config values be added in this package specifically,
 // but the individual opts functions should be in their respective
@@ -28,3 +44,29 @@ func NewOpts() Opts {
 //
 // this allows for easy inference as to which opt applies to what implementation.
 type XDSOption func(*Opts)
+
+// DeactivateLegacyWildcard deactivates legacy wildcard mode for all resource types.
+// In legacy wildcard mode, empty requests to a stream, are treated as wildcard requests as long
+// as there is no request made with resources or explicit wildcard requests on the same stream.
+// When deactivated, empty requests are treated as a request with no subscriptions to any resource.
+// This is recommended for when you are using the go-control-plane to serve grpc-xds clients.
+// These clients never want to treat an empty request as a wildcard subscription.
+func DeactivateLegacyWildcard() XDSOption {
+	return func(o *Opts) {
+		o.legacyWildcardDeactivated = true
+	}
+}
+
+// DeactivateLegacyWildcardForTypes deactivates legacy wildcard mode for specific resource types.
+// In legacy wildcard mode, empty requests to a stream, are treated as wildcard requests as long
+// as there is no request made with resources or explicit wildcard requests on the same stream.
+// When deactivated, empty requests are treated as a request with no subscriptions to any resource.
+func DeactivateLegacyWildcardForTypes(types []string) XDSOption {
+	return func(o *Opts) {
+		typeMap := make(map[string]struct{}, len(types))
+		for _, t := range types {
+			typeMap[t] = struct{}{}
+		}
+		o.legacyWildcardDeactivatedTypes = typeMap
+	}
+}

pkg/server/delta/v3/server.go

@@ -218,7 +218,7 @@ func (s *server) processDelta(str stream.DeltaStream, reqCh <-chan *discovery.De
 				// We also set the subscription as wildcard based on its legacy meaning (no resource name sent in resource_names_subscribe).
 				// If the subscription starts with this legacy mode, adding new resources will not unsubscribe from wildcard.
 				// It can still be done by explicitly unsubscribing from "*"
-				watch.subscription = stream.NewDeltaSubscription(req.GetResourceNamesSubscribe(), req.GetResourceNamesUnsubscribe(), req.GetInitialResourceVersions())
+				watch.subscription = stream.NewDeltaSubscription(req.GetResourceNamesSubscribe(), req.GetResourceNamesUnsubscribe(), req.GetInitialResourceVersions(), s.opts, typeURL)
 			} else {
 				watch.Cancel()
 

pkg/server/sotw/v3/ads.go

@@ -107,7 +107,7 @@ func (s *server) processADS(sw *streamWrapper, reqCh chan *discovery.DiscoveryRe
 				subscription.SetResourceSubscription(req.GetResourceNames())
 			} else {
 				s.opts.Logger.Debugf("[sotw ads] New subscription for type %s and stream %d", typeURL, sw.ID)
-				subscription = stream.NewSotwSubscription(req.GetResourceNames())
+				subscription = stream.NewSotwSubscription(req.GetResourceNames(), s.opts, typeURL)
 			}
 
 			cancel, err := s.cache.CreateWatch(req, subscription, respChan)

pkg/server/sotw/v3/xds.go

@@ -128,7 +128,7 @@ func (s *server) process(str stream.Stream, reqCh chan *discovery.DiscoveryReque
 				subscription.SetResourceSubscription(req.GetResourceNames())
 			} else {
 				s.opts.Logger.Debugf("[sotw] New subscription for type %s and stream %d", typeURL, sw.ID)
-				subscription = stream.NewSotwSubscription(req.GetResourceNames())
+				subscription = stream.NewSotwSubscription(req.GetResourceNames(), s.opts, typeURL)
 			}
 
 			responder := make(chan cache.Response, 1)

pkg/server/stream/v3/subscription.go

@@ -1,5 +1,7 @@
 package stream
 
+import "github.com/envoyproxy/go-control-plane/pkg/server/config"
+
 const (
 	explicitWildcard = "*"
 )
@@ -9,10 +11,10 @@ type Subscription struct {
 	// wildcard indicates if the subscription currently has a wildcard watch.
 	wildcard bool
 
-	// allowLegacyWildcard indicates that the stream never provided any resource
-	// and is de facto wildcard.
-	// As soon as a resource or an explicit subscription to wildcard is provided,
-	// this flag will be set to false
+	// allowLegacyWildcard indicates that an empty request should be treated as a wildcard request.
+	// If in the stream at some point subscribes explicitly a resource or a explicitly makes wildcard
+	// request, then subsequent empty requests should not be treated as wildcard and so this
+	// flag will be set to false.
 	allowLegacyWildcard bool
 
 	// subscribedResourceNames provides the resources explicitly requested by the client
@@ -24,10 +26,19 @@ type Subscription struct {
 }
 
 // newSubscription initializes a subscription state.
-func newSubscription(wildcard bool, initialResourceVersions map[string]string) Subscription {
+func newSubscription(emptyRequest bool, initialResourceVersions map[string]string, opts config.Opts, typeURL string) Subscription {
+	allowLegacyWildcard := emptyRequest
+
+	if opts.LegacyWildcardDeactivated() {
+		allowLegacyWildcard = false
+	} else if typeMap := opts.LegacyWildcardDeactivatedTypes(); len(typeMap) > 0 {
+		if _, found := typeMap[typeURL]; found {
+			allowLegacyWildcard = false
+		}
+	}
+
 	state := Subscription{
-		wildcard:                wildcard,
-		allowLegacyWildcard:     wildcard,
+		allowLegacyWildcard:     allowLegacyWildcard,
 		subscribedResourceNames: map[string]struct{}{},
 		returnedResources:       initialResourceVersions,
 	}
@@ -39,8 +50,8 @@ func newSubscription(wildcard bool, initialResourceVersions map[string]string) S
 	return state
 }
 
-func NewSotwSubscription(subscribed []string) Subscription {
-	sub := newSubscription(len(subscribed) == 0, nil)
+func NewSotwSubscription(subscribed []string, opts config.Opts, typeURL string) Subscription {
+	sub := newSubscription(len(subscribed) == 0, nil, opts, typeURL)
 	sub.SetResourceSubscription(subscribed)
 	return sub
 }
@@ -55,6 +66,7 @@ func (s *Subscription) SetResourceSubscription(subscribed []string) {
 		if len(subscribed) == 0 {
 			// We were wildcard based on legacy behavior and still don't request any resource
 			// The watch remains wildcard
+			s.wildcard = true
 			return
 		}
 
@@ -90,8 +102,8 @@ func (s *Subscription) SetResourceSubscription(subscribed []string) {
 	s.subscribedResourceNames = subscribedResources
 }
 
-func NewDeltaSubscription(subscribed, unsubscribed []string, initialResourceVersions map[string]string) Subscription {
-	sub := newSubscription(len(subscribed) == 0, initialResourceVersions)
+func NewDeltaSubscription(subscribed, unsubscribed []string, initialResourceVersions map[string]string, opts config.Opts, typeURL string) Subscription {
+	sub := newSubscription(len(subscribed) == 0, initialResourceVersions, opts, typeURL)
 	sub.UpdateResourceSubscriptions(subscribed, unsubscribed)
 	return sub
 }
@@ -107,6 +119,7 @@ func (s *Subscription) UpdateResourceSubscriptions(subscribed, unsubscribed []st
 		if len(subscribed) == 0 {
 			// We were wildcard based on legacy behavior and still don't request any resource
 			// The watch remains wildcard
+			s.wildcard = true
 			return
 		}