Merge pull request #15247 from spiffxp/kube-api-limits

Expose kube-api burst/qps settings for kube components

cmd/kube-controller-manager/app/controllermanager.go

@@ -99,6 +99,8 @@ type CMServer struct {
 
 	Master       string
 	Kubeconfig   string
+	KubeApiQps   float32
+	KubeApiBurst int
 }
 
 // NewCMServer creates a new CMServer with a default config.
@@ -128,6 +130,8 @@ func NewCMServer() *CMServer {
 			PersistentVolumeRecyclerMinimumTimeoutHostPath:   60,
 			PersistentVolumeRecyclerIncrementTimeoutHostPath: 30,
 		},
+		KubeApiQps:   20.0,
+		KubeApiBurst: 30,
 	}
 	return &s
 }
@@ -193,6 +197,8 @@ func (s *CMServer) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&s.Kubeconfig, "kubeconfig", s.Kubeconfig, "Path to kubeconfig file with authorization and master location information.")
 	fs.StringVar(&s.RootCAFile, "root-ca-file", s.RootCAFile, "If set, this root certificate authority will be included in service account's token secret. This must be a valid PEM-encoded CA bundle.")
 	fs.BoolVar(&s.EnableExperimental, "enable-experimental", s.EnableExperimental, "Enables experimental controllers (requires enabling experimental API on apiserver).")
+	fs.Float32Var(&s.KubeApiQps, "kube-api-qps", s.KubeApiQps, "QPS to use while talking with kubernetes apiserver")
+	fs.IntVar(&s.KubeApiBurst, "kube-api-burst", s.KubeApiBurst, "Burst to use while talking with kubernetes apiserver")
 }
 
 func (s *CMServer) resyncPeriod() time.Duration {
@@ -215,8 +221,9 @@ func (s *CMServer) Run(_ []string) error {
 		return err
 	}
 
-	kubeconfig.QPS = 20.0
+	// Override kubeconfig qps/burst settings from flags
-	kubeconfig.Burst = 30
+	kubeconfig.QPS = s.KubeApiQps
+	kubeconfig.Burst = s.KubeApiBurst
 
 	kubeClient, err := client.New(kubeconfig)
 	if err != nil {

cmd/kube-proxy/app/server.go

@@ -64,6 +64,8 @@ type ProxyServerConfig struct {
 	nodeRef            *api.ObjectReference // Reference to this node.
 	MasqueradeAll      bool
 	CleanupAndExit     bool
+	KubeApiQps         float32
+	KubeApiBurst       int
 }
 
 type ProxyServer struct {
@@ -88,6 +90,8 @@ func (s *ProxyServerConfig) AddFlags(fs *pflag.FlagSet) {
 	fs.DurationVar(&s.SyncPeriod, "iptables-sync-period", s.SyncPeriod, "How often iptables rules are refreshed (e.g. '5s', '1m', '2h22m').  Must be greater than 0.")
 	fs.BoolVar(&s.MasqueradeAll, "masquerade-all", false, "If using the pure iptables proxy, SNAT everything")
 	fs.BoolVar(&s.CleanupAndExit, "cleanup-iptables", false, "If true cleanup iptables rules and exit.")
+	fs.Float32Var(&s.KubeApiQps, "kube-api-qps", s.KubeApiQps, "QPS to use while talking with kubernetes apiserver")
+	fs.IntVar(&s.KubeApiBurst, "kube-api-burst", s.KubeApiBurst, "Burst to use while talking with kubernetes apiserver")
 }
 
 const (
@@ -112,6 +116,8 @@ func NewProxyConfig() *ProxyServerConfig {
 		OOMScoreAdj:        qos.KubeProxyOOMScoreAdj,
 		ResourceContainer:  "/kube-proxy",
 		SyncPeriod:         30 * time.Second,
+		KubeApiQps:         5.0,
+		KubeApiBurst:       10,
 	}
 }
 
@@ -180,6 +186,11 @@ func NewProxyServerDefault(config *ProxyServerConfig) (*ProxyServer, error) {
 	if err != nil {
 		return nil, err
 	}
+
+	// Override kubeconfig qps/burst settings from flags
+	kubeconfig.QPS = config.KubeApiQps
+	kubeconfig.Burst = config.KubeApiBurst
+
 	client, err := kubeclient.New(kubeconfig)
 	if err != nil {
 		glog.Fatalf("Invalid API configuration: %v", err)

cmd/kubelet/app/server.go

@@ -143,6 +143,9 @@ type KubeletServer struct {
 	ChaosChance float64
 	// Crash immediately, rather than eating panics.
 	ReallyCrashForTesting bool
+
+	KubeApiQps   float32
+	KubeApiBurst int
 }
 
 // bootstrapping interface for kubelet, targets the initialization protocol
@@ -206,6 +209,8 @@ func NewKubeletServer() *KubeletServer {
 		SyncFrequency:       10 * time.Second,
 		SystemContainer:     "",
 		ReconcileCIDR:       true,
+		KubeApiQps:          5.0,
+		KubeApiBurst:        10,
 	}
 }
 
@@ -285,6 +290,8 @@ func (s *KubeletServer) AddFlags(fs *pflag.FlagSet) {
 	fs.Uint64Var(&s.MaxOpenFiles, "max-open-files", 1000000, "Number of files that can be opened by Kubelet process. [default=1000000]")
 	fs.BoolVar(&s.ReconcileCIDR, "reconcile-cidr", s.ReconcileCIDR, "Reconcile node CIDR with the CIDR specified by the API server. No-op if register-node or configure-cbr0 is false. [default=true]")
 	fs.BoolVar(&s.RegisterSchedulable, "register-schedulable", s.RegisterSchedulable, "Register the node as schedulable. No-op if register-node is false. [default=true]")
+	fs.Float32Var(&s.KubeApiQps, "kube-api-qps", s.KubeApiQps, "QPS to use while talking with kubernetes apiserver")
+	fs.IntVar(&s.KubeApiBurst, "kube-api-burst", s.KubeApiBurst, "Burst to use while talking with kubernetes apiserver")
 }
 
 // UnsecuredKubeletConfig returns a KubeletConfig suitable for being run, or an error if the server setup
@@ -580,6 +587,11 @@ func (s *KubeletServer) CreateAPIServerClientConfig() (*client.Config, error) {
 	if err != nil {
 		return nil, err
 	}
+
+	// Override kubeconfig qps/burst settings from flags
+	clientConfig.QPS = s.KubeApiQps
+	clientConfig.Burst = s.KubeApiBurst
+
 	s.addChaosToClientConfig(clientConfig)
 	return clientConfig, nil
 }

hack/verify-flags/known-flags.txt

@@ -129,6 +129,8 @@ ir-user
 jenkins-host
 jenkins-jobs
 km-path
+kube-api-burst
+kube-api-qps
 kubectl-path
 kubelet-cadvisor-port
 kubelet-certificate-authority

plugin/cmd/kube-scheduler/app/server.go

@@ -56,6 +56,8 @@ type SchedulerServer struct {
 	Kubeconfig        string
 	BindPodsQPS       float32
 	BindPodsBurst     int
+	KubeApiQps        float32
+	KubeApiBurst      int
 }
 
 // NewSchedulerServer creates a new SchedulerServer with default parameters
@@ -64,6 +66,10 @@ func NewSchedulerServer() *SchedulerServer {
 		Port:              ports.SchedulerPort,
 		Address:           net.ParseIP("127.0.0.1"),
 		AlgorithmProvider: factory.DefaultProvider,
+		BindPodsQPS:       50.0,
+		BindPodsBurst:     100,
+		KubeApiQps:        50.0,
+		KubeApiBurst:      100,
 	}
 	return &s
 }
@@ -77,8 +83,10 @@ func (s *SchedulerServer) AddFlags(fs *pflag.FlagSet) {
 	fs.BoolVar(&s.EnableProfiling, "profiling", true, "Enable profiling via web interface host:port/debug/pprof/")
 	fs.StringVar(&s.Master, "master", s.Master, "The address of the Kubernetes API server (overrides any value in kubeconfig)")
 	fs.StringVar(&s.Kubeconfig, "kubeconfig", s.Kubeconfig, "Path to kubeconfig file with authorization and master location information.")
-	fs.Float32Var(&s.BindPodsQPS, "bind-pods-qps", 50.0, "Number of bindings per second scheduler is allowed to continuously make")
+	fs.Float32Var(&s.BindPodsQPS, "bind-pods-qps", s.BindPodsQPS, "Number of bindings per second scheduler is allowed to continuously make")
-	fs.IntVar(&s.BindPodsBurst, "bind-pods-burst", 100, "Number of bindings per second scheduler is allowed to make during bursts")
+	fs.IntVar(&s.BindPodsBurst, "bind-pods-burst", s.BindPodsBurst, "Number of bindings per second scheduler is allowed to make during bursts")
+	fs.Float32Var(&s.KubeApiQps, "kube-api-qps", s.KubeApiQps, "QPS to use while talking with kubernetes apiserver")
+	fs.IntVar(&s.KubeApiBurst, "kube-api-burst", s.KubeApiBurst, "Burst to use while talking with kubernetes apiserver")
 }
 
 // Run runs the specified SchedulerServer.  This should never exit.
@@ -95,8 +103,10 @@ func (s *SchedulerServer) Run(_ []string) error {
 	if err != nil {
 		return err
 	}
-	kubeconfig.QPS = 50.0
+
-	kubeconfig.Burst = 100
+	// Override kubeconfig qps/burst settings from flags
+	kubeconfig.QPS = s.KubeApiQps
+	kubeconfig.Burst = s.KubeApiBurst
 
 	kubeClient, err := client.New(kubeconfig)
 	if err != nil {