Add flags for QPS and Burst to *-poollet and *-broker (#1157)

* Add QPS and Burst configuration for Kubernetes client in controller optionsUpdated `Options` struct with `QPS` and `Burst` fields, added corresponding flags in `AddFlags` method, and set these values in the `Run` function for ippoollet, machinepoollet, and volumepoollet controllers. This allows configurable rate limits for Kubernetes client requests and logs the settings on startup.Suggestion: Review the changes to ensure consistency and test the new configuration options. Signed-off-by: Benjamin Alpert <[email protected]> change poollet client specific qps and burst opts implementation to common options pattern change to client config defaults for qps and burst * Bump github.com/onsi/gomega from 1.35.1 to 1.36.0 (#1173) Bumps [github.com/onsi/gomega](https://github.com/onsi/gomega) from 1.35.1 to 1.36.0. - [Release notes](https://github.com/onsi/gomega/releases) - [Changelog](https://github.com/onsi/gomega/blob/master/CHANGELOG.md) - [Commits](onsi/gomega@v1.35.1...v1.36.0) --- updated-dependencies: - dependency-name: github.com/onsi/gomega dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * Bump github.com/bits-and-blooms/bitset from 1.15.0 to 1.17.0 (#1174) Bumps [github.com/bits-and-blooms/bitset](https://github.com/bits-and-blooms/bitset) from 1.15.0 to 1.17.0. - [Release notes](https://github.com/bits-and-blooms/bitset/releases) - [Commits](bits-and-blooms/bitset@v1.15.0...v1.17.0) --- updated-dependencies: - dependency-name: github.com/bits-and-blooms/bitset dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> --------- Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
ironcore-dev · Nov 27, 2024 · dc782b5 · dc782b5
1 parent 5c1ded1
commit dc782b5
Show file tree

Hide file tree

Showing 12 changed files with 186 additions and 65 deletions.
diff --git a/broker/bucketbroker/cmd/bucketbroker/app/app.go b/broker/bucketbroker/cmd/bucketbroker/app/app.go
@@ -9,13 +9,16 @@ import (
 	"fmt"
 	"net"
 
-	"github.com/ironcore-dev/controller-utils/configutils"
-	"github.com/ironcore-dev/ironcore/broker/bucketbroker/server"
-	"github.com/ironcore-dev/ironcore/broker/common"
-	iri "github.com/ironcore-dev/ironcore/iri/apis/bucket/v1alpha1"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	"google.golang.org/grpc"
+
+	"github.com/ironcore-dev/ironcore/broker/bucketbroker/server"
+	"github.com/ironcore-dev/ironcore/broker/common"
+	iri "github.com/ironcore-dev/ironcore/iri/apis/bucket/v1alpha1"
+	"github.com/ironcore-dev/ironcore/utils/client/config"
+
+	"github.com/ironcore-dev/controller-utils/configutils"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 )
@@ -24,6 +27,9 @@ type Options struct {
 	Kubeconfig string
 	Address    string
 
+	QPS   float32
+	Burst int
+
 	Namespace          string
 	BucketPoolName     string
 	BucketPoolSelector map[string]string
@@ -33,6 +39,9 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&o.Kubeconfig, "kubeconfig", o.Kubeconfig, "Path pointing to a kubeconfig file to use.")
 	fs.StringVar(&o.Address, "address", "/var/run/iri-bucketbroker.sock", "Address to listen on.")
 
+	fs.Float32Var(&o.QPS, "qps", config.QPS, "Kubernetes client qps.")
+	fs.IntVar(&o.Burst, "burst", config.Burst, "Kubernetes client burst.")
+
 	fs.StringVar(&o.Namespace, "namespace", o.Namespace, "Target Kubernetes namespace to use.")
 	fs.StringVar(&o.BucketPoolName, "bucket-pool-name", o.BucketPoolName, "Name of the target bucket pool to pin buckets to, if any.")
 	fs.StringToStringVar(&o.BucketPoolSelector, "bucket-pool-selector", o.BucketPoolSelector, "Selector of the target bucket pools to pin buckets to, if any.")
@@ -74,6 +83,10 @@ func Run(ctx context.Context, opts Options) error {
 		return err
 	}
 
+	cfg.QPS = opts.QPS
+	cfg.Burst = opts.Burst
+	setupLog.Info("Kubernetes Client configuration", "QPS", cfg.QPS, "Burst", cfg.Burst)
+
 	srv, err := server.New(cfg, server.Options{
 		Namespace:          opts.Namespace,
 		BucketPoolName:     opts.BucketPoolName,

diff --git a/broker/machinebroker/cmd/machinebroker/app/app.go b/broker/machinebroker/cmd/machinebroker/app/app.go
@@ -13,16 +13,19 @@ import (
 	"net/url"
 
 	"github.com/go-logr/logr"
-	"github.com/ironcore-dev/controller-utils/configutils"
+	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc"
+
 	"github.com/ironcore-dev/ironcore/broker/common"
 	commongrpc "github.com/ironcore-dev/ironcore/broker/common/grpc"
 	machinebrokerhttp "github.com/ironcore-dev/ironcore/broker/machinebroker/http"
 	"github.com/ironcore-dev/ironcore/broker/machinebroker/server"
 	iri "github.com/ironcore-dev/ironcore/iri/apis/machine/v1alpha1"
-	"github.com/spf13/cobra"
-	"github.com/spf13/pflag"
-	"golang.org/x/sync/errgroup"
-	"google.golang.org/grpc"
+	"github.com/ironcore-dev/ironcore/utils/client/config"
+
+	"github.com/ironcore-dev/controller-utils/configutils"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 )
@@ -34,6 +37,9 @@ type Options struct {
 	BaseURL                 string
 	BrokerDownwardAPILabels map[string]string
 
+	QPS   float32
+	Burst int
+
 	Namespace           string
 	MachinePoolName     string
 	MachinePoolSelector map[string]string
@@ -48,6 +54,9 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	fs.StringToStringVar(&o.BrokerDownwardAPILabels, "broker-downward-api-label", nil, "The labels to broker via downward API. "+
 		"Example is for instance to broker \"root-machine-uid\" initially obtained via \"machinepoollet.ironcore.dev/machine-uid\".")
 
+	fs.Float32Var(&o.QPS, "qps", config.QPS, "Kubernetes client qps.")
+	fs.IntVar(&o.Burst, "burst", config.Burst, "Kubernetes client burst.")
+
 	fs.StringVar(&o.Namespace, "namespace", o.Namespace, "Target Kubernetes namespace to use.")
 	fs.StringVar(&o.MachinePoolName, "machine-pool-name", o.MachinePoolName, "Name of the target machine pool to pin machines to, if any.")
 	fs.StringToStringVar(&o.MachinePoolSelector, "machine-pool-selector", o.MachinePoolSelector, "Selector of the target machine pools to pin machines to, if any.")
@@ -102,6 +111,10 @@ func Run(ctx context.Context, opts Options) error {
 		baseURL = u.String()
 	}
 
+	cfg.QPS = opts.QPS
+	cfg.Burst = opts.Burst
+	setupLog.Info("Kubernetes Client configuration", "QPS", cfg.QPS, "Burst", cfg.Burst)
+
 	log.V(1).Info("Creating server",
 		"Namespace", opts.Namespace,
 		"MachinePoolName", opts.MachinePoolName,
@@ -142,7 +155,7 @@ func runServer(ctx context.Context, setupLog, log logr.Logger, srv *server.Serve
 	return nil
 }
 
-func runGRPCServer(ctx context.Context, setupLog logr.Logger, log logr.Logger, srv *server.Server, opts Options) error {
+func runGRPCServer(ctx context.Context, setupLog, log logr.Logger, srv *server.Server, opts Options) error {
 	log.V(1).Info("Cleaning up any previous socket")
 	if err := common.CleanupSocketIfExists(opts.Address); err != nil {
 		return fmt.Errorf("error cleaning up socket: %w", err)

diff --git a/broker/volumebroker/cmd/volumebroker/app/app.go b/broker/volumebroker/cmd/volumebroker/app/app.go
@@ -9,13 +9,16 @@ import (
 	"fmt"
 	"net"
 
-	"github.com/ironcore-dev/controller-utils/configutils"
-	"github.com/ironcore-dev/ironcore/broker/common"
-	"github.com/ironcore-dev/ironcore/broker/volumebroker/server"
-	iri "github.com/ironcore-dev/ironcore/iri/apis/volume/v1alpha1"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	"google.golang.org/grpc"
+
+	"github.com/ironcore-dev/ironcore/broker/common"
+	"github.com/ironcore-dev/ironcore/broker/volumebroker/server"
+	iri "github.com/ironcore-dev/ironcore/iri/apis/volume/v1alpha1"
+	"github.com/ironcore-dev/ironcore/utils/client/config"
+
+	"github.com/ironcore-dev/controller-utils/configutils"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 )
@@ -24,6 +27,9 @@ type Options struct {
 	Kubeconfig string
 	Address    string
 
+	QPS   float32
+	Burst int
+
 	Namespace          string
 	VolumePoolName     string
 	VolumePoolSelector map[string]string
@@ -33,6 +39,9 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&o.Kubeconfig, "kubeconfig", o.Kubeconfig, "Path pointing to a kubeconfig file to use.")
 	fs.StringVar(&o.Address, "address", "/var/run/iri-volumebroker.sock", "Address to listen on.")
 
+	fs.Float32Var(&o.QPS, "qps", config.QPS, "Kubernetes client qps.")
+	fs.IntVar(&o.Burst, "burst", config.Burst, "Kubernetes client burst.")
+
 	fs.StringVar(&o.Namespace, "namespace", o.Namespace, "Target Kubernetes namespace to use.")
 	fs.StringVar(&o.VolumePoolName, "volume-pool-name", o.VolumePoolName, "Name of the target volume pool to pin volumes to, if any.")
 	fs.StringToStringVar(&o.VolumePoolSelector, "volume-pool-selector", o.VolumePoolSelector, "Selector of the target volume pools to pin volumes to, if any.")
@@ -74,6 +83,10 @@ func Run(ctx context.Context, opts Options) error {
 		return err
 	}
 
+	cfg.QPS = opts.QPS
+	cfg.Burst = opts.Burst
+	setupLog.Info("Kubernetes Client configuration", "QPS", cfg.QPS, "Burst", cfg.Burst)
+
 	srv, err := server.New(cfg, server.Options{
 		Namespace:          opts.Namespace,
 		VolumePoolName:     opts.VolumePoolName,

diff --git a/poollet/bucketpoollet/cmd/bucketpoollet/app/app.go b/poollet/bucketpoollet/cmd/bucketpoollet/app/app.go
@@ -65,6 +65,8 @@ type Options struct {
 	RelistThreshold time.Duration
 
 	WatchFilterValue string
+
+	MaxConcurrentReconciles int
 }
 
 func (o *Options) AddFlags(fs *pflag.FlagSet) {
@@ -89,6 +91,8 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	fs.DurationVar(&o.RelistThreshold, "relist-threshold", 3*time.Minute, "event channel relisting threshold.")
 
 	fs.StringVar(&o.WatchFilterValue, "watch-filter", "", "Value to filter for while watching.")
+
+	fs.IntVar(&o.MaxConcurrentReconciles, "max-concurrent-reconciles", 1, "Maximum number of concurrent reconciles.")
 }
 
 func (o *Options) MarkFlagsRequired(cmd *cobra.Command) {
@@ -150,7 +154,8 @@ func Run(ctx context.Context, opts Options) error {
 		return fmt.Errorf("error getting config: %w", err)
 	}
 
-	setupLog.Info("IRI client config", "ChannelCapacity", opts.ChannelCapacity, "RelistPeriod", opts.RelistPeriod, "RelistThreshold", opts.RelistThreshold)
+	setupLog.Info("IRI Client configuration", "ChannelCapacity", opts.ChannelCapacity, "RelistPeriod", opts.RelistPeriod, "RelistThreshold", opts.RelistThreshold)
+	setupLog.Info("Kubernetes Client configuration", "QPS", cfg.QPS, "Burst", cfg.Burst)
 
 	leaderElectionCfg, err := configutils.GetConfig(
 		configutils.Kubeconfig(opts.LeaderElectionKubeconfig),
@@ -213,13 +218,14 @@ func Run(ctx context.Context, opts Options) error {
 		}
 
 		if err := (&controllers.BucketReconciler{
-			EventRecorder:     mgr.GetEventRecorderFor("buckets"),
-			Client:            mgr.GetClient(),
-			Scheme:            scheme,
-			BucketRuntime:     bucketRuntime,
-			BucketClassMapper: bucketClassMapper,
-			BucketPoolName:    opts.BucketPoolName,
-			WatchFilterValue:  opts.WatchFilterValue,
+			EventRecorder:           mgr.GetEventRecorderFor("buckets"),
+			Client:                  mgr.GetClient(),
+			Scheme:                  scheme,
+			BucketRuntime:           bucketRuntime,
+			BucketClassMapper:       bucketClassMapper,
+			BucketPoolName:          opts.BucketPoolName,
+			WatchFilterValue:        opts.WatchFilterValue,
+			MaxConcurrentReconciles: opts.MaxConcurrentReconciles,
 		}).SetupWithManager(mgr); err != nil {
 			return fmt.Errorf("error setting up bucket reconciler with manager: %w", err)
 		}

diff --git a/poollet/bucketpoollet/controllers/bucket_controller.go b/poollet/bucketpoollet/controllers/bucket_controller.go
@@ -10,19 +10,20 @@ import (
 	"fmt"
 
 	"github.com/go-logr/logr"
-	"github.com/ironcore-dev/controller-utils/clientutils"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
 	storagev1alpha1 "github.com/ironcore-dev/ironcore/api/storage/v1alpha1"
 	iriBucket "github.com/ironcore-dev/ironcore/iri/apis/bucket"
 	iri "github.com/ironcore-dev/ironcore/iri/apis/bucket/v1alpha1"
-
 	irimeta "github.com/ironcore-dev/ironcore/iri/apis/meta/v1alpha1"
 	bucketpoolletv1alpha1 "github.com/ironcore-dev/ironcore/poollet/bucketpoollet/api/v1alpha1"
 	"github.com/ironcore-dev/ironcore/poollet/bucketpoollet/bcm"
 	"github.com/ironcore-dev/ironcore/poollet/bucketpoollet/controllers/events"
 	ironcoreclient "github.com/ironcore-dev/ironcore/utils/client"
 	"github.com/ironcore-dev/ironcore/utils/predicates"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
+
+	"github.com/ironcore-dev/controller-utils/clientutils"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -32,6 +33,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
@@ -47,6 +49,8 @@ type BucketReconciler struct {
 
 	BucketPoolName   string
 	WatchFilterValue string
+
+	MaxConcurrentReconciles int
 }
 
 func (r *BucketReconciler) iriBucketLabels(bucket *storagev1alpha1.Bucket) map[string]string {
@@ -432,7 +436,6 @@ func (r *BucketReconciler) updateStatus(ctx context.Context, log logr.Logger, bu
 				Endpoint:  iriAccess.Endpoint,
 			}
 		}
-
 	}
 
 	base := bucket.DeepCopy()
@@ -466,6 +469,10 @@ func (r *BucketReconciler) SetupWithManager(mgr ctrl.Manager) error {
 				predicates.ResourceIsNotExternallyManaged(log),
 			),
 		).
+		WithOptions(
+			controller.Options{
+				MaxConcurrentReconciles: r.MaxConcurrentReconciles,
+			}).
 		Complete(r)
 }
 

diff --git a/poollet/bucketpoollet/controllers/controllers_suite_test.go b/poollet/bucketpoollet/controllers/controllers_suite_test.go
@@ -8,8 +8,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/ironcore-dev/controller-utils/buildutils"
-	"github.com/ironcore-dev/controller-utils/modutils"
 	corev1alpha1 "github.com/ironcore-dev/ironcore/api/core/v1alpha1"
 	storagev1alpha1 "github.com/ironcore-dev/ironcore/api/storage/v1alpha1"
 	storageclient "github.com/ironcore-dev/ironcore/internal/client/storage"
@@ -22,8 +20,9 @@ import (
 	"github.com/ironcore-dev/ironcore/utils/envtest/apiserver"
 	"github.com/ironcore-dev/ironcore/utils/envtest/controllermanager"
 	"github.com/ironcore-dev/ironcore/utils/envtest/process"
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
+
+	"github.com/ironcore-dev/controller-utils/buildutils"
+	"github.com/ironcore-dev/controller-utils/modutils"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -34,10 +33,13 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
-	. "sigs.k8s.io/controller-runtime/pkg/envtest/komega"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	metricserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	. "sigs.k8s.io/controller-runtime/pkg/envtest/komega"
 )
 
 var (

diff --git a/poollet/machinepoollet/cmd/machinepoollet/app/app.go b/poollet/machinepoollet/cmd/machinepoollet/app/app.go
@@ -81,6 +81,8 @@ type Options struct {
 	AddressesOptions addresses.GetOptions
 
 	WatchFilterValue string
+
+	MaxConcurrentReconciles int
 }
 
 func (o *Options) AddFlags(fs *pflag.FlagSet) {
@@ -111,6 +113,8 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	o.AddressesOptions.BindFlags(fs)
 
 	fs.StringVar(&o.WatchFilterValue, "watch-filter", "", "Value to filter for while watching.")
+
+	fs.IntVar(&o.MaxConcurrentReconciles, "max-concurrent-reconciles", 1, "Maximum number of concurrent reconciles.")
 }
 
 func (o *Options) MarkFlagsRequired(cmd *cobra.Command) {
@@ -206,7 +210,8 @@ func Run(ctx context.Context, opts Options) error {
 		return fmt.Errorf("error getting config: %w", err)
 	}
 
-	setupLog.Info("IRI client config", "ChannelCapacity", opts.ChannelCapacity, "RelistPeriod", opts.RelistPeriod, "RelistThreshold", opts.RelistThreshold)
+	setupLog.Info("IRI Client configuration", "ChannelCapacity", opts.ChannelCapacity, "RelistPeriod", opts.RelistPeriod, "RelistThreshold", opts.RelistThreshold)
+	setupLog.Info("Kubernetes Client configuration", "QPS", cfg.QPS, "Burst", cfg.Burst)
 
 	leaderElectionCfg, err := configutils.GetConfig(
 		configutils.Kubeconfig(opts.LeaderElectionKubeconfig),
@@ -313,16 +318,17 @@ func Run(ctx context.Context, opts Options) error {
 		}
 
 		if err := (&controllers.MachineReconciler{
-			EventRecorder:          mgr.GetEventRecorderFor("machines"),
-			Client:                 mgr.GetClient(),
-			MachineRuntime:         machineRuntime,
-			MachineRuntimeName:     version.RuntimeName,
-			MachineRuntimeVersion:  version.RuntimeVersion,
-			MachineClassMapper:     machineClassMapper,
-			MachinePoolName:        opts.MachinePoolName,
-			DownwardAPILabels:      opts.MachineDownwardAPILabels,
-			DownwardAPIAnnotations: opts.MachineDownwardAPIAnnotations,
-			WatchFilterValue:       opts.WatchFilterValue,
+			EventRecorder:           mgr.GetEventRecorderFor("machines"),
+			Client:                  mgr.GetClient(),
+			MachineRuntime:          machineRuntime,
+			MachineRuntimeName:      version.RuntimeName,
+			MachineRuntimeVersion:   version.RuntimeVersion,
+			MachineClassMapper:      machineClassMapper,
+			MachinePoolName:         opts.MachinePoolName,
+			DownwardAPILabels:       opts.MachineDownwardAPILabels,
+			DownwardAPIAnnotations:  opts.MachineDownwardAPIAnnotations,
+			WatchFilterValue:        opts.WatchFilterValue,
+			MaxConcurrentReconciles: opts.MaxConcurrentReconciles,
 		}).SetupWithManager(mgr); err != nil {
 			return fmt.Errorf("error setting up machine reconciler with manager: %w", err)
 		}