fix: copy volume content source to requested destination

Author: acortelyou

pkg/blob/controllerserver.go

@@ -367,9 +367,15 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		mc.ObserveOperationWithResult(isOperationSucceeded, VolumeID, volumeID)
 	}()
 
+	secrets := req.GetSecrets()
+
+	volumeContentSourceSASToken, volumeContentSourcePath, err := d.validateVolumeContentSource(ctx, req, storageEndpointSuffix, secrets, secretName, secretNamespace, getLatestAccountKey)
+	if err != nil {
+		return nil, err
+	}
+
 	var accountKey string
 	accountName := account
-	secrets := req.GetSecrets()
 	if len(secrets) == 0 && accountName == "" {
 		if v, ok := d.volMap.Load(volName); ok {
 			accountName = v.(string)
@@ -435,19 +441,19 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		setKeyValueInMap(parameters, containerNameField, validContainerName)
 	}
 
-	if req.GetVolumeContentSource() != nil {
+	klog.V(2).Infof("begin to create container(%s) on account(%s) type(%s) subsID(%s) rg(%s) location(%s) size(%d)", validContainerName, accountName, storageAccountType, subsID, resourceGroup, location, requestGiB)
+	if err := d.CreateBlobContainer(ctx, subsID, resourceGroup, accountName, validContainerName, secrets); err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to create container(%s) on account(%s) type(%s) rg(%s) location(%s) size(%d), error: %v", validContainerName, accountName, storageAccountType, resourceGroup, location, requestGiB, err)
+	}
+
+	if volumeContentSourcePath != "" {
 		accountSASToken, authAzcopyEnv, err := d.getAzcopyAuth(ctx, accountName, accountKey, storageEndpointSuffix, accountOptions, secrets, secretName, secretNamespace)
 		if err != nil {
 			return nil, status.Errorf(codes.Internal, "failed to getAzcopyAuth on account(%s) rg(%s), error: %v", accountOptions.Name, accountOptions.ResourceGroup, err)
 		}
-		if err := d.copyVolume(req, accountSASToken, authAzcopyEnv, validContainerName, storageEndpointSuffix); err != nil {
+		if err := d.copyVolumeContent(req, volumeContentSourceSASToken, volumeContentSourcePath, accountSASToken, accountName, validContainerName, authAzcopyEnv, storageEndpointSuffix); err != nil {
 			return nil, err
 		}
-	} else {
-		klog.V(2).Infof("begin to create container(%s) on account(%s) type(%s) subsID(%s) rg(%s) location(%s) size(%d)", validContainerName, accountName, storageAccountType, subsID, resourceGroup, location, requestGiB)
-		if err := d.CreateBlobContainer(ctx, subsID, resourceGroup, accountName, validContainerName, secrets); err != nil {
-			return nil, status.Errorf(codes.Internal, "failed to create container(%s) on account(%s) type(%s) rg(%s) location(%s) size(%d), error: %v", validContainerName, accountName, storageAccountType, resourceGroup, location, requestGiB, err)
-		}
 	}
 
 	if storeAccountKey && len(req.GetSecrets()) == 0 {
@@ -755,68 +761,96 @@ func (d *Driver) DeleteBlobContainer(ctx context.Context, subsID, resourceGroupN
 	})
 }
 
-// CopyBlobContainer copies a blob container in the same storage account
-func (d *Driver) copyBlobContainer(req *csi.CreateVolumeRequest, accountSasToken string, authAzcopyEnv []string, dstContainerName, storageEndpointSuffix string) error {
-	var sourceVolumeID string
-	if req.GetVolumeContentSource() != nil && req.GetVolumeContentSource().GetVolume() != nil {
-		sourceVolumeID = req.GetVolumeContentSource().GetVolume().GetVolumeId()
-
-	}
-	resourceGroupName, accountName, srcContainerName, _, _, err := GetContainerInfo(sourceVolumeID) //nolint:dogsled
-	if err != nil {
-		return status.Error(codes.NotFound, err.Error())
+// copyBlobContainer copies a source path to the destination container
+func (d *Driver) copyBlobContainer(srcSasToken string, srcPath string, dstSasToken string, dstAccountName string, dstContainerName string, authAzcopyEnv []string, storageEndpointSuffix string) error {
+	if srcPath == "" {
+		return fmt.Errorf("srcPath(%s) is empty", srcPath)
 	}
-	if srcContainerName == "" || dstContainerName == "" {
-		return fmt.Errorf("srcContainerName(%s) or dstContainerName(%s) is empty", srcContainerName, dstContainerName)
+	if dstAccountName == "" || dstContainerName == "" {
+		return fmt.Errorf("dstAccountName(%s) or dstContainerName(%s) is empty", dstAccountName, dstContainerName)
 	}
 
-	srcPath := fmt.Sprintf("https://%s.blob.%s/%s%s", accountName, storageEndpointSuffix, srcContainerName, accountSasToken)
-	dstPath := fmt.Sprintf("https://%s.blob.%s/%s%s", accountName, storageEndpointSuffix, dstContainerName, accountSasToken)
+	dstPath := fmt.Sprintf("https://%s.blob.%s/%s", dstAccountName, storageEndpointSuffix, dstContainerName)
 
 	jobState, percent, err := d.azcopy.GetAzcopyJob(dstContainerName, authAzcopyEnv)
-	klog.V(2).Infof("azcopy job status: %s, copy percent: %s%%, error: %v", jobState, percent, err)
+	klog.V(2).Infof("copyBlobContainer(%s) status: %s, percent: %s%%, error: %v", dstContainerName, jobState, percent, err)
 	switch jobState {
 	case util.AzcopyJobError, util.AzcopyJobCompleted:
 		return err
 	case util.AzcopyJobRunning:
-		return fmt.Errorf("wait for the existing AzCopy job to complete, current copy percentage is %s%%", percent)
+		return fmt.Errorf("copyBlobContainer(%s) running, percent: %s%%", dstContainerName, percent)
 	case util.AzcopyJobNotFound:
-		klog.V(2).Infof("copy blob container %s to %s", srcContainerName, dstContainerName)
+		klog.V(2).Infof("copyBlobContainer(%s) begin, srcPath: %s, dstPath: %s", dstContainerName, srcPath, dstPath)
 		execFunc := func() error {
-			cmd := exec.Command("azcopy", "copy", srcPath, dstPath, "--recursive", "--check-length=false")
+			cmd := exec.Command("azcopy", "copy", srcPath+srcSasToken, dstPath+dstSasToken, "--recursive", "--check-length=false")
 			if len(authAzcopyEnv) > 0 {
 				cmd.Env = append(os.Environ(), authAzcopyEnv...)
 			}
 			if out, err := cmd.CombinedOutput(); err != nil {
-				return fmt.Errorf("exec error: %v, output: %v", err, string(out))
+				return fmt.Errorf("copyBlobContainer(%s) failed, error: %v, output: %v", dstContainerName, err, string(out))
 			}
 			return nil
 		}
 		timeoutFunc := func() error {
 			_, percent, _ := d.azcopy.GetAzcopyJob(dstContainerName, authAzcopyEnv)
-			return fmt.Errorf("timeout waiting for copy blob container %s to %s complete, current copy percent: %s%%", srcContainerName, dstContainerName, percent)
+			return fmt.Errorf("timeout, percent: %s%%", percent)
 		}
 		copyErr := util.WaitUntilTimeout(time.Duration(d.waitForAzCopyTimeoutMinutes)*time.Minute, execFunc, timeoutFunc)
 		if copyErr != nil {
-			klog.Warningf("CopyBlobContainer(%s, %s, %s) failed with error: %v", resourceGroupName, accountName, dstPath, copyErr)
+			klog.Warningf("copyBlobContainer(%s) failed, error: %v", dstContainerName, copyErr)
 		} else {
-			klog.V(2).Infof("copied blob container %s to %s successfully", srcContainerName, dstContainerName)
+			klog.V(2).Infof("copyBlobContainer(%s) successful", dstContainerName)
 		}
 		return copyErr
 	}
 	return err
 }
 
-// copyVolume copies a volume form volume or snapshot, snapshot is not supported now
-func (d *Driver) copyVolume(req *csi.CreateVolumeRequest, accountSASToken string, authAzcopyEnv []string, dstContainerName, storageEndpointSuffix string) error {
+// validateVolumeContentSource validates source type and returns source path
+func (d *Driver) validateVolumeContentSource(ctx context.Context, req *csi.CreateVolumeRequest, storageEndpointSuffix string, secrets map[string]string, secretName, secretNamespace string, getLatestAccountKey bool) (string, string, error) {
+	vs := req.VolumeContentSource
+	if vs == nil {
+		return "", "", nil
+	}
+	switch vs.Type.(type) {
+	case *csi.VolumeContentSource_Snapshot:
+		return "", "", status.Errorf(codes.InvalidArgument, "volume content source is not supported: snapshot")
+	case *csi.VolumeContentSource_Volume:
+		var srcVolumeID string
+		if vs.GetVolume() != nil {
+			srcVolumeID = vs.GetVolume().GetVolumeId()
+		}
+		srcResourceGroupName, srcAccountName, srcContainerName, _, srcSubscriptionId, err := GetContainerInfo(srcVolumeID) //nolint:dogsled
+		if err != nil {
+			return "", "", status.Error(codes.NotFound, err.Error())
+		}
+		accountOptions := &azure.AccountOptions{
+			Name:                srcAccountName,
+			SubscriptionID:      srcSubscriptionId,
+			ResourceGroup:       srcResourceGroupName,
+			GetLatestAccountKey: getLatestAccountKey,
+		}
+		srcAccountSASToken, _, err := d.getAzcopyAuth(ctx, srcAccountName, "", storageEndpointSuffix, accountOptions, secrets, secretName, secretNamespace)
+		if err != nil {
+			return "", "", status.Errorf(codes.Internal, "failed to getAzcopyAuth on account(%s) rg(%s), error: %v", accountOptions.Name, accountOptions.ResourceGroup, err)
+		}
+		srcPath := fmt.Sprintf("https://%s.blob.%s/%s", srcAccountName, storageEndpointSuffix, srcContainerName)
+		return srcAccountSASToken, srcPath, nil
+	default:
+		return "", "", status.Errorf(codes.InvalidArgument, "volume content source is not recognized: %v", vs)
+	}
+}
+
+// copyVolumeContent copies volume content from supported sources to the destination container
+func (d *Driver) copyVolumeContent(req *csi.CreateVolumeRequest, srcSASToken string, srcPath string, dstSASToken string, dstAccountName string, dstContainerName string, authAzcopyEnv []string, storageEndpointSuffix string) error {
 	vs := req.VolumeContentSource
 	switch vs.Type.(type) {
 	case *csi.VolumeContentSource_Snapshot:
-		return status.Errorf(codes.InvalidArgument, "copy volume from volumeSnapshot is not supported")
+		return status.Errorf(codes.InvalidArgument, "volume content source is not supported: snapshot")
 	case *csi.VolumeContentSource_Volume:
-		return d.copyBlobContainer(req, accountSASToken, authAzcopyEnv, dstContainerName, storageEndpointSuffix)
+		return d.copyBlobContainer(srcSASToken, srcPath, dstSASToken, dstAccountName, dstContainerName, authAzcopyEnv, storageEndpointSuffix)
 	default:
-		return status.Errorf(codes.InvalidArgument, "%v is not a proper volume source", vs)
+		return status.Errorf(codes.InvalidArgument, "volume content source is not recognized: %v", vs)
 	}
 }