run controller by wiring up to a command

Author: deads2k

pkg/cmd/server/kubernetes/master/controller/config.go

@@ -1,19 +1,9 @@
 package controller
 
 import (
-	"fmt"
-	"io/ioutil"
-	"os"
-
-	"k8s.io/apimachinery/pkg/runtime"
-	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	kubecontroller "k8s.io/kubernetes/cmd/kube-controller-manager/app"
-	scheduleroptions "k8s.io/kubernetes/plugin/cmd/kube-scheduler/app/options"
-	schedulerapi "k8s.io/kubernetes/plugin/pkg/scheduler/api"
-	latestschedulerapi "k8s.io/kubernetes/plugin/pkg/scheduler/api/latest"
 
 	configapi "github.com/openshift/origin/pkg/cmd/server/api"
-	cmdflags "github.com/openshift/origin/pkg/cmd/util/flags"
 	"github.com/openshift/origin/pkg/cmd/util/variable"
 	"k8s.io/kubernetes/pkg/volume"
 )
@@ -22,9 +12,6 @@ import (
 // launch the set of kube (not openshift) controllers.
 type KubeControllerConfig struct {
 	HorizontalPodAutoscalerControllerConfig HorizontalPodAutoscalerControllerConfig
-
-	// TODO the scheduler should move out into its own logical component
-	SchedulerControllerConfig SchedulerControllerConfig
 }
 
 // GetControllerInitializers return the controller initializer functions for kube controllers
@@ -36,48 +23,14 @@ func (c KubeControllerConfig) GetControllerInitializers() (map[string]kubecontro
 	// in openshift-infra, and pass it a scale client that knows how to scale DCs
 	ret["horizontalpodautoscaling"] = c.HorizontalPodAutoscalerControllerConfig.RunController
 
-	// FIXME: Move this under openshift controller intialization once we figure out
-	// deployment (options).
-	ret["openshift.io/scheduler"] = c.SchedulerControllerConfig.RunController
-
 	return ret, nil
 }
 
 // BuildKubeControllerConfig builds the struct to create the controller initializers.  Eventually we want this to be fully
 // stock kube with no modification.
 func BuildKubeControllerConfig(options configapi.MasterConfig) (*KubeControllerConfig, error) {
-	var err error
 	ret := &KubeControllerConfig{}
 
-	kubeExternal, _, err := configapi.GetExternalKubeClient(options.MasterClients.OpenShiftLoopbackKubeConfig, options.MasterClients.OpenShiftLoopbackClientConnectionOverrides)
-	if err != nil {
-		return nil, err
-	}
-
-	var schedulerPolicy *schedulerapi.Policy
-	if _, err := os.Stat(options.KubernetesMasterConfig.SchedulerConfigFile); err == nil {
-		schedulerPolicy = &schedulerapi.Policy{}
-		configData, err := ioutil.ReadFile(options.KubernetesMasterConfig.SchedulerConfigFile)
-		if err != nil {
-			return nil, fmt.Errorf("unable to read scheduler config: %v", err)
-		}
-		if err := runtime.DecodeInto(latestschedulerapi.Codec, configData, schedulerPolicy); err != nil {
-			return nil, fmt.Errorf("invalid scheduler configuration: %v", err)
-		}
-	}
-	// resolve extended arguments
-	// TODO: this should be done in config validation (along with the above) so we can provide
-	// proper errors
-	schedulerserver := scheduleroptions.NewSchedulerServer()
-	schedulerserver.PolicyConfigFile = options.KubernetesMasterConfig.SchedulerConfigFile
-	if err := cmdflags.Resolve(options.KubernetesMasterConfig.SchedulerArguments, schedulerserver.AddFlags); len(err) > 0 {
-		return nil, kerrors.NewAggregate(err)
-	}
-	ret.SchedulerControllerConfig = SchedulerControllerConfig{
-		PrivilegedClient: kubeExternal,
-		SchedulerServer:  schedulerserver,
-	}
-
 	imageTemplate := variable.NewDefaultImageTemplate()
 	imageTemplate.Format = options.ImageConfig.Format
 	imageTemplate.Latest = options.ImageConfig.Latest

pkg/cmd/server/kubernetes/master/controller/core.go

@@ -472,6 +472,9 @@ func (m *Master) Start() error {
 		go func() {
 			controllerPlug.WaitForStart()
 
+			// continuously run the scheduler while we have the primary lease
+			go runEmbeddedScheduler(m.config.MasterClients.OpenShiftLoopbackKubeConfig, m.config.KubernetesMasterConfig.SchedulerConfigFile, m.config.KubernetesMasterConfig.SchedulerArguments)
+
 			controllerContext, err := getControllerContext(*m.config, kubeControllerManagerConfig, cloudProvider, informers, utilwait.NeverStop)
 			if err != nil {
 				glog.Fatal(err)

pkg/cmd/server/start/start_master.go

@@ -0,0 +1,53 @@
+package start
+
+import (
+	"github.com/golang/glog"
+
+	kerrors "k8s.io/apimachinery/pkg/util/errors"
+	schedulerapp "k8s.io/kubernetes/plugin/cmd/kube-scheduler/app"
+	scheduleroptions "k8s.io/kubernetes/plugin/cmd/kube-scheduler/app/options"
+	_ "k8s.io/kubernetes/plugin/pkg/scheduler/algorithmprovider"
+
+	cmdflags "github.com/openshift/origin/pkg/cmd/util/flags"
+)
+
+func newScheduler(kubeconfigFile, schedulerConfigFile string, cmdLineArgs map[string][]string) (*scheduleroptions.SchedulerServer, error) {
+	if cmdLineArgs == nil {
+		cmdLineArgs = map[string][]string{}
+	}
+	if len(cmdLineArgs["kubeconfig"]) == 0 {
+		cmdLineArgs["kubeconfig"] = []string{kubeconfigFile}
+	}
+	if len(cmdLineArgs["policy-config-file"]) == 0 {
+		cmdLineArgs["policy-config-file"] = []string{schedulerConfigFile}
+	}
+	// disable serving http since we didn't used to expose it
+	if len(cmdLineArgs["port"]) == 0 {
+		cmdLineArgs["port"] = []string{"-1"}
+	}
+
+	// resolve arguments
+	schedulerServer := scheduleroptions.NewSchedulerServer()
+	if err := cmdflags.Resolve(cmdLineArgs, schedulerServer.AddFlags); len(err) > 0 {
+		return nil, kerrors.NewAggregate(err)
+	}
+
+	return schedulerServer, nil
+}
+
+func runEmbeddedScheduler(kubeconfigFile, schedulerConfigFile string, cmdLineArgs map[string][]string) {
+	for {
+		// TODO we need a real identity for this.  Right now it's just using the loopback connection like it used to.
+		scheduler, err := newScheduler(kubeconfigFile, schedulerConfigFile, 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 := schedulerapp.Run(scheduler); err != nil {
+			glog.Error(err)
+			continue
+		}
+	}
+}