validate CNSRegistreVolume to disallow registering volume not accessible to active clusters on namespace (#3436)

Author: divyenpatel

pkg/syncer/cnsoperator/controller/cnsregistervolume/cnsregistervolume_controller.go

@@ -76,7 +76,9 @@ var (
 	clusterComputeResourceMoIds []string
 	// workloadDomainIsolationEnabled determines if the workload domain
 	// isolation feature is available on a supervisor cluster.
-	workloadDomainIsolationEnabled bool
+	workloadDomainIsolationEnabled          bool
+	isTKGSHAEnabled                         bool
+	isMultipleClustersPerVsphereZoneEnabled bool
 )
 
 // Add creates a new CnsRegisterVolume Controller and adds it to the Manager,
@@ -91,6 +93,9 @@ func Add(mgr manager.Manager, clusterFlavor cnstypes.CnsClusterFlavor,
 	}
 	workloadDomainIsolationEnabled = commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx,
 		common.WorkloadDomainIsolation)
+	isTKGSHAEnabled = commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.TKGsHA)
+	isMultipleClustersPerVsphereZoneEnabled = commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx,
+		common.MultipleClustersPerVsphereZone)
 
 	var volumeInfoService cnsvolumeinfo.VolumeInfoService
 	if clusterFlavor == cnstypes.CnsClusterFlavorWorkload {
@@ -202,7 +207,6 @@ type ReconcileCnsRegisterVolume struct {
 func (r *ReconcileCnsRegisterVolume) Reconcile(ctx context.Context,
 	request reconcile.Request) (reconcile.Result, error) {
 	log := logger.GetLogger(ctx)
-	isTKGSHAEnabled := commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.TKGsHA)
 	// Fetch the CnsRegisterVolume instance.
 	instance := &cnsregistervolumev1alpha1.CnsRegisterVolume{}
 	err := r.client.Get(ctx, request.NamespacedName, instance)
@@ -326,17 +330,55 @@ func (r *ReconcileCnsRegisterVolume) Reconcile(ctx context.Context,
 
 	if syncer.IsPodVMOnStretchSupervisorFSSEnabled {
 		if workloadDomainIsolationEnabled || len(clusterComputeResourceMoIds) > 1 {
-			azClustersMap := topologyMgr.GetAZClustersMap(ctx)
-			isAccessible := isDatastoreAccessibleToAZClusters(ctx, vc, azClustersMap, volume.DatastoreUrl)
-			if !isAccessible {
-				log.Errorf("Volume: %s present on datastore: %s is not accessible to any of the AZ clusters: %v",
-					volumeID, volume.DatastoreUrl, azClustersMap)
-				setInstanceError(ctx, r, instance, "Volume in the spec is not accessible to any of the AZ clusters")
-				_, err = common.DeleteVolumeUtil(ctx, r.volumeManager, volumeID, false)
+			if isMultipleClustersPerVsphereZoneEnabled {
+				// Get zones assigned to the namespace
+				zoneMaps := commonco.ContainerOrchestratorUtility.GetZonesForNamespace(request.Namespace)
+
+				// Convert map keys to slice
+				zones := make([]string, 0, len(zoneMaps))
+				for zone := range zoneMaps {
+					zones = append(zones, zone)
+				}
+				// Get active clusters in the requested zones
+				activeClusters, err := commonco.ContainerOrchestratorUtility.
+					GetActiveClustersForNamespaceInRequestedZones(ctx, request.Namespace, zones)
 				if err != nil {
-					log.Errorf("Failed to untag CNS volume: %s with error: %+v", volumeID, err)
+					msg := fmt.Sprintf("Failed to get active clusters for CnsRegisterVolume request. error: %+v", err)
+					log.Error(msg)
+					setInstanceError(ctx, r, instance, msg)
+					return reconcile.Result{RequeueAfter: timeout}, nil
+				}
+				// Check if volume is accessible in any of the active clusters
+				if !isDatastoreAccessibleToAZClusters(ctx, vc,
+					map[string][]string{"dummy-zone-key": activeClusters}, volume.DatastoreUrl) {
+					log.Errorf("Volume: %s present on datastore: %s is not accessible to any of the active "+
+						"cluster on the namespace: %v", volumeID, volume.DatastoreUrl, activeClusters)
+					setInstanceError(ctx, r, instance, "Volume in the spec is not accessible to any of "+
+						"the active cluster on the namespace")
+
+					// Attempt volume cleanup
+					if _, err = common.DeleteVolumeUtil(ctx, r.volumeManager, volumeID, false); err != nil {
+						log.Errorf("Failed to untag CNS volume: %s with error: %+v", volumeID, err)
+						return reconcile.Result{RequeueAfter: timeout}, nil
+					}
+					// permanent failure and not requeue.
+					return reconcile.Result{}, nil
+				}
+			} else {
+				azClustersMap := topologyMgr.GetAZClustersMap(ctx)
+				isAccessible := isDatastoreAccessibleToAZClusters(ctx, vc, azClustersMap, volume.DatastoreUrl)
+				if !isAccessible {
+					log.Errorf("Volume: %s present on datastore: %s is not accessible to any of the AZ clusters: %v",
+						volumeID, volume.DatastoreUrl, azClustersMap)
+					setInstanceError(ctx, r, instance, "Volume in the spec is not accessible to any of the AZ clusters")
+					_, err = common.DeleteVolumeUtil(ctx, r.volumeManager, volumeID, false)
+					if err != nil {
+						log.Errorf("Failed to untag CNS volume: %s with error: %+v", volumeID, err)
+						return reconcile.Result{RequeueAfter: timeout}, nil
+					}
+					// permanent failure and not requeue.
+					return reconcile.Result{}, nil
 				}
-				return reconcile.Result{RequeueAfter: timeout}, nil
 			}
 		}
 	} else {
@@ -350,8 +392,10 @@ func (r *ReconcileCnsRegisterVolume) Reconcile(ctx context.Context,
 			_, err = common.DeleteVolumeUtil(ctx, r.volumeManager, volumeID, false)
 			if err != nil {
 				log.Errorf("Failed to untag CNS volume: %s with error: %+v", volumeID, err)
+				return reconcile.Result{RequeueAfter: timeout}, nil
 			}
-			return reconcile.Result{RequeueAfter: timeout}, nil
+			// permanent failure and not requeue.
+			return reconcile.Result{}, nil
 		}
 	}
 	// Verify if storage policy is empty.