diff --git a/cmd/clusterctl/client/alias.go b/cmd/clusterctl/client/alias.go
index 3835dbbaad7a..52378dec43a1 100644
--- a/cmd/clusterctl/client/alias.go
+++ b/cmd/clusterctl/client/alias.go
@@ -48,6 +48,9 @@ type CertManagerUpgradePlan cluster.CertManagerUpgradePlan
 // Kubeconfig is a type that specifies inputs related to the actual kubeconfig.
 type Kubeconfig cluster.Kubeconfig
 
+// RESTThrottle is a type that specifies inputs related to the throttle on the rest.Config.
+type RESTThrottle cluster.RESTThrottle
+
 // Processor defines the methods necessary for creating a specific yaml
 // processor.
 type Processor yaml.Processor
diff --git a/cmd/clusterctl/client/client.go b/cmd/clusterctl/client/client.go
index 1f7d4b093d65..57cb22ab4c91 100644
--- a/cmd/clusterctl/client/client.go
+++ b/cmd/clusterctl/client/client.go
@@ -117,8 +117,9 @@ type RepositoryClientFactory func(RepositoryClientFactoryInput) (repository.Clie
 
 // ClusterClientFactoryInput represents the inputs required by the factory.
 type ClusterClientFactoryInput struct {
-	Kubeconfig Kubeconfig
-	Processor  Processor
+	Kubeconfig   Kubeconfig
+	Processor    Processor
+	RESTThrottle RESTThrottle
 }
 
 // ClusterClientFactory is a factory of cluster.Client from a given input.
@@ -207,11 +208,24 @@ func defaultRepositoryFactory(configClient config.Client) RepositoryClientFactor
 // defaultClusterFactory is a ClusterClientFactory func the uses the default client provided by the cluster low level library.
 func defaultClusterFactory(configClient config.Client) ClusterClientFactory {
 	return func(input ClusterClientFactoryInput) (cluster.Client, error) {
+		var proxyOpts []cluster.ProxyOption
+		if input.RESTThrottle.QPS > 0 {
+			proxyOpts = append(proxyOpts, cluster.InjectRESTConfigQPS(input.RESTThrottle.QPS))
+		}
+		if input.RESTThrottle.Burst > 0 {
+			proxyOpts = append(proxyOpts, cluster.InjectRESTConfigBurst(input.RESTThrottle.Burst))
+		}
 		return cluster.New(
 			// Kubeconfig is a type alias to cluster.Kubeconfig
 			cluster.Kubeconfig(input.Kubeconfig),
 			configClient,
 			cluster.InjectYamlProcessor(input.Processor),
+			cluster.InjectProxy(
+				cluster.NewProxy(
+					cluster.Kubeconfig(input.Kubeconfig),
+					proxyOpts...,
+				),
+			),
 		), nil
 	}
 }
diff --git a/cmd/clusterctl/client/cluster/client.go b/cmd/clusterctl/client/cluster/client.go
index bd34d912581b..3cc4f549ddf9 100644
--- a/cmd/clusterctl/client/cluster/client.go
+++ b/cmd/clusterctl/client/cluster/client.go
@@ -204,7 +204,7 @@ func newClusterClient(kubeconfig Kubeconfig, configClient config.Client, options
 
 	// if there is an injected proxy, use it, otherwise use a default one
 	if client.proxy == nil {
-		client.proxy = newProxy(client.kubeconfig)
+		client.proxy = NewProxy(client.kubeconfig)
 	}
 
 	// if there is an injected repositoryClientFactory, use it, otherwise use the default one
diff --git a/cmd/clusterctl/client/cluster/ownergraph.go b/cmd/clusterctl/client/cluster/ownergraph.go
index 6d1601b743c9..bc640172cc0d 100644
--- a/cmd/clusterctl/client/cluster/ownergraph.go
+++ b/cmd/clusterctl/client/cluster/ownergraph.go
@@ -43,7 +43,7 @@ type OwnerGraphNode struct {
 // own owner references; there is no guarantee about the stability of this API. Using this test with providers may require
 // a custom implementation of this function, or the OwnerGraph it returns.
 func GetOwnerGraph(namespace, kubeconfigPath string) (OwnerGraph, error) {
-	p := newProxy(Kubeconfig{Path: kubeconfigPath, Context: ""})
+	p := NewProxy(Kubeconfig{Path: kubeconfigPath, Context: ""})
 	invClient := newInventoryClient(p, nil)
 
 	graph := newObjectGraph(p, invClient)
diff --git a/cmd/clusterctl/client/cluster/proxy.go b/cmd/clusterctl/client/cluster/proxy.go
index 5c114beac458..f7a106ef9f0d 100644
--- a/cmd/clusterctl/client/cluster/proxy.go
+++ b/cmd/clusterctl/client/cluster/proxy.go
@@ -41,6 +41,13 @@ import (
 	"sigs.k8s.io/cluster-api/version"
 )
 
+const (
+	// DefaultRESTConfigQPS is the default maximum queries per second for the Kubernetes client.
+	DefaultRESTConfigQPS float32 = 20
+	// DefaultRESTConfigBurst is the default maximum request burst for the Kubernetes client.
+	DefaultRESTConfigBurst = 300
+)
+
 var (
 	localScheme = scheme.Scheme
 )
@@ -77,10 +84,17 @@ type Proxy interface {
 	GetResourceNames(groupVersion, kind string, options []client.ListOption, prefix string) ([]string, error)
 }
 
+// RESTThrottle is a type that specifies inputs related to the throttle on the rest.Config.
+type RESTThrottle struct {
+	QPS   float32
+	Burst int
+}
+
 type proxy struct {
 	kubeconfig         Kubeconfig
 	timeout            time.Duration
 	configLoadingRules *clientcmd.ClientConfigLoadingRules
+	throttle           RESTThrottle
 }
 
 var _ Proxy = &proxy{}
@@ -148,9 +162,8 @@ func (k *proxy) GetConfig() (*rest.Config, error) {
 	}
 	restConfig.UserAgent = fmt.Sprintf("clusterctl/%s (%s)", version.Get().GitVersion, version.Get().Platform)
 
-	// Set QPS and Burst to a threshold that ensures the controller runtime client/client go doesn't generate throttling log messages
-	restConfig.QPS = 20
-	restConfig.Burst = 100
+	restConfig.QPS = k.throttle.QPS
+	restConfig.Burst = k.throttle.Burst
 
 	return restConfig, nil
 }
@@ -373,7 +386,22 @@ func InjectKubeconfigPaths(paths []string) ProxyOption {
 	}
 }
 
-func newProxy(kubeconfig Kubeconfig, opts ...ProxyOption) Proxy {
+// InjectRESTConfigQPS sets the QPS for the REST config.
+func InjectRESTConfigQPS(qps float32) ProxyOption {
+	return func(p *proxy) {
+		p.throttle.QPS = qps
+	}
+}
+
+// InjectRESTConfigBurst sets the burst for the REST config.
+func InjectRESTConfigBurst(burst int) ProxyOption {
+	return func(p *proxy) {
+		p.throttle.Burst = burst
+	}
+}
+
+// NewProxy creates a new proxy.
+func NewProxy(kubeconfig Kubeconfig, opts ...ProxyOption) Proxy {
 	// If a kubeconfig file isn't provided, find one in the standard locations.
 	rules := clientcmd.NewDefaultClientConfigLoadingRules()
 	if kubeconfig.Path != "" {
@@ -383,6 +411,11 @@ func newProxy(kubeconfig Kubeconfig, opts ...ProxyOption) Proxy {
 		kubeconfig:         kubeconfig,
 		timeout:            30 * time.Second,
 		configLoadingRules: rules,
+		// Set QPS and Burst to a threshold that ensures the controller runtime client/client go doesn't get throttled. This is especially important in environments with many CustomResourceDefinitions.
+		throttle: RESTThrottle{
+			QPS:   DefaultRESTConfigQPS,
+			Burst: DefaultRESTConfigBurst,
+		},
 	}
 
 	for _, o := range opts {
diff --git a/cmd/clusterctl/client/cluster/proxy_test.go b/cmd/clusterctl/client/cluster/proxy_test.go
index dfa55f029512..bab8515dee49 100644
--- a/cmd/clusterctl/client/cluster/proxy_test.go
+++ b/cmd/clusterctl/client/cluster/proxy_test.go
@@ -69,7 +69,7 @@ func TestProxyGetConfig(t *testing.T) {
 				configFile := filepath.Join(dir, ".test-kubeconfig.yaml")
 				g.Expect(os.WriteFile(configFile, []byte(tt.kubeconfigContents), 0600)).To(Succeed())
 
-				proxy := newProxy(Kubeconfig{Path: configFile, Context: tt.context})
+				proxy := NewProxy(Kubeconfig{Path: configFile, Context: tt.context})
 				conf, err := proxy.GetConfig()
 				if tt.expectErr {
 					g.Expect(err).To(HaveOccurred())
@@ -82,7 +82,7 @@ func TestProxyGetConfig(t *testing.T) {
 				g.Expect(conf.Host).To(Equal(tt.expectedHost))
 				g.Expect(conf.UserAgent).To(Equal(fmt.Sprintf("clusterctl/%s (%s)", version.Get().GitVersion, version.Get().Platform)))
 				g.Expect(conf.QPS).To(BeEquivalentTo(20))
-				g.Expect(conf.Burst).To(BeEquivalentTo(100))
+				g.Expect(conf.Burst).To(BeEquivalentTo(300))
 				g.Expect(conf.Timeout.String()).To(Equal("30s"))
 			})
 		}
@@ -96,11 +96,30 @@ func TestProxyGetConfig(t *testing.T) {
 		configFile := filepath.Join(dir, ".test-kubeconfig.yaml")
 		g.Expect(os.WriteFile(configFile, []byte(kubeconfig("management", "default")), 0600)).To(Succeed())
 
-		proxy := newProxy(Kubeconfig{Path: configFile, Context: "management"}, InjectProxyTimeout(23*time.Second))
+		proxy := NewProxy(Kubeconfig{Path: configFile, Context: "management"}, InjectProxyTimeout(23*time.Second))
 		conf, err := proxy.GetConfig()
 		g.Expect(err).ToNot(HaveOccurred())
 		g.Expect(conf.Timeout.String()).To(Equal("23s"))
 	})
+
+	t.Run("configure throttle", func(t *testing.T) {
+		g := NewWithT(t)
+		dir, err := os.MkdirTemp("", "clusterctl")
+		g.Expect(err).NotTo(HaveOccurred())
+		defer os.RemoveAll(dir)
+		configFile := filepath.Join(dir, ".test-kubeconfig.yaml")
+		g.Expect(os.WriteFile(configFile, []byte(kubeconfig("management", "default")), 0600)).To(Succeed())
+
+		proxy := NewProxy(
+			Kubeconfig{Path: configFile, Context: "management"},
+			InjectRESTConfigQPS(30),
+			InjectRESTConfigBurst(400),
+		)
+		conf, err := proxy.GetConfig()
+		g.Expect(err).ToNot(HaveOccurred())
+		g.Expect(conf.QPS).To(Equal(float32(30)))
+		g.Expect(conf.Burst).To(Equal(400))
+	})
 }
 
 // These tests are emulating the files passed in via KUBECONFIG env var by
@@ -123,7 +142,7 @@ func TestKUBECONFIGEnvVar(t *testing.T) {
 		configFile := filepath.Join(dir, ".test-kubeconfig.yaml")
 		g.Expect(os.WriteFile(configFile, []byte(kubeconfigContents), 0600)).To(Succeed())
 
-		proxy := newProxy(
+		proxy := NewProxy(
 			// dont't give an explicit path but rather define the file in the
 			// configLoadingRules precedence chain.
 			Kubeconfig{Path: "", Context: context},
@@ -151,7 +170,7 @@ func TestKUBECONFIGEnvVar(t *testing.T) {
 		configFile := filepath.Join(dir, ".test-kubeconfig.yaml")
 		g.Expect(os.WriteFile(configFile, []byte(kubeconfigContents), 0600)).To(Succeed())
 
-		proxy := newProxy(
+		proxy := NewProxy(
 			// dont't give an explicit path but rather define the file in the
 			// configLoadingRules precedence chain.
 			Kubeconfig{Path: "", Context: context},
@@ -229,7 +248,7 @@ func TestProxyCurrentNamespace(t *testing.T) {
 				g.Expect(os.WriteFile(configFile, []byte(tt.kubeconfigContents), 0600)).To(Succeed())
 			}
 
-			proxy := newProxy(Kubeconfig{Path: configFile, Context: tt.kubeconfigContext})
+			proxy := NewProxy(Kubeconfig{Path: configFile, Context: tt.kubeconfigContext})
 			ns, err := proxy.CurrentNamespace()
 			if tt.expectErr {
 				g.Expect(err).To(HaveOccurred())
diff --git a/cmd/clusterctl/client/config.go b/cmd/clusterctl/client/config.go
index 4140c2cf1495..2b1876061d8d 100644
--- a/cmd/clusterctl/client/config.go
+++ b/cmd/clusterctl/client/config.go
@@ -152,6 +152,9 @@ type GetClusterTemplateOptions struct {
 	// YamlProcessor defines the yaml processor to use for the cluster
 	// template processing. If not defined, SimpleProcessor will be used.
 	YamlProcessor Processor
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 // numSources return the number of template sources currently set on a GetClusterTemplateOptions.
@@ -217,7 +220,11 @@ func (c *clusterctlClient) GetClusterTemplate(options GetClusterTemplateOptions)
 	}
 
 	// Gets  the client for the current management cluster
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{options.Kubeconfig, options.YamlProcessor})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		Processor:    options.YamlProcessor,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/clusterctl/client/delete.go b/cmd/clusterctl/client/delete.go
index 77ee268a3b15..b883cbe4bd2d 100644
--- a/cmd/clusterctl/client/delete.go
+++ b/cmd/clusterctl/client/delete.go
@@ -60,10 +60,16 @@ type DeleteOptions struct {
 
 	// SkipInventory forces the deletion of the inventory items used by clusterctl to track providers.
 	SkipInventory bool
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 func (c *clusterctlClient) Delete(options DeleteOptions) error {
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return err
 	}
diff --git a/cmd/clusterctl/client/describe.go b/cmd/clusterctl/client/describe.go
index 2bdf1464537d..a5a938afa64f 100644
--- a/cmd/clusterctl/client/describe.go
+++ b/cmd/clusterctl/client/describe.go
@@ -57,12 +57,18 @@ type DescribeClusterOptions struct {
 	// Grouping groups machines objects in case the ready conditions
 	// have the same Status, Severity and Reason.
 	Grouping bool
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 // DescribeCluster returns the object tree representing the status of a Cluster API cluster.
 func (c *clusterctlClient) DescribeCluster(options DescribeClusterOptions) (*tree.ObjectTree, error) {
 	// gets access to the management cluster
-	cluster, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	cluster, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/clusterctl/client/get_kubeconfig.go b/cmd/clusterctl/client/get_kubeconfig.go
index 7f33402d7d8f..81aae6eac38a 100644
--- a/cmd/clusterctl/client/get_kubeconfig.go
+++ b/cmd/clusterctl/client/get_kubeconfig.go
@@ -31,11 +31,17 @@ type GetKubeconfigOptions struct {
 
 	// WorkloadClusterName is the name of the workload cluster.
 	WorkloadClusterName string
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 func (c *clusterctlClient) GetKubeconfig(options GetKubeconfigOptions) (string, error) {
 	// gets access to the management cluster
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return "", err
 	}
diff --git a/cmd/clusterctl/client/init.go b/cmd/clusterctl/client/init.go
index 3c4540a6401d..df1f01110d88 100644
--- a/cmd/clusterctl/client/init.go
+++ b/cmd/clusterctl/client/init.go
@@ -83,6 +83,9 @@ type InitOptions struct {
 	// allowMissingProviderCRD is used to allow for a missing provider CRD when listing images.
 	// It is set to false to enforce that provider CRD is available when performing the standard init operation.
 	allowMissingProviderCRD bool
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 // Init initializes a management cluster by adding the requested list of providers.
@@ -90,7 +93,10 @@ func (c *clusterctlClient) Init(options InitOptions) ([]Components, error) {
 	log := logf.Log
 
 	// gets access to the management cluster
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -167,7 +173,10 @@ func (c *clusterctlClient) Init(options InitOptions) ([]Components, error) {
 // InitImages returns the list of images required for init.
 func (c *clusterctlClient) InitImages(options InitOptions) ([]string, error) {
 	// gets access to the management cluster
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/clusterctl/client/move.go b/cmd/clusterctl/client/move.go
index 32d90c65a578..9db14077df67 100644
--- a/cmd/clusterctl/client/move.go
+++ b/cmd/clusterctl/client/move.go
@@ -46,6 +46,14 @@ type MoveOptions struct {
 
 	// DryRun means the move action is a dry run, no real action will be performed.
 	DryRun bool
+
+	// FromRESTThrottle defines parameters for the rest.Config's throttle for
+	// the source management cluster's Kubernetes client.
+	FromRESTThrottle RESTThrottle
+
+	// ToRESTThrottle defines parameters for the rest.Config's throttle for
+	// the destination management cluster's Kubernetes client.
+	ToRESTThrottle RESTThrottle
 }
 
 func (c *clusterctlClient) Move(options MoveOptions) error {
@@ -72,7 +80,7 @@ func (c *clusterctlClient) Move(options MoveOptions) error {
 
 func (c *clusterctlClient) move(options MoveOptions) error {
 	// Get the client for interacting with the source management cluster.
-	fromCluster, err := c.getClusterClient(options.FromKubeconfig)
+	fromCluster, err := c.getClusterClient(options.FromKubeconfig, options.FromRESTThrottle)
 	if err != nil {
 		return err
 	}
@@ -89,7 +97,7 @@ func (c *clusterctlClient) move(options MoveOptions) error {
 	var toCluster cluster.Client
 	if !options.DryRun {
 		// Get the client for interacting with the target management cluster.
-		if toCluster, err = c.getClusterClient(options.ToKubeconfig); err != nil {
+		if toCluster, err = c.getClusterClient(options.ToKubeconfig, options.ToRESTThrottle); err != nil {
 			return err
 		}
 	}
@@ -98,7 +106,7 @@ func (c *clusterctlClient) move(options MoveOptions) error {
 }
 
 func (c *clusterctlClient) fromDirectory(options MoveOptions) error {
-	toCluster, err := c.getClusterClient(options.ToKubeconfig)
+	toCluster, err := c.getClusterClient(options.ToKubeconfig, options.ToRESTThrottle)
 	if err != nil {
 		return err
 	}
@@ -111,7 +119,7 @@ func (c *clusterctlClient) fromDirectory(options MoveOptions) error {
 }
 
 func (c *clusterctlClient) toDirectory(options MoveOptions) error {
-	fromCluster, err := c.getClusterClient(options.FromKubeconfig)
+	fromCluster, err := c.getClusterClient(options.FromKubeconfig, options.FromRESTThrottle)
 	if err != nil {
 		return err
 	}
@@ -132,8 +140,11 @@ func (c *clusterctlClient) toDirectory(options MoveOptions) error {
 	return fromCluster.ObjectMover().ToDirectory(options.Namespace, options.ToDirectory)
 }
 
-func (c *clusterctlClient) getClusterClient(kubeconfig Kubeconfig) (cluster.Client, error) {
-	cluster, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: kubeconfig})
+func (c *clusterctlClient) getClusterClient(kubeconfig Kubeconfig, throttle RESTThrottle) (cluster.Client, error) {
+	cluster, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   kubeconfig,
+		RESTThrottle: throttle,
+	})
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/clusterctl/client/rollout.go b/cmd/clusterctl/client/rollout.go
index c69ef2edab4b..667e41c3cad2 100644
--- a/cmd/clusterctl/client/rollout.go
+++ b/cmd/clusterctl/client/rollout.go
@@ -38,6 +38,9 @@ type RolloutRestartOptions struct {
 	// Namespace where the resource(s) live. If unspecified, the namespace name will be inferred
 	// from the current configuration.
 	Namespace string
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 // RolloutPauseOptions carries the options supported by RolloutPause.
@@ -52,6 +55,9 @@ type RolloutPauseOptions struct {
 	// Namespace where the resource(s) live. If unspecified, the namespace name will be inferred
 	// from the current configuration.
 	Namespace string
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 // RolloutResumeOptions carries the options supported by RolloutResume.
@@ -66,6 +72,9 @@ type RolloutResumeOptions struct {
 	// Namespace where the resource(s) live. If unspecified, the namespace name will be inferred
 	// from the current configuration.
 	Namespace string
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 // RolloutUndoOptions carries the options supported by RolloutUndo.
@@ -83,10 +92,16 @@ type RolloutUndoOptions struct {
 
 	// Revision number to rollback to when issuing the undo command.
 	ToRevision int64
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 func (c *clusterctlClient) RolloutRestart(options RolloutRestartOptions) error {
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return err
 	}
@@ -103,7 +118,10 @@ func (c *clusterctlClient) RolloutRestart(options RolloutRestartOptions) error {
 }
 
 func (c *clusterctlClient) RolloutPause(options RolloutPauseOptions) error {
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return err
 	}
@@ -120,7 +138,10 @@ func (c *clusterctlClient) RolloutPause(options RolloutPauseOptions) error {
 }
 
 func (c *clusterctlClient) RolloutResume(options RolloutResumeOptions) error {
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return err
 	}
@@ -137,7 +158,10 @@ func (c *clusterctlClient) RolloutResume(options RolloutResumeOptions) error {
 }
 
 func (c *clusterctlClient) RolloutUndo(options RolloutUndoOptions) error {
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return err
 	}
diff --git a/cmd/clusterctl/client/topology.go b/cmd/clusterctl/client/topology.go
index 8e65bb3547db..43c9fc3e67a2 100644
--- a/cmd/clusterctl/client/topology.go
+++ b/cmd/clusterctl/client/topology.go
@@ -39,6 +39,9 @@ type TopologyPlanOptions struct {
 	// This namespace is used as default for objects with missing namespaces.
 	// If the namespace of any of the input objects conflicts with Namespace an error is returned.
 	Namespace string
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 // TopologyPlanOutput defines the output of the topology plan operation.
@@ -47,7 +50,10 @@ type TopologyPlanOutput = cluster.TopologyPlanOutput
 // TopologyPlan performs a dry run execution of the topology reconciler using the given inputs.
 // It returns a summary of the changes observed during the execution.
 func (c *clusterctlClient) TopologyPlan(options TopologyPlanOptions) (*TopologyPlanOutput, error) {
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/clusterctl/client/upgrade.go b/cmd/clusterctl/client/upgrade.go
index 849ed23cc889..a481cf2e5a04 100644
--- a/cmd/clusterctl/client/upgrade.go
+++ b/cmd/clusterctl/client/upgrade.go
@@ -36,11 +36,17 @@ const upgradeItemProviderNameError = "invalid provider name %q. Provider name sh
 type PlanUpgradeOptions struct {
 	// Kubeconfig defines the kubeconfig to use for accessing the management cluster. If empty, default discovery rules apply.
 	Kubeconfig Kubeconfig
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 func (c *clusterctlClient) PlanCertManagerUpgrade(options PlanUpgradeOptions) (CertManagerUpgradePlan, error) {
 	// Get the client for interacting with the management cluster.
-	cluster, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	cluster, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return CertManagerUpgradePlan{}, err
 	}
@@ -52,7 +58,10 @@ func (c *clusterctlClient) PlanCertManagerUpgrade(options PlanUpgradeOptions) (C
 
 func (c *clusterctlClient) PlanUpgrade(options PlanUpgradeOptions) ([]UpgradePlan, error) {
 	// Get the client for interacting with the management cluster.
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -126,6 +135,9 @@ type ApplyUpgradeOptions struct {
 
 	// WaitProviderTimeout sets the timeout per provider upgrade.
 	WaitProviderTimeout time.Duration
+
+	// RESTThrottle defines parameters for the rest.Config's throttle.
+	RESTThrottle RESTThrottle
 }
 
 func (c *clusterctlClient) ApplyUpgrade(options ApplyUpgradeOptions) error {
@@ -134,7 +146,10 @@ func (c *clusterctlClient) ApplyUpgrade(options ApplyUpgradeOptions) error {
 	}
 
 	// Get the client for interacting with the management cluster.
-	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{Kubeconfig: options.Kubeconfig})
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{
+		Kubeconfig:   options.Kubeconfig,
+		RESTThrottle: options.RESTThrottle,
+	})
 	if err != nil {
 		return err
 	}
diff --git a/cmd/clusterctl/cmd/delete.go b/cmd/clusterctl/cmd/delete.go
index 662a505b93b3..d2f074cbabcd 100644
--- a/cmd/clusterctl/cmd/delete.go
+++ b/cmd/clusterctl/cmd/delete.go
@@ -21,6 +21,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 )
 
 type deleteOptions struct {
@@ -35,6 +36,8 @@ type deleteOptions struct {
 	includeNamespace          bool
 	includeCRDs               bool
 	deleteAll                 bool
+	restQPS                   float32
+	restBurst                 int
 }
 
 var dd = &deleteOptions{}
@@ -92,6 +95,11 @@ func init() {
 	deleteCmd.Flags().StringVar(&dd.kubeconfigContext, "kubeconfig-context", "",
 		"Context to be used within the kubeconfig file. If empty, current context will be used.")
 
+	deleteCmd.Flags().Float32Var(&dd.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	deleteCmd.Flags().IntVar(&dd.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
+
 	deleteCmd.Flags().BoolVar(&dd.includeNamespace, "include-namespace", false,
 		"Forces the deletion of the namespace where the providers are hosted (and of all the contained objects)")
 	deleteCmd.Flags().BoolVar(&dd.includeCRDs, "include-crd", false,
@@ -148,5 +156,9 @@ func runDelete() error {
 		IPAMProviders:             dd.ipamProviders,
 		RuntimeExtensionProviders: dd.runtimeExtensionProviders,
 		DeleteAll:                 dd.deleteAll,
+		RESTThrottle: client.RESTThrottle{
+			QPS:   dd.restQPS,
+			Burst: dd.restBurst,
+		},
 	})
 }
diff --git a/cmd/clusterctl/cmd/describe_cluster.go b/cmd/clusterctl/cmd/describe_cluster.go
index b73cdf3e75d0..d4fe4dae1d67 100644
--- a/cmd/clusterctl/cmd/describe_cluster.go
+++ b/cmd/clusterctl/cmd/describe_cluster.go
@@ -34,6 +34,7 @@ import (
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/tree"
 )
 
@@ -65,6 +66,8 @@ type describeClusterOptions struct {
 	grouping                bool
 	disableGrouping         bool
 	color                   bool
+	restQPS                 float32
+	restBurst               int
 }
 
 var dc = &describeClusterOptions{}
@@ -114,6 +117,11 @@ func init() {
 	describeClusterClusterCmd.Flags().StringVarP(&dc.namespace, "namespace", "n", "",
 		"The namespace where the workload cluster is located. If unspecified, the current namespace will be used.")
 
+	describeClusterClusterCmd.Flags().Float32Var(&dc.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	describeClusterClusterCmd.Flags().IntVar(&dc.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
+
 	describeClusterClusterCmd.Flags().StringVar(&dc.showOtherConditions, "show-conditions", "",
 		"list of comma separated kind or kind/name for which the command should show all the object's conditions (use 'all' to show conditions for everything).")
 	describeClusterClusterCmd.Flags().BoolVar(&dc.showMachineSets, "show-machinesets", false,
@@ -162,6 +170,10 @@ func runDescribeCluster(cmd *cobra.Command, name string) error {
 		AddTemplateVirtualNode:  true,
 		Echo:                    dc.echo,
 		Grouping:                dc.grouping && !dc.disableGrouping,
+		RESTThrottle: client.RESTThrottle{
+			QPS:   dc.restQPS,
+			Burst: dc.restBurst,
+		},
 	})
 	if err != nil {
 		return err
diff --git a/cmd/clusterctl/cmd/generate_cluster.go b/cmd/clusterctl/cmd/generate_cluster.go
index 066a24a7e0cc..fca7ececcf62 100644
--- a/cmd/clusterctl/cmd/generate_cluster.go
+++ b/cmd/clusterctl/cmd/generate_cluster.go
@@ -23,6 +23,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 )
 
 type generateClusterOptions struct {
@@ -44,6 +45,9 @@ type generateClusterOptions struct {
 	listVariables bool
 
 	output string
+
+	restQPS   float32
+	restBurst int
 }
 
 var gc = &generateClusterOptions{}
@@ -108,6 +112,11 @@ func init() {
 	generateClusterClusterCmd.Flags().StringVar(&gc.kubeconfigContext, "kubeconfig-context", "",
 		"Context to be used within the kubeconfig file. If empty, current context will be used.")
 
+	generateClusterClusterCmd.Flags().Float32Var(&gc.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	generateClusterClusterCmd.Flags().IntVar(&gc.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
+
 	// flags for the template variables
 	generateClusterClusterCmd.Flags().StringVarP(&gc.targetNamespace, "target-namespace", "n", "",
 		"The namespace to use for the workload cluster. If unspecified, the current namespace will be used.")
@@ -156,6 +165,10 @@ func runGenerateClusterTemplate(cmd *cobra.Command, name string) error {
 		TargetNamespace:   gc.targetNamespace,
 		KubernetesVersion: gc.kubernetesVersion,
 		ListVariablesOnly: gc.listVariables,
+		RESTThrottle: client.RESTThrottle{
+			QPS:   gc.restQPS,
+			Burst: gc.restBurst,
+		},
 	}
 
 	if cmd.Flags().Changed("control-plane-machine-count") {
diff --git a/cmd/clusterctl/cmd/get_kubeconfig.go b/cmd/clusterctl/cmd/get_kubeconfig.go
index e2e95fda048a..7e061f4922ce 100644
--- a/cmd/clusterctl/cmd/get_kubeconfig.go
+++ b/cmd/clusterctl/cmd/get_kubeconfig.go
@@ -24,12 +24,15 @@ import (
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 )
 
 type getKubeconfigOptions struct {
 	kubeconfig        string
 	kubeconfigContext string
 	namespace         string
+	restQPS           float32
+	restBurst         int
 }
 
 var gk = &getKubeconfigOptions{}
@@ -66,6 +69,11 @@ func init() {
 	getKubeconfigCmd.Flags().StringVar(&gk.kubeconfigContext, "kubeconfig-context", "",
 		"Context to be used within the kubeconfig file. If empty, current context will be used.")
 
+	getKubeconfigCmd.Flags().Float32Var(&gk.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	getKubeconfigCmd.Flags().IntVar(&gk.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
+
 	// completions
 	getKubeconfigCmd.ValidArgsFunction = resourceNameCompletionFunc(
 		getKubeconfigCmd.Flags().Lookup("kubeconfig"),
@@ -88,6 +96,10 @@ func runGetKubeconfig(workloadClusterName string) error {
 		Kubeconfig:          client.Kubeconfig{Path: gk.kubeconfig, Context: gk.kubeconfigContext},
 		WorkloadClusterName: workloadClusterName,
 		Namespace:           gk.namespace,
+		RESTThrottle: client.RESTThrottle{
+			QPS:   gk.restQPS,
+			Burst: gk.restBurst,
+		},
 	}
 
 	out, err := c.GetKubeconfig(options)
diff --git a/cmd/clusterctl/cmd/init.go b/cmd/clusterctl/cmd/init.go
index 9d6eafbdbf9d..e4b573c91ac5 100644
--- a/cmd/clusterctl/cmd/init.go
+++ b/cmd/clusterctl/cmd/init.go
@@ -22,6 +22,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 )
 
 type initOptions struct {
@@ -37,6 +38,8 @@ type initOptions struct {
 	validate                  bool
 	waitProviders             bool
 	waitProviderTimeout       int
+	restQPS                   float32
+	restBurst                 int
 }
 
 var initOpts = &initOptions{}
@@ -103,6 +106,10 @@ func init() {
 		"IPAM providers and versions (e.g. infoblox:v0.0.1) to add to the management cluster.")
 	initCmd.PersistentFlags().StringSliceVar(&initOpts.runtimeExtensionProviders, "runtime-extension", nil,
 		"Runtime extension providers and versions (e.g. test:v0.0.1) to add to the management cluster.")
+	initCmd.PersistentFlags().Float32Var(&initOpts.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	initCmd.PersistentFlags().IntVar(&initOpts.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
 	initCmd.Flags().StringVarP(&initOpts.targetNamespace, "target-namespace", "n", "",
 		"The target namespace where the providers should be deployed. If unspecified, the provider components' default namespace is used.")
 	initCmd.Flags().BoolVar(&initOpts.waitProviders, "wait-providers", false,
@@ -135,6 +142,10 @@ func runInit() error {
 		WaitProviders:             initOpts.waitProviders,
 		WaitProviderTimeout:       time.Duration(initOpts.waitProviderTimeout) * time.Second,
 		IgnoreValidationErrors:    !initOpts.validate,
+		RESTThrottle: client.RESTThrottle{
+			QPS:   initOpts.restQPS,
+			Burst: initOpts.restBurst,
+		},
 	}
 
 	if _, err := c.Init(options); err != nil {
diff --git a/cmd/clusterctl/cmd/move.go b/cmd/clusterctl/cmd/move.go
index c75557e0aa10..04a21cc70520 100644
--- a/cmd/clusterctl/cmd/move.go
+++ b/cmd/clusterctl/cmd/move.go
@@ -21,6 +21,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 )
 
 type moveOptions struct {
@@ -32,6 +33,10 @@ type moveOptions struct {
 	fromDirectory         string
 	toDirectory           string
 	dryRun                bool
+	fromRestQPS           float32
+	toRestQPS             float32
+	fromRestBurst         int
+	toRestBurst           int
 }
 
 var mo = &moveOptions{}
@@ -79,6 +84,15 @@ func init() {
 	moveCmd.Flags().StringVar(&mo.fromDirectory, "from-directory", "",
 		"Read Cluster API objects and all dependencies from a directory into a management cluster.")
 
+	moveCmd.Flags().Float32Var(&mo.fromRestQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver of the source management cluster.")
+	moveCmd.Flags().Float32Var(&mo.toRestQPS, "to-kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver of the destination management cluster.")
+	moveCmd.Flags().IntVar(&mo.fromRestBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver of the source management cluster.")
+	moveCmd.Flags().IntVar(&mo.toRestBurst, "to-kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver of the destination management cluster.")
+
 	moveCmd.MarkFlagsMutuallyExclusive("to-directory", "to-kubeconfig")
 	moveCmd.MarkFlagsMutuallyExclusive("from-directory", "to-directory")
 	moveCmd.MarkFlagsMutuallyExclusive("from-directory", "kubeconfig")
@@ -106,5 +120,13 @@ func runMove() error {
 		ToDirectory:    mo.toDirectory,
 		Namespace:      mo.namespace,
 		DryRun:         mo.dryRun,
+		FromRESTThrottle: client.RESTThrottle{
+			QPS:   mo.fromRestQPS,
+			Burst: mo.fromRestBurst,
+		},
+		ToRESTThrottle: client.RESTThrottle{
+			QPS:   mo.toRestQPS,
+			Burst: mo.toRestBurst,
+		},
 	})
 }
diff --git a/cmd/clusterctl/cmd/rollout/pause.go b/cmd/clusterctl/cmd/rollout/pause.go
index 8e29ed200fec..af22627fb361 100644
--- a/cmd/clusterctl/cmd/rollout/pause.go
+++ b/cmd/clusterctl/cmd/rollout/pause.go
@@ -22,6 +22,7 @@ import (
 	"k8s.io/kubectl/pkg/util/templates"
 
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 )
 
 // pauseOptions is the start of the data required to perform the operation.
@@ -30,6 +31,8 @@ type pauseOptions struct {
 	kubeconfigContext string
 	resources         []string
 	namespace         string
+	restQPS           float32
+	restBurst         int
 }
 
 var pauseOpt = &pauseOptions{}
@@ -65,6 +68,10 @@ func NewCmdRolloutPause(cfgFile string) *cobra.Command {
 	cmd.Flags().StringVar(&pauseOpt.kubeconfigContext, "kubeconfig-context", "",
 		"Context to be used within the kubeconfig file. If empty, current context will be used.")
 	cmd.Flags().StringVarP(&pauseOpt.namespace, "namespace", "n", "", "Namespace where the resource(s) reside. If unspecified, the defult namespace will be used.")
+	cmd.Flags().Float32Var(&pauseOpt.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	cmd.Flags().IntVar(&pauseOpt.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
 
 	return cmd
 }
@@ -81,5 +88,9 @@ func runPause(cfgFile string, args []string) error {
 		Kubeconfig: client.Kubeconfig{Path: pauseOpt.kubeconfig, Context: pauseOpt.kubeconfigContext},
 		Namespace:  pauseOpt.namespace,
 		Resources:  pauseOpt.resources,
+		RESTThrottle: client.RESTThrottle{
+			QPS:   pauseOpt.restQPS,
+			Burst: pauseOpt.restBurst,
+		},
 	})
 }
diff --git a/cmd/clusterctl/cmd/rollout/restart.go b/cmd/clusterctl/cmd/rollout/restart.go
index 7f46fac317fd..633a345cf1f0 100644
--- a/cmd/clusterctl/cmd/rollout/restart.go
+++ b/cmd/clusterctl/cmd/rollout/restart.go
@@ -21,6 +21,7 @@ import (
 	"k8s.io/kubectl/pkg/util/templates"
 
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 )
 
 // restartOptions is the start of the data required to perform the operation.
@@ -29,6 +30,8 @@ type restartOptions struct {
 	kubeconfigContext string
 	resources         []string
 	namespace         string
+	restQPS           float32
+	restBurst         int
 }
 
 var restartOpt = &restartOptions{}
@@ -64,6 +67,10 @@ func NewCmdRolloutRestart(cfgFile string) *cobra.Command {
 	cmd.Flags().StringVar(&restartOpt.kubeconfigContext, "kubeconfig-context", "",
 		"Context to be used within the kubeconfig file. If empty, current context will be used.")
 	cmd.Flags().StringVarP(&restartOpt.namespace, "namespace", "n", "", "Namespace where the resource(s) reside. If unspecified, the defult namespace will be used.")
+	cmd.Flags().Float32Var(&restartOpt.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	cmd.Flags().IntVar(&restartOpt.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
 
 	return cmd
 }
@@ -80,5 +87,9 @@ func runRestart(cfgFile string, _ *cobra.Command, args []string) error {
 		Kubeconfig: client.Kubeconfig{Path: restartOpt.kubeconfig, Context: restartOpt.kubeconfigContext},
 		Namespace:  restartOpt.namespace,
 		Resources:  restartOpt.resources,
+		RESTThrottle: client.RESTThrottle{
+			QPS:   restartOpt.restQPS,
+			Burst: restartOpt.restBurst,
+		},
 	})
 }
diff --git a/cmd/clusterctl/cmd/rollout/resume.go b/cmd/clusterctl/cmd/rollout/resume.go
index 07dc4bc1fa5f..abb4b2579356 100644
--- a/cmd/clusterctl/cmd/rollout/resume.go
+++ b/cmd/clusterctl/cmd/rollout/resume.go
@@ -21,6 +21,7 @@ import (
 	"k8s.io/kubectl/pkg/util/templates"
 
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 )
 
 // resumeOptions is the start of the data required to perform the operation.
@@ -29,6 +30,8 @@ type resumeOptions struct {
 	kubeconfigContext string
 	resources         []string
 	namespace         string
+	restQPS           float32
+	restBurst         int
 }
 
 var resumeOpt = &resumeOptions{}
@@ -64,6 +67,10 @@ func NewCmdRolloutResume(cfgFile string) *cobra.Command {
 	cmd.Flags().StringVar(&resumeOpt.kubeconfigContext, "kubeconfig-context", "",
 		"Context to be used within the kubeconfig file. If empty, current context will be used.")
 	cmd.Flags().StringVarP(&resumeOpt.namespace, "namespace", "n", "", "Namespace where the resource(s) reside. If unspecified, the defult namespace will be used.")
+	cmd.Flags().Float32Var(&resumeOpt.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	cmd.Flags().IntVar(&resumeOpt.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
 
 	return cmd
 }
@@ -80,5 +87,9 @@ func runResume(cfgFile string, args []string) error {
 		Kubeconfig: client.Kubeconfig{Path: resumeOpt.kubeconfig, Context: resumeOpt.kubeconfigContext},
 		Namespace:  resumeOpt.namespace,
 		Resources:  resumeOpt.resources,
+		RESTThrottle: client.RESTThrottle{
+			QPS:   resumeOpt.restQPS,
+			Burst: resumeOpt.restBurst,
+		},
 	})
 }
diff --git a/cmd/clusterctl/cmd/rollout/undo.go b/cmd/clusterctl/cmd/rollout/undo.go
index cfa6603009b4..809b1ae61cb3 100644
--- a/cmd/clusterctl/cmd/rollout/undo.go
+++ b/cmd/clusterctl/cmd/rollout/undo.go
@@ -21,6 +21,7 @@ import (
 	"k8s.io/kubectl/pkg/util/templates"
 
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 )
 
 // undoOptions is the start of the data required to perform the operation.
@@ -30,6 +31,8 @@ type undoOptions struct {
 	resources         []string
 	namespace         string
 	toRevision        int64
+	restQPS           float32
+	restBurst         int
 }
 
 var undoOpt = &undoOptions{}
@@ -64,6 +67,10 @@ func NewCmdRolloutUndo(cfgFile string) *cobra.Command {
 		"Context to be used within the kubeconfig file. If empty, current context will be used.")
 	cmd.Flags().StringVarP(&undoOpt.namespace, "namespace", "n", "", "Namespace where the resource(s) reside. If unspecified, the defult namespace will be used.")
 	cmd.Flags().Int64Var(&undoOpt.toRevision, "to-revision", undoOpt.toRevision, "The revision to rollback to. Default to 0 (last revision).")
+	cmd.Flags().Float32Var(&undoOpt.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	cmd.Flags().IntVar(&undoOpt.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
 
 	return cmd
 }
@@ -81,5 +88,9 @@ func runUndo(cfgFile string, args []string) error {
 		Namespace:  undoOpt.namespace,
 		Resources:  undoOpt.resources,
 		ToRevision: undoOpt.toRevision,
+		RESTThrottle: client.RESTThrottle{
+			QPS:   undoOpt.restQPS,
+			Burst: undoOpt.restBurst,
+		},
 	})
 }
diff --git a/cmd/clusterctl/cmd/topology_plan.go b/cmd/clusterctl/cmd/topology_plan.go
index 8a8ede41bfe5..79d6cb2eb83f 100644
--- a/cmd/clusterctl/cmd/topology_plan.go
+++ b/cmd/clusterctl/cmd/topology_plan.go
@@ -45,6 +45,8 @@ type topologyPlanOptions struct {
 	cluster           string
 	namespace         string
 	outDir            string
+	restQPS           float32
+	restBurst         int
 }
 
 var tp = &topologyPlanOptions{}
@@ -88,11 +90,16 @@ var topologyPlanCmd = &cobra.Command{
 }
 
 func init() {
-	topologyPlanCmd.Flags().StringVar(&initOpts.kubeconfig, "kubeconfig", "",
+	topologyPlanCmd.Flags().StringVar(&tp.kubeconfig, "kubeconfig", "",
 		"Path to the kubeconfig for the management cluster. If unspecified, default discovery rules apply.")
-	topologyPlanCmd.Flags().StringVar(&initOpts.kubeconfigContext, "kubeconfig-context", "",
+	topologyPlanCmd.Flags().StringVar(&tp.kubeconfigContext, "kubeconfig-context", "",
 		"Context to be used within the kubeconfig file. If empty, current context will be used.")
 
+	topologyPlanCmd.Flags().Float32Var(&tp.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	topologyPlanCmd.Flags().IntVar(&tp.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
+
 	topologyPlanCmd.Flags().StringArrayVarP(&tp.files, "file", "f", nil, "path to the file with new or modified resources to be applied; the file should not contain more than one Cluster or more than one ClusterClass")
 	topologyPlanCmd.Flags().StringVarP(&tp.cluster, "cluster", "c", "", "name of the target cluster; this parameter is required when more than one cluster is affected")
 	topologyPlanCmd.Flags().StringVarP(&tp.namespace, "namespace", "n", "", "target namespace for the operation. If specified, it is used as default namespace for objects with missing namespace")
@@ -132,6 +139,10 @@ func runTopologyPlan() error {
 		Objs:       convertToPtrSlice(objs),
 		Cluster:    tp.cluster,
 		Namespace:  tp.namespace,
+		RESTThrottle: client.RESTThrottle{
+			QPS:   tp.restQPS,
+			Burst: tp.restBurst,
+		},
 	})
 	if err != nil {
 		return err
diff --git a/cmd/clusterctl/cmd/upgrade_apply.go b/cmd/clusterctl/cmd/upgrade_apply.go
index 41175669f4f7..9dc353aad67b 100644
--- a/cmd/clusterctl/cmd/upgrade_apply.go
+++ b/cmd/clusterctl/cmd/upgrade_apply.go
@@ -23,6 +23,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 )
 
 type upgradeApplyOptions struct {
@@ -37,6 +38,8 @@ type upgradeApplyOptions struct {
 	runtimeExtensionProviders []string
 	waitProviders             bool
 	waitProviderTimeout       int
+	restQPS                   float32
+	restBurst                 int
 }
 
 var ua = &upgradeApplyOptions{}
@@ -73,6 +76,11 @@ func init() {
 	upgradeApplyCmd.Flags().StringVar(&ua.contract, "contract", "",
 		"The API Version of Cluster API (contract, e.g. v1alpha4) the management cluster should upgrade to")
 
+	upgradeApplyCmd.Flags().Float32Var(&ua.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	upgradeApplyCmd.Flags().IntVar(&ua.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
+
 	upgradeApplyCmd.Flags().StringVar(&ua.coreProvider, "core", "",
 		"Core provider instance version (e.g. cluster-api:v1.1.5) to upgrade to. This flag can be used as alternative to --contract.")
 	upgradeApplyCmd.Flags().StringSliceVarP(&ua.infrastructureProviders, "infrastructure", "i", nil,
@@ -122,5 +130,9 @@ func runUpgradeApply() error {
 		RuntimeExtensionProviders: ua.runtimeExtensionProviders,
 		WaitProviders:             ua.waitProviders,
 		WaitProviderTimeout:       time.Duration(ua.waitProviderTimeout) * time.Second,
+		RESTThrottle: client.RESTThrottle{
+			QPS:   ua.restQPS,
+			Burst: ua.restBurst,
+		},
 	})
 }
diff --git a/cmd/clusterctl/cmd/upgrade_plan.go b/cmd/clusterctl/cmd/upgrade_plan.go
index 00f711e8a53c..5fef04912fc7 100644
--- a/cmd/clusterctl/cmd/upgrade_plan.go
+++ b/cmd/clusterctl/cmd/upgrade_plan.go
@@ -25,11 +25,14 @@ import (
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 )
 
 type upgradePlanOptions struct {
 	kubeconfig        string
 	kubeconfigContext string
+	restQPS           float32
+	restBurst         int
 }
 
 var up = &upgradePlanOptions{}
@@ -62,6 +65,10 @@ func init() {
 		"Path to the kubeconfig file to use for accessing the management cluster. If empty, default discovery rules apply.")
 	upgradePlanCmd.Flags().StringVar(&up.kubeconfigContext, "kubeconfig-context", "",
 		"Context to be used within the kubeconfig file. If empty, current context will be used.")
+	upgradePlanCmd.Flags().Float32Var(&up.restQPS, "kube-api-qps", cluster.DefaultRESTConfigQPS,
+		"QPS to use while talking with kubernetes apiserver.")
+	upgradePlanCmd.Flags().IntVar(&up.restBurst, "kube-api-burst", cluster.DefaultRESTConfigBurst,
+		"Burst to use while talking with kubernetes apiserver.")
 }
 
 func runUpgradePlan() error {
@@ -72,6 +79,10 @@ func runUpgradePlan() error {
 
 	certManUpgradePlan, err := c.PlanCertManagerUpgrade(client.PlanUpgradeOptions{
 		Kubeconfig: client.Kubeconfig{Path: up.kubeconfig, Context: up.kubeconfigContext},
+		RESTThrottle: client.RESTThrottle{
+			QPS:   up.restQPS,
+			Burst: up.restBurst,
+		},
 	})
 	if err != nil {
 		return err
@@ -86,6 +97,10 @@ func runUpgradePlan() error {
 
 	upgradePlans, err := c.PlanUpgrade(client.PlanUpgradeOptions{
 		Kubeconfig: client.Kubeconfig{Path: up.kubeconfig, Context: up.kubeconfigContext},
+		RESTThrottle: client.RESTThrottle{
+			QPS:   up.restQPS,
+			Burst: up.restBurst,
+		},
 	})
 
 	if err != nil {