Add command-line flags to allow setting client rate limits (QPS/Burst) (#2797)

* Add `-clientRateLimitQPS` and `-clientRateLimitBurst` flags to allow controlling client rate limits.

Signed-off-by: Sebastian Widmer <sebastian.widmer@vshn.net>

* Return error if QPS is higher than max value  of float32

Signed-off-by: Sebastian Widmer <sebastian.widmer@vshn.net>

cmd/initContainer/main.go

@@ -6,7 +6,6 @@ package main
 import (
 	"context"
 	"flag"
-	"fmt"
 	"os"
 	"sync"
 	"time"
@@ -20,16 +19,16 @@ import (
 	coord "k8s.io/api/coordination/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	rest "k8s.io/client-go/rest"
-	clientcmd "k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
 	"k8s.io/klog/v2/klogr"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 var (
-	kubeconfig string
-	setupLog   = log.Log.WithName("setup")
+	kubeconfig           string
+	setupLog             = log.Log.WithName("setup")
+	clientRateLimitQPS   float64
+	clientRateLimitBurst int
 )
 
 const (
@@ -46,6 +45,8 @@ func main() {
 	log.SetLogger(klogr.New())
 	// arguments
 	flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
+	flag.Float64Var(&clientRateLimitQPS, "clientRateLimitQPS", 0, "Configure the maximum QPS to the master from Kyverno. Uses the client default if zero.")
+	flag.IntVar(&clientRateLimitBurst, "clientRateLimitBurst", 0, "Configure the maximum burst for throttle. Uses the client default if zero.")
 	if err := flag.Set("v", "2"); err != nil {
 		klog.Fatalf("failed to set log level: %v", err)
 	}
@@ -55,7 +56,7 @@ func main() {
 	// os signal handler
 	stopCh := signal.SetupSignalHandler()
 	// create client config
-	clientConfig, err := createClientConfig(kubeconfig)
+	clientConfig, err := config.CreateClientConfig(kubeconfig, clientRateLimitQPS, clientRateLimitBurst, log.Log)
 	if err != nil {
 		setupLog.Error(err, "Failed to build kubeconfig")
 		os.Exit(1)
@@ -69,7 +70,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	pclientConfig, err := config.CreateClientConfig(kubeconfig, log.Log)
+	pclientConfig, err := config.CreateClientConfig(kubeconfig, clientRateLimitQPS, clientRateLimitBurst, log.Log)
 	if err != nil {
 		setupLog.Error(err, "Failed to build client config")
 		os.Exit(1)
@@ -182,17 +183,6 @@ func executeRequest(client *client.Client, pclient *kyvernoclient.Clientset, req
 	return nil
 }
 
-func createClientConfig(kubeconfig string) (*rest.Config, error) {
-	logger := log.Log
-	if kubeconfig == "" {
-		logger.Info("Using in-cluster configuration")
-		return rest.InClusterConfig()
-	}
-
-	logger.Info(fmt.Sprintf("Using configuration from '%s'", kubeconfig))
-	return clientcmd.BuildConfigFromFlags("", kubeconfig)
-}
-
 type request struct {
 	kind string
 	name string

cmd/kyverno/main.go

@@ -64,6 +64,8 @@ var (
 	policyControllerResyncPeriod time.Duration
 	imagePullSecrets             string
 	imageSignatureRepository     string
+	clientRateLimitQPS           float64
+	clientRateLimitBurst         int
 	setupLog                     = log.Log.WithName("setup")
 )
 
@@ -96,6 +98,8 @@ func main() {
 	flag.StringVar(&imagePullSecrets, "imagePullSecrets", "", "Secret resource names for image registry access credentials.")
 	flag.StringVar(&imageSignatureRepository, "imageSignatureRepository", "", "Alternate repository for image signatures. Can be overridden per rule via `verifyImages.Repository`.")
 	flag.BoolVar(&autoUpdateWebhooks, "autoUpdateWebhooks", true, "Set this flag to 'false' to disable auto-configuration of the webhook.")
+	flag.Float64Var(&clientRateLimitQPS, "clientRateLimitQPS", 0, "Configure the maximum QPS to the master from Kyverno. Uses the client default if zero.")
+	flag.IntVar(&clientRateLimitBurst, "clientRateLimitBurst", 0, "Configure the maximum burst for throttle. Uses the client default if zero.")
 
 	if err := flag.Set("v", "2"); err != nil {
 		setupLog.Error(err, "failed to set log level")
@@ -107,7 +111,7 @@ func main() {
 	version.PrintVersionInfo(log.Log)
 	cleanUp := make(chan struct{})
 	stopCh := signal.SetupSignalHandler()
-	clientConfig, err := config.CreateClientConfig(kubeconfig, log.Log)
+	clientConfig, err := config.CreateClientConfig(kubeconfig, clientRateLimitQPS, clientRateLimitBurst, log.Log)
 	if err != nil {
 		setupLog.Error(err, "Failed to build kubeconfig")
 		os.Exit(1)

pkg/config/config.go

@@ -1,6 +1,8 @@
 package config
 
 import (
+	"fmt"
+	"math"
 	"os"
 
 	"github.com/go-logr/logr"
@@ -97,14 +99,31 @@ var (
 	ReadinessServicePath = "/health/readiness"
 )
 
-//CreateClientConfig creates client config
-func CreateClientConfig(kubeconfig string, log logr.Logger) (*rest.Config, error) {
+//CreateClientConfig creates client config and applies rate limit QPS and burst
+func CreateClientConfig(kubeconfig string, qps float64, burst int, log logr.Logger) (*rest.Config, error) {
 	logger := log.WithName("CreateClientConfig")
+
+	clientConfig, err := createClientConfig(kubeconfig, logger)
+	if err != nil {
+		return nil, err
+	}
+
+	if qps > math.MaxFloat32 {
+		return nil, fmt.Errorf("client rate limit QPS must not be higher than %e", math.MaxFloat32)
+	}
+	clientConfig.Burst = burst
+	clientConfig.QPS = float32(qps)
+
+	return clientConfig, nil
+}
+
+// createClientConfig creates client config
+func createClientConfig(kubeconfig string, log logr.Logger) (*rest.Config, error) {
 	if kubeconfig == "" {
-		logger.Info("Using in-cluster configuration")
+		log.Info("Using in-cluster configuration")
 		return rest.InClusterConfig()
 	}
-	logger.V(4).Info("Using specified kubeconfig", "kubeconfig", kubeconfig)
+	log.V(4).Info("Using specified kubeconfig", "kubeconfig", kubeconfig)
 	return clientcmd.BuildConfigFromFlags("", kubeconfig)
 }
 

pkg/config/config_test.go

@@ -0,0 +1,72 @@
+package config_test
+
+import (
+	"math"
+	"os"
+	"testing"
+
+	"github.com/go-logr/logr"
+	"gotest.tools/assert"
+	"k8s.io/apimachinery/pkg/runtime"
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
+	clientcmdlatest "k8s.io/client-go/tools/clientcmd/api/latest"
+
+	"github.com/kyverno/kyverno/pkg/config"
+)
+
+func Test_CreateClientConfig_WithKubeConfig(t *testing.T) {
+	cf := createMinimalKubeconfig(t)
+	defer os.Remove(cf)
+	_, err := config.CreateClientConfig(cf, 0, 0, logr.Discard())
+	assert.NilError(t, err)
+}
+
+func Test_CreateClientConfig_SetBurstQPS(t *testing.T) {
+	const (
+		qps   = 55
+		burst = 99
+	)
+
+	cf := createMinimalKubeconfig(t)
+	defer os.Remove(cf)
+	c, err := config.CreateClientConfig(cf, qps, burst, logr.Discard())
+	assert.NilError(t, err)
+	assert.Equal(t, float32(qps), c.QPS)
+	assert.Equal(t, burst, c.Burst)
+}
+
+func Test_CreateClientConfig_LimitQPStoFloat32(t *testing.T) {
+	qps := float64(math.MaxFloat32) * 2
+
+	cf := createMinimalKubeconfig(t)
+	defer os.Remove(cf)
+	_, err := config.CreateClientConfig(cf, qps, 0, logr.Discard())
+	assert.ErrorContains(t, err, "QPS")
+}
+
+func createMinimalKubeconfig(t *testing.T) string {
+	t.Helper()
+
+	minimalConfig := clientcmdapi.Config{
+		Clusters: map[string]*clientcmdapi.Cluster{
+			"test": {Server: "http://localhost:7777"},
+		},
+		AuthInfos: map[string]*clientcmdapi.AuthInfo{
+			"test": {},
+		},
+		Contexts: map[string]*clientcmdapi.Context{
+			"test": {AuthInfo: "test", Cluster: "test"},
+		},
+		CurrentContext: "test",
+	}
+
+	f, err := os.CreateTemp("", "")
+	assert.NilError(t, err)
+	enc, err := runtime.Encode(clientcmdlatest.Codec, &minimalConfig)
+	assert.NilError(t, err)
+	_, err = f.Write(enc)
+	assert.NilError(t, err)
+	assert.NilError(t, f.Close())
+
+	return f.Name()
+}