feat(platform): controller support crd mode (#2026)

api/platform/types.go

@@ -113,6 +113,8 @@ const (
 	AnywhereLocalizationsAnno = "tkestack.io/anywhere-localizations"
 	// AnywhereMachinesAnno contains base64 machines json data
 	AnywhereMachinesAnno = "tkestack.io/anywhere-machines"
+	// ClusterNameLable contains related cluster's name for no-cluster resources
+	ClusterNameLable = "tkestack.io/cluster-name"
 )
 
 // KubeVendorType describe the kubernetes provider of the cluster

api/platform/v1/types.go

@@ -124,6 +124,8 @@ const (
 	AnywhereLocalizationsAnno = "tkestack.io/anywhere-localizations"
 	// AnywhereMachinesAnno contains base64 machines json data
 	AnywhereMachinesAnno = "tkestack.io/anywhere-machines"
+	// ClusterNameLable contains related cluster's name for no-cluster resources
+	ClusterNameLable = "tkestack.io/cluster-name"
 )
 
 // KubeVendorType describe the kubernetes provider of the cluster

cmd/tke-platform-controller/app/options/clustercontroller.go

@@ -33,6 +33,7 @@ const (
 	flagUpperLimitofRandomHealthCheckPeriod = "upper-limit-random-healthcheck-period"
 	flagClusterRateLimiterLimit             = "cluster-rate-limiter-limit"
 	flagClusterRateLimiterBurst             = "cluster-rate-limiter-burst"
+	flagClusterIsCRDMode                    = "cluster-is-crd-mode"
 )
 
 const (
@@ -43,6 +44,7 @@ const (
 	configUpperLimitofRandomHealthCheckPeriod = "controller.upper-limit-random-healthcheck-period"
 	configClusterRateLimiterLimit             = "controller.cluster_rate_limiter_limit"
 	configClusterRateLimiterBurst             = "controller.cluster_rate_limiter_burst"
+	configClusterIsCRDMode                    = "controller.cluster_is_crd_mode"
 )
 
 // ClusterControllerOptions holds the ClusterController options.
@@ -61,6 +63,7 @@ func NewClusterControllerOptions() *ClusterControllerOptions {
 			RandomeRangeUpperLimitForHealthCheckPeriod: defaultRandomeRangeUpperLimitForHealthCheckPeriod,
 			BucketRateLimiterLimit:                     defaultBucketRateLimiterLimit,
 			BucketRateLimiterBurst:                     defaultBucketRateLimiterBurst,
+			IsCRDMode:                                  false,
 		},
 	}
 }
@@ -91,6 +94,9 @@ func (o *ClusterControllerOptions) AddFlags(fs *pflag.FlagSet) {
 
 	fs.IntVar(&o.BucketRateLimiterBurst, flagClusterRateLimiterBurst, o.BucketRateLimiterBurst, "The number of bursts of at most b tokens.")
 	_ = viper.BindPFlag(configClusterRateLimiterBurst, fs.Lookup(flagClusterRateLimiterBurst))
+
+	fs.BoolVar(&o.IsCRDMode, flagClusterIsCRDMode, o.IsCRDMode, "Whether the controller is using CRD mode")
+	_ = viper.BindPFlag(configClusterIsCRDMode, fs.Lookup(flagClusterIsCRDMode))
 }
 
 // ApplyTo fills up ClusterController config with options.
@@ -106,6 +112,7 @@ func (o *ClusterControllerOptions) ApplyTo(cfg *clusterconfig.ClusterControllerC
 	cfg.RandomeRangeUpperLimitForHealthCheckPeriod = o.RandomeRangeUpperLimitForHealthCheckPeriod
 	cfg.BucketRateLimiterLimit = o.BucketRateLimiterLimit
 	cfg.BucketRateLimiterBurst = o.BucketRateLimiterBurst
+	cfg.IsCRDMode = o.IsCRDMode
 
 	return nil
 }
@@ -130,5 +137,6 @@ func (o *ClusterControllerOptions) ApplyFlags() []error {
 	o.RandomeRangeUpperLimitForHealthCheckPeriod = viper.GetDuration(configUpperLimitofRandomHealthCheckPeriod)
 	o.BucketRateLimiterLimit = viper.GetInt(configClusterRateLimiterLimit)
 	o.BucketRateLimiterBurst = viper.GetInt(configClusterRateLimiterBurst)
+	o.IsCRDMode = viper.GetBool(configClusterIsCRDMode)
 	return nil
 }

pkg/controller/config/client.go

@@ -20,6 +20,8 @@ package config
 
 import (
 	"fmt"
+
+	"k8s.io/client-go/rest"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	"tkestack.io/tke/pkg/controller/options"
@@ -29,9 +31,15 @@ import (
 func BuildClientConfig(opts *options.APIServerClientOptions) (cfg *restclient.Config, ok bool, err error) {
 	if opts.Required {
 		if opts.Server == "" && opts.ServerClientConfig == "" {
-			err = fmt.Errorf("either %s or %s should be specified",
-				options.FlagAPIClientServer(opts.Name),
-				options.FlagAPIClientServerClientConfig(opts.Name))
+			cfg, err = rest.InClusterConfig()
+			if err != nil {
+				err = fmt.Errorf("%s or %s is not specified, try to use in cluster config failed %v",
+					options.FlagAPIClientServer(opts.Name),
+					options.FlagAPIClientServerClientConfig(opts.Name),
+					err)
+			} else {
+				ok = true
+			}
 			return
 		}
 	}

pkg/controller/options/apiclient.go

@@ -20,6 +20,7 @@ package options
 
 import (
 	"fmt"
+
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
 )
@@ -113,12 +114,6 @@ func (o *APIServerClientOptions) ApplyFlags() []error {
 	o.QPS = float32(viper.GetFloat64(o.configAPIClientQPS))
 	o.ContentType = viper.GetString(o.configAPIClientContentType)
 
-	if o.Required {
-		if o.ServerClientConfig == "" && o.Server == "" {
-			errs = append(errs, fmt.Errorf("must specify either `%s` or `%s`", FlagAPIClientServer(o.Name), FlagAPIClientServerClientConfig(o.Name)))
-		}
-	}
-
 	return errs
 }
 

pkg/platform/controller/cluster/cluster_controller.go

@@ -71,6 +71,7 @@ type Controller struct {
 	healthCheckPeriod                          time.Duration
 	randomeRangeLowerLimitForHealthCheckPeriod time.Duration
 	randomeRangeUpperLimitForHealthCheckPeriod time.Duration
+	isCRDMode                                  bool
 }
 
 // NewController creates a new Controller object.
@@ -88,6 +89,7 @@ func NewController(
 			platformClient,
 			finalizerToken,
 			true),
+		isCRDMode: configuration.IsCRDMode,
 	}
 	rateLimit := workqueue.NewMaxOfRateLimiter(
 		workqueue.NewItemExponentialFailureRateLimiter(5*time.Millisecond, 1000*time.Second),
@@ -345,6 +347,10 @@ func (c *Controller) reconcile(ctx context.Context, key string, cluster *platfor
 	var err error
 
 	switch cluster.Status.Phase {
+	// empty string is for crd without mutating webhook
+	case "":
+		cluster.Status.Phase = platformv1.ClusterInitializing
+		err = c.onCreate(ctx, cluster)
 	case platformv1.ClusterInitializing:
 		err = c.onCreate(ctx, cluster)
 	case platformv1.ClusterRunning, platformv1.ClusterFailed:
@@ -389,18 +395,37 @@ func (c *Controller) onCreate(ctx context.Context, cluster *platformv1.Cluster)
 		if err != nil {
 			// Update status, ignore failure
 			_, _ = c.platformClient.ClusterCredentials().Update(ctx, clusterWrapper.ClusterCredential, metav1.UpdateOptions{})
-			_, _ = c.platformClient.Clusters().Update(ctx, clusterWrapper.Cluster, metav1.UpdateOptions{})
+			updatedCls, _ := c.platformClient.Clusters().Update(ctx, clusterWrapper.Cluster, metav1.UpdateOptions{})
+			// if using crd, cluster status cannot be updated through update cluster
+			if c.isCRDMode {
+				var clsStatus *platformv1.Cluster
+				if updatedCls == nil {
+					clsStatus = clusterWrapper.Cluster
+				} else {
+					clsStatus = updatedCls
+					clsStatus.Status = clusterWrapper.Cluster.Status
+				}
+				_, _ = c.platformClient.Clusters().UpdateStatus(ctx, clsStatus, metav1.UpdateOptions{})
+			}
 			return err
 		}
 		clusterWrapper.ClusterCredential, err = c.platformClient.ClusterCredentials().Update(ctx, clusterWrapper.ClusterCredential, metav1.UpdateOptions{})
 		if err != nil {
 			return err
 		}
 		clusterWrapper.RegisterRestConfig(clusterWrapper.ClusterCredential.RESTConfig(cluster))
-		clusterWrapper.Cluster, err = c.platformClient.Clusters().Update(ctx, clusterWrapper.Cluster, metav1.UpdateOptions{})
+		cls, err := c.platformClient.Clusters().Update(ctx, clusterWrapper.Cluster, metav1.UpdateOptions{})
 		if err != nil {
 			return err
 		}
+		// if using crd, cluster status cannot be updated through update cluster
+		if c.isCRDMode {
+			cls.Status = clusterWrapper.Cluster.Status
+			clusterWrapper.Cluster, err = c.platformClient.Clusters().UpdateStatus(ctx, cls, metav1.UpdateOptions{})
+			if err != nil {
+				return err
+			}
+		}
 	}
 
 	return nil
@@ -487,10 +512,20 @@ func (c *Controller) ensureCreateClusterCredential(ctx context.Context, cluster
 	}
 
 	// TODO use informer search by labels.
-	fieldSelector := fields.OneTermEqualSelector("clusterName", cluster.Name).String()
-	clustercredentials, err := c.platformClient.ClusterCredentials().List(ctx, metav1.ListOptions{FieldSelector: fieldSelector})
-	if err != nil {
-		return nil, err
+	var clustercredentials *platformv1.ClusterCredentialList
+	var err error
+	if c.isCRDMode {
+		labelSelector := fields.OneTermEqualSelector(platformv1.ClusterNameLable, cluster.Name).String()
+		clustercredentials, err = c.platformClient.ClusterCredentials().List(ctx, metav1.ListOptions{LabelSelector: labelSelector})
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		fieldSelector := fields.OneTermEqualSelector("clusterName", cluster.Name).String()
+		clustercredentials, err = c.platformClient.ClusterCredentials().List(ctx, metav1.ListOptions{FieldSelector: fieldSelector})
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	// [Idempotent] if not found cluster credentials, create one for next logic
@@ -500,6 +535,8 @@ func (c *Controller) ensureCreateClusterCredential(ctx context.Context, cluster
 			TenantID:    cluster.Spec.TenantID,
 			ClusterName: cluster.Name,
 			ObjectMeta: metav1.ObjectMeta{
+				Labels:       map[string]string{platformv1.ClusterNameLable: cluster.Name},
+				GenerateName: "cc-",
 				OwnerReferences: []metav1.OwnerReference{
 					*metav1.NewControllerRef(cluster, platformv1.SchemeGroupVersion.WithKind("Cluster"))},
 			},

pkg/platform/controller/cluster/config/types.go

@@ -39,4 +39,6 @@ type ClusterControllerConfiguration struct {
 	BucketRateLimiterLimit int
 	// BucketRateLimiterBurst bursts of at most b tokens.
 	BucketRateLimiterBurst int
+	// IsCRDMode Whether the controller is using CRD mode
+	IsCRDMode bool
 }