Update vm-operator api version and parallelize sync volume api calls (#3604)

Author: nikhilbarge

go.mod

@@ -23,7 +23,7 @@ require (
 	github.com/pkg/sftp v1.13.6
 	github.com/prometheus/client_golang v1.22.0
 	github.com/stretchr/testify v1.10.0
-	github.com/vmware-tanzu/vm-operator/api v1.8.7-0.20250820184450-53a697d52f9c
+	github.com/vmware-tanzu/vm-operator/api v1.9.1-0.20250923172217-bf5a74e51c65
 	github.com/vmware-tanzu/vm-operator/external/byok v0.0.0-20250509154507-b93e51fc90fa
 	github.com/vmware/govmomi v0.52.0-alpha.0.0.20250807230438-0eee109f1f2c
 	go.uber.org/zap v1.27.0

go.sum

@@ -287,8 +287,8 @@ github.com/thecodeteam/gofsutil v0.1.2 h1:FL87mBzZeeuDMZm8hpYLFcYylQdq6bbm8UQ1oc
 github.com/thecodeteam/gofsutil v0.1.2/go.mod h1:7bDOpr2aMnmdm9RTdxBEeqdOr+8RpnQhsB/VUEI3DgM=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20220101234140-673ab2c3ae75 h1:6fotK7otjonDflCTK0BCfls4SPy3NcCVb5dqqmbRknE=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20220101234140-673ab2c3ae75/go.mod h1:KO6IkyS8Y3j8OdNO85qEYBsRPuteD+YciPomcXdrMnk=
-github.com/vmware-tanzu/vm-operator/api v1.8.7-0.20250820184450-53a697d52f9c h1:+lECoxyxQcMsxU2CJ53YvaLEwqfGduS/3yQaLLRDYDo=
-github.com/vmware-tanzu/vm-operator/api v1.8.7-0.20250820184450-53a697d52f9c/go.mod h1:hkc/QZCSHcosWWMPS6VWWR12WenZcNE3BaTJ/8A8sNE=
+github.com/vmware-tanzu/vm-operator/api v1.9.1-0.20250923172217-bf5a74e51c65 h1:oby5iJHxU5KgtajXxT8B8VxUUPH20Zh0KTYlI+18AYs=
+github.com/vmware-tanzu/vm-operator/api v1.9.1-0.20250923172217-bf5a74e51c65/go.mod h1:nWTPpxfe4gHuuYuFcrs86+NMxfkqPk3a3IlvI8TCWak=
 github.com/vmware-tanzu/vm-operator/external/byok v0.0.0-20250509154507-b93e51fc90fa h1:4MKu14YJ7J54O6QKmT4ds5EUpysWLLtQRMff73cVkmU=
 github.com/vmware-tanzu/vm-operator/external/byok v0.0.0-20250509154507-b93e51fc90fa/go.mod h1:8tiuyYslzjLIUmOlXZuGKQdQP2ZgWGCVhVeyptmZYnk=
 github.com/vmware/govmomi v0.52.0-alpha.0.0.20250807230438-0eee109f1f2c h1:1nMVFr1CBMSNLLjsfx3QPfZ5k0R1/O29QX/A2X0w3RQ=

pkg/syncer/cnsoperator/controller/virtualmachinesnapshot/virtualmachinesnapshot_controller.go

@@ -29,6 +29,7 @@ import (
 	vmoperatortypes "github.com/vmware-tanzu/vm-operator/api/v1alpha5"
 	cnstypes "github.com/vmware/govmomi/cns/types"
 	"go.uber.org/zap"
+	syncgroup "golang.org/x/sync/errgroup"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -292,7 +293,7 @@ func (r *ReconcileVirtualMachineSnapshot) reconcileNormal(ctx context.Context, l
 		// if found fetch vmsnapshot and pvcs and pvs
 		vmKey := apitypes.NamespacedName{
 			Namespace: vmsnapshot.Namespace,
-			Name:      vmsnapshot.Spec.VMRef.Name,
+			Name:      vmsnapshot.Spec.VMName,
 		}
 		log.Infof("reconcileNormal: get virtulal machine %s/%s", vmKey.Namespace, vmKey.Name)
 		virtualMachine, _, err := utils.GetVirtualMachineAllApiVersions(ctx, vmKey,
@@ -426,6 +427,7 @@ func (r *ReconcileVirtualMachineSnapshot) syncVolumesAndUpdateCNSVolumeInfo(ctx
 	var err error
 	cnsVolumeIds := []cnstypes.CnsVolumeId{}
 	syncMode := []string{string(cnstypes.CnsSyncVolumeModeSPACE_USAGE)}
+	syncgrp, sgctx := syncgroup.WithContext(ctx)
 	for _, vmVolume := range vm.Spec.Volumes {
 		if vmVolume.VirtualMachineVolumeSource.PersistentVolumeClaim == nil {
 			continue
@@ -461,15 +463,10 @@ func (r *ReconcileVirtualMachineSnapshot) syncVolumesAndUpdateCNSVolumeInfo(ctx
 						SyncMode: syncMode,
 					},
 				}
-				// Trigger CNS VolumeSync API for identified volume-lds and Fetch Latest Aggregated snapshot size
-				log.Infof("syncVolumesAndUpdateCNSVolumeInfo: Trigger CNS VolumeSync API for volume %s",
-					cnsVolId)
-				syncVolumeFaultType, err := r.volumeManager.SyncVolume(ctx, syncVolumeSpecs)
-				if err != nil {
-					log.Errorf("syncVolumesAndUpdateCNSVolumeInfo: error while sync volume %s "+
-						"cnsfault %s. error: %v", cnsVolId, syncVolumeFaultType, err)
-					return err
-				}
+				// parallelize sync volume api calls
+				syncgrp.Go(func() error {
+					return r.invokeSyncVolume(sgctx, log, syncVolumeSpecs)
+				})
 			}
 		} else {
 			err = fmt.Errorf("could not find the PV associated with PVC %s/%s",
@@ -478,10 +475,19 @@ func (r *ReconcileVirtualMachineSnapshot) syncVolumesAndUpdateCNSVolumeInfo(ctx
 			return err
 		}
 	}
+	// if syncvolume api is not invoked for any volume, no need to update CNSVolumeInfos
 	if len(cnsVolumeIds) == 0 {
-		log.Infof("syncVolumesAndUpdateCNSVolumeInfo: no volumes found to sync, skipping volume sync")
+		log.Info("syncVolumesAndUpdateCNSVolumeInfo: no volumes found to sync, skipped volume sync")
 		return nil
 	}
+	log.Info("syncVolumesAndUpdateCNSVolumeInfo: wait for syncvolume operation to be completed")
+	if err := syncgrp.Wait(); err != nil {
+		log.Errorf("syncVolumesAndUpdateCNSVolumeInfo: error while sync volume "+
+			"error: %v", err)
+		return err
+	}
+	log.Info("syncVolumesAndUpdateCNSVolumeInfo: volumes are synced successfully. " +
+		"will update related CNSVolumeInfos")
 	// fetch updated cns volumes
 	queryFilter := cnstypes.CnsQueryFilter{
 		VolumeIds: cnsVolumeIds,
@@ -500,7 +506,7 @@ func (r *ReconcileVirtualMachineSnapshot) syncVolumesAndUpdateCNSVolumeInfo(ctx
 
 				//  Update CNSVolumeInfo with latest aggregated Size and Update SPU used value.
 				patch, err := common.GetCNSVolumeInfoPatch(ctx, val.AggregatedSnapshotCapacityInMb,
-					cnsvolume.VolumeId.Id) // TODO: UDPATE to value returned
+					cnsvolume.VolumeId.Id)
 				if err != nil {
 					log.Errorf("syncVolumesAndUpdateCNSVolumeInfo: failed to get cnsvolumeinfo patch for "+
 						"volume %s, error: %v", cnsvolume.VolumeId.Id, err)
@@ -529,3 +535,24 @@ func (r *ReconcileVirtualMachineSnapshot) syncVolumesAndUpdateCNSVolumeInfo(ctx
 	}
 	return nil
 }
+
+func (r *ReconcileVirtualMachineSnapshot) invokeSyncVolume(ctx context.Context, log *zap.SugaredLogger,
+	syncVolumeSpecs []cnstypes.CnsSyncVolumeSpec) error {
+	select {
+	case <-ctx.Done():
+		log.Infof("invokeSyncVolume: Sync Volume Operation cancelled for volume %s",
+			syncVolumeSpecs[0])
+		return ctx.Err()
+	default:
+		// Trigger CNS VolumeSync API for identified volume-lds and Fetch Latest Aggregated snapshot size
+		log.Infof("invokeSyncVolume: Trigger CNS SyncVolume API for volume %s",
+			syncVolumeSpecs[0])
+		syncVolumeFaultType, err := r.volumeManager.SyncVolume(ctx, syncVolumeSpecs)
+		if err != nil {
+			log.Errorf("invokeSyncVolume: error while sync volume %s "+
+				"cnsfault %s. error: %v", syncVolumeSpecs[0], syncVolumeFaultType, err)
+			return err
+		}
+		return nil
+	}
+}