Support to set QPS and burst by configuration. (#3969)

Signed-off-by: KunWuLuan <[email protected]>
Signed-off-by: Future-Outlier <[email protected]>
Signed-off-by: Kai-Hsun Chen <[email protected]>
Co-authored-by: Future-Outlier <[email protected]>
Co-authored-by: yueming.wk <[email protected]>
Co-authored-by: Kai-Hsun Chen <[email protected]>

Author: 4 people

ray-operator/apis/config/v1alpha1/configuration_types.go

@@ -12,25 +12,26 @@ import (
 
 // Configuration is the Schema for Ray operator config.
 type Configuration struct {
-	metav1.TypeMeta `json:",inline"`
+	// Burst allows temporary exceeding of QPS limit for handling request spikes.
+	// Default: 200
+	Burst *int `json:"burst,omitempty"`
 
-	// MetricsAddr is the address the metrics endpoint binds to.
-	MetricsAddr string `json:"metricsAddr,omitempty"`
-
-	// ProbeAddr is the address the probe endpoint binds to.
-	ProbeAddr string `json:"probeAddr,omitempty"`
+	// QPS controls the maximum requests per second to the Kubernetes API server.
+	// Default: 100.0
+	QPS *float64 `json:"qps,omitempty"`
 
 	// EnableLeaderElection enables leader election. Enabling this will ensure
 	// there is only one active instance of the operator.
 	EnableLeaderElection *bool `json:"enableLeaderElection,omitempty"`
 
-	// LeaderElectionNamespace is the namespace where the leader election
-	// resources live. Defaults to the pod namesapce if not set.
-	LeaderElectionNamespace string `json:"leaderElectionNamespace,omitempty"`
+	metav1.TypeMeta `json:",inline"`
 
-	// WatchNamespace specifies a list of namespaces to watch for custom resources, separated by commas.
-	// If empty, all namespaces will be watched.
-	WatchNamespace string `json:"watchNamespace,omitempty"`
+	// LogStdoutEncoder is the encoder to use when logging to stdout. Valid values are "json" and "console".
+	// Defaults to `json` if empty.
+	LogStdoutEncoder string `json:"logStdoutEncoder,omitempty"`
+
+	// ProbeAddr is the address the probe endpoint binds to.
+	ProbeAddr string `json:"probeAddr,omitempty"`
 
 	// LogFile is a path to a local file for synchronizing logs.
 	LogFile string `json:"logFile,omitempty"`
@@ -39,22 +40,29 @@ type Configuration struct {
 	// Defaults to `json` if empty.
 	LogFileEncoder string `json:"logFileEncoder,omitempty"`
 
-	// LogFileEncoder is the encoder to use when logging to a file. Valid values are "json" and "console".
-	// Defaults to `json` if empty.
-	LogStdoutEncoder string `json:"logStdoutEncoder,omitempty"`
+	// LeaderElectionNamespace is the namespace where the leader election
+	// resources live. Defaults to the pod namesapce if not set.
+	LeaderElectionNamespace string `json:"leaderElectionNamespace,omitempty"`
 
 	// BatchScheduler enables the batch scheduler integration with a specific scheduler
 	// based on the given name, currently, supported values are volcano, yunikorn, kai-scheduler.
 	BatchScheduler string `json:"batchScheduler,omitempty"`
 
-	// HeadSidecarContainers includes specification for a sidecar container
-	// to inject into every Head pod.
-	HeadSidecarContainers []corev1.Container `json:"headSidecarContainers,omitempty"`
+	// MetricsAddr is the address the metrics endpoint binds to.
+	MetricsAddr string `json:"metricsAddr,omitempty"`
+
+	// WatchNamespace specifies a list of namespaces to watch for custom resources, separated by commas.
+	// If empty, all namespaces will be watched.
+	WatchNamespace string `json:"watchNamespace,omitempty"`
 
 	// WorkerSidecarContainers includes specification for a sidecar container
 	// to inject into every Worker pod.
 	WorkerSidecarContainers []corev1.Container `json:"workerSidecarContainers,omitempty"`
 
+	// HeadSidecarContainers includes specification for a sidecar container
+	// to inject into every Head pod.
+	HeadSidecarContainers []corev1.Container `json:"headSidecarContainers,omitempty"`
+
 	// ReconcileConcurrency is the max concurrency for each reconciler.
 	ReconcileConcurrency int `json:"reconcileConcurrency,omitempty"`
 

ray-operator/apis/config/v1alpha1/defaults.go

@@ -10,6 +10,8 @@ const (
 	DefaultProbeAddr            = ":8082"
 	DefaultEnableLeaderElection = true
 	DefaultReconcileConcurrency = 1
+	DefaultQPS                  = float64(100)
+	DefaultBurst                = 200
 )
 
 func addDefaultingFuncs(scheme *runtime.Scheme) error {
@@ -36,4 +38,12 @@ func SetDefaults_Configuration(cfg *Configuration) {
 	if cfg.ReconcileConcurrency == 0 {
 		cfg.ReconcileConcurrency = DefaultReconcileConcurrency
 	}
+
+	if cfg.QPS == nil {
+		cfg.QPS = ptr.To(DefaultQPS)
+	}
+
+	if cfg.Burst == nil {
+		cfg.Burst = ptr.To(DefaultBurst)
+	}
 }

ray-operator/apis/config/v1alpha1/zz_generated.deepcopy.go

@@ -6,6 +6,7 @@ require (
 	github.com/Masterminds/semver/v3 v3.3.1
 	github.com/go-logr/logr v1.4.3
 	github.com/go-logr/zapr v1.3.0
+	github.com/google/go-cmp v0.7.0
 	github.com/jarcoal/httpmock v1.4.0
 	github.com/onsi/ginkgo/v2 v2.23.4
 	github.com/onsi/gomega v1.37.0
@@ -49,7 +50,6 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/btree v1.1.3 // indirect
 	github.com/google/gnostic-models v0.6.9 // indirect
-	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect

ray-operator/go.mod

@@ -70,6 +70,8 @@ func main() {
 	var enableBatchScheduler bool
 	var batchScheduler string
 	var enableMetrics bool
+	var qps float64
+	var burst int
 
 	// TODO: remove flag-based config once Configuration API graduates to v1.
 	flag.StringVar(&metricsAddr, "metrics-addr", configapi.DefaultMetricsAddr, "The address the metric endpoint binds to.")
@@ -101,6 +103,8 @@ func main() {
 		"Use Kubernetes proxy subresource when connecting to the Ray Head node.")
 	flag.StringVar(&featureGates, "feature-gates", "", "A set of key=value pairs that describe feature gates. E.g. FeatureOne=true,FeatureTwo=false,...")
 	flag.BoolVar(&enableMetrics, "enable-metrics", false, "Enable the emission of control plane metrics.")
+	flag.Float64Var(&qps, "qps", float64(configapi.DefaultQPS), "The QPS value for the client communicating with the Kubernetes API server.")
+	flag.IntVar(&burst, "burst", configapi.DefaultBurst, "The maximum burst for throttling requests from this client to the Kubernetes API server.")
 
 	opts := k8szap.Options{
 		TimeEncoder: zapcore.ISO8601TimeEncoder,
@@ -131,6 +135,8 @@ func main() {
 		config.UseKubernetesProxy = useKubernetesProxy
 		config.DeleteRayJobAfterJobFinishes = os.Getenv(utils.DELETE_RAYJOB_CR_AFTER_JOB_FINISHES) == "true"
 		config.EnableMetrics = enableMetrics
+		config.QPS = &qps
+		config.Burst = &burst
 	}
 
 	stdoutEncoder, err := newLogEncoder(logStdoutEncoder)
@@ -228,6 +234,8 @@ func main() {
 	setupLog.Info("Setup manager")
 	restConfig := ctrl.GetConfigOrDie()
 	restConfig.UserAgent = userAgent
+	restConfig.QPS = float32(*config.QPS)
+	restConfig.Burst = *config.Burst
 	mgr, err := ctrl.NewManager(restConfig, options)
 	exitOnError(err, "unable to start manager")
 

ray-operator/main.go

@@ -34,6 +34,8 @@ kind: Configuration
 				ProbeAddr:            ":8082",
 				EnableLeaderElection: ptr.To(true),
 				ReconcileConcurrency: 1,
+				QPS:                  ptr.To(configapi.DefaultQPS),
+				Burst:                ptr.To(configapi.DefaultBurst),
 			},
 			expectErr: false,
 		},
@@ -55,6 +57,8 @@ reconcileConcurrency: 1
 				ProbeAddr:            ":8082",
 				EnableLeaderElection: ptr.To(true),
 				ReconcileConcurrency: 1,
+				QPS:                  ptr.To(configapi.DefaultQPS),
+				Burst:                ptr.To(configapi.DefaultBurst),
 			},
 			expectErr: false,
 		},
@@ -94,6 +98,8 @@ workerSidecarContainers:
 						Image: "fluent/fluent-bit:1.9.6",
 					},
 				},
+				QPS:   ptr.To(configapi.DefaultQPS),
+				Burst: ptr.To(configapi.DefaultBurst),
 			},
 			expectErr: false,
 		},
@@ -116,6 +122,8 @@ unknownfield: 1
 				ProbeAddr:            ":8082",
 				EnableLeaderElection: ptr.To(true),
 				ReconcileConcurrency: 1,
+				QPS:                  ptr.To(configapi.DefaultQPS),
+				Burst:                ptr.To(configapi.DefaultBurst),
 			},
 			expectErr: false,
 		},
@@ -137,6 +145,8 @@ reconcileConcurrency: true
 				ProbeAddr:            ":8082",
 				EnableLeaderElection: ptr.To(true),
 				ReconcileConcurrency: 0,
+				QPS:                  ptr.To(configapi.DefaultQPS),
+				Burst:                ptr.To(configapi.DefaultBurst),
 			},
 			expectErr:   true,
 			errContains: "json: cannot unmarshal bool into Go struct field Configuration.reconcileConcurrency of type int",
@@ -154,6 +164,30 @@ reconcileConcurrency: true
 				MetricsAddr:          ":8080",
 				ProbeAddr:            ":8082",
 				EnableLeaderElection: ptr.To(true),
+				QPS:                  ptr.To(configapi.DefaultQPS),
+				Burst:                ptr.To(configapi.DefaultBurst),
+			},
+			expectErr:   true,
+			errContains: `no kind "Configuration" is registered for version "config.ray.io/v1beta1" in scheme`,
+		},
+		{
+			name: "set QPS and Burst",
+			configData: `apiVersion: config.ray.io/v1beta1
+kind: Configuration
+qps: 150
+burst: 300
+`,
+			expectedConfig: configapi.Configuration{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "Configuration",
+					APIVersion: "config.ray.io/v1alpha1",
+				},
+				MetricsAddr:          ":8080",
+				ProbeAddr:            ":8082",
+				EnableLeaderElection: ptr.To(true),
+				ReconcileConcurrency: 1,
+				QPS:                  ptr.To((150.0)),
+				Burst:                ptr.To(300),
 			},
 			expectErr:   true,
 			errContains: `no kind "Configuration" is registered for version "config.ray.io/v1beta1" in scheme`,