separate controller initialization for easy controllers

cmd/kube-controller-manager/app/BUILD

@@ -13,8 +13,15 @@ load(
 go_library(
     name = "go_default_library",
     srcs = [
+        "apps.go",
+        "autoscaling.go",
+        "batch.go",
+        "certificates.go",
         "controllermanager.go",
+        "core.go",
+        "extensions.go",
         "plugins.go",
+        "policy.go",
     ],
     tags = ["automanaged"],
     deps = [

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

+/*
+Copyright 2016 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 app implements a server that runs a set of active
+// components.  This includes replication controllers, service endpoints and
+// nodes.
+//
+package app
+
+import (
+	petset "k8s.io/kubernetes/pkg/controller/petset"
+	"k8s.io/kubernetes/pkg/runtime/schema"
+)
+
+func startStatefulSetController(ctx ControllerContext) (bool, error) {
+	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "apps", Version: "v1beta1", Resource: "statefulsets"}] {
+		return false, nil
+	}
+	resyncPeriod := ResyncPeriod(&ctx.Options)()
+	go petset.NewStatefulSetController(
+		ctx.InformerFactory.Pods().Informer(),
+		ctx.ClientBuilder.ClientOrDie("statefulset-controller"),
+		resyncPeriod,
+	).Run(1, ctx.Stop)
+	return true, nil
+}

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

+/*
+Copyright 2016 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 app implements a server that runs a set of active
+// components.  This includes replication controllers, service endpoints and
+// nodes.
+//
+package app
+
+import (
+	"k8s.io/kubernetes/pkg/controller/podautoscaler"
+	"k8s.io/kubernetes/pkg/controller/podautoscaler/metrics"
+	"k8s.io/kubernetes/pkg/runtime/schema"
+)
+
+func startHPAController(ctx ControllerContext) (bool, error) {
+	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "autoscaling", Version: "v1", Resource: "horizontalpodautoscalers"}] {
+		return false, nil
+	}
+	hpaClient := ctx.ClientBuilder.ClientOrDie("horizontal-pod-autoscaler")
+	metricsClient := metrics.NewHeapsterMetricsClient(
+		hpaClient,
+		metrics.DefaultHeapsterNamespace,
+		metrics.DefaultHeapsterScheme,
+		metrics.DefaultHeapsterService,
+		metrics.DefaultHeapsterPort,
+	)
+	replicaCalc := podautoscaler.NewReplicaCalculator(metricsClient, hpaClient.Core())
+	go podautoscaler.NewHorizontalController(
+		hpaClient.Core(),
+		hpaClient.Extensions(),
+		hpaClient.Autoscaling(),
+		replicaCalc,
+		ctx.Options.HorizontalPodAutoscalerSyncPeriod.Duration,
+	).Run(ctx.Stop)
+	return true, nil
+}

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

+/*
+Copyright 2016 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 app implements a server that runs a set of active
+// components.  This includes replication controllers, service endpoints and
+// nodes.
+//
+package app
+
+import (
+	"k8s.io/kubernetes/pkg/apis/batch"
+	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_5"
+	"k8s.io/kubernetes/pkg/controller/cronjob"
+	"k8s.io/kubernetes/pkg/runtime/schema"
+)
+
+func startCronJobController(ctx ControllerContext) (bool, error) {
+	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "batch", Version: "v2alpha1", Resource: "cronjobs"}] {
+		return false, nil
+	}
+	// TODO: this is a temp fix for allowing kubeClient list v2alpha1 sj, should switch to using clientset
+	cronjobConfig := ctx.ClientBuilder.ConfigOrDie("cronjob-controller")
+	cronjobConfig.ContentConfig.GroupVersion = &schema.GroupVersion{Group: batch.GroupName, Version: "v2alpha1"}
+	go cronjob.NewCronJobController(
+		clientset.NewForConfigOrDie(cronjobConfig),
+	).Run(ctx.Stop)
+	return true, nil
+}

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

+/*
+Copyright 2016 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 app implements a server that runs a set of active
+// components.  This includes replication controllers, service endpoints and
+// nodes.
+//
+package app
+
+import (
+	"github.com/golang/glog"
+
+	certcontroller "k8s.io/kubernetes/pkg/controller/certificates"
+	"k8s.io/kubernetes/pkg/runtime/schema"
+)
+
+func startCSRController(ctx ControllerContext) (bool, error) {
+	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "certificates.k8s.io", Version: "v1alpha1", Resource: "certificatesigningrequests"}] {
+		return false, nil
+	}
+	resyncPeriod := ResyncPeriod(&ctx.Options)()
+	c := ctx.ClientBuilder.ClientOrDie("certificate-controller")
+	certController, err := certcontroller.NewCertificateController(
+		c,
+		resyncPeriod,
+		ctx.Options.ClusterSigningCertFile,
+		ctx.Options.ClusterSigningKeyFile,
+		certcontroller.NewGroupApprover(c.Certificates().CertificateSigningRequests(), ctx.Options.ApproveAllKubeletCSRsForGroup),
+	)
+	if err != nil {
+		// TODO this is failing consistently in test-cmd and local-up-cluster.sh.  Fix them and make it consistent with all others which
+		// cause a crash loop
+		glog.Errorf("Failed to start certificate controller: %v", err)
+		return false, nil
+	}
+	go certController.Run(1, ctx.Stop)
+	return true, nil
+}

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

+/*
+Copyright 2016 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 app implements a server that runs a set of active
+// components.  This includes replication controllers, service endpoints and
+// nodes.
+//
+package app
+
+import (
+	"fmt"
+
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/v1"
+	"k8s.io/kubernetes/pkg/apimachinery/registered"
+	"k8s.io/kubernetes/pkg/apis/extensions"
+	metav1 "k8s.io/kubernetes/pkg/apis/meta/v1"
+	"k8s.io/kubernetes/pkg/client/typed/discovery"
+	"k8s.io/kubernetes/pkg/client/typed/dynamic"
+	"k8s.io/kubernetes/pkg/controller"
+	endpointcontroller "k8s.io/kubernetes/pkg/controller/endpoint"
+	"k8s.io/kubernetes/pkg/controller/garbagecollector"
+	"k8s.io/kubernetes/pkg/controller/garbagecollector/metaonly"
+	namespacecontroller "k8s.io/kubernetes/pkg/controller/namespace"
+	"k8s.io/kubernetes/pkg/controller/podgc"
+	replicationcontroller "k8s.io/kubernetes/pkg/controller/replication"
+	resourcequotacontroller "k8s.io/kubernetes/pkg/controller/resourcequota"
+	serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
+	quotainstall "k8s.io/kubernetes/pkg/quota/install"
+	"k8s.io/kubernetes/pkg/runtime/schema"
+	"k8s.io/kubernetes/pkg/runtime/serializer"
+)
+
+func startEndpointController(ctx ControllerContext) (bool, error) {
+	go endpointcontroller.NewEndpointController(
+		ctx.InformerFactory.Pods().Informer(),
+		ctx.ClientBuilder.ClientOrDie("endpoint-controller"),
+	).Run(int(ctx.Options.ConcurrentEndpointSyncs), ctx.Stop)
+	return true, nil
+}
+
+func startReplicationController(ctx ControllerContext) (bool, error) {
+	go replicationcontroller.NewReplicationManager(
+		ctx.InformerFactory.Pods().Informer(),
+		ctx.ClientBuilder.ClientOrDie("replication-controller"),
+		ResyncPeriod(&ctx.Options),
+		replicationcontroller.BurstReplicas,
+		int(ctx.Options.LookupCacheSizeForRC),
+		ctx.Options.EnableGarbageCollector,
+	).Run(int(ctx.Options.ConcurrentRCSyncs), ctx.Stop)
+	return true, nil
+}
+
+func startPodGCController(ctx ControllerContext) (bool, error) {
+	go podgc.NewPodGC(
+		ctx.ClientBuilder.ClientOrDie("pod-garbage-collector"),
+		ctx.InformerFactory.Pods().Informer(),
+		int(ctx.Options.TerminatedPodGCThreshold),
+	).Run(ctx.Stop)
+	return true, nil
+}
+
+func startResourceQuotaController(ctx ControllerContext) (bool, error) {
+	resourceQuotaControllerClient := ctx.ClientBuilder.ClientOrDie("resourcequota-controller")
+	resourceQuotaRegistry := quotainstall.NewRegistry(resourceQuotaControllerClient, ctx.InformerFactory)
+	groupKindsToReplenish := []schema.GroupKind{
+		api.Kind("Pod"),
+		api.Kind("Service"),
+		api.Kind("ReplicationController"),
+		api.Kind("PersistentVolumeClaim"),
+		api.Kind("Secret"),
+		api.Kind("ConfigMap"),
+	}
+	resourceQuotaControllerOptions := &resourcequotacontroller.ResourceQuotaControllerOptions{
+		KubeClient:                resourceQuotaControllerClient,
+		ResyncPeriod:              controller.StaticResyncPeriodFunc(ctx.Options.ResourceQuotaSyncPeriod.Duration),
+		Registry:                  resourceQuotaRegistry,
+		ControllerFactory:         resourcequotacontroller.NewReplenishmentControllerFactory(ctx.InformerFactory, resourceQuotaControllerClient),
+		ReplenishmentResyncPeriod: ResyncPeriod(&ctx.Options),
+		GroupKindsToReplenish:     groupKindsToReplenish,
+	}
+	go resourcequotacontroller.NewResourceQuotaController(
+		resourceQuotaControllerOptions,
+	).Run(int(ctx.Options.ConcurrentResourceQuotaSyncs), ctx.Stop)
+	return true, nil
+}
+
+func startNamespaceController(ctx ControllerContext) (bool, error) {
+	// TODO: should use a dynamic RESTMapper built from the discovery results.
+	restMapper := registered.RESTMapper()
+
+	// Find the list of namespaced resources via discovery that the namespace controller must manage
+	namespaceKubeClient := ctx.ClientBuilder.ClientOrDie("namespace-controller")
+	namespaceClientPool := dynamic.NewClientPool(ctx.ClientBuilder.ConfigOrDie("namespace-controller"), restMapper, dynamic.LegacyAPIPathResolverFunc)
+	// TODO: consider using a list-watch + cache here rather than polling
+	resources, err := namespaceKubeClient.Discovery().ServerResources()
+	if err != nil {
+		return true, fmt.Errorf("failed to get preferred server resources: %v", err)
+	}
+	gvrs, err := discovery.GroupVersionResources(resources)
+	if err != nil {
+		return true, fmt.Errorf("failed to parse preferred server resources: %v", err)
+	}
+	discoverResourcesFn := namespaceKubeClient.Discovery().ServerPreferredNamespacedResources
+	if _, found := gvrs[extensions.SchemeGroupVersion.WithResource("thirdpartyresource")]; found {
+		// make discovery static
+		snapshot, err := discoverResourcesFn()
+		if err != nil {
+			return true, fmt.Errorf("failed to get server resources: %v", err)
+		}
+		discoverResourcesFn = func() ([]*metav1.APIResourceList, error) {
+			return snapshot, nil
+		}
+	}
+	namespaceController := namespacecontroller.NewNamespaceController(namespaceKubeClient, namespaceClientPool, discoverResourcesFn, ctx.Options.NamespaceSyncPeriod.Duration, v1.FinalizerKubernetes)
+	go namespaceController.Run(int(ctx.Options.ConcurrentNamespaceSyncs), ctx.Stop)
+
+	return true, nil
+
+}
+
+func startServiceAccountController(ctx ControllerContext) (bool, error) {
+	go serviceaccountcontroller.NewServiceAccountsController(
+		ctx.InformerFactory.ServiceAccounts(),
+		ctx.InformerFactory.Namespaces(),
+		ctx.ClientBuilder.ClientOrDie("service-account-controller"),
+		serviceaccountcontroller.DefaultServiceAccountsControllerOptions(),
+	).Run(1, ctx.Stop)
+	return true, nil
+}
+
+func startGarbageCollectorController(ctx ControllerContext) (bool, error) {
+	if !ctx.Options.EnableGarbageCollector {
+		return false, nil
+	}
+
+	// TODO: should use a dynamic RESTMapper built from the discovery results.
+	restMapper := registered.RESTMapper()
+
+	gcClientset := ctx.ClientBuilder.ClientOrDie("generic-garbage-collector")
+	preferredResources, err := gcClientset.Discovery().ServerPreferredResources()
+	if err != nil {
+		return true, fmt.Errorf("failed to get supported resources from server: %v", err)
+	}
+	deletableResources := discovery.FilteredBy(discovery.SupportsAllVerbs{Verbs: []string{"delete"}}, preferredResources)
+	deletableGroupVersionResources, err := discovery.GroupVersionResources(deletableResources)
+	if err != nil {
+		return true, fmt.Errorf("Failed to parse resources from server: %v", err)
+	}
+
+	config := ctx.ClientBuilder.ConfigOrDie("generic-garbage-collector")
+	config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
+	metaOnlyClientPool := dynamic.NewClientPool(config, restMapper, dynamic.LegacyAPIPathResolverFunc)
+	config.ContentConfig = dynamic.ContentConfig()
+	clientPool := dynamic.NewClientPool(config, restMapper, dynamic.LegacyAPIPathResolverFunc)
+	garbageCollector, err := garbagecollector.NewGarbageCollector(metaOnlyClientPool, clientPool, restMapper, deletableGroupVersionResources)
+	if err != nil {
+		return true, fmt.Errorf("Failed to start the generic garbage collector: %v", err)
+	}
+	workers := int(ctx.Options.ConcurrentGCSyncs)
+	go garbageCollector.Run(workers, ctx.Stop)
+
+	return true, nil
+}

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

+/*
+Copyright 2016 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 app implements a server that runs a set of active
+// components.  This includes replication controllers, service endpoints and
+// nodes.
+//
+package app
+
+import (
+	"k8s.io/kubernetes/pkg/controller/daemon"
+	"k8s.io/kubernetes/pkg/controller/deployment"
+	"k8s.io/kubernetes/pkg/controller/job"
+	replicaset "k8s.io/kubernetes/pkg/controller/replicaset"
+	"k8s.io/kubernetes/pkg/runtime/schema"
+)
+
+func startDaemonSetController(ctx ControllerContext) (bool, error) {
+	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "daemonsets"}] {
+		return false, nil
+	}
+	go daemon.NewDaemonSetsController(
+		ctx.InformerFactory.DaemonSets(),
+		ctx.InformerFactory.Pods(),
+		ctx.InformerFactory.Nodes(),
+		ctx.ClientBuilder.ClientOrDie("daemon-set-controller"),
+		int(ctx.Options.LookupCacheSizeForDaemonSet),
+	).Run(int(ctx.Options.ConcurrentDaemonSetSyncs), ctx.Stop)
+	return true, nil
+}
+
+func startJobController(ctx ControllerContext) (bool, error) {
+	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "jobs"}] {
+		return false, nil
+	}
+	go job.NewJobController(
+		ctx.InformerFactory.Pods().Informer(),
+		ctx.InformerFactory.Jobs(),
+		ctx.ClientBuilder.ClientOrDie("job-controller"),
+	).Run(int(ctx.Options.ConcurrentJobSyncs), ctx.Stop)
+	return true, nil
+}
+
+func startDeploymentController(ctx ControllerContext) (bool, error) {
+	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "deployments"}] {
+		return false, nil
+	}
+	go deployment.NewDeploymentController(
+		ctx.InformerFactory.Deployments(),
+		ctx.InformerFactory.ReplicaSets(),
+		ctx.InformerFactory.Pods(),
+		ctx.ClientBuilder.ClientOrDie("deployment-controller"),
+	).Run(int(ctx.Options.ConcurrentDeploymentSyncs), ctx.Stop)
+	return true, nil
+}
+
+func startReplicaSetController(ctx ControllerContext) (bool, error) {
+	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "replicasets"}] {
+		return false, nil
+	}
+	go replicaset.NewReplicaSetController(
+		ctx.InformerFactory.ReplicaSets(),
+		ctx.InformerFactory.Pods(),
+		ctx.ClientBuilder.ClientOrDie("replicaset-controller"),
+		replicaset.BurstReplicas,
+		int(ctx.Options.LookupCacheSizeForRS),
+		ctx.Options.EnableGarbageCollector,
+	).Run(int(ctx.Options.ConcurrentRSSyncs), ctx.Stop)
+	return true, nil
+}

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

+/*
+Copyright 2016 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 app implements a server that runs a set of active
+// components.  This includes replication controllers, service endpoints and
+// nodes.
+//
+package app
+
+import (
+	"k8s.io/kubernetes/pkg/controller/disruption"
+	"k8s.io/kubernetes/pkg/runtime/schema"
+)
+
+func startDisruptionController(ctx ControllerContext) (bool, error) {
+	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "policy", Version: "v1beta1", Resource: "poddisruptionbudgets"}] {
+		return false, nil
+	}
+	go disruption.NewDisruptionController(
+		ctx.InformerFactory.Pods().Informer(),
+		ctx.ClientBuilder.ClientOrDie("disruption-controller"),
+	).Run(ctx.Stop)
+	return true, nil
+}