Repository: kubernetes-sigs/azurefile-csi-driver
Pull request #1751 (2024-03-14 07:37:54)

fix: fix azcopy exec timeout func
<!--  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?**

<!--
Add one of the following kinds:
/kind bug
/kind cleanup
/kind documentation
/kind feature
/kind test
/kind design

Optionally add one or more of the following kinds if applicable:
/kind api-change
/kind deprecation
/kind failing-test
/kind flake
/kind regression
-->

/kind bug

**What this PR does / why we need it**:

fix: fix azcopy exec timeout func, use `WaitForExecCompletion` func in `util.go` waits for the exec function to complete or times out

**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 #

**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**:
```
none
```


k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/azurefile-csi-driver/pull/1751#" title="Author self-approved">umagnus</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign [andyzhangx](https://github.com/andyzhangx) for approval. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fazurefile-csi-driver).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes-sigs/azurefile-csi-driver/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["andyzhangx"]} -->
k8s-ci-robot: Hi @umagnus. Thanks for your PR.

I'm waiting for a [kubernetes-sigs](https://github.com/orgs/kubernetes-sigs/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://git.k8s.io/community/community-membership.md#member) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fazurefile-csi-driver).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>


Patch:
diff --git a/.trivyignore b/.trivyignore
deleted file mode 100644
index b0f56dcff..000000000
--- a/.trivyignore
+++ /dev/null
@@ -1,1 +0,0 @@
-CVE-2024-24786
diff --git a/pkg/azurefile/azurefile.go b/pkg/azurefile/azurefile.go
index 4868cbbb2..49e8d7085 100644
--- a/pkg/azurefile/azurefile.go
+++ b/pkg/azurefile/azurefile.go
@@ -187,8 +187,6 @@ const (
 	SnapshotID       = "snapshot_id"
 
 	FSGroupChangeNone = "None"
-
-	waitForAzCopyInterval = 2 * time.Second
 )
 
 var (
@@ -1011,55 +1009,59 @@ func (d *Driver) copyFileShare(ctx context.Context, req *csi.CreateVolumeRequest
 		return fmt.Errorf("srcFileShareName(%s) or dstFileShareName(%s) is empty", srcFileShareName, dstFileShareName)
 	}
 
-	timeAfter := time.After(time.Duration(d.waitForAzCopyTimeoutMinutes) * time.Minute)
-	timeTick := time.Tick(waitForAzCopyInterval)
 	srcPath := fmt.Sprintf("https://%s.file.%s/%s%s", accountName, storageEndpointSuffix, srcFileShareName, accountSASToken)
 	dstPath := fmt.Sprintf("https://%s.file.%s/%s%s", accountName, storageEndpointSuffix, dstFileShareName, accountSASToken)
 
 	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 jobState == fileutil.AzcopyJobError || jobState == fileutil.AzcopyJobCompleted {
+	switch jobState {
+	case fileutil.AzcopyJobError, fileutil.AzcopyJobCompleted:
 		return err
-	}
-	klog.V(2).Infof("begin to copy fileshare %s to %s", srcFileShareName, dstFileShareName)
-	for {
-		select {
-		case <-timeTick:
-			jobState, percent, err := d.azcopy.GetAzcopyJob(dstFileShareName, authAzcopyEnv)
-			klog.V(2).Infof("azcopy job status: %s, copy percent: %s%%, error: %v", jobState, percent, err)
-			switch jobState {
-			case fileutil.AzcopyJobError, fileutil.AzcopyJobCompleted:
+	case fileutil.AzcopyJobRunning:
+		return fmt.Errorf("wait for the existing AzCopy job to complete, current copy percentage is %s%%", percent)
+	case fileutil.AzcopyJobNotFound:
+		klog.V(2).Infof("copy fileshare %s to %s", srcFileShareName, dstFileShareName)
+		execFuncWithAuth := func() error {
+			cmd := exec.Command("azcopy", "copy", srcPath, dstPath)
+			cmd.Args = append(cmd.Args, defaultAzcopyCopyOptions...)
+			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 nil
+		}
+		timeoutFunc := func() error {
+			_, percent, _ := d.azcopy.GetAzcopyJob(dstFileShareName, authAzcopyEnv)
+			return fmt.Errorf("timeout waiting for copy blob container %s to %s complete, current copy percent: %s%%", srcFileShareName, dstFileShareName, percent)
+		}
+		copyErr := fileutil.WaitForExecCompletion(time.Duration(d.waitForAzCopyTimeoutMinutes)*time.Minute, execFuncWithAuth, timeoutFunc)
+		if accountSASToken == "" && copyErr != nil && strings.Contains(copyErr.Error(), authorizationPermissionMismatch) {
+			klog.Warningf("azcopy list failed with AuthorizationPermissionMismatch error, should assign \"Storage File Data SMB Share Elevated Contributor\" role to controller identity, fall back to use sas token, original error: %v", copyErr)
+			d.azcopySasTokenCache.Set(accountName, "")
+			var sasToken string
+			if sasToken, _, err = d.getAzcopyAuth(ctx, accountName, "", storageEndpointSuffix, accountOptions, secrets, secretName, secretNamespace, true); err != nil {
 				return err
-			case fileutil.AzcopyJobNotFound:
-				klog.V(2).Infof("copy fileshare %s to %s", srcFileShareName, dstFileShareName)
-				cmd := exec.Command("azcopy", "copy", srcPath, dstPath)
+			}
+			execFuncWithSasToken := func() error {
+				cmd := exec.Command("azcopy", "copy", srcPath+sasToken, dstPath+sasToken)
 				cmd.Args = append(cmd.Args, defaultAzcopyCopyOptions...)
-				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))
 				}
-				out, copyErr := cmd.CombinedOutput()
-				if accountSASToken == "" && strings.Contains(string(out), authorizationPermissionMismatch) && copyErr != nil {
-					klog.Warningf("azcopy list failed with AuthorizationPermissionMismatch error, should assign \"Storage File Data SMB Share Elevated Contributor\" role to controller identity, fall back to use sas token, original output: %v", string(out))
-					d.azcopySasTokenCache.Set(accountName, "")
-					var sasToken string
-					if sasToken, _, err = d.getAzcopyAuth(ctx, accountName, "", storageEndpointSuffix, accountOptions, secrets, secretName, secretNamespace, true); err != nil {
-						return err
-					}
-					cmd := exec.Command("azcopy", "copy", srcPath+sasToken, dstPath+sasToken)
-					cmd.Args = append(cmd.Args, defaultAzcopyCopyOptions...)
-					out, copyErr = cmd.CombinedOutput()
-				}
-				if copyErr != nil {
-					klog.Warningf("CopyFileShare(%s, %s, %s) failed with error(%v): %v", resourceGroupName, accountName, dstFileShareName, copyErr, string(out))
-				} else {
-					klog.V(2).Infof("copied fileshare %s to %s successfully", srcFileShareName, dstFileShareName)
-				}
-				return copyErr
+				return nil
 			}
-		case <-timeAfter:
-			return fmt.Errorf("timeout waiting for copy fileshare %s to %s succeed", srcFileShareName, dstFileShareName)
+			copyErr = fileutil.WaitForExecCompletion(time.Duration(d.waitForAzCopyTimeoutMinutes)*time.Minute, execFuncWithSasToken, timeoutFunc)
 		}
+		if copyErr != nil {
+			klog.Warningf("CopyFileShare(%s, %s, %s) failed with error: %v", resourceGroupName, accountName, dstFileShareName, copyErr)
+		} else {
+			klog.V(2).Infof("copied fileshare %s to %s successfully", srcFileShareName, dstFileShareName)
+		}
+		return copyErr
 	}
+	return err
 }
 
 // GetTotalAccountQuota returns the total quota in GB of all file shares in the storage account and the number of file shares
diff --git a/pkg/azurefile/controllerserver.go b/pkg/azurefile/controllerserver.go
index 47a4c026e..6ad3c21b0 100644
--- a/pkg/azurefile/controllerserver.go
+++ b/pkg/azurefile/controllerserver.go
@@ -117,7 +117,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		// logging the job status if it's volume cloning
 		if req.GetVolumeContentSource() != nil {
 			jobState, percent, err := d.azcopy.GetAzcopyJob(volName, []string{})
-			klog.V(2).Infof("azcopy job status: %s, copy percent: %s%%, error: %v", jobState, percent, err)
+			return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsWithAzcopyFmt, volName, jobState, percent, err)
 		}
 		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, volName)
 	}
diff --git a/pkg/azurefile/volume_lock.go b/pkg/azurefile/volume_lock.go
index b244f918a..72a3b82b1 100644
--- a/pkg/azurefile/volume_lock.go
+++ b/pkg/azurefile/volume_lock.go
@@ -23,7 +23,8 @@ import (
 )
 
 const (
-	volumeOperationAlreadyExistsFmt = "An operation with the given Volume ID %s already exists"
+	volumeOperationAlreadyExistsFmt           = "An operation with the given Volume ID %s already exists"
+	volumeOperationAlreadyExistsWithAzcopyFmt = "An operation using azcopy with the given Volume ID %s already exists. Azcopy job status: %s, copy percent: %s%%, error: %v"
 )
 
 // VolumeLocks implements a map with atomic operations. It stores a set of all volume IDs
diff --git a/pkg/util/util.go b/pkg/util/util.go
index fd8f29bad..199daa6c1 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -22,6 +22,7 @@ import (
 	"os/exec"
 	"strings"
 	"sync"
+	"time"
 
 	"k8s.io/klog/v2"
 )
@@ -205,3 +206,30 @@ func parseAzcopyJobShow(jobshow string) (AzcopyJobState, string, error) {
 	}
 	return AzcopyJobRunning, strings.ReplaceAll(segments[1], "\n", ""), nil
 }
+
+// ExecFunc returns a exec function's output and error
+type ExecFunc func() (err error)
+
+// TimeoutFunc returns output and error if an ExecFunc timeout
+type TimeoutFunc func() (err error)
+
+// WaitForExecCompletion waits for the exec function to complete or return timeout error
+func WaitForExecCompletion(timeout time.Duration, execFunc ExecFunc, timeoutFunc TimeoutFunc) error {
+	// Create a channel to receive the result of the azcopy exec function
+	done := make(chan bool)
+	var err error
+
+	// Start the azcopy exec function in a goroutine
+	go func() {
+		err = execFunc()
+		done <- true
+	}()
+
+	// Wait for the function to complete or time out
+	select {
+	case <-done:
+		return err
+	case <-time.After(timeout):
+		return timeoutFunc()
+	}
+}


Test patch:
diff --git a/pkg/azurefile/controllerserver_test.go b/pkg/azurefile/controllerserver_test.go
index 5f94c8e1c..7bac6b004 100644
--- a/pkg/azurefile/controllerserver_test.go
+++ b/pkg/azurefile/controllerserver_test.go
@@ -1890,7 +1890,7 @@ func TestCopyVolume(t *testing.T) {
 			},
 		},
 		{
-			name: "azcopy job is first in progress and then be completed",
+			name: "azcopy job is in progress",
 			testFunc: func(t *testing.T) {
 				d := NewFakeDriver()
 				mp := map[string]string{}
@@ -1919,15 +1919,12 @@ func TestCopyVolume(t *testing.T) {
 
 				m := util.NewMockEXEC(ctrl)
 				listStr1 := "JobId: ed1c3833-eaff-fe42-71d7-513fb065a9d9\nStart Time: Monday, 07-Aug-23 03:29:54 UTC\nStatus: InProgress\nCommand: copy https://{accountName}.file.core.windows.net/{srcFileshare}{SAStoken} https://{accountName}.file.core.windows.net/{dstFileshare}{SAStoken} --recursive --check-length=false"
-				listStr2 := "JobId: ed1c3833-eaff-fe42-71d7-513fb065a9d9\nStart Time: Monday, 07-Aug-23 03:29:54 UTC\nStatus: Completed\nCommand: copy https://{accountName}.file.core.windows.net/{srcFileshare}{SAStoken} https://{accountName}.file.core.windows.net/{dstFileshare}{SAStoken} --recursive --check-length=false"
-				o1 := m.EXPECT().RunCommand(gomock.Eq("azcopy jobs list | grep dstFileshare -B 3"), gomock.Any()).Return(listStr1, nil).Times(1)
+				m.EXPECT().RunCommand(gomock.Eq("azcopy jobs list | grep dstFileshare -B 3"), gomock.Any()).Return(listStr1, nil).Times(1)
 				m.EXPECT().RunCommand(gomock.Not("azcopy jobs list | grep dstFileshare -B 3"), gomock.Any()).Return("Percent Complete (approx): 50.0", nil)
-				o2 := m.EXPECT().RunCommand(gomock.Eq("azcopy jobs list | grep dstFileshare -B 3"), gomock.Any()).Return(listStr2, nil)
-				gomock.InOrder(o1, o2)
 
 				d.azcopy.ExecCmd = m
 
-				var expectedErr error
+				expectedErr := fmt.Errorf("wait for the existing AzCopy job to complete, current copy percentage is 50.0%%")
 				err := d.copyVolume(ctx, req, "sastoken", []string{}, "", "", secret, &fileclient.ShareOptions{Name: "dstFileshare"}, nil, "core.windows.net")
 				if !reflect.DeepEqual(err, expectedErr) {
 					t.Errorf("Unexpected error: %v", err)
diff --git a/pkg/util/util_test.go b/pkg/util/util_test.go
index 7e2716997..1417ca678 100644
--- a/pkg/util/util_test.go
+++ b/pkg/util/util_test.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"reflect"
 	"testing"
+	"time"
 
 	gomock "go.uber.org/mock/gomock"
 )
@@ -260,3 +261,55 @@ func TestParseAzcopyJobShow(t *testing.T) {
 		}
 	}
 }
+
+func TestWaitForExecCompletion(t *testing.T) {
+	tests := []struct {
+		desc        string
+		timeout     time.Duration
+		execFunc    ExecFunc
+		timeoutFunc TimeoutFunc
+		expectedErr error
+	}{
+		{
+			desc:    "execFunc returns error",
+			timeout: 1 * time.Second,
+			execFunc: func() error {
+				return fmt.Errorf("execFunc error")
+			},
+			timeoutFunc: func() error {
+				return fmt.Errorf("timeout error")
+			},
+			expectedErr: fmt.Errorf("execFunc error"),
+		},
+		{
+			desc:    "execFunc timeout",
+			timeout: 1 * time.Second,
+			execFunc: func() error {
+				time.Sleep(2 * time.Second)
+				return nil
+			},
+			timeoutFunc: func() error {
+				return fmt.Errorf("timeout error")
+			},
+			expectedErr: fmt.Errorf("timeout error"),
+		},
+		{
+			desc:    "execFunc completed successfully",
+			timeout: 1 * time.Second,
+			execFunc: func() error {
+				return nil
+			},
+			timeoutFunc: func() error {
+				return fmt.Errorf("timeout error")
+			},
+			expectedErr: nil,
+		},
+	}
+
+	for _, test := range tests {
+		err := WaitForExecCompletion(test.timeout, test.execFunc, test.timeoutFunc)
+		if err != nil && (err.Error() != test.expectedErr.Error()) {
+			t.Errorf("unexpected error: %v, expected error: %v", err, test.expectedErr)
+		}
+	}
+}