Make number of workers configurable

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

@@ -378,7 +378,7 @@ func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc
 	controllers["clusterrole-aggregation"] = startClusterRoleAggregrationController
 	controllers["pvc-protection"] = startPVCProtectionController
 	controllers["pv-protection"] = startPVProtectionController
-	controllers["ttl-after-finished-controller"] = startTTLAfterFinishedController
+	controllers["ttl-after-finished"] = startTTLAfterFinishedController
 
 	return controllers
 }

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

@@ -426,6 +426,6 @@ func startTTLAfterFinishedController(ctx ControllerContext) (http.Handler, bool,
 	go ttlafterfinished.New(
 		ctx.InformerFactory.Batch().V1().Jobs(),
 		ctx.ClientBuilder.ClientOrDie("ttl-after-finished-controller"),
-	).Run(5, ctx.Stop)
+	).Run(int(ctx.ComponentConfig.TTLAfterFinishedController.ConcurrentTTLSyncs), 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",
+        "ttlafterfinishedcontroller.go",
     ],
     importpath = "k8s.io/kubernetes/cmd/kube-controller-manager/app/options",
     deps = [

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

@@ -77,6 +77,7 @@ type KubeControllerManagerOptions struct {
 	ReplicationController            *ReplicationControllerOptions
 	ResourceQuotaController          *ResourceQuotaControllerOptions
 	SAController                     *SAControllerOptions
+	TTLAfterFinishedController       *TTLAfterFinishedControllerOptions
 
 	SecureServing *apiserveroptions.SecureServingOptionsWithLoopback
 	// TODO: remove insecure serving mode
@@ -172,6 +173,9 @@ func NewKubeControllerManagerOptions() (*KubeControllerManagerOptions, error) {
 		ServiceController: &cmoptions.ServiceControllerOptions{
 			ConcurrentServiceSyncs: componentConfig.ServiceController.ConcurrentServiceSyncs,
 		},
+		TTLAfterFinishedController: &TTLAfterFinishedControllerOptions{
+			ConcurrentTTLSyncs: componentConfig.TTLAfterFinishedController.ConcurrentTTLSyncs,
+		},
 		SecureServing: apiserveroptions.NewSecureServingOptions().WithLoopback(),
 		InsecureServing: (&apiserveroptions.DeprecatedInsecureServingOptions{
 			BindAddress: net.ParseIP(componentConfig.Generic.Address),
@@ -251,6 +255,7 @@ func (s *KubeControllerManagerOptions) Flags(allControllers []string, disabledBy
 	s.ReplicationController.AddFlags(fss.FlagSet("replicationcontroller"))
 	s.ResourceQuotaController.AddFlags(fss.FlagSet("resourcequota controller"))
 	s.SAController.AddFlags(fss.FlagSet("serviceaccount controller"))
+	s.TTLAfterFinishedController.AddFlags(fss.FlagSet("ttl-after-finished controller"))
 
 	fs := fss.FlagSet("misc")
 	fs.StringVar(&s.Master, "master", s.Master, "The address of the Kubernetes API server (overrides any value in kubeconfig).")
@@ -328,6 +333,9 @@ func (s *KubeControllerManagerOptions) ApplyTo(c *kubecontrollerconfig.Config) e
 	if err := s.ServiceController.ApplyTo(&c.ComponentConfig.ServiceController); err != nil {
 		return err
 	}
+	if err := s.TTLAfterFinishedController.ApplyTo(&c.ComponentConfig.TTLAfterFinishedController); err != nil {
+		return err
+	}
 	if err := s.InsecureServing.ApplyTo(&c.InsecureServing, &c.LoopbackClientConfig); err != nil {
 		return err
 	}
@@ -376,6 +384,7 @@ func (s *KubeControllerManagerOptions) Validate(allControllers []string, disable
 	errs = append(errs, s.ResourceQuotaController.Validate()...)
 	errs = append(errs, s.SAController.Validate()...)
 	errs = append(errs, s.ServiceController.Validate()...)
+	errs = append(errs, s.TTLAfterFinishedController.Validate()...)
 	errs = append(errs, s.SecureServing.Validate()...)
 	errs = append(errs, s.InsecureServing.Validate()...)
 	errs = append(errs, s.Authentication.Validate()...)

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

@@ -116,6 +116,7 @@ func TestAddFlags(t *testing.T) {
 		"--cert-dir=/a/b/c",
 		"--bind-address=192.168.4.21",
 		"--secure-port=10001",
+		"--concurrent-ttl-after-finished-syncs=8",
 	}
 	fs.Parse(args)
 	// Sort GCIgnoredResources because it's built from a map, which means the
@@ -255,6 +256,9 @@ func TestAddFlags(t *testing.T) {
 		ServiceController: &cmoptions.ServiceControllerOptions{
 			ConcurrentServiceSyncs: 2,
 		},
+		TTLAfterFinishedController: &TTLAfterFinishedControllerOptions{
+			ConcurrentTTLSyncs: 8,
+		},
 		SecureServing: (&apiserveroptions.SecureServingOptions{
 			BindPort:    10001,
 			BindAddress: net.ParseIP("192.168.4.21"),

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

@@ -0,0 +1,58 @@
+/*
+Copyright 2018 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 (
+	"github.com/spf13/pflag"
+
+	kubectrlmgrconfig "k8s.io/kubernetes/pkg/controller/apis/config"
+)
+
+// TTLAfterFinishedControllerOptions holds the TTLAfterFinishedController options.
+type TTLAfterFinishedControllerOptions struct {
+	ConcurrentTTLSyncs int32
+}
+
+// AddFlags adds flags related to TTLAfterFinishedController for controller manager to the specified FlagSet.
+func (o *TTLAfterFinishedControllerOptions) AddFlags(fs *pflag.FlagSet) {
+	if o == nil {
+		return
+	}
+
+	fs.Int32Var(&o.ConcurrentTTLSyncs, "concurrent-ttl-after-finished-syncs", o.ConcurrentTTLSyncs, "The number of TTL-after-finished controller workers that are allowed to sync concurrently.")
+}
+
+// ApplyTo fills up TTLAfterFinishedController config with options.
+func (o *TTLAfterFinishedControllerOptions) ApplyTo(cfg *kubectrlmgrconfig.TTLAfterFinishedControllerConfiguration) error {
+	if o == nil {
+		return nil
+	}
+
+	cfg.ConcurrentTTLSyncs = o.ConcurrentTTLSyncs
+
+	return nil
+}
+
+// Validate checks validation of TTLAfterFinishedControllerOptions.
+func (o *TTLAfterFinishedControllerOptions) Validate() []error {
+	if o == nil {
+		return nil
+	}
+
+	errs := []error{}
+	return errs
+}