feat(destionation): add --stream-queue-capacity arg

controller/api/destination/endpoint_profile_translator.go

@@ -54,6 +54,7 @@ func newEndpointProfileTranslator(
 	stream pb.Destination_GetProfileServer,
 	endStream chan struct{},
 	log *logging.Entry,
+	queueCapacity int,
 ) *endpointProfileTranslator {
 	return &endpointProfileTranslator{
 		forceOpaqueTransport: forceOpaqueTransport,
@@ -66,7 +67,7 @@ func newEndpointProfileTranslator(
 
 		stream:    stream,
 		endStream: endStream,
-		updates:   make(chan *watcher.Address, updateQueueCapacity),
+		updates:   make(chan *watcher.Address, queueCapacity),
 		stop:      make(chan struct{}),
 
 		log: log.WithField("component", "endpoint-profile-translator"),

controller/api/destination/endpoint_profile_translator_test.go

@@ -44,6 +44,7 @@ func TestEndpointProfileTranslator(t *testing.T) {
 			mockGetProfileServer,
 			nil,
 			log,
+			DefaultStreamQueueCapacity,
 		)
 		translator.Start()
 		defer translator.Stop()
@@ -83,17 +84,19 @@ func TestEndpointProfileTranslator(t *testing.T) {
 		}
 		log := logging.WithField("test", t.Name())
 		endStream := make(chan struct{})
+		queueCapacity := DefaultStreamQueueCapacity
 		translator := newEndpointProfileTranslator(
 			true, true, "cluster", "identity", make(map[uint32]struct{}), nil,
 			mockGetProfileServer,
 			endStream,
 			log,
+			queueCapacity,
 		)
 
 		// We avoid starting the translator so that it doesn't drain its update
 		// queue and we can test the overflow behavior.
 
-		for i := 0; i < updateQueueCapacity/2; i++ {
+		for i := 0; i < queueCapacity/2; i++ {
 			if err := translator.Update(podAddr); err != nil {
 				t.Fatal("Expected update")
 			}
@@ -114,10 +117,10 @@ func TestEndpointProfileTranslator(t *testing.T) {
 		}
 
 		// The queue should be full and the next update should fail.
-		t.Logf("Queue length=%d capacity=%d", translator.queueLen(), updateQueueCapacity)
+		t.Logf("Queue length=%d capacity=%d", translator.queueLen(), queueCapacity)
 		if err := translator.Update(podAddr); err == nil {
 			if !errors.Is(err, http.ErrServerClosed) {
-				t.Fatalf("Expected update to fail; queue=%d; capacity=%d", translator.queueLen(), updateQueueCapacity)
+				t.Fatalf("Expected update to fail; queue=%d; capacity=%d", translator.queueLen(), queueCapacity)
 			}
 		}
 

controller/api/destination/endpoint_translator.go

@@ -27,8 +27,6 @@ const (
 	envAdminListenAddr   = "LINKERD2_PROXY_ADMIN_LISTEN_ADDR"
 	envControlListenAddr = "LINKERD2_PROXY_CONTROL_LISTEN_ADDR"
 
-	updateQueueCapacity = 100
-
 	defaultProxyInboundPort = 4143
 )
 
@@ -101,6 +99,7 @@ func newEndpointTranslator(
 	stream pb.Destination_GetServer,
 	endStream chan struct{},
 	log *logging.Entry,
+	queueCapacity int,
 ) (*endpointTranslator, error) {
 	log = log.WithFields(logging.Fields{
 		"component": "endpoint-translator",
@@ -139,7 +138,7 @@ func newEndpointTranslator(
 		endStream,
 		log,
 		counter,
-		make(chan interface{}, updateQueueCapacity),
+		make(chan interface{}, queueCapacity),
 		make(chan struct{}),
 	}, nil
 }

controller/api/destination/federated_service_watcher.go

@@ -364,6 +364,7 @@ func (fs *federatedService) remoteDiscoverySubscribe(
 		subscriber.stream,
 		subscriber.endStream,
 		fs.log,
+		fs.config.StreamQueueCapacity,
 	)
 	if err != nil {
 		fs.log.Errorf("Failed to create endpoint translator for remote discovery service %q in cluster %s: %s", id.service.Name, id.cluster, err)
@@ -418,6 +419,7 @@ func (fs *federatedService) localDiscoverySubscribe(
 		subscriber.stream,
 		subscriber.endStream,
 		fs.log,
+		fs.config.StreamQueueCapacity,
 	)
 	if err != nil {
 		fs.log.Errorf("Failed to create endpoint translator for %s: %s", localDiscovery, err)

controller/api/destination/federated_service_watcher_test.go

@@ -151,7 +151,7 @@ func mockFederatedServiceWatcher(t *testing.T) (*federatedServiceWatcher, error)
 	if err != nil {
 		return nil, fmt.Errorf("NewClusterStoreWithDecoder returned an error: %w", err)
 	}
-	fsw, err := newFederatedServiceWatcher(k8sAPI, metadataAPI, &Config{}, clusterStore, localEndpoints, logging.WithField("test", t.Name()))
+	fsw, err := newFederatedServiceWatcher(k8sAPI, metadataAPI, &Config{StreamQueueCapacity: DefaultStreamQueueCapacity}, clusterStore, localEndpoints, logging.WithField("test", t.Name()))
 	if err != nil {
 		return nil, fmt.Errorf("newFederatedServiceWatcher returned an error: %w", err)
 	}

controller/api/destination/profile_translator.go

@@ -47,6 +47,10 @@ var profileUpdatesQueueOverflowCounter = promauto.NewCounterVec(
 )
 
 func newProfileTranslator(serviceID watcher.ServiceID, stream pb.Destination_GetProfileServer, log *logging.Entry, fqn string, port uint32, endStream chan struct{}) (*profileTranslator, error) {
+	return newProfileTranslatorWithCapacity(serviceID, stream, log, fqn, port, endStream, DefaultStreamQueueCapacity)
+}
+
+func newProfileTranslatorWithCapacity(serviceID watcher.ServiceID, stream pb.Destination_GetProfileServer, log *logging.Entry, fqn string, port uint32, endStream chan struct{}, queueCapacity int) (*profileTranslator, error) {
 	parentRef := &meta.Metadata{
 		Kind: &meta.Metadata_Resource{
 			Resource: &meta.Resource{
@@ -73,7 +77,7 @@ func newProfileTranslator(serviceID watcher.ServiceID, stream pb.Destination_Get
 		endStream:       endStream,
 		log:             log.WithField("component", "profile-translator"),
 		overflowCounter: overflowCounter,
-		updates:         make(chan *sp.ServiceProfile, updateQueueCapacity),
+		updates:         make(chan *sp.ServiceProfile, queueCapacity),
 		stop:            make(chan struct{}),
 	}, nil
 }

controller/api/destination/server.go

@@ -24,6 +24,10 @@ import (
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 )
 
+// DefaultStreamQueueCapacity defines the default maximum number of pending
+// updates buffered per stream before the stream is closed.
+const DefaultStreamQueueCapacity = 100
+
 type (
 	Config struct {
 		ControllerNS,
@@ -39,6 +43,8 @@ type (
 		MeshedHttp2ClientParams *pb.Http2ClientParams
 
 		DefaultOpaquePorts map[uint32]struct{}
+
+		StreamQueueCapacity int
 	}
 
 	server struct {
@@ -219,6 +225,7 @@ func (s *server) Get(dest *pb.GetDestination, stream pb.Destination_GetServer) e
 			stream,
 			streamEnd,
 			log,
+			s.config.StreamQueueCapacity,
 		)
 		if err != nil {
 			return status.Errorf(codes.Internal, "Failed to create endpoint translator: %s", err)
@@ -257,6 +264,7 @@ func (s *server) Get(dest *pb.GetDestination, stream pb.Destination_GetServer) e
 			stream,
 			streamEnd,
 			log,
+			s.config.StreamQueueCapacity,
 		)
 		if err != nil {
 			return status.Errorf(codes.Internal, "Failed to create endpoint translator: %s", err)
@@ -403,7 +411,7 @@ func (s *server) subscribeToServiceProfile(
 	// We build up the pipeline of profile updaters backwards, starting from
 	// the translator which takes profile updates, translates them to protobuf
 	// and pushes them onto the gRPC stream.
-	translator, err := newProfileTranslator(service, stream, log, fqn, port, streamEnd)
+	translator, err := newProfileTranslatorWithCapacity(service, stream, log, fqn, port, streamEnd, s.config.StreamQueueCapacity)
 	if err != nil {
 		return status.Errorf(codes.Internal, "Failed to create profile translator: %s", err)
 	}
@@ -552,6 +560,7 @@ func (s *server) subscribeToEndpointProfile(
 		stream,
 		streamEnd,
 		log,
+		s.config.StreamQueueCapacity,
 	)
 	translator.Start()
 	defer translator.Stop()

controller/api/destination/test_util.go

@@ -1062,7 +1062,7 @@ spec:
 		t.Fatalf("can't create cluster store: %s", err)
 	}
 
-	federatedServices, err := newFederatedServiceWatcher(k8sAPI, metadataAPI, &Config{}, clusterStore, endpoints, log)
+	federatedServices, err := newFederatedServiceWatcher(k8sAPI, metadataAPI, &Config{StreamQueueCapacity: DefaultStreamQueueCapacity}, clusterStore, endpoints, log)
 	if err != nil {
 		t.Fatalf("can't create federated service watcher: %s", err)
 	}
@@ -1082,6 +1082,7 @@ spec:
 			ClusterDomain:       "mycluster.local",
 			IdentityTrustDomain: "trust.domain",
 			DefaultOpaquePorts:  defaultOpaquePorts,
+			StreamQueueCapacity: DefaultStreamQueueCapacity,
 		},
 		workloads,
 		endpoints,
@@ -1191,6 +1192,7 @@ metadata:
 		mockGetServer,
 		nil,
 		logging.WithField("test", t.Name()),
+		DefaultStreamQueueCapacity,
 	)
 	if err != nil {
 		t.Fatalf("failed to create endpoint translator: %s", err)

controller/cmd/destination/main.go

@@ -47,6 +47,7 @@ func Main(args []string) {
 	// This will default to true. It can be overridden with experimental CLI
 	// flags. Currently not exposed as a configuration value through Helm.
 	exportControllerQueueMetrics := cmd.Bool("export-queue-metrics", true, "Exports queue metrics for the external workload controller")
+	streamQueueCapacity := cmd.Int("stream-queue-capacity", destination.DefaultStreamQueueCapacity, "Maximum number of updates buffered per stream before the stream is closed")
 
 	traceCollector := flags.AddTraceFlags(cmd)
 
@@ -66,6 +67,10 @@ func Main(args []string) {
 
 	flags.ConfigureAndParse(cmd, args)
 
+	if *streamQueueCapacity <= 0 {
+		log.Fatalf("--stream-queue-capacity must be greater than 0")
+	}
+
 	if *enableIPv6 && !*enableEndpointSlices {
 		log.Fatal("If --enable-ipv6=true then --enable-endpoint-slices needs to be true")
 	}
@@ -190,6 +195,7 @@ func Main(args []string) {
 		EnableIPv6:              *enableIPv6,
 		ExtEndpointZoneWeights:  *extEndpointZoneWeights,
 		MeshedHttp2ClientParams: meshedHTTP2ClientParams,
+		StreamQueueCapacity:     *streamQueueCapacity,
 	}
 	server, err := destination.NewServer(
 		*addr,