[Backport 1.21] Move cloud-controller-manager into an embedded executor (#3530)

pkg/daemons/control/server.go

@@ -20,23 +20,16 @@ import (
 	"github.com/rancher/k3s/pkg/version"
 	"github.com/rancher/wrangler-api/pkg/generated/controllers/rbac"
 	"github.com/sirupsen/logrus"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/authentication/authenticator"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/clientcmd"
-	ccm "k8s.io/cloud-provider"
-	cloudprovider "k8s.io/cloud-provider"
-	ccmapp "k8s.io/cloud-provider/app"
-	cloudcontrollerconfig "k8s.io/cloud-provider/app/config"
-	ccmopt "k8s.io/cloud-provider/options"
-	cliflag "k8s.io/component-base/cli/flag"
 	app2 "k8s.io/controller-manager/app"
 	"k8s.io/kubernetes/pkg/kubeapiserver/authorizer/modes"
 	"k8s.io/kubernetes/pkg/proxy/util"
 
-	// registering k3s cloud provider
-	_ "github.com/rancher/k3s/pkg/cloudprovider"
 	// for client metric registration
 	_ "k8s.io/component-base/metrics/prometheus/restclient"
 )
@@ -90,7 +83,9 @@ func Server(ctx context.Context, cfg *config.Control) error {
 	}
 
 	if !cfg.DisableCCM {
-		cloudControllerManager(ctx, cfg, runtime)
+		if err := cloudControllerManager(ctx, cfg, runtime); err != nil {
+			return err
+		}
 	}
 
 	return nil
@@ -347,89 +342,79 @@ func setupStorageBackend(argsMap map[string]string, cfg *config.Control) {
 	}
 }
 
-func cloudControllerManager(ctx context.Context, cfg *config.Control, runtime *config.ControlRuntime) {
-	// Reference: https://github.com/kubernetes/kubernetes/pull/93764
-	// The above-linked change made the in-tree cloud-controller-manager command much more flexible
-	// but much more complicated to wrap. It now validates some of the configuration very early on, before
-	// the CLI args are parsed, so some of the configuration needs to be hardcoded instead of set via flags.
-	// Reference: https://github.com/kubernetes/kubernetes/pull/98210
-	// The above-linked change further clarifies the intent of the example cloud-controller-manager.
-
+func cloudControllerManager(ctx context.Context, cfg *config.Control, runtime *config.ControlRuntime) error {
 	argsMap := map[string]string{
-		"profiling": "false",
-	}
-	args := config.GetArgsList(argsMap, cfg.ExtraCloudControllerArgs)
-
-	ccmOptions, err := ccmopt.NewCloudControllerManagerOptions()
-	if err != nil {
-		logrus.Fatalf("Unable to initialize cloudcontroller options: %v", err)
+		"profiling":                    "false",
+		"allocate-node-cidrs":          "true",                                // ccmOptions.KubeCloudShared.AllocateNodeCIDRs = true
+		"cloud-provider":               version.Program,                       // ccmOptions.KubeCloudShared.CloudProvider.Name = version.Program
+		"cluster-cidr":                 util2.JoinIPNets(cfg.ClusterIPRanges), // ccmOptions.KubeCloudShared.ClusterCIDR = util2.JoinIPNets(cfg.ClusterIPRanges)
+		"configure-cloud-routes":       "false",                               // ccmOptions.KubeCloudShared.ConfigureCloudRoutes = false
+		"kubeconfig":                   runtime.KubeConfigCloudController,     // ccmOptions.Kubeconfig = runtime.KubeConfigCloudController
+		"node-status-update-frequency": "1m0s",                                // ccmOptions.NodeStatusUpdateFrequency = metav1.Duration{Duration: 1 * time.Minute}
+		"bind-address":                 "127.0.0.1",                           // ccmOptions.SecureServing.BindAddress = localhostIP
+		"port":                         "0",                                   // ccmOptions.SecureServing.BindPort = 0
 	}
-
-	ccmOptions.KubeCloudShared.AllocateNodeCIDRs = true
-	ccmOptions.KubeCloudShared.CloudProvider.Name = version.Program
-	ccmOptions.KubeCloudShared.ClusterCIDR = util2.JoinIPNets(cfg.ClusterIPRanges)
-	ccmOptions.KubeCloudShared.ConfigureCloudRoutes = false
-	ccmOptions.Kubeconfig = runtime.KubeConfigCloudController
-	ccmOptions.NodeStatusUpdateFrequency = metav1.Duration{Duration: 1 * time.Minute}
-	ccmOptions.SecureServing.BindAddress = localhostIP
-	ccmOptions.SecureServing.BindPort = 0
-
 	if cfg.NoLeaderElect {
-		ccmOptions.Generic.LeaderElection.LeaderElect = false
+		argsMap["leader-elect"] = "false"
 	}
+	args := config.GetArgsList(argsMap, cfg.ExtraCloudControllerArgs)
 
-	controllerInitializers := ccmapp.DefaultInitFuncConstructors
-	delete(controllerInitializers, "service")
-	delete(controllerInitializers, "route")
+	logrus.Infof("Running cloud-controller-manager %s", config.ArgString(args))
 
-	cloudInitializer := func(config *cloudcontrollerconfig.CompletedConfig) cloudprovider.Interface {
-		cloud, err := ccm.InitCloudProvider(version.Program, runtime.KubeConfigCloudController)
-		if err != nil {
-			logrus.Fatalf("Cloud provider could not be initialized: %v", err)
-		}
-		if cloud == nil {
-			logrus.Fatalf("Cloud provider is nil")
-		}
+	ccmRBACReady := make(chan struct{})
 
-		cloud.Initialize(config.ClientBuilder, make(chan struct{}))
-		if informerUserCloud, ok := cloud.(ccm.InformerUser); ok {
-			informerUserCloud.SetInformers(config.SharedInformers)
-		}
-
-		return cloud
-	}
+	go func() {
+		defer close(ccmRBACReady)
 
-	command := ccmapp.NewCloudControllerManagerCommand(ccmOptions, cloudInitializer, controllerInitializers, cliflag.NamedFlagSets{}, wait.NeverStop)
-	command.SetArgs(args)
+	apiReadyLoop:
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-runtime.APIServerReady:
+				break apiReadyLoop
+			case <-time.After(30 * time.Second):
+				logrus.Infof("Waiting for API server to become available")
+			}
+		}
 
-	go func() {
+		logrus.Infof("Waiting for cloud-controller-manager privileges to become available")
 		for {
-			// check for the cloud controller rbac binding
-			if err := checkForCloudControllerPrivileges(runtime); err != nil {
-				logrus.Infof("Waiting for cloudcontroller rbac role to be created")
-				select {
-				case <-ctx.Done():
-					logrus.Fatalf("cloud-controller-manager context canceled: %v", ctx.Err())
-				case <-time.After(5 * time.Second):
+			select {
+			case <-ctx.Done():
+				return
+			case err := <-promise(func() error { return checkForCloudControllerPrivileges(runtime, 5*time.Second) }):
+				if err != nil {
+					logrus.Infof("Waiting for cloud-controller-manager privileges to become available")
 					continue
 				}
+				return
 			}
-			break
 		}
-		logrus.Infof("Running cloud-controller-manager with args %v", config.ArgString(args))
-		logrus.Fatalf("cloud-controller-manager exited: %v", command.ExecuteContext(ctx))
 	}()
+
+	return executor.CloudControllerManager(ccmRBACReady, args)
 }
 
-func checkForCloudControllerPrivileges(runtime *config.ControlRuntime) error {
+func checkForCloudControllerPrivileges(runtime *config.ControlRuntime, timeout time.Duration) error {
 	restConfig, err := clientcmd.BuildConfigFromFlags("", runtime.KubeConfigAdmin)
 	if err != nil {
 		return err
 	}
-	crb := rbac.NewFactoryFromConfigOrDie(restConfig).Rbac().V1().ClusterRoleBinding()
-	_, err = crb.Get(version.Program+"-cloud-controller-manager", metav1.GetOptions{})
+	err = wait.PollImmediate(time.Second, timeout, func() (bool, error) {
+		crb := rbac.NewFactoryFromConfigOrDie(restConfig).Rbac().V1().ClusterRoleBinding()
+		_, err = crb.Get(version.Program+"-cloud-controller-manager", metav1.GetOptions{})
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				return false, nil
+			}
+			return false, err
+		}
+		return true, nil
+	})
+
 	if err != nil {
-		return err
+		logrus.Errorf("error encountered waitng for cloud-controller-manager privileges: %v", err)
 	}
 	return nil
 }

pkg/daemons/executor/embed.go

@@ -6,17 +6,26 @@ import (
 	"context"
 	"net/http"
 
-	"k8s.io/apiserver/pkg/authentication/authenticator"
-
-	proxy "k8s.io/kubernetes/cmd/kube-proxy/app"
-	kubelet "k8s.io/kubernetes/cmd/kubelet/app"
-
 	"github.com/rancher/k3s/pkg/cli/cmds"
 	daemonconfig "github.com/rancher/k3s/pkg/daemons/config"
+	"github.com/rancher/k3s/pkg/version"
 	"github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apiserver/pkg/authentication/authenticator"
+	ccm "k8s.io/cloud-provider"
+	cloudprovider "k8s.io/cloud-provider"
+	ccmapp "k8s.io/cloud-provider/app"
+	cloudcontrollerconfig "k8s.io/cloud-provider/app/config"
+	ccmopt "k8s.io/cloud-provider/options"
+	cliflag "k8s.io/component-base/cli/flag"
 	"k8s.io/kubernetes/cmd/kube-apiserver/app"
 	cmapp "k8s.io/kubernetes/cmd/kube-controller-manager/app"
+	proxy "k8s.io/kubernetes/cmd/kube-proxy/app"
 	sapp "k8s.io/kubernetes/cmd/kube-scheduler/app"
+	kubelet "k8s.io/kubernetes/cmd/kubelet/app"
+
+	// registering k3s cloud provider
+	_ "github.com/rancher/k3s/pkg/cloudprovider"
 )
 
 func init() {
@@ -87,3 +96,41 @@ func (Embedded) ControllerManager(apiReady <-chan struct{}, args []string) error
 
 	return nil
 }
+
+func (Embedded) CloudControllerManager(ccmRBACReady <-chan struct{}, args []string) error {
+	ccmOptions, err := ccmopt.NewCloudControllerManagerOptions()
+	if err != nil {
+		logrus.Fatalf("unable to initialize command options: %v", err)
+	}
+
+	cloudInitializer := func(config *cloudcontrollerconfig.CompletedConfig) cloudprovider.Interface {
+		cloud, err := ccm.InitCloudProvider(version.Program, "")
+		if err != nil {
+			logrus.Fatalf("Cloud provider could not be initialized: %v", err)
+		}
+		if cloud == nil {
+			logrus.Fatalf("Cloud provider is nil")
+		}
+
+		cloud.Initialize(config.ClientBuilder, make(chan struct{}))
+		if informerUserCloud, ok := cloud.(ccm.InformerUser); ok {
+			informerUserCloud.SetInformers(config.SharedInformers)
+		}
+
+		return cloud
+	}
+
+	controllerInitializers := ccmapp.DefaultInitFuncConstructors
+	delete(controllerInitializers, "service")
+	delete(controllerInitializers, "route")
+
+	command := ccmapp.NewCloudControllerManagerCommand(ccmOptions, cloudInitializer, controllerInitializers, cliflag.NamedFlagSets{}, wait.NeverStop)
+	command.SetArgs(args)
+
+	go func() {
+		<-ccmRBACReady
+		logrus.Fatalf("cloud-controller-manager exited: %v", command.Execute())
+	}()
+
+	return nil
+}

pkg/daemons/executor/executor.go

@@ -27,6 +27,7 @@ type Executor interface {
 	ControllerManager(apiReady <-chan struct{}, args []string) error
 	CurrentETCDOptions() (InitialOptions, error)
 	ETCD(args ETCDConfig) error
+	CloudControllerManager(ccmRBACReady <-chan struct{}, args []string) error
 }
 
 type ETCDConfig struct {
@@ -115,3 +116,7 @@ func CurrentETCDOptions() (InitialOptions, error) {
 func ETCD(args ETCDConfig) error {
 	return executor.ETCD(args)
 }
+
+func CloudControllerManager(ccmRBACReady <-chan struct{}, args []string) error {
+	return executor.CloudControllerManager(ccmRBACReady, args)
+}