Repository: akuity/kargo

Issue:
Add support of custom QPS and Burst values for client-go
# Proposed Feature

Can you please add support of custom configuration for overriding QPS and Burst in client-go? 

# Motivation

During experimenting with MAX_CONCURRENT_WAREHOUSE_RECONCILES, after I changed default 4 to 32, I see often error messages related to client-side throttle:

`I0526 14:46:06.880713       7 request.go:729] Waited for 1.882506858s due to client-side throttling, not priority and fairness, request: GET:https://172.30.0.1:443/api/v1/namespaces/example-namespace/secrets?labelSelector=kargo.akuity.io%2Fcred-type%3Dgit`

Please correct me if I am wrong, but maybe client-go's default QPS and Burst values are conservative, and there might be cases where it needs to be adjusted.


Discussion:
@hiddeco: The usefulness of QPS and burst overrides depend on if your cluster has ["priority and fairness" enabled](https://kubernetes.io/docs/concepts/cluster-administration/flow-control/).

But I do agree we should improve the construction of our client config, both to detect this and to allow overrides if it isn't.

Solution patch:
diff --git a/cmd/controlplane/api.go b/cmd/controlplane/api.go
index 7b7cbe11..ad30a1b1 100644
--- a/cmd/controlplane/api.go
+++ b/cmd/controlplane/api.go
@@ -15,11 +15,14 @@ import (
 	"github.com/akuity/kargo/internal/server/config"
 	"github.com/akuity/kargo/internal/server/kubernetes"
 	"github.com/akuity/kargo/internal/server/rbac"
+	"github.com/akuity/kargo/internal/types"
 	versionpkg "github.com/akuity/kargo/pkg/x/version"
 )
 
 type apiOptions struct {
 	KubeConfig string
+	QPS        float32
+	Burst      int
 
 	BindAddress string
 	Port        string
@@ -51,6 +54,8 @@ func newAPICommand() *cobra.Command {
 
 func (o *apiOptions) complete() {
 	o.KubeConfig = os.GetEnv("KUBECONFIG", "")
+	o.QPS = types.MustParseFloat32(os.GetEnv("KUBE_API_QPS", "50.0"))
+	o.Burst = types.MustParseInt(os.GetEnv("KUBE_API_BURST", "300"))
 
 	o.BindAddress = os.GetEnv("BIND_ADDRESS", "0.0.0.0")
 	o.Port = os.GetEnv("PORT", "8080")
@@ -72,6 +77,8 @@ func (o *apiOptions) run(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("error getting Kubernetes client REST config: %w", err)
 	}
+	kubernetes.ConfigureQPSBurst(ctx, restCfg, o.QPS, o.Burst)
+
 	kubeClientOptions := kubernetes.ClientOptions{}
 	if serverCfg.OIDCConfig != nil {
 		kubeClientOptions.GlobalServiceAccountNamespaces = serverCfg.OIDCConfig.GlobalServiceAccountNamespaces
diff --git a/cmd/controlplane/controller.go b/cmd/controlplane/controller.go
index 9c9f0031..3423839c 100644
--- a/cmd/controlplane/controller.go
+++ b/cmd/controlplane/controller.go
@@ -40,8 +40,11 @@ import (
 )
 
 type controllerOptions struct {
-	ShardName  string
+	ShardName string
+
 	KubeConfig string
+	QPS        float32
+	Burst      int
 
 	ArgoCDEnabled       bool
 	ArgoCDKubeConfig    string
@@ -77,10 +80,15 @@ func newControllerCommand() *cobra.Command {
 
 func (o *controllerOptions) complete() {
 	o.ShardName = os.GetEnv("SHARD_NAME", "")
+
 	o.KubeConfig = os.GetEnv("KUBECONFIG", "")
+	o.QPS = types.MustParseFloat32(os.GetEnv("KUBE_API_QPS", "50.0"))
+	o.Burst = types.MustParseInt(os.GetEnv("KUBE_API_BURST", "300"))
+
 	o.ArgoCDEnabled = types.MustParseBool(os.GetEnv("ARGOCD_INTEGRATION_ENABLED", "true"))
 	o.ArgoCDKubeConfig = os.GetEnv("ARGOCD_KUBECONFIG", "")
 	o.ArgoCDNamespaceOnly = types.MustParseBool(os.GetEnv("ARGOCD_WATCH_ARGOCD_NAMESPACE_ONLY", "false"))
+
 	o.MetricsBindAddress = os.GetEnv("METRICS_BIND_ADDRESS", "0")
 	o.PprofBindAddress = os.GetEnv("PPROF_BIND_ADDRESS", "")
 }
@@ -148,6 +156,7 @@ func (o *controllerOptions) setupKargoManager(
 		return nil, stagesReconcilerCfg,
 			fmt.Errorf("error loading REST config for Kargo controller manager: %w", err)
 	}
+	kubernetes.ConfigureQPSBurst(ctx, restCfg, o.QPS, o.Burst)
 	restCfg.ContentType = runtime.ContentTypeJSON
 
 	scheme := runtime.NewScheme()
@@ -259,6 +268,7 @@ func (o *controllerOptions) setupArgoCDManager(ctx context.Context) (manager.Man
 	if err != nil {
 		return nil, fmt.Errorf("error loading REST config for Argo CD controller manager: %w", err)
 	}
+	kubernetes.ConfigureQPSBurst(ctx, restCfg, o.QPS, o.Burst)
 	restCfg.ContentType = runtime.ContentTypeJSON
 
 	argocdNamespace := libargocd.Namespace()
diff --git a/cmd/controlplane/external_webhooks.go b/cmd/controlplane/external_webhooks.go
index 5e650ce6..3b5d8701 100644
--- a/cmd/controlplane/external_webhooks.go
+++ b/cmd/controlplane/external_webhooks.go
@@ -14,12 +14,15 @@ import (
 	"github.com/akuity/kargo/internal/logging"
 	"github.com/akuity/kargo/internal/os"
 	"github.com/akuity/kargo/internal/server/kubernetes"
+	"github.com/akuity/kargo/internal/types"
 	"github.com/akuity/kargo/internal/webhook/external"
 	versionpkg "github.com/akuity/kargo/pkg/x/version"
 )
 
 type externalWebhooksServerOptions struct {
 	KubeConfig string
+	QPS        float32
+	Burst      int
 
 	BindAddress string
 	Port        string
@@ -51,6 +54,9 @@ func newExternalWebhooksServerCommand() *cobra.Command {
 
 func (o *externalWebhooksServerOptions) complete() {
 	o.KubeConfig = os.GetEnv("KUBECONFIG", "")
+	o.QPS = types.MustParseFloat32(os.GetEnv("KUBE_API_QPS", "50.0"))
+	o.Burst = types.MustParseInt(os.GetEnv("KUBE_API_BURST", "300"))
+
 	o.BindAddress = os.GetEnv("BIND_ADDRESS", "0.0.0.0")
 	o.Port = os.GetEnv("PORT", "8080")
 }
@@ -71,6 +77,7 @@ func (o *externalWebhooksServerOptions) run(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("error getting Kubernetes client REST config: %w", err)
 	}
+	kubernetes.ConfigureQPSBurst(ctx, restCfg, o.QPS, o.Burst)
 
 	cluster, err := libCluster.New(restCfg)
 	if err != nil {
diff --git a/cmd/controlplane/garbage_collector.go b/cmd/controlplane/garbage_collector.go
index 95190490..a485dc5c 100644
--- a/cmd/controlplane/garbage_collector.go
+++ b/cmd/controlplane/garbage_collector.go
@@ -19,11 +19,14 @@ import (
 	"github.com/akuity/kargo/internal/logging"
 	"github.com/akuity/kargo/internal/os"
 	"github.com/akuity/kargo/internal/server/kubernetes"
+	"github.com/akuity/kargo/internal/types"
 	versionpkg "github.com/akuity/kargo/pkg/x/version"
 )
 
 type garbageCollectorOptions struct {
 	KubeConfig string
+	QPS        float32
+	Burst      int
 
 	MetricsBindAddress string
 	PprofBindAddress   string
@@ -55,6 +58,9 @@ func newGarbageCollectorCommand() *cobra.Command {
 
 func (o *garbageCollectorOptions) complete() {
 	o.KubeConfig = os.GetEnv("KUBECONFIG", "")
+	o.QPS = types.MustParseFloat32(os.GetEnv("KUBE_API_QPS", "50.0"))
+	o.Burst = types.MustParseInt(os.GetEnv("KUBE_API_BURST", "300"))
+
 	o.MetricsBindAddress = os.GetEnv("METRICS_BIND_ADDRESS", "0")
 	o.PprofBindAddress = os.GetEnv("PPROF_BIND_ADDRESS", "")
 }
@@ -97,6 +103,7 @@ func (o *garbageCollectorOptions) setupManager(ctx context.Context) (manager.Man
 	if err != nil {
 		return nil, fmt.Errorf("error loading REST config: %w", err)
 	}
+	kubernetes.ConfigureQPSBurst(ctx, restCfg, o.QPS, o.Burst)
 
 	scheme := runtime.NewScheme()
 	if err = corev1.AddToScheme(scheme); err != nil {
diff --git a/cmd/controlplane/kubernetes_webhooks.go b/cmd/controlplane/kubernetes_webhooks.go
index acd35cf6..d1ae92ed 100644
--- a/cmd/controlplane/kubernetes_webhooks.go
+++ b/cmd/controlplane/kubernetes_webhooks.go
@@ -19,6 +19,7 @@ import (
 	"github.com/akuity/kargo/internal/logging"
 	"github.com/akuity/kargo/internal/os"
 	"github.com/akuity/kargo/internal/server/kubernetes"
+	"github.com/akuity/kargo/internal/types"
 	libWebhook "github.com/akuity/kargo/internal/webhook/kubernetes"
 	"github.com/akuity/kargo/internal/webhook/kubernetes/freight"
 	"github.com/akuity/kargo/internal/webhook/kubernetes/project"
@@ -32,6 +33,8 @@ import (
 
 type kubernetesWebhooksServerOptions struct {
 	KubeConfig string
+	QPS        float32
+	Burst      int
 
 	MetricsBindAddress string
 	PprofBindAddress   string
@@ -63,6 +66,9 @@ func newKubernetesWebhooksServerCommand() *cobra.Command {
 
 func (o *kubernetesWebhooksServerOptions) complete() {
 	o.KubeConfig = os.GetEnv("KUBECONFIG", "")
+	o.QPS = types.MustParseFloat32(os.GetEnv("KUBE_API_QPS", "50.0"))
+	o.Burst = types.MustParseInt(os.GetEnv("KUBE_API_BURST", "300"))
+
 	o.MetricsBindAddress = os.GetEnv("METRICS_BIND_ADDRESS", "0")
 	o.PprofBindAddress = os.GetEnv("PPROF_BIND_ADDRESS", "")
 }
@@ -83,6 +89,7 @@ func (o *kubernetesWebhooksServerOptions) run(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("error getting REST config: %w", err)
 	}
+	kubernetes.ConfigureQPSBurst(ctx, restCfg, o.QPS, o.Burst)
 
 	scheme := runtime.NewScheme()
 	if err = corev1.AddToScheme(scheme); err != nil {
diff --git a/cmd/controlplane/management_controller.go b/cmd/controlplane/management_controller.go
index 03afd34b..d17367e1 100644
--- a/cmd/controlplane/management_controller.go
+++ b/cmd/controlplane/management_controller.go
@@ -28,6 +28,8 @@ import (
 
 type managementControllerOptions struct {
 	KubeConfig string
+	QPS        float32
+	Burst      int
 
 	KargoNamespace               string
 	ManageControllerRoleBindings bool
@@ -62,8 +64,12 @@ func newManagementControllerCommand() *cobra.Command {
 
 func (o *managementControllerOptions) complete() {
 	o.KubeConfig = os.GetEnv("KUBECONFIG", "")
+	o.QPS = types.MustParseFloat32(os.GetEnv("KUBE_API_QPS", "50.0"))
+	o.Burst = types.MustParseInt(os.GetEnv("KUBE_API_BURST", "300"))
+
 	o.KargoNamespace = os.GetEnv("KARGO_NAMESPACE", "kargo")
 	o.ManageControllerRoleBindings = types.MustParseBool(os.GetEnv("MANAGE_CONTROLLER_ROLE_BINDINGS", "true"))
+
 	o.MetricsBindAddress = os.GetEnv("METRICS_BIND_ADDRESS", "0")
 	o.PprofBindAddress = os.GetEnv("PPROF_BIND_ADDRESS", "")
 }
@@ -121,6 +127,7 @@ func (o *managementControllerOptions) setupManager(ctx context.Context) (manager
 	if err != nil {
 		return nil, fmt.Errorf("error loading REST config for Kargo controller manager: %w", err)
 	}
+	kubernetes.ConfigureQPSBurst(ctx, restCfg, o.QPS, o.Burst)
 	restCfg.ContentType = runtime.ContentTypeJSON
 
 	scheme := runtime.NewScheme()
diff --git a/docs/docs/20-quickstart/img/kargo-dashboard-projects.png b/docs/docs/20-quickstart/img/kargo-dashboard-projects.png
index f7a90158..6a0dcb70 100644
Binary files a/docs/docs/20-quickstart/img/kargo-dashboard-projects.png and b/docs/docs/20-quickstart/img/kargo-dashboard-projects.png differ
diff --git a/docs/docs/20-quickstart/img/kargo-promote-option.png b/docs/docs/20-quickstart/img/kargo-promote-option.png
index bf33188d..147caf55 100644
Binary files a/docs/docs/20-quickstart/img/kargo-promote-option.png and b/docs/docs/20-quickstart/img/kargo-promote-option.png differ
diff --git a/docs/docs/20-quickstart/img/kargo-stage-arrow-button.png b/docs/docs/20-quickstart/img/kargo-stage-arrow-button.png
deleted file mode 100644
index 6f38b519..00000000
Binary files a/docs/docs/20-quickstart/img/kargo-stage-arrow-button.png and /dev/null differ
diff --git a/docs/docs/20-quickstart/img/kargo-stage-staggered-bars-button.png b/docs/docs/20-quickstart/img/kargo-stage-staggered-bars-button.png
new file mode 100644
index 00000000..863c7307
Binary files /dev/null and b/docs/docs/20-quickstart/img/kargo-stage-staggered-bars-button.png differ
diff --git a/docs/docs/20-quickstart/index.md b/docs/docs/20-quickstart/index.md
index 88a11631..9dbf4b25 100644
--- a/docs/docs/20-quickstart/index.md
+++ b/docs/docs/20-quickstart/index.md
@@ -677,10 +677,10 @@ pipeline.
     
 Now is a good time to explore the UI some more.
 
-* To see more information about the `test` `Stage`, click the arrow icon in the
-  header of the `Stage`'s node in the pipeline:
+* To see more information about the `test` `Stage`, click the staggered bars
+  icon in the header of the `Stage`'s node in the pipeline:
 
-  ![Kargo Stage Arrow Button](img/kargo-stage-arrow-button.png)
+  ![Kargo Stage Staggered Bars Button](img/kargo-stage-staggered-bars-button.png)
 
   This view provides a comprehensive overview of the `Stage`s health, current
   `Freight`, histories of `Promotion`s and of `Freight` previously promoted to
diff --git a/go.mod b/go.mod
index eec3947d..e31651f2 100644
--- a/go.mod
+++ b/go.mod
@@ -78,6 +78,7 @@ require (
 	k8s.io/klog/v2 v2.130.1
 	k8s.io/utils v0.0.0-20250502105355-0f33e8f1c979
 	oras.land/oras-go v1.2.6
+	sigs.k8s.io/cli-utils v0.37.2
 	sigs.k8s.io/controller-runtime v0.20.4
 	sigs.k8s.io/kustomize/api v0.19.0
 	sigs.k8s.io/kustomize/kyaml v0.19.0
diff --git a/go.sum b/go.sum
index 0a3062be..5be29502 100644
--- a/go.sum
+++ b/go.sum
@@ -690,6 +690,8 @@ k8s.io/utils v0.0.0-20250502105355-0f33e8f1c979 h1:jgJW5IePPXLGB8e/1wvd0Ich9QE97
 k8s.io/utils v0.0.0-20250502105355-0f33e8f1c979/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 oras.land/oras-go v1.2.6 h1:z8cmxQXBU8yZ4mkytWqXfo6tZcamPwjsuxYU81xJ8Lk=
 oras.land/oras-go v1.2.6/go.mod h1:OVPc1PegSEe/K8YiLfosrlqlqTN9PUyFvOw5Y9gwrT8=
+sigs.k8s.io/cli-utils v0.37.2 h1:GOfKw5RV2HDQZDJlru5KkfLO1tbxqMoyn1IYUxqBpNg=
+sigs.k8s.io/cli-utils v0.37.2/go.mod h1:V+IZZr4UoGj7gMJXklWBg6t5xbdThFBcpj4MrZuCYco=
 sigs.k8s.io/controller-runtime v0.20.4 h1:X3c+Odnxz+iPTRobG4tp092+CvBU9UK0t/bRf+n0DGU=
 sigs.k8s.io/controller-runtime v0.20.4/go.mod h1:xg2XB0K5ShQzAgsoujxuKN4LNXR2LfwwHsPj7Iaw+XY=
 sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 h1:gBQPwqORJ8d8/YNZWEjoZs7npUVDpVXUUOFfW6CgAqE=
diff --git a/internal/server/kubernetes/client.go b/internal/server/kubernetes/client.go
index 1cd42972..172dfc48 100644
--- a/internal/server/kubernetes/client.go
+++ b/internal/server/kubernetes/client.go
@@ -21,6 +21,7 @@ import (
 	kubescheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
+	"sigs.k8s.io/cli-utils/pkg/flowcontrol"
 	libClient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 	libCluster "sigs.k8s.io/controller-runtime/pkg/cluster"
@@ -711,6 +712,27 @@ func GetRestConfig(ctx context.Context, path string) (*rest.Config, error) {
 	return cfg, nil
 }
 
+// ConfigureQPSBurst configures the provided REST config to use the specified
+// QPS and burst values, unless PriorityAndFairness flow control is enabled in
+// the cluster, in which case, it disables QPS and burst.
+//
+// For more information on PriorityAndFairness flow control, see:
+// https://kubernetes.io/docs/concepts/cluster-administration/flow-control/
+func ConfigureQPSBurst(ctx context.Context, cfg *rest.Config, qps float32, burst int) {
+	logger := logging.LoggerFromContext(ctx)
+
+	if ok, err := flowcontrol.IsEnabled(ctx, cfg); err != nil && ok {
+		logger.Debug("PriorityAndFairness flow control is enabled; disabling QPS and burst")
+		cfg.QPS = -1
+		cfg.Burst = -1
+		return
+	}
+
+	logger.Debug("configuring QPS and burst for REST config", "qps", qps, "burst", burst)
+	cfg.QPS = qps
+	cfg.Burst = burst
+}
+
 // gvrAndKeyFromObj extracts the group, version, and plural resource type
 // information from the provided object.
 func gvrAndKeyFromObj(
diff --git a/internal/types/types.go b/internal/types/types.go
index f21d7c0c..e3dccc8f 100644
--- a/internal/types/types.go
+++ b/internal/types/types.go
@@ -2,6 +2,8 @@ package types
 
 import "strconv"
 
+// MustParseBool parses a string into a boolean value, panicking if the string
+// cannot be parsed as a boolean.
 func MustParseBool(s string) bool {
 	b, err := strconv.ParseBool(s)
 	if err != nil {
@@ -9,3 +11,23 @@ func MustParseBool(s string) bool {
 	}
 	return b
 }
+
+// MustParseFloat32 parses a string into a float32 value, panicking if the
+// string cannot be parsed as a float32.
+func MustParseFloat32(s string) float32 {
+	f, err := strconv.ParseFloat(s, 32)
+	if err != nil {
+		panic(err)
+	}
+	return float32(f)
+}
+
+// MustParseInt parses a string into an int value, panicking if the string
+// cannot be parsed as an int.
+func MustParseInt(s string) int {
+	i, err := strconv.Atoi(s)
+	if err != nil {
+		panic(err)
+	}
+	return i
+}



Test patch:
diff --git a/internal/types/types_test.go b/internal/types/types_test.go
index e642b905..06a9fa3d 100644
--- a/internal/types/types_test.go
+++ b/internal/types/types_test.go
@@ -6,7 +6,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestParseBool(t *testing.T) {
+func TestMustParseBool(t *testing.T) {
 	t.Parallel()
 	testSets := map[string]struct {
 		Input     string
@@ -38,3 +38,61 @@ func TestParseBool(t *testing.T) {
 		})
 	}
 }
+
+func TestMustParseFloat32(t *testing.T) {
+	t.Parallel()
+	testSets := map[string]struct {
+		Input     string
+		Expected  float32
+		MustPanic bool
+	}{
+		"valid float32": {
+			Input:    "1.23",
+			Expected: 1.23,
+		},
+		"invalid float32": {
+			Input:     "float",
+			MustPanic: true,
+		},
+	}
+	for name, ts := range testSets {
+		t.Run(name, func(t *testing.T) {
+			if ts.MustPanic {
+				require.Panics(t, func() {
+					_ = MustParseFloat32(ts.Input)
+				})
+			} else {
+				require.Equal(t, ts.Expected, MustParseFloat32(ts.Input))
+			}
+		})
+	}
+}
+
+func TestMustParseInt(t *testing.T) {
+	t.Parallel()
+	testSets := map[string]struct {
+		Input     string
+		Expected  int
+		MustPanic bool
+	}{
+		"valid int": {
+			Input:    "123",
+			Expected: 123,
+		},
+		"invalid int": {
+			Input:     "int",
+			MustPanic: true,
+		},
+	}
+	for name, ts := range testSets {
+		t.Run(name, func(t *testing.T) {
+			if ts.MustPanic {
+				require.Panics(t, func() {
+					_ = MustParseInt(ts.Input)
+				})
+			} else {
+				require.Equal(t, ts.Expected, MustParseInt(ts.Input))
+			}
+		})
+	}
+}