✨ Expose QPS and Burst flags on agent command line

pkg/agent/manager.go

@@ -24,6 +24,7 @@ import (
  // Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
  // to ensure that exec-entrypoint and run can make use of them.
  "github.com/spf13/cobra"
+ "github.com/spf13/pflag"
  _ "k8s.io/client-go/plugin/pkg/client/auth"
  "k8s.io/client-go/rest"
  "k8s.io/component-base/version"
@@ -38,6 +39,7 @@ import (
  "sigs.k8s.io/controller-runtime/pkg/log/zap"
 
  v1alpha1 "github.com/kubestellar/ocm-status-addon/api/v1alpha1"
+ clientopts "github.com/kubestellar/ocm-status-addon/pkg/client-options"
 )
 
 var (
@@ -79,11 +81,18 @@ type AgentOptions struct {
  SpokeClusterName string
  AddonName string
  AddonNamespace string
+ LocalLimits clientopts.ClientLimits[*pflag.FlagSet]
+ HubLimits clientopts.ClientLimits[*pflag.FlagSet]
 }
 
 // NewAgentOptions returns the flags with default value set
 func NewAgentOptions(addonName string) *AgentOptions {
- return &AgentOptions{ZapOpts: zap.Options{Development: true}, AddonName: addonName}
+ return &AgentOptions{
+ ZapOpts: zap.Options{Development: true},
+ AddonName: addonName,
+ LocalLimits: clientopts.NewClientLimits[*pflag.FlagSet]("local", "accessing the local cluster"),
+ HubLimits: clientopts.NewClientLimits[*pflag.FlagSet]("hub", "accessing the hub"),
+ }
 }
 
 func (o *AgentOptions) AddFlags(cmd *cobra.Command) {
@@ -102,6 +111,8 @@ func (o *AgentOptions) AddFlags(cmd *cobra.Command) {
  flag.BoolVar(&o.EnableLeaderElection, "leader-elect", false,
  "Enable leader election for controller manager. "+
  "Enabling this will ensure there is only one active controller manager.")
+ o.LocalLimits.AddFlags(flags)
+ o.HubLimits.AddFlags(flags)
 }
 
 func (o *AgentOptions) RunAgent(ctx context.Context, kubeconfig *rest.Config) error {
@@ -112,6 +123,7 @@ func (o *AgentOptions) RunAgent(ctx context.Context, kubeconfig *rest.Config) er
  // setup manager
  // manager here is mainly used for leader election and health checks
  managedConfig := ctrl.GetConfigOrDie()
+ managedConfig = o.LocalLimits.LimitConfig(managedConfig)
  mgr, err := ctrl.NewManager(managedConfig, ctrl.Options{
  Scheme: scheme,
  MetricsBindAddress: o.MetricsAddr,
@@ -150,6 +162,7 @@ func (o *AgentOptions) RunAgent(ctx context.Context, kubeconfig *rest.Config) er
  setupLog.Error(err, "could not build resr.Config")
  os.Exit(1)
  }
+ hubConfig = o.HubLimits.LimitConfig(hubConfig)
 
  // start the agent
  agent, err := NewAgent(mgr, managedConfig, hubConfig, o.SpokeClusterName, o.AddonName)

pkg/client-options/client-options.go

@@ -0,0 +1,88 @@
+/*
+Copyright 2023 The KubeStellar Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package clientopts
+
+import (
+ "k8s.io/client-go/rest"
+ "k8s.io/client-go/tools/clientcmd"
+ "k8s.io/client-go/util/flowcontrol"
+)
+
+type FlagSet interface {
+ Float64Var(p *float64, name string, value float64, usage string)
+ IntVar(p *int, name string, value int, usage string)
+ StringVar(p *string, name string, value string, usage string)
+}
+
+type ClientLimits[FS FlagSet] struct {
+ name string
+ description string
+ QPS float64
+ Burst int
+}
+
+type ClientOptions[FS FlagSet] struct {
+ ClientLimits[FS]
+ loadingRules *clientcmd.ClientConfigLoadingRules
+ overrides clientcmd.ConfigOverrides
+}
+
+func NewClientLimits[FS FlagSet](name, description string) ClientLimits[FS] {
+ return ClientLimits[FS]{
+ name: name,
+ description: description,
+ QPS: float64(rest.DefaultQPS),
+ Burst: rest.DefaultBurst,
+ }
+}
+
+func NewClientOptions[FS FlagSet](name string, description string) *ClientOptions[FS] {
+ return &ClientOptions[FS]{
+ ClientLimits: NewClientLimits[FS](name, description),
+ loadingRules: clientcmd.NewDefaultClientConfigLoadingRules(),
+ overrides: clientcmd.ConfigOverrides{},
+ }
+}
+
+func (opts *ClientLimits[FS]) AddFlags(flags FS) {
+ flags.Float64Var(&opts.QPS, opts.name+"-qps", opts.QPS, "Max average requests/sec for "+opts.description)
+ flags.IntVar(&opts.Burst, opts.name+"-burst", opts.Burst, "Allowed burst in requests/sec for "+opts.description)
+}
+
+func (opts *ClientOptions[FS]) AddFlags(flags FS) {
+ opts.ClientLimits.AddFlags(flags)
+ flags.StringVar(&opts.loadingRules.ExplicitPath, opts.name+"-kubeconfig", opts.loadingRules.ExplicitPath, "Path to the kubeconfig file to use for "+opts.description)
+ flags.StringVar(&opts.overrides.CurrentContext, opts.name+"-context", opts.overrides.CurrentContext, "The name of the kubeconfig context to use for "+opts.description)
+ flags.StringVar(&opts.overrides.Context.AuthInfo, opts.name+"-user", opts.overrides.Context.AuthInfo, "The name of the kubeconfig user to use for "+opts.description)
+ flags.StringVar(&opts.overrides.Context.Cluster, opts.name+"-cluster", opts.overrides.Context.Cluster, "The name of the kubeconfig cluster to use for "+opts.description)
+
+}
+
+func (opts *ClientOptions[FS]) ToRESTConfig() (*rest.Config, error) {
+ clientConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(opts.loadingRules, &opts.overrides)
+ base, err := clientConfig.ClientConfig()
+ if err != nil {
+ return base, err
+ }
+ return opts.ClientLimits.LimitConfig(base), nil
+}
+
+func (opts *ClientLimits[FS]) LimitConfig(base *rest.Config) *rest.Config {
+ ans := *base
+ ans.RateLimiter = flowcontrol.NewTokenBucketRateLimiter(float32(opts.QPS), opts.Burst)
+ return &ans
+}