dont pass in opts to NewSubscription interface, just pass in if legacy mode is true

Author: zhiyanfoo

pkg/cache/v3/delta_test.go

@@ -19,7 +19,6 @@ 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"
 )
@@ -40,7 +39,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, config.NewOpts(), typ)
+		subscriptions[typ] = stream.NewDeltaSubscription(nil, nil, nil, true)
 		_, err := c.CreateDeltaWatch(&discovery.DeltaDiscoveryRequest{
 			Node: &core.Node{
 				Id: "node",
@@ -121,7 +120,7 @@ func TestDeltaRemoveResources(t *testing.T) {
 
 	for _, typ := range testTypes {
 		watches[typ] = make(chan cache.DeltaResponse, 1)
-		sub := stream.NewDeltaSubscription(nil, nil, nil, config.NewOpts(), typ)
+		sub := stream.NewDeltaSubscription(nil, nil, nil, true)
 		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.
@@ -211,7 +210,7 @@ func TestConcurrentSetDeltaWatch(t *testing.T) {
 						},
 						TypeUrl:                rsrc.EndpointType,
 						ResourceNamesSubscribe: []string{clusterName},
-					}, stream.NewDeltaSubscription([]string{clusterName}, nil, nil, config.NewOpts(), rsrc.EndpointType), responses)
+					}, stream.NewDeltaSubscription([]string{clusterName}, nil, nil, true), responses)
 
 					require.NoError(t, err)
 					defer cancel()
@@ -228,7 +227,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, config.NewOpts(), rsrc.EndpointType)
+	sub := stream.NewDeltaSubscription(names[rsrc.EndpointType], nil, nil, true)
 	_, err := c.CreateDeltaWatch(&discovery.DeltaDiscoveryRequest{
 		Node: &core.Node{
 			Id: key,
@@ -278,7 +277,7 @@ func TestSnapshotCacheDeltaWatchCancel(t *testing.T) {
 			},
 			TypeUrl:                typ,
 			ResourceNamesSubscribe: names[typ],
-		}, stream.NewDeltaSubscription(names[typ], nil, nil, config.NewOpts(), typ), responses)
+		}, stream.NewDeltaSubscription(names[typ], nil, nil, true), responses)
 		require.NoError(t, err)
 
 		// Cancel the watch

pkg/cache/v3/linear_test.go

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

pkg/cache/v3/simple_test.go

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

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(), s.opts, typeURL)
+				watch.subscription = stream.NewDeltaSubscription(req.GetResourceNamesSubscribe(), req.GetResourceNamesUnsubscribe(), req.GetInitialResourceVersions(), s.opts.IsLegacyWildcardActive(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(), s.opts, typeURL)
+				subscription = stream.NewSotwSubscription(req.GetResourceNames(), s.opts.IsLegacyWildcardActive(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(), s.opts, typeURL)
+				subscription = stream.NewSotwSubscription(req.GetResourceNames(), s.opts.IsLegacyWildcardActive(typeURL))
 			}
 
 			responder := make(chan cache.Response, 1)

pkg/server/stream/v3/subscription.go

@@ -1,7 +1,5 @@
 package stream
 
-import "github.com/envoyproxy/go-control-plane/pkg/server/config"
-
 const (
 	explicitWildcard = "*"
 )
@@ -48,8 +46,8 @@ func newSubscription(emptyRequest, allowLegacyWildcard bool, initialResourceVers
 	return state
 }
 
-func NewSotwSubscription(subscribed []string, opts config.Opts, typeURL string) Subscription {
-	sub := newSubscription(len(subscribed) == 0, opts.IsLegacyWildcardActive(typeURL), nil)
+func NewSotwSubscription(subscribed []string, allowLegacyWildcard bool) Subscription {
+	sub := newSubscription(len(subscribed) == 0, allowLegacyWildcard, nil)
 	sub.SetResourceSubscription(subscribed)
 	return sub
 }
@@ -99,8 +97,8 @@ func (s *Subscription) SetResourceSubscription(subscribed []string) {
 	s.subscribedResourceNames = subscribedResources
 }
 
-func NewDeltaSubscription(subscribed, unsubscribed []string, initialResourceVersions map[string]string, opts config.Opts, typeURL string) Subscription {
-	sub := newSubscription(len(subscribed) == 0, opts.IsLegacyWildcardActive(typeURL), initialResourceVersions)
+func NewDeltaSubscription(subscribed, unsubscribed []string, initialResourceVersions map[string]string, allowLegacyWildcard bool) Subscription {
+	sub := newSubscription(len(subscribed) == 0, allowLegacyWildcard, initialResourceVersions)
 	sub.UpdateResourceSubscriptions(subscribed, unsubscribed)
 	return sub
 }

pkg/server/stream/v3/subscription_test.go

@@ -10,7 +10,7 @@ import (
 
 func TestSotwSubscriptions(t *testing.T) {
 	t.Run("legacy mode properly handled", func(t *testing.T) {
-		sub := NewSotwSubscription([]string{}, config.NewOpts(), "")
+		sub := NewSotwSubscription([]string{}, true)
 		assert.True(t, sub.IsWildcard())
 
 		// Requests always set empty in legacy mode
@@ -37,7 +37,7 @@ func TestSotwSubscriptions(t *testing.T) {
 
 	t.Run("new wildcard mode from start", func(t *testing.T) {
 		// A resource is provided so the subscription was created in wildcard
-		sub := NewSotwSubscription([]string{"*"}, config.NewOpts(), "")
+		sub := NewSotwSubscription([]string{"*"}, true)
 		assert.True(t, sub.IsWildcard())
 		assert.Empty(t, sub.SubscribedResources())
 
@@ -75,7 +75,7 @@ func TestSotwSubscriptions(t *testing.T) {
 
 func TestDeltaSubscriptions(t *testing.T) {
 	t.Run("legacy mode properly handled", func(t *testing.T) {
-		sub := NewDeltaSubscription([]string{}, []string{}, map[string]string{"resource": "version"}, config.NewOpts(), "")
+		sub := NewDeltaSubscription([]string{}, []string{}, map[string]string{"resource": "version"}, true)
 		assert.True(t, sub.IsWildcard())
 		assert.Empty(t, sub.SubscribedResources())
 		assert.Equal(t, map[string]string{"resource": "version"}, sub.ReturnedResources())
@@ -104,7 +104,7 @@ func TestDeltaSubscriptions(t *testing.T) {
 
 	t.Run("new wildcard mode", func(t *testing.T) {
 		// A resource is provided so the subscription was created in wildcard
-		sub := NewDeltaSubscription([]string{"*"}, []string{}, map[string]string{"resource": "version"}, config.NewOpts(), "")
+		sub := NewDeltaSubscription([]string{"*"}, []string{}, map[string]string{"resource": "version"}, true)
 		assert.True(t, sub.IsWildcard())
 		assert.Empty(t, sub.SubscribedResources())
 
@@ -166,8 +166,9 @@ func TestSotwSubscriptionsWithDeactivatedLegacyWildcard(t *testing.T) {
 		deactivateOpt := config.DeactivateLegacyWildcard()
 		deactivateOpt(&opts)
 
+		typeURL := "type.googleapis.com/envoy.config.cluster.v3.Cluster"
 		// Create subscription with empty resource list (would normally be legacy wildcard)
-		sub := NewSotwSubscription([]string{}, opts, "type.googleapis.com/envoy.config.cluster.v3.Cluster")
+		sub := NewSotwSubscription([]string{}, opts.IsLegacyWildcardActive(typeURL))
 
 		// With deactivated legacy wildcard, subscription should NOT be wildcard initially
 		// because allowLegacyWildcard=false means empty list doesn't trigger legacy behavior
@@ -194,11 +195,11 @@ func TestSotwSubscriptionsWithDeactivatedLegacyWildcardForTypes(t *testing.T) {
 		deactivateOpt(&opts)
 
 		// Both cluster and endpoint should have legacy wildcard deactivated
-		subCluster := NewSotwSubscription([]string{}, opts, clusterType)
+		subCluster := NewSotwSubscription([]string{}, opts.IsLegacyWildcardActive(clusterType))
 		subCluster.SetResourceSubscription([]string{})
 		assert.False(t, subCluster.IsWildcard())
 
-		subEndpoint := NewSotwSubscription([]string{}, opts, endpointType)
+		subEndpoint := NewSotwSubscription([]string{}, opts.IsLegacyWildcardActive(endpointType))
 		subEndpoint.SetResourceSubscription([]string{})
 		assert.False(t, subEndpoint.IsWildcard())
 
@@ -207,7 +208,7 @@ func TestSotwSubscriptionsWithDeactivatedLegacyWildcardForTypes(t *testing.T) {
 		assert.True(t, subEndpoint.IsWildcard())
 
 		// Route should still have legacy wildcard enabled
-		subRoute := NewSotwSubscription([]string{}, opts, routeType)
+		subRoute := NewSotwSubscription([]string{}, opts.IsLegacyWildcardActive(routeType))
 		subRoute.SetResourceSubscription([]string{})
 		assert.True(t, subRoute.IsWildcard())
 	})
@@ -219,8 +220,9 @@ func TestDeltaSubscriptionsWithDeactivatedLegacyWildcard(t *testing.T) {
 		deactivateOpt := config.DeactivateLegacyWildcard()
 		deactivateOpt(&opts)
 
+		typeURL := "type.googleapis.com/envoy.config.cluster.v3.Cluster"
 		// Create subscription with empty resource list (would normally be legacy wildcard)
-		sub := NewDeltaSubscription([]string{}, []string{}, map[string]string{"resource": "version"}, opts, "type.googleapis.com/envoy.config.cluster.v3.Cluster")
+		sub := NewDeltaSubscription([]string{}, []string{}, map[string]string{"resource": "version"}, opts.IsLegacyWildcardActive(typeURL))
 
 		// With deactivated legacy wildcard, subscription should NOT be wildcard initially
 		assert.False(t, sub.IsWildcard())
@@ -248,11 +250,11 @@ func TestDeltaSubscriptionsWithDeactivatedLegacyWildcardForTypes(t *testing.T) {
 		deactivateOpt(&opts)
 
 		// Both cluster and endpoint should have legacy wildcard deactivated
-		subCluster := NewDeltaSubscription([]string{}, []string{}, map[string]string{}, opts, clusterType)
+		subCluster := NewDeltaSubscription([]string{}, []string{}, map[string]string{}, opts.IsLegacyWildcardActive(clusterType))
 		subCluster.UpdateResourceSubscriptions(nil, nil)
 		assert.False(t, subCluster.IsWildcard())
 
-		subEndpoint := NewDeltaSubscription([]string{}, []string{}, map[string]string{}, opts, endpointType)
+		subEndpoint := NewDeltaSubscription([]string{}, []string{}, map[string]string{}, opts.IsLegacyWildcardActive(endpointType))
 		subEndpoint.UpdateResourceSubscriptions(nil, nil)
 		assert.False(t, subEndpoint.IsWildcard())
 
@@ -261,7 +263,7 @@ func TestDeltaSubscriptionsWithDeactivatedLegacyWildcardForTypes(t *testing.T) {
 		assert.True(t, subEndpoint.IsWildcard())
 
 		// Route should still have legacy wildcard enabled
-		subRoute := NewDeltaSubscription([]string{}, []string{}, map[string]string{}, opts, routeType)
+		subRoute := NewDeltaSubscription([]string{}, []string{}, map[string]string{}, opts.IsLegacyWildcardActive(routeType))
 		subRoute.UpdateResourceSubscriptions(nil, nil)
 		assert.True(t, subRoute.IsWildcard())
 	})