Merge pull request #1849 from andyzhangx/clean-azcopy-jobs

fix: cleanup azcopy jobs after job complete

Author: andyzhangx

pkg/blob/blob.go

@@ -282,7 +282,7 @@ func NewDriver(options *DriverOptions, kubeClient kubernetes.Interface, cloud *p
 		sasTokenExpirationMinutes:              options.SasTokenExpirationMinutes,
 		waitForAzCopyTimeoutMinutes:            options.WaitForAzCopyTimeoutMinutes,
 		fsGroupChangePolicy:                    options.FSGroupChangePolicy,
-		azcopy:                                 &util.Azcopy{},
+		azcopy:                                 &util.Azcopy{ExecCmd: &util.ExecCommand{}},
 		KubeClient:                             kubeClient,
 		cloud:                                  cloud,
 	}

pkg/blob/controllerserver.go

@@ -803,7 +803,7 @@ func (d *Driver) copyBlobContainer(ctx context.Context, req *csi.CreateVolumeReq
 	jobState, percent, err := d.azcopy.GetAzcopyJob(dstContainerName, authAzcopyEnv)
 	klog.V(2).Infof("azcopy job status: %s, copy percent: %s%%, error: %v", jobState, percent, err)
 	switch jobState {
-	case util.AzcopyJobError, util.AzcopyJobCompleted:
+	case util.AzcopyJobError, util.AzcopyJobCompleted, util.AzcopyJobCompletedWithErrors, util.AzcopyJobCompletedWithSkipped, util.AzcopyJobCompletedWithErrorsAndSkipped:
 		return err
 	case util.AzcopyJobRunning:
 		err = wait.PollImmediate(20*time.Second, time.Duration(d.waitForAzCopyTimeoutMinutes)*time.Minute, func() (bool, error) {
@@ -835,6 +835,9 @@ func (d *Driver) copyBlobContainer(ctx context.Context, req *csi.CreateVolumeReq
 		klog.Warningf("CopyBlobContainer(%s, %s, %s) failed with error: %v", accountOptions.ResourceGroup, dstAccountName, dstContainerName, err)
 	} else {
 		klog.V(2).Infof("copied blob container %s to %s successfully", srcContainerName, dstContainerName)
+		if out, err := d.azcopy.CleanJobs(); err != nil {
+			klog.Warningf("clean azcopy jobs failed with error: %v, output: %s", err, string(out))
+		}
 	}
 	return err
 }

pkg/util/util.go

@@ -45,10 +45,13 @@ const (
 type AzcopyJobState string
 
 const (
-	AzcopyJobError     AzcopyJobState = "Error"
-	AzcopyJobNotFound  AzcopyJobState = "NotFound"
-	AzcopyJobRunning   AzcopyJobState = "Running"
-	AzcopyJobCompleted AzcopyJobState = "Completed"
+	AzcopyJobError                         AzcopyJobState = "Error"
+	AzcopyJobNotFound                      AzcopyJobState = "NotFound"
+	AzcopyJobRunning                       AzcopyJobState = "Running"
+	AzcopyJobCompleted                     AzcopyJobState = "Completed"
+	AzcopyJobCompletedWithErrors           AzcopyJobState = "CompletedWithErrors"
+	AzcopyJobCompletedWithSkipped          AzcopyJobState = "CompletedWithSkipped"
+	AzcopyJobCompletedWithErrorsAndSkipped AzcopyJobState = "CompletedWithErrorsAndSkipped"
 )
 
 // RoundUpBytes rounds up the volume size in bytes up to multiplications of GiB
@@ -242,9 +245,6 @@ func (ac *Azcopy) GetAzcopyJob(dstBlobContainer string, authAzcopyEnv []string)
 	// Start Time: Wednesday, 09-Aug-23 09:09:03 UTC
 	// Status: Cancelled
 	// Command: copy https://{accountName}.file.core.windows.net/{srcBlobContainer}{SAStoken} https://{accountName}.file.core.windows.net/{dstBlobContainer}{SAStoken} --recursive --check-length=false
-	if ac.ExecCmd == nil {
-		ac.ExecCmd = &ExecCommand{}
-	}
 	out, err := ac.ExecCmd.RunCommand(cmdStr, authAzcopyEnv)
 	// if grep command returns nothing, the exec will return exit status 1 error, so filter this error
 	if err != nil && err.Error() != "exit status 1" {
@@ -278,13 +278,8 @@ func (ac *Azcopy) GetAzcopyJob(dstBlobContainer string, authAzcopyEnv []string)
 	return jobState, percent, nil
 }
 
-// TestListJobs test azcopy jobs list command with authAzcopyEnv
-func (ac *Azcopy) TestListJobs(accountName, storageEndpointSuffix string, authAzcopyEnv []string) (string, error) {
-	cmdStr := fmt.Sprintf("azcopy list %s", fmt.Sprintf("https://%s.blob.%s", accountName, storageEndpointSuffix))
-	if ac.ExecCmd == nil {
-		ac.ExecCmd = &ExecCommand{}
-	}
-	return ac.ExecCmd.RunCommand(cmdStr, authAzcopyEnv)
+func (ac *Azcopy) CleanJobs() (string, error) {
+	return ac.ExecCmd.RunCommand("azcopy jobs clean", nil)
 }
 
 // parseAzcopyJobList parse command azcopy jobs list, get jobid and state from joblist

pkg/util/util_test.go

@@ -353,6 +353,41 @@ func TestTrimDuplicatedSpace(t *testing.T) {
 	}
 }
 
+func TestCleanJobs(t *testing.T) {
+	tests := []struct {
+		desc        string
+		execStr     string
+		execErr     error
+		expectedErr error
+	}{
+		{
+			desc:        "run exec get error",
+			execStr:     "",
+			execErr:     fmt.Errorf("error"),
+			expectedErr: fmt.Errorf("error"),
+		},
+		{
+			desc:        "run exec succeed",
+			execStr:     "cleaned",
+			execErr:     nil,
+			expectedErr: nil,
+		},
+	}
+	for _, test := range tests {
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		m := NewMockEXEC(ctrl)
+		m.EXPECT().RunCommand(gomock.Eq("azcopy jobs clean"), nil).Return(test.execStr, test.execErr)
+
+		azcopyFunc := &Azcopy{ExecCmd: m}
+		_, err := azcopyFunc.CleanJobs()
+		if !reflect.DeepEqual(err, test.expectedErr) {
+			t.Errorf("test[%s]: unexpected err: %v, expected err: %v", test.desc, err, test.expectedErr)
+		}
+	}
+}
+
 func TestGetAzcopyJob(t *testing.T) {
 	tests := []struct {
 		desc             string