run kube controllers separately based on their command

pkg/cmd/server/start/start_kube_controller_manager.go

@@ -1,10 +1,24 @@
 package start
 
 import (
-	"github.com/openshift/origin/pkg/cmd/server/bootstrappolicy"
+	"strconv"
+
+	"github.com/golang/glog"
+	"github.com/spf13/pflag"
+
+	kerrors "k8s.io/apimachinery/pkg/util/errors"
+	controllerapp "k8s.io/kubernetes/cmd/kube-controller-manager/app"
+	controlleroptions "k8s.io/kubernetes/cmd/kube-controller-manager/app/options"
 	"k8s.io/kubernetes/pkg/api/v1"
 	kapiv1 "k8s.io/kubernetes/pkg/api/v1"
+	kubeexternalinformers "k8s.io/kubernetes/pkg/client/informers/informers_generated/externalversions"
+	"k8s.io/kubernetes/pkg/controller"
 	"k8s.io/kubernetes/pkg/volume"
+	_ "k8s.io/kubernetes/plugin/pkg/scheduler/algorithmprovider"
+
+	"github.com/openshift/origin/pkg/cmd/server/bootstrappolicy"
+	cmdflags "github.com/openshift/origin/pkg/cmd/util/flags"
+	"k8s.io/kubernetes/pkg/apis/componentconfig"
 )
 
 // newPersistentVolumeRecyclerPodTemplate provides a function which makes our recycler pod template for use in the kube-controller-manager
@@ -26,3 +40,143 @@ func newPersistentVolumeRecyclerPodTemplate(recyclerImageName string) func() *v1
 		return defaultScrubPod
 	}
 }
+
+// newControllerContext provides a function which overrides the default and plugs a different set of informers in
+func newControllerContext(kubeExternalInformers kubeexternalinformers.SharedInformerFactory) func(s *controlleroptions.CMServer, rootClientBuilder, clientBuilder controller.ControllerClientBuilder, stop <-chan struct{}) (controllerapp.ControllerContext, error) {
+	oldContextFunc := controllerapp.CreateControllerContext
+	return func(s *controlleroptions.CMServer, rootClientBuilder, clientBuilder controller.ControllerClientBuilder, stop <-chan struct{}) (controllerapp.ControllerContext, error) {
+		ret, err := oldContextFunc(s, rootClientBuilder, clientBuilder, stop)
+		if err != nil {
+			return controllerapp.ControllerContext{}, err
+		}
+
+		// Overwrite the informers.  Since nothing accessed the existing informers that we're overwriting, they are inert.
+		// TODO Remove this.  It keeps in-process memory utilization down, but we shouldn't do it.
+		ret.InformerFactory = kubeExternalInformers
+
+		return ret, nil
+	}
+}
+
+func kubeControllerManagerAddFlags(cmserver *controlleroptions.CMServer) func(flags *pflag.FlagSet) {
+	return func(flags *pflag.FlagSet) {
+		cmserver.AddFlags(flags, controllerapp.KnownControllers(), controllerapp.ControllersDisabledByDefault.List())
+	}
+}
+
+func newKubeControllerManager(kubeconfigFile, saPrivateKeyFile, saRootCAFile, podEvictionTimeout string, dynamicProvisioningEnabled bool, cmdLineArgs map[string][]string) (*controlleroptions.CMServer, error) {
+	if cmdLineArgs == nil {
+		cmdLineArgs = map[string][]string{}
+	}
+
+	if _, ok := cmdLineArgs["controllers"]; !ok {
+		cmdLineArgs["controllers"] = []string{
+			"*", // start everything but the exceptions}
+			// we don't appear to use this
+			"-ttl",
+			// we have to configure this separately until it is generic
+			"-horizontalpodautoscaling",
+			// we carry patches on this. For now....
+			"-serviceaccount-token",
+		}
+	}
+	if _, ok := cmdLineArgs["service-account-private-key-file"]; !ok {
+		cmdLineArgs["service-account-private-key-file"] = []string{saPrivateKeyFile}
+	}
+	if _, ok := cmdLineArgs["root-ca-file"]; !ok {
+		cmdLineArgs["root-ca-file"] = []string{saRootCAFile}
+	}
+	if _, ok := cmdLineArgs["kubeconfig"]; !ok {
+		cmdLineArgs["kubeconfig"] = []string{kubeconfigFile}
+	}
+	if _, ok := cmdLineArgs["pod-eviction-timeout"]; !ok {
+		cmdLineArgs["pod-eviction-timeout"] = []string{podEvictionTimeout}
+	}
+	if _, ok := cmdLineArgs["enable-dynamic-provisioning"]; !ok {
+		cmdLineArgs["enable-dynamic-provisioning"] = []string{strconv.FormatBool(dynamicProvisioningEnabled)}
+	}
+
+	// disable serving http since we didn't used to expose it
+	if _, ok := cmdLineArgs["port"]; !ok {
+		cmdLineArgs["port"] = []string{"-1"}
+	}
+
+	// these force "default" values to match what we want
+	if _, ok := cmdLineArgs["use-service-account-credentials"]; !ok {
+		cmdLineArgs["use-service-account-credentials"] = []string{"true"}
+	}
+	if _, ok := cmdLineArgs["cluster-signing-cert-file"]; !ok {
+		cmdLineArgs["cluster-signing-cert-file"] = []string{""}
+	}
+	if _, ok := cmdLineArgs["cluster-signing-key-file"]; !ok {
+		cmdLineArgs["cluster-signing-key-file"] = []string{""}
+	}
+	if _, ok := cmdLineArgs["experimental-cluster-signing-duration"]; !ok {
+		cmdLineArgs["experimental-cluster-signing-duration"] = []string{"0s"}
+	}
+	if _, ok := cmdLineArgs["leader-elect-retry-period"]; !ok {
+		cmdLineArgs["leader-elect-retry-period"] = []string{"3s"}
+	}
+	if _, ok := cmdLineArgs["leader-elect-resource-lock"]; !ok {
+		cmdLineArgs["leader-elect-resource-lock"] = []string{"configmaps"}
+	}
+
+	// resolve arguments
+	controllerManager := controlleroptions.NewCMServer()
+	if err := cmdflags.Resolve(cmdLineArgs, kubeControllerManagerAddFlags(controllerManager)); len(err) > 0 {
+		return nil, kerrors.NewAggregate(err)
+	}
+
+	// TODO make this configurable or discoverable.  This is going to prevent us from running the stock GC controller
+	// IF YOU ADD ANYTHING TO THIS LIST, MAKE SURE THAT YOU UPDATE THEIR STRATEGIES TO PREVENT GC FINALIZERS
+	controllerManager.GCIgnoredResources = append(controllerManager.GCIgnoredResources,
+		// explicitly disabled from GC for now - not enough value to track them
+		componentconfig.GroupResource{Group: "authorization.openshift.io", Resource: "rolebindingrestrictions"},
+		componentconfig.GroupResource{Group: "network.openshift.io", Resource: "clusternetworks"},
+		componentconfig.GroupResource{Group: "network.openshift.io", Resource: "egressnetworkpolicies"},
+		componentconfig.GroupResource{Group: "network.openshift.io", Resource: "hostsubnets"},
+		componentconfig.GroupResource{Group: "network.openshift.io", Resource: "netnamespaces"},
+		componentconfig.GroupResource{Group: "oauth.openshift.io", Resource: "oauthclientauthorizations"},
+		componentconfig.GroupResource{Group: "oauth.openshift.io", Resource: "oauthclients"},
+		componentconfig.GroupResource{Group: "quota.openshift.io", Resource: "clusterresourcequotas"},
+		componentconfig.GroupResource{Group: "user.openshift.io", Resource: "groups"},
+		componentconfig.GroupResource{Group: "user.openshift.io", Resource: "identities"},
+		componentconfig.GroupResource{Group: "user.openshift.io", Resource: "users"},
+		componentconfig.GroupResource{Group: "image.openshift.io", Resource: "images"},
+
+		// virtual resource
+		componentconfig.GroupResource{Group: "project.openshift.io", Resource: "projects"},
+		// these resources contain security information in their names, and we don't need to track them
+		componentconfig.GroupResource{Group: "oauth.openshift.io", Resource: "oauthaccesstokens"},
+		componentconfig.GroupResource{Group: "oauth.openshift.io", Resource: "oauthauthorizetokens"},
+		// exposed already as cronjobs
+		componentconfig.GroupResource{Group: "batch", Resource: "scheduledjobs"},
+		// exposed already as extensions v1beta1 by other controllers
+		componentconfig.GroupResource{Group: "apps", Resource: "deployments"},
+		// exposed as autoscaling v1
+		componentconfig.GroupResource{Group: "extensions", Resource: "horizontalpodautoscalers"},
+	)
+
+	return controllerManager, nil
+}
+
+func runEmbeddedKubeControllerManager(kubeconfigFile, saPrivateKeyFile, saRootCAFile, podEvictionTimeout string, dynamicProvisioningEnabled bool, cmdLineArgs map[string][]string,
+	recyclerImage string, kubeExternalInformers kubeexternalinformers.SharedInformerFactory) {
+	volume.NewPersistentVolumeRecyclerPodTemplate = newPersistentVolumeRecyclerPodTemplate(recyclerImage)
+	controllerapp.CreateControllerContext = newControllerContext(kubeExternalInformers)
+
+	for {
+		// TODO we need a real identity for this.  Right now it's just using the loopback connection like it used to.
+		controllerManager, err := newKubeControllerManager(kubeconfigFile, saPrivateKeyFile, saRootCAFile, podEvictionTimeout, dynamicProvisioningEnabled, cmdLineArgs)
+		if err != nil {
+			glog.Error(err)
+			continue
+		}
+		// this does a second leader election, but doing the second leader election will allow us to move out process in
+		// 3.8 if we so choose.
+		if err := controllerapp.Run(controllerManager); err != nil {
+			glog.Error(err)
+			continue
+		}
+	}
+}

pkg/cmd/server/start/start_master.go

@@ -27,15 +27,13 @@ import (
 	clientgoclientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 	aggregatorinstall "k8s.io/kube-aggregator/pkg/apis/apiregistration/install"
-	kubecontroller "k8s.io/kubernetes/cmd/kube-controller-manager/app"
 	kapi "k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/capabilities"
 	kinformers "k8s.io/kubernetes/pkg/client/informers/informers_generated/externalversions"
 	"k8s.io/kubernetes/pkg/kubectl/cmd/templates"
 	kcmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
 	kubelettypes "k8s.io/kubernetes/pkg/kubelet/types"
 	"k8s.io/kubernetes/pkg/master"
-	"k8s.io/kubernetes/pkg/volume"
 	kutilerrors "k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/errors"
 
 	assetapiserver "github.com/openshift/origin/pkg/assets/apiserver"
@@ -410,10 +408,10 @@ func (m *Master) Start() error {
 		imageTemplate := variable.NewDefaultImageTemplate()
 		imageTemplate.Format = m.config.ImageConfig.Format
 		imageTemplate.Latest = m.config.ImageConfig.Latest
-		volume.NewPersistentVolumeRecyclerPodTemplate = newPersistentVolumeRecyclerPodTemplate(imageTemplate.ExpandOrDie("recycler"))
+		recyclerImage := imageTemplate.ExpandOrDie("recycler")
 
+		// you can't double run healthz, so only do this next bit if we aren't starting the API
 		if !m.api {
-			// you can't double run healthz, so only do this next bit if we aren't starting the API
 
 			glog.Infof("Starting controllers on %s (%s)", m.config.ServingInfo.BindAddress, version.Get().String())
 			if len(m.config.DisabledFeatures) > 0 {
@@ -478,6 +476,16 @@ func (m *Master) Start() error {
 			// continuously run the scheduler while we have the primary lease
 			go runEmbeddedScheduler(m.config.MasterClients.OpenShiftLoopbackKubeConfig, m.config.KubernetesMasterConfig.SchedulerConfigFile, m.config.KubernetesMasterConfig.SchedulerArguments)
 
+			go runEmbeddedKubeControllerManager(
+				m.config.MasterClients.OpenShiftLoopbackKubeConfig,
+				m.config.ServiceAccountConfig.PrivateKeyFile,
+				m.config.ServiceAccountConfig.MasterCA,
+				m.config.KubernetesMasterConfig.PodEvictionTimeout,
+				m.config.VolumeConfig.DynamicProvisioningEnabled,
+				m.config.KubernetesMasterConfig.ControllerArguments,
+				recyclerImage,
+				informers.GetExternalKubeInformers())
+
 			controllerContext, err := getControllerContext(*m.config, kubeControllerManagerConfig, cloudProvider, informers, utilwait.NeverStop)
 			if err != nil {
 				glog.Fatal(err)
@@ -702,20 +710,10 @@ func startControllers(options configapi.MasterConfig, allocationController origi
 
 	allocationController.RunSecurityAllocationController()
 
-	// set the upstream default until it is configurable
-	kubernetesControllerInitializers := kubecontroller.NewControllerInitializers()
 	openshiftControllerInitializers, err := openshiftControllerConfig.GetControllerInitializers()
 	if err != nil {
 		return err
 	}
-	// Add kubernetes controllers initialized from Origin
-	for name, initFn := range kubernetesControllerInitializers {
-		if _, exists := openshiftControllerInitializers[name]; exists {
-			// don't overwrite, openshift takes priority
-			continue
-		}
-		openshiftControllerInitializers[name] = origincontrollers.FromKubeInitFunc(initFn)
-	}
 
 	excludedControllers := getExcludedControllers(options)
 
@@ -749,19 +747,11 @@ func startControllers(options configapi.MasterConfig, allocationController origi
 }
 
 func getExcludedControllers(options configapi.MasterConfig) sets.String {
-	excludedControllers := sets.NewString(
-		// not used in openshift.  Yet?
-		"ttl",
-		"bootstrapsigner",
-		"tokencleaner",
-		// remove the HPA controller until it is generic
-		"horizontalpodautoscaling",
-	)
+	excludedControllers := sets.NewString()
 	if !configapi.IsBuildEnabled(&options) {
 		excludedControllers.Insert("openshift.io/build")
 		excludedControllers.Insert("openshift.io/build-config-change")
 	}
-
 	return excludedControllers
 }