Repository: kubernetes-sigs/azurefile-csi-driver (go)

Issue:
Remove plan and log files when azcopy jobs are complete
**Is your feature request related to a problem?/Why is this needed**
<!-- A clear and concise description of what the problem is. Ex. I'm always frustrated when [...] -->

When we could **make sure that there is no azcopy job running**, we could use `azcopy jobs clean` to clean all azcopy plan and log files after snapshot restore or volume cloning complete:
https://learn.microsoft.com/en-us/azure/storage/common/storage-use-azcopy-configure?source=recommendations#remove-plan-and-log-files

**Describe the solution you'd like in detail**
<!-- A clear and concise description of what you want to happen. -->

**Describe alternatives you've considered**
<!-- A clear and concise description of any alternative solutions or features you've considered. -->

**Additional context**
<!-- Add any other context or screenshots about the feature request here. -->


Pull request:
fix: cleanup azcopy jobs after job complete
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Please label this pull request according to what type of issue you are addressing, especially if this is a release targeted pull request. For reference on required PR/issue labels, read here:
https://git.k8s.io/community/contributors/devel/release.md#issue-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://git.k8s.io/community/contributors/devel/testing.md
4. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
5. Follow the instructions for writing a release note: https://git.k8s.io/community/contributors/guide/release-notes.md
6. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What type of PR is this?**
/kind bug

**What this PR does / why we need it**:
fix: cleanup azcopy jobs after job complete
this could reduce .azopy/plans folder files

**Which issue(s) this PR fixes**:
<!-- 
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #2387

**Requirements**:
- [ ] uses [conventional commit messages](https://www.conventionalcommits.org/)
  <!-- Common commit types:
        build: Build 🏭
        chore: Maintenance 🔧
        ci: Continuous Integration 💜
        docs: Documentation 📘
        feat: Features 🌈
        fix: Bug Fixes 🐞
        perf: Performance Improvements 🚀
        refactor: Code Refactoring 💎
        revert: Revert Change ◀️
        style: Code Style 🎶
        security: Security Fix 🛡️
        test: Testing 💚 -->
- [ ] includes documentation
- [ ] adds unit tests
- [ ] tested upgrade from previous version

**Special notes for your reviewer**:


**Release note**:
```
fix: cleanup azcopy jobs after job complete
```


Patch:
diff --git a/pkg/azurefile/azurefile.go b/pkg/azurefile/azurefile.go
index 455cf7474..c6a60f272 100644
--- a/pkg/azurefile/azurefile.go
+++ b/pkg/azurefile/azurefile.go
@@ -325,7 +325,7 @@ func NewDriver(options *DriverOptions) *Driver {
 	driver.volLockMap = newLockMap()
 	driver.subnetLockMap = newLockMap()
 	driver.volumeLocks = newVolumeLocks()
-	driver.azcopy = &fileutil.Azcopy{}
+	driver.azcopy = &fileutil.Azcopy{ExecCmd: &fileutil.ExecCommand{}}
 	driver.kubeconfig = options.KubeConfig
 	driver.endpoint = options.Endpoint
 	driver.resolver = new(NetResolver)
diff --git a/pkg/azurefile/controllerserver.go b/pkg/azurefile/controllerserver.go
index fd83c18d4..da3b85d03 100644
--- a/pkg/azurefile/controllerserver.go
+++ b/pkg/azurefile/controllerserver.go
@@ -26,7 +26,7 @@ import (
 	"strings"
 	"time"
 
-	volumehelper "sigs.k8s.io/azurefile-csi-driver/pkg/util"
+	"sigs.k8s.io/azurefile-csi-driver/pkg/util"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/storage/armstorage"
@@ -95,7 +95,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	}
 
 	capacityBytes := req.GetCapacityRange().GetRequiredBytes()
-	requestGiB := volumehelper.RoundUpGiB(capacityBytes)
+	requestGiB := util.RoundUpGiB(capacityBytes)
 	if requestGiB == 0 {
 		requestGiB = defaultAzureFileQuota
 		klog.Warningf("no quota specified, set as default value(%d GiB)", defaultAzureFileQuota)
@@ -642,7 +642,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 			// use uuid as vhd disk name if file share specified
 			diskName = uuid.NewString() + vhdSuffix
 		}
-		diskSizeBytes := volumehelper.GiBToBytes(requestGiB)
+		diskSizeBytes := util.GiBToBytes(requestGiB)
 		klog.V(2).Infof("begin to create vhd file(%s) size(%d) on share(%s) on account(%s) type(%s) rg(%s) location(%s)",
 			diskName, diskSizeBytes, validFileShareName, account, sku, resourceGroup, location)
 		if err := createDisk(ctx, accountName, accountKey, d.getStorageEndPointSuffix(), validFileShareName, diskName, diskSizeBytes); err != nil {
@@ -914,7 +914,7 @@ func (d *Driver) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequ
 		klog.V(2).Infof("snapshot(%s) already exists", snapshotName)
 		return &csi.CreateSnapshotResponse{
 			Snapshot: &csi.Snapshot{
-				SizeBytes:      volumehelper.GiBToBytes(int64(itemSnapshotQuota)),
+				SizeBytes:      util.GiBToBytes(int64(itemSnapshotQuota)),
 				SnapshotId:     sourceVolumeID + "#" + itemSnapshot,
 				SourceVolumeId: sourceVolumeID,
 				CreationTime:   timestamppb.New(itemSnapshotTime),
@@ -996,7 +996,7 @@ func (d *Driver) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequ
 
 	createResp := &csi.CreateSnapshotResponse{
 		Snapshot: &csi.Snapshot{
-			SizeBytes:      volumehelper.GiBToBytes(int64(itemSnapshotQuota)),
+			SizeBytes:      util.GiBToBytes(int64(itemSnapshotQuota)),
 			SnapshotId:     sourceVolumeID + "#" + itemSnapshot,
 			SourceVolumeId: sourceVolumeID,
 			CreationTime:   timestamppb.New(itemSnapshotTime),
@@ -1134,21 +1134,21 @@ func (d *Driver) copyFileShareByAzcopy(srcFileShareName, dstFileShareName, srcPa
 	klog.V(2).Infof("azcopy job status: %s, copy percent: %s%%, error: %v", jobState, percent, err)
 
 	switch jobState {
-	case volumehelper.AzcopyJobError, volumehelper.AzcopyJobCompleted:
+	case util.AzcopyJobError, util.AzcopyJobCompleted, util.AzcopyJobCompletedWithErrors, util.AzcopyJobCompletedWithSkipped, util.AzcopyJobCompletedWithErrorsAndSkipped:
 		return err
-	case volumehelper.AzcopyJobRunning:
+	case util.AzcopyJobRunning:
 		err = wait.PollImmediate(20*time.Second, time.Duration(d.waitForAzCopyTimeoutMinutes)*time.Minute, func() (bool, error) {
 			jobState, percent, err := d.azcopy.GetAzcopyJob(dstFileShareName, authAzcopyEnv)
 			klog.V(2).Infof("azcopy job status: %s, copy percent: %s%%, error: %v", jobState, percent, err)
 			if err != nil {
 				return false, err
 			}
-			if jobState == volumehelper.AzcopyJobRunning {
+			if jobState == util.AzcopyJobRunning {
 				return false, nil
 			}
 			return true, nil
 		})
-	case volumehelper.AzcopyJobNotFound:
+	case util.AzcopyJobNotFound:
 		klog.V(2).Infof("copy fileshare %s:%s to %s:%s", srcAccountName, srcFileShareName, dstAccountName, dstFileShareName)
 		execAzcopyJob := func() error {
 			if out, err := d.execAzcopyCopy(srcPathAuth, dstPath, azcopyCopyOptions, authAzcopyEnv); err != nil {
@@ -1160,13 +1160,16 @@ func (d *Driver) copyFileShareByAzcopy(srcFileShareName, dstFileShareName, srcPa
 			jobState, percent, _ := d.azcopy.GetAzcopyJob(dstFileShareName, authAzcopyEnv)
 			return fmt.Errorf("azcopy job status: %s, timeout waiting for copy fileshare %s:%s to %s:%s complete, current copy percent: %s%%", jobState, srcAccountName, srcFileShareName, dstAccountName, dstFileShareName, percent)
 		}
-		err = volumehelper.WaitUntilTimeout(time.Duration(d.waitForAzCopyTimeoutMinutes)*time.Minute, execAzcopyJob, timeoutFunc)
+		err = util.WaitUntilTimeout(time.Duration(d.waitForAzCopyTimeoutMinutes)*time.Minute, execAzcopyJob, timeoutFunc)
 	}
 
 	if err != nil {
 		klog.Warningf("CopyFileShare(%s, %s, %s) failed with error: %v", accountOptions.ResourceGroup, dstAccountName, dstFileShareName, err)
 	} else {
 		klog.V(2).Infof("copied fileshare %s to %s successfully", srcFileShareName, dstFileShareName)
+		if out, err := d.azcopy.CleanJobs(); err != nil {
+			klog.Warningf("clean azcopy jobs failed with error: %v, output: %s", err, string(out))
+		}
 	}
 	return err
 }
@@ -1191,7 +1194,7 @@ func (d *Driver) ControllerExpandVolume(ctx context.Context, req *csi.Controller
 	if capacityBytes == 0 {
 		return nil, status.Error(codes.InvalidArgument, "volume capacity range missing in request")
 	}
-	requestGiB := volumehelper.RoundUpGiB(capacityBytes)
+	requestGiB := util.RoundUpGiB(capacityBytes)
 	if err := d.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_EXPAND_VOLUME); err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "invalid expand volume request: %v", req)
 	}
diff --git a/pkg/util/util.go b/pkg/util/util.go
index 23e251774..416e9fce4 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -34,10 +34,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
@@ -107,9 +110,6 @@ func (ac *Azcopy) GetAzcopyJob(dstFileshare string, authAzcopyEnv []string) (Azc
 	// Start Time: Wednesday, 09-Aug-23 09:09:03 UTC
 	// Status: Cancelled
 	// Command: copy https://{accountName}.file.core.windows.net/{srcFileshare}{SAStoken} https://{accountName}.file.core.windows.net/{dstFileshare}{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" {
@@ -143,13 +143,8 @@ func (ac *Azcopy) GetAzcopyJob(dstFileshare string, authAzcopyEnv []string) (Azc
 	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.file.%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


Test patch:
diff --git a/pkg/util/util_test.go b/pkg/util/util_test.go
index fcf05f800..0e7c6417e 100644
--- a/pkg/util/util_test.go
+++ b/pkg/util/util_test.go
@@ -162,8 +162,7 @@ func TestGetAzcopyJob(t *testing.T) {
 			m.EXPECT().RunCommand(gomock.Not("azcopy jobs list | grep dstFileshare -B 3"), []string{}).Return(test.showStr, test.showErr)
 		}
 
-		azcopyFunc := &Azcopy{}
-		azcopyFunc.ExecCmd = m
+		azcopyFunc := &Azcopy{ExecCmd: m}
 		jobState, percent, err := azcopyFunc.GetAzcopyJob(dstFileshare, []string{})
 		if jobState != test.expectedJobState || percent != test.expectedPercent || !reflect.DeepEqual(err, test.expectedErr) {
 			t.Errorf("test[%s]: unexpected jobState: %v, percent: %v, err: %v, expected jobState: %v, percent: %v, err: %v", test.desc, jobState, percent, err, test.expectedJobState, test.expectedPercent, test.expectedErr)
@@ -171,6 +170,41 @@ func TestGetAzcopyJob(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 TestParseAzcopyJobList(t *testing.T) {
 	tests := []struct {
 		desc             string