feat(kuma-cp) change KDS max message limit (#2265)

Signed-off-by: Jakub Dyszkiewicz <jakub.dyszkiewicz@gmail.com>

pkg/api-server/config_ws_test.go

@@ -157,14 +157,15 @@ var _ = Describe("Config WS", func() {
 				"refreshInterval": "1s",
 				"tlsCertFile": "",
 				"tlsKeyFile": "",
-				"zoneInsightFlushInterval": "10s"
-			  },
-			  "pollTimeout": "500ms"
+				"zoneInsightFlushInterval": "10s",
+				"maxMsgSize": 10485760
+			  }
 			},
 			"zone": {
 			  "kds": {
 				"refreshInterval": "1s",
-				"rootCaFile": ""
+				"rootCaFile": "",
+				"maxMsgSize": 10485760
 			  }
 			}
 		  },

pkg/config/app/kuma-cp/kuma-cp.defaults.yaml

@@ -313,8 +313,6 @@ dnsServer:
 # Multizone mode
 multizone:
   global:
-    # The Zone status poller timeout
-    pollTimeout: 500ms # ENV: KUMA_MULTIZONE_GLOBAL_POLL_TIMEOUT
     kds:
       # Port of a gRPC server that serves Kuma Discovery Service (KDS).
       grpcPort: 5685 # ENV: KUMA_MULTIZONE_GLOBAL_KDS_GRPC_PORT
@@ -326,6 +324,9 @@ multizone:
       tlsCertFile: # ENV: KUMA_MULTIZONE_GLOBAL_KDS_TLS_CERT_FILE
       # TTlsKeyFile defines a path to a file with PEM-encoded TLS key.
       tlsKeyFile: # ENV: KUMA_MULTIZONE_GLOBAL_KDS_TLS_KEY_FILE
+      # MaxMsgSize defines a maximum size of the message in bytes that is exchanged using KDS.
+      # In practice this means a limit on full list of one resource type.
+      maxMsgSize: 10485760 # ENV: KUMA_MULTIZONE_GLOBAL_KDS_MAX_MSG_SIZE
   zone:
     # Kuma Zone name used to mark the zone dataplane resources
     name: "" # ENV: KUMA_MULTIZONE_ZONE_NAME
@@ -336,6 +337,9 @@ multizone:
       refreshInterval: 1s # ENV: KUMA_MULTIZONE_ZONE_KDS_REFRESH_INTERVAL
       # RootCAFile defines a path to a file with PEM-encoded Root CA. Client will verify server by using it.
       rootCaFile: # ENV: KUMA_MULTIZONE_ZONE_KDS_ROOT_CA_FILE
+      # MaxMsgSize defines a maximum size of the message in bytes that is exchanged using KDS.
+      # In practice this means a limit on full list of one resource type.
+      maxMsgSize: 10485760 # ENV: KUMA_MULTIZONE_ZONE_KDS_MAX_MSG_SIZE
 
 # Diagnostics configuration
 diagnostics:

pkg/config/loader_test.go

@@ -192,16 +192,17 @@ var _ = Describe("Config loader", func() {
 			Expect(cfg.Mode).To(Equal(config_core.Zone))
 			Expect(cfg.Multizone.Zone.Name).To(Equal("zone-1"))
 
-			Expect(cfg.Multizone.Global.PollTimeout).To(Equal(750 * time.Millisecond))
 			Expect(cfg.Multizone.Global.KDS.GrpcPort).To(Equal(uint32(1234)))
 			Expect(cfg.Multizone.Global.KDS.RefreshInterval).To(Equal(time.Second * 2))
 			Expect(cfg.Multizone.Global.KDS.ZoneInsightFlushInterval).To(Equal(time.Second * 5))
 			Expect(cfg.Multizone.Global.KDS.TlsCertFile).To(Equal("/cert"))
 			Expect(cfg.Multizone.Global.KDS.TlsKeyFile).To(Equal("/key"))
+			Expect(cfg.Multizone.Global.KDS.MaxMsgSize).To(Equal(uint32(1)))
 			Expect(cfg.Multizone.Zone.GlobalAddress).To(Equal("grpc://1.1.1.1:5685"))
 			Expect(cfg.Multizone.Zone.Name).To(Equal("zone-1"))
 			Expect(cfg.Multizone.Zone.KDS.RootCAFile).To(Equal("/rootCa"))
 			Expect(cfg.Multizone.Zone.KDS.RefreshInterval).To(Equal(9 * time.Second))
+			Expect(cfg.Multizone.Zone.KDS.MaxMsgSize).To(Equal(uint32(2)))
 
 			Expect(cfg.Defaults.SkipMeshCreation).To(BeTrue())
 
@@ -374,19 +375,20 @@ guiServer:
 mode: zone
 multizone:
   global:
-    pollTimeout: 750ms
     kds:
       grpcPort: 1234
       refreshInterval: 2s
       zoneInsightFlushInterval: 5s
       tlsCertFile: /cert
       tlsKeyFile: /key
+      maxMsgSize: 1
   zone:
     globalAddress: "grpc://1.1.1.1:5685"
     name: "zone-1"
     kds:
       refreshInterval: 9s
       rootCaFile: /rootCa
+      maxMsgSize: 2
 dnsServer:
   domain: test-domain
   port: 15653
@@ -525,15 +527,16 @@ sdsServer:
 				"KUMA_DNS_SERVER_PORT":                                                                     "15653",
 				"KUMA_DNS_SERVER_CIDR":                                                                     "127.1.0.0/16",
 				"KUMA_MODE":                                                                                "zone",
-				"KUMA_MULTIZONE_GLOBAL_POLL_TIMEOUT":                                                       "750ms",
 				"KUMA_MULTIZONE_GLOBAL_KDS_GRPC_PORT":                                                      "1234",
 				"KUMA_MULTIZONE_GLOBAL_KDS_REFRESH_INTERVAL":                                               "2s",
 				"KUMA_MULTIZONE_GLOBAL_KDS_TLS_CERT_FILE":                                                  "/cert",
 				"KUMA_MULTIZONE_GLOBAL_KDS_TLS_KEY_FILE":                                                   "/key",
+				"KUMA_MULTIZONE_GLOBAL_KDS_MAX_MSG_SIZE":                                                   "1",
 				"KUMA_MULTIZONE_ZONE_GLOBAL_ADDRESS":                                                       "grpc://1.1.1.1:5685",
 				"KUMA_MULTIZONE_ZONE_NAME":                                                                 "zone-1",
 				"KUMA_MULTIZONE_ZONE_KDS_ROOT_CA_FILE":                                                     "/rootCa",
 				"KUMA_MULTIZONE_ZONE_KDS_REFRESH_INTERVAL":                                                 "9s",
+				"KUMA_MULTIZONE_ZONE_KDS_MAX_MSG_SIZE":                                                     "2",
 				"KUMA_MULTIZONE_GLOBAL_KDS_ZONE_INSIGHT_FLUSH_INTERVAL":                                    "5s",
 				"KUMA_DEFAULTS_SKIP_MESH_CREATION":                                                         "true",
 				"KUMA_DIAGNOSTICS_SERVER_PORT":                                                             "5003",

pkg/config/multizone/kds.go

@@ -20,6 +20,9 @@ type KdsServerConfig struct {
 	TlsCertFile string `yaml:"tlsCertFile" envconfig:"kuma_multizone_global_kds_tls_cert_file"`
 	// TlsKeyFile defines a path to a file with PEM-encoded TLS key.
 	TlsKeyFile string `yaml:"tlsKeyFile" envconfig:"kuma_multizone_global_kds_tls_key_file"`
+	// MaxMsgSize defines a maximum size of the message that is exchanged using KDS.
+	// In practice this means a limit on full list of one resource type.
+	MaxMsgSize uint32 `yaml:"maxMsgSize" envconfig:"kuma_multizone_global_kds_max_msg_size"`
 }
 
 var _ config.Config = &KdsServerConfig{}
@@ -51,6 +54,9 @@ type KdsClientConfig struct {
 	RefreshInterval time.Duration `yaml:"refreshInterval" envconfig:"kuma_multizone_zone_kds_refresh_interval"`
 	// RootCAFile defines a path to a file with PEM-encoded Root CA. Client will verify the server by using it.
 	RootCAFile string `yaml:"rootCaFile" envconfig:"kuma_multizone_zone_kds_root_ca_file"`
+	// MaxMsgSize defines a maximum size of the message that is exchanged using KDS.
+	// In practice this means a limit on full list of one resource type.
+	MaxMsgSize uint32 `yaml:"maxMsgSize" envconfig:"kuma_multizone_zone_kds_max_msg_size"`
 }
 
 var _ config.Config = &KdsClientConfig{}

pkg/config/multizone/multicluster.go

@@ -16,7 +16,6 @@ var _ config.Config = &MultizoneConfig{}
 
 // Global configuration
 type GlobalConfig struct {
-	PollTimeout time.Duration `yaml:"pollTimeout,omitempty" envconfig:"kuma_multizone_global_poll_timeout"`
 	// KDS Configuration
 	KDS *KdsServerConfig `yaml:"kds,omitempty"`
 }
@@ -31,11 +30,11 @@ func (g *GlobalConfig) Validate() error {
 
 func DefaultGlobalConfig() *GlobalConfig {
 	return &GlobalConfig{
-		PollTimeout: 500 * time.Millisecond,
 		KDS: &KdsServerConfig{
 			GrpcPort:                 5685,
 			RefreshInterval:          1 * time.Second,
 			ZoneInsightFlushInterval: 10 * time.Second,
+			MaxMsgSize:               10 * 1024 * 1024,
 		},
 	}
 }
@@ -94,6 +93,7 @@ func DefaultZoneConfig() *ZoneConfig {
 		Name:          "",
 		KDS: &KdsClientConfig{
 			RefreshInterval: 1 * time.Second,
+			MaxMsgSize:      10 * 1024 * 1024,
 		},
 	}
 }

pkg/kds/mux/client.go

@@ -44,6 +44,10 @@ func (c *client) Start(stop <-chan struct{}) (errs error) {
 		return err
 	}
 	dialOpts := c.metrics.GRPCClientInterceptors()
+	dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(
+		grpc.MaxCallSendMsgSize(int(c.config.MaxMsgSize)),
+		grpc.MaxCallRecvMsgSize(int(c.config.MaxMsgSize))),
+	)
 	switch u.Scheme {
 	case "grpc":
 		dialOpts = append(dialOpts, grpc.WithInsecure())

pkg/kds/mux/server.go

@@ -66,6 +66,8 @@ func (s *server) Start(stop <-chan struct{}) error {
 			MinTime:             grpcKeepAliveTime,
 			PermitWithoutStream: true,
 		}),
+		grpc.MaxRecvMsgSize(int(s.config.MaxMsgSize)),
+		grpc.MaxSendMsgSize(int(s.config.MaxMsgSize)),
 	}
 	grpcOptions = append(grpcOptions, s.metrics.GRPCServerInterceptors()...)
 	useTLS := s.config.TlsCertFile != ""