add flag concurrent-statefulset-syncs to kube-controller-manager (#79169)

* add flag `concurrent-statefulset-syncs` to set number of concurrent workers for statefulset controller

* change default value of ConcurrentStatefulSetSyncs from 1 to 5

* 1. fix doc comment of statefulset config types.go
2. add missing deps k8s.io/kubernetes/pkg/controller/statefulset/config

* add missing dep k8s.io/kubernetes/pkg/controller/statefulset/config/v1alpha1

* updated bazel BUILD files

* update kube-controller-manager options testcase

* fix codegen

* fix golint error

* fix testcase

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

@@ -62,7 +62,7 @@ func startStatefulSetController(ctx ControllerContext) (http.Handler, bool, erro
 		ctx.InformerFactory.Core().V1().PersistentVolumeClaims(),
 		ctx.InformerFactory.Apps().V1().ControllerRevisions(),
 		ctx.ClientBuilder.ClientOrDie("statefulset-controller"),
-	).Run(1, ctx.Stop)
+	).Run(int(ctx.ComponentConfig.StatefulSetController.ConcurrentStatefulSetSyncs), ctx.Stop)
 	return nil, true, nil
 }
 

cmd/kube-controller-manager/app/options/BUILD

@@ -28,6 +28,7 @@ go_library(
         "replicationcontroller.go",
         "resourcequotacontroller.go",
         "serviceaccountcontroller.go",
+        "statefulsetcontroller.go",
         "ttlafterfinishedcontroller.go",
     ],
     importpath = "k8s.io/kubernetes/cmd/kube-controller-manager/app/options",
@@ -52,6 +53,7 @@ go_library(
         "//pkg/controller/replication/config:go_default_library",
         "//pkg/controller/resourcequota/config:go_default_library",
         "//pkg/controller/serviceaccount/config:go_default_library",
+        "//pkg/controller/statefulset/config:go_default_library",
         "//pkg/controller/ttlafterfinished/config:go_default_library",
         "//pkg/controller/volume/attachdetach/config:go_default_library",
         "//pkg/controller/volume/persistentvolume/config:go_default_library",
@@ -110,6 +112,7 @@ go_test(
         "//pkg/controller/resourcequota/config:go_default_library",
         "//pkg/controller/service/config:go_default_library",
         "//pkg/controller/serviceaccount/config:go_default_library",
+        "//pkg/controller/statefulset/config:go_default_library",
         "//pkg/controller/ttlafterfinished/config:go_default_library",
         "//pkg/controller/volume/attachdetach/config:go_default_library",
         "//pkg/controller/volume/persistentvolume/config:go_default_library",

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

@@ -63,6 +63,7 @@ type KubeControllerManagerOptions struct {
 	CSRSigningController             *CSRSigningControllerOptions
 	DaemonSetController              *DaemonSetControllerOptions
 	DeploymentController             *DeploymentControllerOptions
+	StatefulSetController            *StatefulSetControllerOptions
 	DeprecatedFlags                  *DeprecatedControllerOptions
 	EndpointController               *EndpointControllerOptions
 	GarbageCollectorController       *GarbageCollectorControllerOptions
@@ -114,6 +115,9 @@ func NewKubeControllerManagerOptions() (*KubeControllerManagerOptions, error) {
 		DeploymentController: &DeploymentControllerOptions{
 			&componentConfig.DeploymentController,
 		},
+		StatefulSetController: &StatefulSetControllerOptions{
+			&componentConfig.StatefulSetController,
+		},
 		DeprecatedFlags: &DeprecatedControllerOptions{
 			&componentConfig.DeprecatedController,
 		},
@@ -216,6 +220,7 @@ func (s *KubeControllerManagerOptions) Flags(allControllers []string, disabledBy
 	s.AttachDetachController.AddFlags(fss.FlagSet("attachdetach controller"))
 	s.CSRSigningController.AddFlags(fss.FlagSet("csrsigning controller"))
 	s.DeploymentController.AddFlags(fss.FlagSet("deployment controller"))
+	s.StatefulSetController.AddFlags(fss.FlagSet("statefulset controller"))
 	s.DaemonSetController.AddFlags(fss.FlagSet("daemonset controller"))
 	s.DeprecatedFlags.AddFlags(fss.FlagSet("deprecated"))
 	s.EndpointController.AddFlags(fss.FlagSet("endpoint controller"))
@@ -261,6 +266,9 @@ func (s *KubeControllerManagerOptions) ApplyTo(c *kubecontrollerconfig.Config) e
 	if err := s.DeploymentController.ApplyTo(&c.ComponentConfig.DeploymentController); err != nil {
 		return err
 	}
+	if err := s.StatefulSetController.ApplyTo(&c.ComponentConfig.StatefulSetController); err != nil {
+		return err
+	}
 	if err := s.DeprecatedFlags.ApplyTo(&c.ComponentConfig.DeprecatedController); err != nil {
 		return err
 	}
@@ -342,6 +350,7 @@ func (s *KubeControllerManagerOptions) Validate(allControllers []string, disable
 	errs = append(errs, s.CSRSigningController.Validate()...)
 	errs = append(errs, s.DaemonSetController.Validate()...)
 	errs = append(errs, s.DeploymentController.Validate()...)
+	errs = append(errs, s.StatefulSetController.Validate()...)
 	errs = append(errs, s.DeprecatedFlags.Validate()...)
 	errs = append(errs, s.EndpointController.Validate()...)
 	errs = append(errs, s.GarbageCollectorController.Validate()...)

cmd/kube-controller-manager/app/options/options_test.go

@@ -47,6 +47,7 @@ import (
 	resourcequotaconfig "k8s.io/kubernetes/pkg/controller/resourcequota/config"
 	serviceconfig "k8s.io/kubernetes/pkg/controller/service/config"
 	serviceaccountconfig "k8s.io/kubernetes/pkg/controller/serviceaccount/config"
+	statefulsetconfig "k8s.io/kubernetes/pkg/controller/statefulset/config"
 	ttlafterfinishedconfig "k8s.io/kubernetes/pkg/controller/ttlafterfinished/config"
 	attachdetachconfig "k8s.io/kubernetes/pkg/controller/volume/attachdetach/config"
 	persistentvolumeconfig "k8s.io/kubernetes/pkg/controller/volume/persistentvolume/config"
@@ -71,6 +72,7 @@ func TestAddFlags(t *testing.T) {
 		"--cluster-signing-cert-file=/cluster-signing-cert",
 		"--cluster-signing-key-file=/cluster-signing-key",
 		"--concurrent-deployment-syncs=10",
+		"--concurrent-statefulset-syncs=15",
 		"--concurrent-endpoint-syncs=10",
 		"--concurrent-gc-syncs=30",
 		"--concurrent-namespace-syncs=20",
@@ -216,6 +218,11 @@ func TestAddFlags(t *testing.T) {
 				DeploymentControllerSyncPeriod: metav1.Duration{Duration: 45 * time.Second},
 			},
 		},
+		StatefulSetController: &StatefulSetControllerOptions{
+			&statefulsetconfig.StatefulSetControllerConfiguration{
+				ConcurrentStatefulSetSyncs: 15,
+			},
+		},
 		DeprecatedFlags: &DeprecatedControllerOptions{
 			&kubectrlmgrconfig.DeprecatedControllerConfiguration{
 				DeletingPodsQPS:    0.1,

cmd/kube-controller-manager/app/options/statefulsetcontroller.go

@@ -0,0 +1,63 @@
+/*
+Copyright 2019 The Kubernetes 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 options
+
+import (
+	"fmt"
+
+	"github.com/spf13/pflag"
+
+	statefulsetconfig "k8s.io/kubernetes/pkg/controller/statefulset/config"
+)
+
+// StatefulSetControllerOptions holds the StatefulSetController options.
+type StatefulSetControllerOptions struct {
+	*statefulsetconfig.StatefulSetControllerConfiguration
+}
+
+// AddFlags adds flags related to StatefulSetController for controller manager to the specified FlagSet.
+func (o *StatefulSetControllerOptions) AddFlags(fs *pflag.FlagSet) {
+	if o == nil {
+		return
+	}
+
+	fs.Int32Var(&o.ConcurrentStatefulSetSyncs, "concurrent-statefulset-syncs", o.ConcurrentStatefulSetSyncs, "The number of statefulset objects that are allowed to sync concurrently. Larger number = more responsive statefulsets, but more CPU (and network) load")
+}
+
+// ApplyTo fills up StatefulSetController config with options.
+func (o *StatefulSetControllerOptions) ApplyTo(cfg *statefulsetconfig.StatefulSetControllerConfiguration) error {
+	if o == nil {
+		return nil
+	}
+
+	cfg.ConcurrentStatefulSetSyncs = o.ConcurrentStatefulSetSyncs
+
+	return nil
+}
+
+// Validate checks validation of StatefulSetControllerOptions.
+func (o *StatefulSetControllerOptions) Validate() []error {
+	if o == nil {
+		return nil
+	}
+
+	errs := []error{}
+	if o.ConcurrentStatefulSetSyncs < 1 {
+		errs = append(errs, fmt.Errorf("concurrent-statefulset-syncs must be greater than 0, but got %d", o.ConcurrentStatefulSetSyncs))
+	}
+	return errs
+}