feat: support excluded sync resources

KubeKyrie · KubeKyrie · commit cf3696757b3d · 2023-12-14T18:10:00.000+08:00
Signed-off-by: KubeKyrie &lt;shaolong.qin@daocloud.io&gt;
diff --git a/kustomize/crds/cluster.clusterpedia.io_clustersyncresources.yaml b/kustomize/crds/cluster.clusterpedia.io_clustersyncresources.yaml
@@ -32,6 +32,25 @@ spec:
             type: object
           spec:
             properties:
+              excludedSyncResources:
+                items:
+                  properties:
+                    group:
+                      type: string
+                    resources:
+                      items:
+                        type: string
+                      minItems: 1
+                      type: array
+                    versions:
+                      items:
+                        type: string
+                      type: array
+                  required:
+                  - group
+                  - resources
+                  type: object
+                type: array
               syncResources:
                 items:
                   properties:
diff --git a/kustomize/crds/cluster.clusterpedia.io_pediaclusters.yaml b/kustomize/crds/cluster.clusterpedia.io_pediaclusters.yaml
@@ -66,6 +66,25 @@ spec:
               certData:
                 format: byte
                 type: string
+              excludedSyncResources:
+                items:
+                  properties:
+                    group:
+                      type: string
+                    resources:
+                      items:
+                        type: string
+                      minItems: 1
+                      type: array
+                    versions:
+                      items:
+                        type: string
+                      type: array
+                  required:
+                  - group
+                  - resources
+                  type: object
+                type: array
               keyData:
                 format: byte
                 type: string
diff --git a/pkg/generated/openapi/zz_generated.openapi.go b/pkg/generated/openapi/zz_generated.openapi.go
diff --git a/pkg/synchromanager/clustersynchro/cluster_synchro.go b/pkg/synchromanager/clustersynchro/cluster_synchro.go
@@ -59,10 +59,11 @@ type ClusterSynchro struct {
 	storageResourceVersions map[schema.GroupVersionResource]map[string]interface{}
 	storageResourceSynchros sync.Map
 
-	syncResources       atomic.Value // []clusterv1alpha2.ClusterGroupResources
-	setSyncResourcesCh  chan struct{}
-	resourceNegotiator  *ResourceNegotiator
-	groupResourceStatus atomic.Value // *GroupResourceStatus
+	syncResources         atomic.Value // []clusterv1alpha2.ClusterGroupResources
+	excludedSyncResources atomic.Value // []clusterv1alpha2.ClusterGroupResources
+	setSyncResourcesCh    chan struct{}
+	resourceNegotiator    *ResourceNegotiator
+	groupResourceStatus   atomic.Value // *GroupResourceStatus
 
 	runningCondition atomic.Value // metav1.Condition
 	healthyCondition atomic.Value // metav1.Condition
@@ -278,9 +279,12 @@ func (s *ClusterSynchro) Shutdown(updateStatus bool) {
 	<-s.closed
 }
 
-func (s *ClusterSynchro) SetResources(syncResources []clusterv1alpha2.ClusterGroupResources, syncAllCustomResources bool) {
+func (s *ClusterSynchro) SetResources(syncResources, excludedSyncResources []clusterv1alpha2.ClusterGroupResources, syncAllCustomResources bool) {
 	s.syncResources.Store(syncResources)
 	s.resourceNegotiator.SetSyncAllCustomResources(syncAllCustomResources)
+	if excludedSyncResources != nil {
+		s.excludedSyncResources.Store(excludedSyncResources)
+	}
 
 	s.resetSyncResources()
 }
@@ -315,7 +319,13 @@ func (s *ClusterSynchro) refreshSyncResources() {
 	if syncResources == nil {
 		return
 	}
-	groupResourceStatus, storageResourceSyncConfigs := s.resourceNegotiator.NegotiateSyncResources(syncResources)
+
+	excludedSyncResources := make([]clusterv1alpha2.ClusterGroupResources, 0)
+	loadValue := s.excludedSyncResources.Load()
+	if loadValue != nil {
+		excludedSyncResources = loadValue.([]clusterv1alpha2.ClusterGroupResources)
+	}
+	groupResourceStatus, storageResourceSyncConfigs := s.resourceNegotiator.NegotiateSyncResources(syncResources, excludedSyncResources)
 
 	lastGroupResourceStatus := s.groupResourceStatus.Load().(*GroupResourceStatus)
 	deleted := groupResourceStatus.Merge(lastGroupResourceStatus)
diff --git a/pkg/synchromanager/clustersynchro/resource_negotiator.go b/pkg/synchromanager/clustersynchro/resource_negotiator.go
@@ -39,7 +39,7 @@ func (negotiator *ResourceNegotiator) SetSyncAllCustomResources(sync bool) {
 	negotiator.syncAllCustomResources = sync
 }
 
-func (negotiator *ResourceNegotiator) NegotiateSyncResources(syncResources []clusterv1alpha2.ClusterGroupResources) (*GroupResourceStatus, map[schema.GroupVersionResource]syncConfig) {
+func (negotiator *ResourceNegotiator) NegotiateSyncResources(syncResources, excludedSyncResources []clusterv1alpha2.ClusterGroupResources) (*GroupResourceStatus, map[schema.GroupVersionResource]syncConfig) {
 	var syncAllResources bool
 	var watchKubeVersion, watchAggregatorResourceTypes bool
 	for i, syncResource := range syncResources {
@@ -96,8 +96,24 @@ func (negotiator *ResourceNegotiator) NegotiateSyncResources(syncResources []clu
 				continue
 			}
 
+			// remove excludedSyncResources, groupResources.Versions or excludedGroupResources.Versions may be wildcard
+			availableVersions := supportedVersions
+			for _, excludedGroupResources := range excludedSyncResources {
+				excludedResources := sets.New(excludedGroupResources.Resources...)
+				if groupResources.Group != excludedGroupResources.Group {
+					continue
+				}
+				if excludedResources.Has("*") {
+					availableVersions = []string{}
+					break
+				}
+				if excludedResources.Has(resource) {
+					availableVersions = getNewAvailableVersions(availableVersions, excludedGroupResources.Versions)
+				}
+			}
+
 			syncGK := schema.GroupKind{Group: syncGR.Group, Kind: apiResource.Kind}
-			syncVersions, isLegacyResource, err := negotiateSyncVersions(syncGK, groupResources.Versions, supportedVersions)
+			syncVersions, isLegacyResource, err := negotiateSyncVersions(syncGK, groupResources.Versions, availableVersions)
 			if err != nil {
 				klog.InfoS("Skip resource sync", "cluster", negotiator.name, "resource", resource, "reason", err)
 				continue
@@ -412,3 +428,21 @@ func (set GVRSet) Insert(gvrs ...schema.GroupVersionResource) {
 		set[gvr] = struct{}{}
 	}
 }
+
+func getNewAvailableVersions(supportedVersions, excludedVersions []string) []string {
+	if len(excludedVersions) == 0 {
+		return []string{}
+	}
+	excludedVersionSet := sets.New(excludedVersions...)
+	if excludedVersionSet.Has("*") {
+		return []string{}
+	}
+
+	res := make([]string, 0)
+	for _, version := range supportedVersions {
+		if !excludedVersionSet.Has(version) {
+			res = append(res, version)
+		}
+	}
+	return res
+}
diff --git a/pkg/synchromanager/clustersynchro_manager.go b/pkg/synchromanager/clustersynchro_manager.go
@@ -340,6 +340,7 @@ func (manager *Manager) reconcileCluster(cluster *clusterv1alpha2.PediaCluster)
 
 	var warnMsg string
 	syncResources := cluster.Spec.SyncResources
+	excludedSyncResources := cluster.Spec.ExcludedSyncResources
 	if refName := cluster.Spec.SyncResourcesRefName; refName != "" {
 		if ref, err := manager.clusterSyncResourcesLister.Get(refName); err != nil {
 			if !apierrors.IsNotFound(err) {
@@ -354,6 +355,7 @@ func (manager *Manager) reconcileCluster(cluster *clusterv1alpha2.PediaCluster)
 			warnMsg = "Warning: sync resource ref is not found"
 		} else {
 			syncResources = append(syncResources, ref.Spec.SyncResources...)
+			excludedSyncResources = append(excludedSyncResources, ref.Spec.ExcludedSyncResources...)
 		}
 	}
 
@@ -439,7 +441,7 @@ func (manager *Manager) reconcileCluster(cluster *clusterv1alpha2.PediaCluster)
 		manager.synchrolock.Unlock()
 	}
 
-	synchro.SetResources(syncResources, cluster.Spec.SyncAllCustomResources)
+	synchro.SetResources(syncResources, excludedSyncResources, cluster.Spec.SyncAllCustomResources)
 	return controller.NoRequeueResult
 }
 
diff --git a/staging/src/github.com/clusterpedia-io/api/cluster/v1alpha2/types.go b/staging/src/github.com/clusterpedia-io/api/cluster/v1alpha2/types.go
@@ -89,6 +89,9 @@ type ClusterSpec struct {
 	// +required
 	SyncResources []ClusterGroupResources `json:"syncResources"`
 
+	// +optional
+	ExcludedSyncResources []ClusterGroupResources `json:"excludedSyncResources"`
+
 	// +optional
 	SyncAllCustomResources bool `json:"syncAllCustomResources,omitempty"`
 
@@ -242,6 +245,8 @@ type ClusterSyncResources struct {
 type ClusterSyncResourcesSpec struct {
 	// +required
 	SyncResources []ClusterGroupResources `json:"syncResources"`
+	// +optional
+	ExcludedSyncResources []ClusterGroupResources `json:"excludedSyncResources"`
 }
 
 // +kubebuilder:object:root=true
diff --git a/staging/src/github.com/clusterpedia-io/api/cluster/v1alpha2/zz_generated.deepcopy.go b/staging/src/github.com/clusterpedia-io/api/cluster/v1alpha2/zz_generated.deepcopy.go

Original file line number	Diff line number	Diff line change
`@@ -340,6 +340,7 @@ func (manager Manager) reconcileCluster(cluster clusterv1alpha2.PediaCluster)`
`340`	`340`
`341`	`341`	`var warnMsg string`
`342`	`342`	`syncResources := cluster.Spec.SyncResources`
	`343`	`+ excludedSyncResources := cluster.Spec.ExcludedSyncResources`
`343`	`344`	`if refName := cluster.Spec.SyncResourcesRefName; refName != "" {`
`344`	`345`	`if ref, err := manager.clusterSyncResourcesLister.Get(refName); err != nil {`
`345`	`346`	`if !apierrors.IsNotFound(err) {`
`@@ -354,6 +355,7 @@ func (manager Manager) reconcileCluster(cluster clusterv1alpha2.PediaCluster)`
`354`	`355`	`warnMsg = "Warning: sync resource ref is not found"`
`355`	`356`	`} else {`
`356`	`357`	`syncResources = append(syncResources, ref.Spec.SyncResources...)`
	`358`	`+ excludedSyncResources = append(excludedSyncResources, ref.Spec.ExcludedSyncResources...)`
`357`	`359`	`}`
`358`	`360`	`}`
`359`	`361`
`@@ -439,7 +441,7 @@ func (manager Manager) reconcileCluster(cluster clusterv1alpha2.PediaCluster)`
`439`	`441`	`manager.synchrolock.Unlock()`
`440`	`442`	`}`
`441`	`443`
`442`		`- synchro.SetResources(syncResources, cluster.Spec.SyncAllCustomResources)`
	`444`	`+ synchro.SetResources(syncResources, excludedSyncResources, cluster.Spec.SyncAllCustomResources)`
`443`	`445`	`return controller.NoRequeueResult`
`444`	`446`	`}`
`445`	`447`