Add dynamic BackingType detection via VirtualDiskManager

When the disk-backing annotation is not present on a PVC, dynamically
query the backing type from vSphere VirtualDiskManager API and cache
it on the PVC annotation for future attach operations.

Changes:
- Add queryBackingTypeFromVirtualDiskInfo() to query disk type via
  CNS QueryVolume API and VirtualDiskManager.QueryVirtualDiskInfo()
- Add convertDiskTypeToBackingType() to map vSphere disk types
  (thin, preallocated, eagerZeroedThick, etc.) to backing types
- Add patchPVCBackingTypeAnnotation() to persist queried backing type
  on PVC annotation for caching
- Update constructBatchAttachRequest() to query and cache backing type
  when annotation is missing
- Add unit tests for convertDiskTypeToBackingType and
  patchPVCBackingTypeAnnotation functions

Author: divyenpatel

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/go-co-op/gocron v1.37.0
 	github.com/go-logr/zapr v1.3.0
 	github.com/golang/protobuf v1.5.4
+	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
 	github.com/hashicorp/go-version v1.6.0
 	github.com/kubernetes-csi/csi-proxy/v2 v2.0.0-alpha.1
@@ -100,7 +101,6 @@ require (
 	github.com/google/btree v1.1.3 // indirect
 	github.com/google/cadvisor v0.52.1 // indirect
 	github.com/google/cel-go v0.26.0 // indirect
-	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db // indirect
 	github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 // indirect
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect

pkg/syncer/cnsoperator/controller/cnsnodevmbatchattachment/cnsnodevmbatchattachment_controller.go

@@ -524,7 +524,7 @@ func (r *Reconciler) processBatchAttach(ctx context.Context, k8sClient kubernete
 
 	// Construct batch attach request
 	pvcsInAttachList, volumeIdsInAttachList, batchAttachRequest, err := constructBatchAttachRequest(ctx,
-		volumesToAttach, instance)
+		volumesToAttach, instance, r.volumeManager, r.configInfo, k8sClient)
 	if err != nil {
 		log.Errorf("failed to construct batch attach request. Err: %s", err)
 		return err

pkg/syncer/cnsoperator/controller/cnsnodevmbatchattachment/cnsnodevmbatchattachment_helper.go

@@ -32,6 +32,8 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/common"
 
+	cnstypes "github.com/vmware/govmomi/cns/types"
+	"github.com/vmware/govmomi/object"
 	vimtypes "github.com/vmware/govmomi/vim25/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -506,12 +508,198 @@ func listAttachedFcdsForVM(ctx context.Context,
 	return attachedFCDs, nil
 }
 
+// convertDiskTypeToBackingType converts the diskType returned by QueryVirtualDiskInfo
+// to the appropriate CnsVolumeBackingType string used for batch attach operations.
+// The returned values correspond to VirtualDevice.FileBackingInfo subclasses as defined in
+// github.com/vmware/govmomi/cns/types (CnsVolumeBackingType constants).
+func convertDiskTypeToBackingType(diskType string) string {
+	switch diskType {
+	case "thin", "preallocated", "thick", "eagerZeroedThick", "thick2Gb", "flatMonolithic":
+		// All flat/thick disk types use FlatVer2BackingInfo
+		// thin -> FlatVer2BackingInfo (thinProvisioned=true)
+		// preallocated/thick -> FlatVer2BackingInfo (thinProvisioned=false, eagerlyScrub=false)
+		// eagerZeroedThick -> FlatVer2BackingInfo (thinProvisioned=false, eagerlyScrub=true)
+		// thick2Gb/flatMonolithic -> FlatVer2BackingInfo (split variations)
+		return string(cnstypes.CnsVolumeBackingTypeFlatVer2BackingInfo)
+	case "sparse2Gb", "sparseMonolithic", "delta", "vmfsSparse":
+		// sparse types -> SparseVer2BackingInfo
+		return string(cnstypes.CnsVolumeBackingTypeSparseVer2BackingInfo)
+	case "seSparse":
+		// seSparse -> SeSparseBackingInfo
+		return string(cnstypes.CnsVolumeBackingTypeSeSparseBackingInfo)
+	case "rdm", "rdmp":
+		// rdm -> RawDiskMappingVer1BackingInfo (compatibilityMode="virtualMode")
+		// rdmp -> RawDiskMappingVer1BackingInfo (compatibilityMode="physicalMode")
+		return string(cnstypes.CnsVolumeBackingTypeRawDiskMappingVer1BackingInfo)
+	default:
+		// Unknown disk type, return empty string
+		return ""
+	}
+}
+
+// queryBackingTypeFromVirtualDiskInfo queries the backing type of a volume using
+// VirtualDiskManager's QueryVirtualDiskInfo API.
+// It first queries the volume to get the DatastoreUrl, then uses that URL to call
+// QueryVirtualDiskInfo which returns the disk type information.
+func queryBackingTypeFromVirtualDiskInfo(ctx context.Context,
+	volumeManager volumes.Manager,
+	configInfo config.ConfigurationInfo,
+	volumeID string) (string, error) {
+	log := logger.GetLogger(ctx)
+
+	// Step 1: Query the volume to get the DatastoreUrl
+	queryFilter := cnstypes.CnsQueryFilter{
+		VolumeIds: []cnstypes.CnsVolumeId{
+			{
+				Id: volumeID,
+			},
+		},
+	}
+
+	querySelection := cnstypes.CnsQuerySelection{
+		Names: []string{string(cnstypes.QuerySelectionNameTypeDataStoreUrl)},
+	}
+
+	queryResult, err := volumeManager.QueryVolumeAsync(ctx, queryFilter, &querySelection)
+	if err != nil {
+		// Fall back to QueryVolume if QueryVolumeAsync is not supported
+		if err.Error() == cnsvsphere.ErrNotSupported.Error() {
+			log.Warn("QueryVolumeAsync is not supported. Invoking QueryVolume API")
+			queryResult, err = volumeManager.QueryVolume(ctx, queryFilter)
+			if err != nil {
+				return "", fmt.Errorf("failed to query volume %s: %w", volumeID, err)
+			}
+		} else {
+			return "", fmt.Errorf("failed to query volume async %s: %w", volumeID, err)
+		}
+	}
+
+	if queryResult == nil || len(queryResult.Volumes) == 0 {
+		return "", fmt.Errorf("no volume found for volumeID %s", volumeID)
+	}
+
+	datastoreUrl := queryResult.Volumes[0].DatastoreUrl
+	if datastoreUrl == "" {
+		return "", fmt.Errorf("datastore URL not found for volumeID %s", volumeID)
+	}
+
+	log.Debugf("Retrieved DatastoreUrl %s for volumeID %s", datastoreUrl, volumeID)
+
+	// Step 2: Get VirtualCenter instance
+	vc, err := cnsvsphere.GetVirtualCenterInstance(ctx, &configInfo, false)
+	if err != nil {
+		return "", fmt.Errorf("failed to get VirtualCenter instance: %w", err)
+	}
+
+	err = vc.Connect(ctx)
+	if err != nil {
+		return "", fmt.Errorf("failed to connect to VirtualCenter: %w", err)
+	}
+
+	// Step 3: Get the datacenter
+	dcs, err := vc.GetDatacenters(ctx)
+	if err != nil {
+		return "", fmt.Errorf("failed to get datacenters: %w", err)
+	}
+
+	if len(dcs) == 0 {
+		return "", fmt.Errorf("no datacenters found")
+	}
+
+	// Step 4: Create VirtualDiskManager and query disk info
+	virtualDiskManager := object.NewVirtualDiskManager(vc.Client.Client)
+
+	// The name parameter for QueryVirtualDiskInfo should be the datastore path
+	// Format: [datastoreName] path/to/disk.vmdk
+	// Since we have the DatastoreUrl, we need to construct the disk path
+	// The DatastoreUrl format is: ds:///vmfs/volumes/<datastore-uuid>/
+	// We need to query the backing object details to get the file path
+	queryFilterForPath := cnstypes.CnsQueryFilter{
+		VolumeIds: []cnstypes.CnsVolumeId{
+			{
+				Id: volumeID,
+			},
+		},
+	}
+
+	querySelectionForPath := cnstypes.CnsQuerySelection{
+		Names: []string{string(cnstypes.QuerySelectionNameTypeBackingObjectDetails)},
+	}
+
+	queryResultForPath, err := volumeManager.QueryVolumeAsync(ctx, queryFilterForPath, &querySelectionForPath)
+	if err != nil {
+		if err.Error() == cnsvsphere.ErrNotSupported.Error() {
+			queryResultForPath, err = volumeManager.QueryVolume(ctx, queryFilterForPath)
+			if err != nil {
+				return "", fmt.Errorf("failed to query volume for backing details %s: %w", volumeID, err)
+			}
+		} else {
+			return "", fmt.Errorf("failed to query volume async for backing details %s: %w", volumeID, err)
+		}
+	}
+
+	if queryResultForPath == nil || len(queryResultForPath.Volumes) == 0 {
+		return "", fmt.Errorf("no volume found for volumeID %s when querying backing details", volumeID)
+	}
+
+	backingObjectDetails := queryResultForPath.Volumes[0].BackingObjectDetails
+	if backingObjectDetails == nil {
+		return "", fmt.Errorf("backing object details not found for volumeID %s", volumeID)
+	}
+
+	// Get the backing file path from BackingObjectDetails
+	// BackingDiskPath is available on CnsBlockBackingDetails
+	blockBackingDetails, ok := backingObjectDetails.(*cnstypes.CnsBlockBackingDetails)
+	if !ok {
+		return "", fmt.Errorf("backing object details is not of type CnsBlockBackingDetails for volumeID %s", volumeID)
+	}
+
+	backingFilePath := blockBackingDetails.BackingDiskPath
+	if backingFilePath == "" {
+		return "", fmt.Errorf("backing disk path not found for volumeID %s", volumeID)
+	}
+
+	log.Debugf("Retrieved backing disk path %s for volumeID %s", backingFilePath, volumeID)
+
+	// Step 5: Call QueryVirtualDiskInfo
+	// Use the first datacenter for the query
+	var dc *object.Datacenter
+	if len(dcs) > 0 {
+		dc = dcs[0].Datacenter
+	}
+
+	diskInfoList, err := virtualDiskManager.QueryVirtualDiskInfo(ctx, backingFilePath, dc, false)
+	if err != nil {
+		return "", fmt.Errorf("failed to query virtual disk info for %s: %w", backingFilePath, err)
+	}
+
+	if len(diskInfoList) == 0 {
+		return "", fmt.Errorf("no disk info returned for %s", backingFilePath)
+	}
+
+	diskType := diskInfoList[0].DiskType
+	log.Infof("Retrieved diskType %s for volumeID %s from VirtualDiskManager", diskType, volumeID)
+
+	// Step 6: Convert diskType to backing type
+	backingType := convertDiskTypeToBackingType(diskType)
+	if backingType == "" {
+		log.Warnf("Unknown diskType %s for volumeID %s, using diskType as backingType", diskType, volumeID)
+		backingType = diskType
+	}
+
+	log.Infof("Converted diskType %s to backingType %s for volumeID %s", diskType, backingType, volumeID)
+	return backingType, nil
+}
+
 // constructBatchAttachRequest goes through all volumes in instance spec and
 // constructs the batchAttach request for each of them.
 // It also validates each of the requests to make sure user input is correct.
 func constructBatchAttachRequest(ctx context.Context,
 	volumesToAttach map[string]string,
-	instance *v1alpha1.CnsNodeVMBatchAttachment) (pvcsInSpec map[string]string,
+	instance *v1alpha1.CnsNodeVMBatchAttachment,
+	volumeManager volumes.Manager,
+	configInfo config.ConfigurationInfo,
+	k8sClient kubernetes.Interface) (pvcsInSpec map[string]string,
 	volumeIdsInSpec map[string]string,
 	batchAttachRequest []volumes.BatchAttachRequest, err error) {
 	log := logger.GetLogger(ctx)
@@ -546,14 +734,39 @@ func constructBatchAttachRequest(ctx context.Context,
 			isPvcEncrypted := isPvcEncrypted(pvcObj.Annotations)
 			log.Infof("PVC %s has encryption enabled: %t", pvcName, isPvcEncrypted)
 
+			// Get BackingType from PVC annotation, if not available query from VirtualDiskManager
+			backingType := pvcObj.GetAnnotations()[common.AnnKeyBackingDiskType]
+			if backingType == "" {
+				log.Infof("BackingType annotation not found on PVC %s, querying from VirtualDiskManager", pvcName)
+				queriedBackingType, queryErr := queryBackingTypeFromVirtualDiskInfo(ctx, volumeManager, configInfo, volumeID)
+				if queryErr != nil {
+					log.Warnf("Failed to query BackingType from VirtualDiskManager for volumeID %s: %v. "+
+						"Proceeding with empty BackingType", volumeID, queryErr)
+				} else {
+					backingType = queriedBackingType
+					log.Infof("Successfully retrieved BackingType %s for PVC %s from VirtualDiskManager",
+						backingType, pvcName)
+
+					// Update the PVC annotation with the queried BackingType so it can be reused in future attach operations
+					patchErr := patchPVCBackingTypeAnnotation(ctx, k8sClient, pvcObj, backingType)
+					if patchErr != nil {
+						log.Warnf("Failed to update BackingType annotation on PVC %s: %v. "+
+							"The BackingType will be queried again on next attach", pvcName, patchErr)
+					} else {
+						log.Infof("Successfully updated BackingType annotation %s on PVC %s",
+							backingType, pvcName)
+					}
+				}
+			}
+
 			// Populate values for attach request.
 			currentBatchAttachRequest := volumes.BatchAttachRequest{
 				VolumeID:        volumeID,
 				SharingMode:     string(volume.PersistentVolumeClaim.SharingMode),
 				DiskMode:        string(volume.PersistentVolumeClaim.DiskMode),
 				ControllerKey:   volume.PersistentVolumeClaim.ControllerKey,
 				UnitNumber:      volume.PersistentVolumeClaim.UnitNumber,
-				BackingType:     pvcObj.GetAnnotations()[common.AnnKeyBackingDiskType],
+				BackingType:     backingType,
 				VolumeEncrypted: &isPvcEncrypted,
 			}
 			batchAttachRequest = append(batchAttachRequest, currentBatchAttachRequest)
@@ -755,6 +968,53 @@ func patchPVCAnnotations(ctx context.Context, k8sClient kubernetes.Interface,
 	return nil
 }
 
+// patchPVCBackingTypeAnnotation updates the BackingType annotation on the PVC.
+// This is used to cache the backing type so it doesn't need to be queried again on future attach operations.
+func patchPVCBackingTypeAnnotation(ctx context.Context, k8sClient kubernetes.Interface,
+	pvc *v1.PersistentVolumeClaim, backingType string) error {
+	log := logger.GetLogger(ctx)
+
+	patchAnnotations := make(map[string]interface{})
+	if pvc.Annotations != nil {
+		for k, v := range pvc.Annotations {
+			patchAnnotations[k] = v
+		}
+	}
+
+	log.Infof("Setting BackingType annotation %s=%s on PVC %s", common.AnnKeyBackingDiskType, backingType, pvc.Name)
+	patchAnnotations[common.AnnKeyBackingDiskType] = backingType
+
+	// Build patch structure
+	patch := map[string]interface{}{
+		"metadata": map[string]interface{}{
+			"annotations": patchAnnotations,
+		},
+	}
+
+	patchBytes, err := json.Marshal(patch)
+	if err != nil {
+		log.Errorf("failed to marshal BackingType annotation for PVC %s. Err: %s", pvc.Name, err)
+		return fmt.Errorf("failed to marshal patch: %v", err)
+	}
+
+	log.Infof("Patching PVC %s with BackingType annotation", pvc.Name)
+
+	// Apply the patch
+	updatedpvc, err := k8sClient.CoreV1().PersistentVolumeClaims(pvc.Namespace).Patch(
+		ctx,
+		pvc.Name,
+		types.MergePatchType,
+		patchBytes,
+		metav1.PatchOptions{},
+	)
+	if err != nil {
+		log.Errorf("failed to patch PVC %s with BackingType annotation. Err: %s", pvc.Name, err)
+		return fmt.Errorf("failed to patch PVC %s: %v", pvc.Name, err)
+	}
+	log.Debugf("Successfully patched PVC: %s with BackingType annotation %+v", pvc.Name, updatedpvc.Annotations)
+	return nil
+}
+
 // pvcHasUsedByAnnotaion goes through all annotations on the PVC to find out if the PVC is used by any VM or not.
 func pvcHasUsedByAnnotaion(ctx context.Context, pvc *v1.PersistentVolumeClaim) bool {
 	log := logger.GetLogger(ctx)