rollouts: clusterstore enhancements to support local kind clusters (#…

…3875)

rollouts/api/v1alpha1/rollout_types.go

@@ -53,29 +53,57 @@ type ClusterTargetSelector struct {
 // ClusterReference contains the identify information
 // need to refer a cluster.
 type ClusterRef struct {
-	Name string `json:"name"`
+	APIVersion string `json:"apiVersion"`
+	Kind       string `json:"kind"`
+	Namespace  string `json:"namespace"`
+	Name       string `json:"name"`
+}
+
+func (r *ClusterRef) GetKind() string {
+	return r.Kind
+}
+
+func (r *ClusterRef) GetName() string {
+	return r.Name
+}
+
+func (r *ClusterRef) GetNamespace() string {
+	return r.Namespace
+}
+
+func (r *ClusterRef) GetAPIVersion() string {
+	return r.APIVersion
 }
 
 // different types of cluster sources
 const (
-	KCC      ClusterSourceType = "KCC"
-	GCPFleet ClusterSourceType = "GCPFleet"
+	KCC         ClusterSourceType = "KCC"
+	GCPFleet    ClusterSourceType = "GCPFleet"
+	KindCluster ClusterSourceType = "Kind"
 )
 
-// +kubebuilder:validation:Enum=KCC;GCPFleet
+// +kubebuilder:validation:Enum=KCC;GCPFleet;Kind
 type ClusterSourceType string
 
 // ClusterDiscovery represents configuration needed to discover clusters.
 type ClusterDiscovery struct {
 	SourceType ClusterSourceType      `json:"sourceType"`
 	GCPFleet   *ClusterSourceGCPFleet `json:"gcpFleet,omitempty"`
+	Kind       *ClusterSourceKind     `json:"kind,omitempty"`
 }
 
 // ClusterSourceGCPFleet represents configuration needed to discover gcp fleet clusters.
 type ClusterSourceGCPFleet struct {
 	ProjectIds []string `json:"projectIds"`
 }
 
+// ClusterSourceKind contains configuration needed to discover kind clusters.
+type ClusterSourceKind struct {
+	// Namespace where configmaps corresponding to kind clusters live
+	// defaults to `kind-clusters`
+	Namespace string `json:"namespace,omitempty"`
+}
+
 const (
 	GitHub PackageSourceType = "GitHub"
 	GitLab PackageSourceType = "GitLab"

rollouts/config/crd/bases/gitops.kpt.dev_remotesyncs.yaml

@@ -53,10 +53,19 @@ spec:
                 description: ClusterReference contains the identify information need
                   to refer a cluster.
                 properties:
+                  apiVersion:
+                    type: string
+                  kind:
+                    type: string
                   name:
                     type: string
+                  namespace:
+                    type: string
                 required:
+                - apiVersion
+                - kind
                 - name
+                - namespace
                 type: object
               template:
                 properties:

rollouts/config/crd/bases/gitops.kpt.dev_rollouts.yaml

@@ -63,10 +63,20 @@ spec:
                     required:
                     - projectIds
                     type: object
+                  kind:
+                    description: ClusterSourceKind contains configuration needed to
+                      discover kind clusters.
+                    properties:
+                      namespace:
+                        description: Namespace where configmaps corresponding to kind
+                          clusters live defaults to `kind-clusters`
+                        type: string
+                    type: object
                   sourceType:
                     enum:
                     - KCC
                     - GCPFleet
+                    - Kind
                     type: string
                 required:
                 - sourceType

rollouts/controllers/remotesync_controller.go

@@ -386,7 +386,7 @@ func (r *RemoteSyncReconciler) deleteExternalResources(ctx context.Context, remo
 }
 
 func (r *RemoteSyncReconciler) getDynamicClientForCluster(ctx context.Context, clusterRef *gitopsv1alpha1.ClusterRef) (dynamic.Interface, error) {
-	restConfig, err := r.store.GetRESTConfig(ctx, clusterRef.Name)
+	restConfig, err := r.store.GetRESTConfig(ctx, clusterRef)
 	if err != nil {
 		return nil, err
 	}

rollouts/controllers/rollout_controller.go

@@ -223,7 +223,7 @@ func (r *RolloutReconciler) validateProgressiveRolloutStrategy(ctx context.Conte
 
 	clusterWaveMap := make(map[string]string)
 	for _, cluster := range allClusters {
-		clusterWaveMap[cluster.Name] = ""
+		clusterWaveMap[cluster.Ref.Name] = ""
 	}
 
 	pauseAfterWaveName := ""
@@ -244,26 +244,26 @@ func (r *RolloutReconciler) validateProgressiveRolloutStrategy(ctx context.Conte
 		}
 
 		for _, cluster := range waveClusters {
-			currentClusterWave, found := clusterWaveMap[cluster.Name]
+			currentClusterWave, found := clusterWaveMap[cluster.Ref.Name]
 			if !found {
 				// this should never happen
-				return fmt.Errorf("wave %q references cluster %s not selected by the rollout", wave.Name, cluster.Name)
+				return fmt.Errorf("wave %q references cluster %s not selected by the rollout", wave.Name, cluster.Ref.Name)
 			}
 
 			if currentClusterWave != "" {
-				return fmt.Errorf("a cluster cannot be selected by more than one wave - cluster %s is selected by waves %q and %q", cluster.Name, currentClusterWave, wave.Name)
+				return fmt.Errorf("a cluster cannot be selected by more than one wave - cluster %s is selected by waves %q and %q", cluster.Ref.Name, currentClusterWave, wave.Name)
 			}
 
-			clusterWaveMap[cluster.Name] = wave.Name
+			clusterWaveMap[cluster.Ref.Name] = wave.Name
 		}
 
 		pauseWaveNameFound = pauseWaveNameFound || pauseAfterWaveName == wave.Name
 	}
 
 	for _, cluster := range allClusters {
-		wave, _ := clusterWaveMap[cluster.Name]
+		wave, _ := clusterWaveMap[cluster.Ref.Name]
 		if wave == "" {
-			return fmt.Errorf("waves should cover all clusters selected by the rollout - cluster %s is not covered by any waves", cluster.Name)
+			return fmt.Errorf("waves should cover all clusters selected by the rollout - cluster %s is not covered by any waves", cluster.Ref.Name)
 		}
 	}
 
@@ -409,7 +409,7 @@ func (r *RolloutReconciler) computeTargets(ctx context.Context,
 			continue
 		}
 		cluster := &clusterPackages[idx].Cluster
-		clusterName := cluster.Name[strings.LastIndex(cluster.Name, "/")+1:]
+		clusterName := cluster.Ref.Name[strings.LastIndex(cluster.Ref.Name, "/")+1:]
 		pkg := &clusterPkg.Packages[0]
 		rs := gitopsv1alpha1.RemoteSync{}
 		key := client.ObjectKey{
@@ -481,14 +481,14 @@ func (r *RolloutReconciler) getWaveTargets(ctx context.Context, rollout *gitopsv
 		}
 
 		for _, cluster := range waveClusters {
-			clusterNameToWaveTarget[cluster.Name] = &thisWaveTarget
+			clusterNameToWaveTarget[cluster.Ref.Name] = &thisWaveTarget
 		}
 
 		allWaveTargets = append(allWaveTargets, thisWaveTarget)
 	}
 
 	for _, toCreate := range allTargets.ToBeCreated {
-		wavetTargets := clusterNameToWaveTarget[toCreate.cluster.Name].Targets
+		wavetTargets := clusterNameToWaveTarget[toCreate.cluster.Ref.Name].Targets
 		wavetTargets.ToBeCreated = append(wavetTargets.ToBeCreated, toCreate)
 	}
 
@@ -545,7 +545,6 @@ func (r *RolloutReconciler) rolloutTargets(ctx context.Context, rollout *gitopsv
 
 	for _, target := range targets.ToBeCreated {
 		rs := newRemoteSync(rollout, target)
-
 		if maxConcurrent > concurrentUpdates {
 			if err := r.Create(ctx, rs); err != nil {
 				logger.Info("Warning, error creating RemoteSync", "remoteSync", klog.KRef(rs.Namespace, rs.Name), "err", err)
@@ -716,7 +715,7 @@ func isRSErrored(rss *gitopsv1alpha1.RemoteSync) bool {
 // Given a package identifier and cluster, create a RemoteSync object.
 func newRemoteSync(rollout *gitopsv1alpha1.Rollout, target *clusterPackagePair) *gitopsv1alpha1.RemoteSync {
 	t := true
-	clusterRef := gitopsv1alpha1.ClusterRef{Name: target.cluster.Name}
+	clusterRef := target.cluster.Ref
 	clusterName := clusterRef.Name[strings.LastIndex(clusterRef.Name, "/")+1:]
 
 	templateType := gitopsv1alpha1.TemplateTypeRootSync

rollouts/pkg/clusterstore/clusterstore.go

@@ -17,21 +17,36 @@ package clusterstore
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	gkeclusterapis "github.com/GoogleCloudPlatform/k8s-config-connector/pkg/clients/generated/apis/container/v1beta1"
+	gkehubapis "github.com/GoogleCloudPlatform/k8s-config-connector/pkg/clients/generated/apis/gkehub/v1beta1"
 	gitopsv1alpha1 "github.com/GoogleContainerTools/kpt/rollouts/api/v1alpha1"
 )
 
+var (
+	KCCClusterGVK         = gkeclusterapis.ContainerClusterGVK
+	GKEFleetMembershipGVK = gkehubapis.GKEHubMembershipGVK
+	KindClusterGVK        = schema.GroupVersionKind{
+		Group:   "clusters.gitops.kpt.dev",
+		Version: "v1",
+		Kind:    "KindCluster",
+	}
+)
+
 type ClusterStore struct {
 	containerClusterStore *ContainerClusterStore
 	gcpFleetClusterStore  *GCPFleetClusterStore
+	kindClusterStore      *KindClusterStore
 }
+
 type Cluster struct {
-	Name   string
+	// Ref is the reference to the target cluster
+	Ref    gitopsv1alpha1.ClusterRef
 	Labels map[string]string
 }
 
@@ -47,6 +62,7 @@ func NewClusterStore(client client.Client, config *rest.Config) (*ClusterStore,
 	clusterStore := &ClusterStore{
 		containerClusterStore: containerClusterStore,
 		gcpFleetClusterStore:  &GCPFleetClusterStore{},
+		kindClusterStore:      &KindClusterStore{Client: client},
 	}
 
 	return clusterStore, nil
@@ -62,17 +78,24 @@ func (cs *ClusterStore) ListClusters(ctx context.Context, clusterDiscovery *gito
 	case gitopsv1alpha1.KCC:
 		return cs.containerClusterStore.ListClusters(ctx, selector)
 
+	case gitopsv1alpha1.KindCluster:
+		return cs.kindClusterStore.ListClusters(ctx, selector)
 	default:
 		return nil, fmt.Errorf("%v cluster source not supported", clusterSourceType)
 	}
 }
 
-func (cs *ClusterStore) GetRESTConfig(ctx context.Context, name string) (*rest.Config, error) {
-	switch {
-	case strings.Contains(name, "memberships") || strings.Contains(name, "gkeMemberships"):
-		return cs.gcpFleetClusterStore.GetRESTConfig(ctx, name)
-
+func (cs *ClusterStore) GetRESTConfig(ctx context.Context, clusterRef *gitopsv1alpha1.ClusterRef) (*rest.Config, error) {
+	// TODO (droot): Using kind property of the clusterRef for now but in the future
+	// expand it to use the other properties (seems like an overkill for now).
+	switch clusterKind := clusterRef.Kind; clusterKind {
+	case GKEFleetMembershipGVK.Kind:
+		return cs.gcpFleetClusterStore.GetRESTConfig(ctx, clusterRef.GetName())
+	case KindClusterGVK.Kind:
+		return cs.kindClusterStore.GetRESTConfig(ctx, clusterRef.GetName())
+	case KCCClusterGVK.Kind:
+		return cs.containerClusterStore.GetRESTConfig(ctx, clusterRef.GetName())
 	default:
-		return cs.containerClusterStore.GetRESTConfig(ctx, name)
+		return nil, fmt.Errorf("unknown cluster kind %s", clusterKind)
 	}
 }

rollouts/pkg/clusterstore/containerclusterstore.go

@@ -25,6 +25,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	gkeclusterapis "github.com/GoogleCloudPlatform/k8s-config-connector/pkg/clients/generated/apis/container/v1beta1"
+	gitopsv1alpha1 "github.com/GoogleContainerTools/kpt/rollouts/api/v1alpha1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -174,9 +175,12 @@ func (cs *ContainerClusterStore) getConfigConnectorContextTokenSource(ctx contex
 
 func (cs *ContainerClusterStore) toCluster(containerCluster *gkeclusterapis.ContainerCluster) Cluster {
 	cluster := Cluster{
-		Name:   containerCluster.Name,
+		Ref: gitopsv1alpha1.ClusterRef{
+			APIVersion: KCCClusterGVK.GroupVersion().String(),
+			Kind:       KCCClusterGVK.Kind,
+			Name:       containerCluster.Name,
+		},
 		Labels: containerCluster.Labels,
 	}
-
 	return cluster
 }

rollouts/pkg/clusterstore/gcpfleetclusterstore.go

@@ -198,7 +198,11 @@ func (cs *GCPFleetClusterStore) listMemberships(ctx context.Context, projectId s
 
 func (cs *GCPFleetClusterStore) toCluster(membership *gkehubv1.Membership) Cluster {
 	cluster := Cluster{
-		Name:   membership.Name,
+		Ref: gitopsv1alpha1.ClusterRef{
+			APIVersion: GKEFleetMembershipGVK.GroupVersion().String(),
+			Kind:       GKEFleetMembershipGVK.Kind,
+			Name:       membership.Name,
+		},
 		Labels: membership.Labels,
 	}