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

Issue:
Handle concurrent operation of NodeUnstage with NodeStage and NodeUnpublish with NodePublish
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!
-->


**What happened**:
we don't correctly handle concurrent processing of NodeUnstage call if a previous NodeStage was in-progress and same thing applies to NodeUnpublish call too.

The specs says, if an operation is in-progress for the volume then SP should return error code "10 ABORTED" - https://github.com/container-storage-interface/spec/blob/master/spec.md#timeouts

/kind bug

**What you expected to happen**:

**How to reproduce it**:

**Anything else we need to know?**:

**Environment**:
- CSI Driver version:
- Kubernetes version (use `kubectl version`):
- OS (e.g. from /etc/os-release):
- Kernel (e.g. `uname -a`):
- Install tools:
- Others:


Pull request description:
fix: Handle concurrent operation of NodeUnstage with NodeStage and NodeUnpublish with NodePublish
<!--  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
<!--
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
-->

**What this PR does / why we need it**:
we don't correctly handle concurrent processing of NodeUnstage call if a previous NodeStage was in-progress and same thing applies to NodeUnpublish call too.

The specs says, if an operation is in-progress for the volume then SP should return error code "10 ABORTED".

**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 #253

**Requirements**:
- [x] 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**:
```
none
```

**Release note**:
```
none
```


Patch:
diff --git a/deploy/example/statefulset-nonroot.yaml b/deploy/example/statefulset-nonroot.yaml
index 4f4a01174..a2c1f9098 100644
--- a/deploy/example/statefulset-nonroot.yaml
+++ b/deploy/example/statefulset-nonroot.yaml
@@ -23,9 +23,9 @@ spec:
         - name: statefulset-azurefile
           image: andyzhangx/ubuntu1604:nonroot
           command:
-            - "/bin/bash"
+            - "/bin/sh"
             - "-c"
-            - set -euo pipefail; while true; do echo $(date) >> /mnt/azurefile/outfile; sleep 1; done
+            - while true; do echo $(date) >> /mnt/azurefile/outfile; sleep 1; done
           volumeMounts:
             - name: persistent-storage
               mountPath: /mnt/azurefile
diff --git a/pkg/azurefile/azurefile.go b/pkg/azurefile/azurefile.go
index 62492b2f6..4acaaec1e 100644
--- a/pkg/azurefile/azurefile.go
+++ b/pkg/azurefile/azurefile.go
@@ -130,6 +130,9 @@ type Driver struct {
 	volLockMap *lockMap
 	// only for nfs feature
 	subnetLockMap *lockMap
+	// A map storing all volumes with ongoing operations so that additional operations
+	// for that same volume (as defined by VolumeID) return an Aborted error
+	volumeLocks *volumeLocks
 }
 
 // NewDriver Creates a NewCSIDriver object. Assumes vendor version is equal to driver version &
@@ -141,6 +144,7 @@ func NewDriver(nodeID string) *Driver {
 	driver.NodeID = nodeID
 	driver.volLockMap = newLockMap()
 	driver.subnetLockMap = newLockMap()
+	driver.volumeLocks = newVolumeLocks()
 	return &driver
 }
 
diff --git a/pkg/azurefile/nodeserver.go b/pkg/azurefile/nodeserver.go
index afd2c1c9a..af92d1ed3 100644
--- a/pkg/azurefile/nodeserver.go
+++ b/pkg/azurefile/nodeserver.go
@@ -56,6 +56,13 @@ func (d *Driver) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolu
 		return nil, status.Error(codes.InvalidArgument, "Staging target not provided")
 	}
 
+	volumeID := req.GetVolumeId()
+
+	if acquired := d.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer d.volumeLocks.Release(volumeID)
+
 	mountOptions := []string{"bind"}
 	if req.GetReadonly() {
 		mountOptions = append(mountOptions, "ro")
@@ -98,6 +105,11 @@ func (d *Driver) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublish
 	targetPath := req.GetTargetPath()
 	volumeID := req.GetVolumeId()
 
+	if acquired := d.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer d.volumeLocks.Release(volumeID)
+
 	klog.V(2).Infof("NodeUnpublishVolume: unmounting volume %s on %s", volumeID, targetPath)
 	err := CleanupMountPoint(d.mounter, targetPath, false)
 	if err != nil {
@@ -152,6 +164,11 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		}
 	}
 
+	if acquired := d.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer d.volumeLocks.Release(volumeID)
+
 	osSeparator := string(os.PathSeparator)
 	if strings.TrimSpace(server) == "" {
 		// server address is "accountname.file.core.windows.net" by default
@@ -252,6 +269,11 @@ func (d *Driver) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolu
 		return nil, status.Error(codes.InvalidArgument, "Staging target not provided")
 	}
 
+	if acquired := d.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer d.volumeLocks.Release(volumeID)
+
 	klog.V(2).Infof("NodeUnstageVolume: CleanupMountPoint volume %s on %s", volumeID, stagingTargetPath)
 	if err := CleanupSMBMountPoint(d.mounter, stagingTargetPath, false); err != nil {
 		return nil, status.Errorf(codes.Internal, "failed to unmount staging target %q: %v", stagingTargetPath, err)
diff --git a/pkg/azurefile/volume_lock.go b/pkg/azurefile/volume_lock.go
new file mode 100644
index 000000000..b244f918a
--- /dev/null
+++ b/pkg/azurefile/volume_lock.go
@@ -0,0 +1,58 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package azurefile
+
+import (
+	"sync"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+const (
+	volumeOperationAlreadyExistsFmt = "An operation with the given Volume ID %s already exists"
+)
+
+// VolumeLocks implements a map with atomic operations. It stores a set of all volume IDs
+// with an ongoing operation.
+type volumeLocks struct {
+	locks sets.String
+	mux   sync.Mutex
+}
+
+func newVolumeLocks() *volumeLocks {
+	return &volumeLocks{
+		locks: sets.NewString(),
+	}
+}
+
+// TryAcquire tries to acquire the lock for operating on volumeID and returns true if successful.
+// If another operation is already using volumeID, returns false.
+func (vl *volumeLocks) TryAcquire(volumeID string) bool {
+	vl.mux.Lock()
+	defer vl.mux.Unlock()
+	if vl.locks.Has(volumeID) {
+		return false
+	}
+	vl.locks.Insert(volumeID)
+	return true
+}
+
+func (vl *volumeLocks) Release(volumeID string) {
+	vl.mux.Lock()
+	defer vl.mux.Unlock()
+	vl.locks.Delete(volumeID)
+}


Test patch:
diff --git a/pkg/azurefile/nodeserver_test.go b/pkg/azurefile/nodeserver_test.go
index 824c33e4c..790037159 100644
--- a/pkg/azurefile/nodeserver_test.go
+++ b/pkg/azurefile/nodeserver_test.go
@@ -94,6 +94,7 @@ func TestNodeGetCapabilities(t *testing.T) {
 }
 
 func TestNodePublishVolume(t *testing.T) {
+	d := NewFakeDriver()
 	volumeCap := csi.VolumeCapability_AccessMode{Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER}
 	var (
 		errorMountSource     = testutil.GetWorkDirPath("error_mount_source", t)
@@ -106,8 +107,10 @@ func TestNodePublishVolume(t *testing.T) {
 
 	tests := []struct {
 		desc        string
+		setup       func()
 		req         csi.NodePublishVolumeRequest
 		expectedErr testutil.TestError
+		cleanup     func()
 	}{
 		{
 			desc: "[Error] Volume capabilities missing",
@@ -140,6 +143,23 @@ func TestNodePublishVolume(t *testing.T) {
 				DefaultError: status.Error(codes.InvalidArgument, "Staging target not provided"),
 			},
 		},
+		{
+			desc: "[Error] Volume operation in progress",
+			setup: func() {
+				d.volumeLocks.TryAcquire("vol_1")
+			},
+			req: csi.NodePublishVolumeRequest{VolumeCapability: &csi.VolumeCapability{AccessMode: &volumeCap},
+				VolumeId:          "vol_1",
+				TargetPath:        targetTest,
+				StagingTargetPath: sourceTest,
+				Readonly:          true},
+			expectedErr: testutil.TestError{
+				DefaultError: status.Error(codes.Aborted, fmt.Sprintf(volumeOperationAlreadyExistsFmt, "vol_1")),
+			},
+			cleanup: func() {
+				d.volumeLocks.Release("vol_1")
+			},
+		},
 		{
 			desc: "[Error] Not a directory",
 			req: csi.NodePublishVolumeRequest{VolumeCapability: &csi.VolumeCapability{AccessMode: &volumeCap},
@@ -194,7 +214,6 @@ func TestNodePublishVolume(t *testing.T) {
 
 	// Setup
 	_ = makeDir(alreadyMountedTarget)
-	d := NewFakeDriver()
 	mounter, err := NewFakeMounter()
 	if err != nil {
 		t.Fatalf(fmt.Sprintf("failed to get fake mounter: %v", err))
@@ -205,10 +224,16 @@ func TestNodePublishVolume(t *testing.T) {
 	d.mounter = mounter
 
 	for _, test := range tests {
+		if test.setup != nil {
+			test.setup()
+		}
 		_, err := d.NodePublishVolume(context.Background(), &test.req)
 		if !testutil.AssertError(err, &test.expectedErr) {
 			t.Errorf("test case: %s, \nUnexpected error: %v\nExpected error: %v", test.desc, err, test.expectedErr.GetExpectedError())
 		}
+		if test.cleanup != nil {
+			test.cleanup()
+		}
 	}
 
 	// Clean up
@@ -221,12 +246,15 @@ func TestNodePublishVolume(t *testing.T) {
 func TestNodeUnpublishVolume(t *testing.T) {
 	errorTarget := testutil.GetWorkDirPath("error_is_likely_target", t)
 	targetFile := testutil.GetWorkDirPath("abc.go", t)
+	d := NewFakeDriver()
 
 	tests := []struct {
 		desc         string
+		setup        func()
 		req          csi.NodeUnpublishVolumeRequest
 		skipOnDarwin bool
 		expectedErr  testutil.TestError
+		cleanup      func()
 	}{
 		{
 			desc: "[Error] Volume ID missing",
@@ -242,6 +270,19 @@ func TestNodeUnpublishVolume(t *testing.T) {
 				DefaultError: status.Error(codes.InvalidArgument, "Target path missing in request"),
 			},
 		},
+		{
+			desc: "[Error] Volume operation in progress",
+			setup: func() {
+				d.volumeLocks.TryAcquire("vol_1")
+			},
+			req: csi.NodeUnpublishVolumeRequest{TargetPath: targetFile, VolumeId: "vol_1"},
+			expectedErr: testutil.TestError{
+				DefaultError: status.Error(codes.Aborted, fmt.Sprintf(volumeOperationAlreadyExistsFmt, "vol_1")),
+			},
+			cleanup: func() {
+				d.volumeLocks.Release("vol_1")
+			},
+		},
 		{
 			desc:         "[Error] Unmount error mocked by IsLikelyNotMountPoint",
 			skipOnDarwin: true,
@@ -259,7 +300,6 @@ func TestNodeUnpublishVolume(t *testing.T) {
 
 	// Setup
 	_ = makeDir(errorTarget)
-	d := NewFakeDriver()
 	mounter, err := NewFakeMounter()
 	if err != nil {
 		t.Fatalf(fmt.Sprintf("failed to get fake mounter: %v", err))
@@ -270,6 +310,9 @@ func TestNodeUnpublishVolume(t *testing.T) {
 	d.mounter = mounter
 
 	for _, test := range tests {
+		if test.setup != nil {
+			test.setup()
+		}
 		if test.skipOnDarwin && runtime.GOOS == "darwin" {
 			continue
 		}
@@ -277,6 +320,9 @@ func TestNodeUnpublishVolume(t *testing.T) {
 		if !testutil.AssertError(err, &test.expectedErr) {
 			t.Errorf("test case: %s, \nUnexpected error: %v\nExpected error: %v", test.desc, err, test.expectedErr.GetExpectedError())
 		}
+		if test.cleanup != nil {
+			test.cleanup()
+		}
 	}
 
 	// Clean up
@@ -290,6 +336,7 @@ func TestNodeStageVolume(t *testing.T) {
 			Mount: &csi.VolumeCapability_MountVolume{},
 		},
 	}
+	d := NewFakeDriver()
 
 	var (
 		errorMountSensSource   = testutil.GetWorkDirPath("error_mount_sens_source", t)
@@ -338,6 +385,7 @@ func TestNodeStageVolume(t *testing.T) {
 
 	tests := []struct {
 		desc         string
+		setup        func()
 		req          csi.NodeStageVolumeRequest
 		execScripts  []ExecArgs
 		skipOnDarwin bool
@@ -349,6 +397,7 @@ func TestNodeStageVolume(t *testing.T) {
 		// that is common amongst all other flaky
 		// error messages
 		flakyWindowsErrorMessage string
+		cleanup                  func()
 	}{
 		{
 			desc:        "[Error] Volume ID missing",
@@ -380,6 +429,22 @@ func TestNodeStageVolume(t *testing.T) {
 				DefaultError: status.Error(codes.InvalidArgument, "failed to get account name from vol_1"),
 			},
 		},
+		{
+			desc: "[Error] Volume operation in progress",
+			setup: func() {
+				d.volumeLocks.TryAcquire("vol_1##")
+			},
+			req: csi.NodeStageVolumeRequest{VolumeId: "vol_1##", StagingTargetPath: sourceTest,
+				VolumeCapability: &stdVolCap,
+				VolumeContext:    volContext,
+				Secrets:          secrets},
+			expectedErr: testutil.TestError{
+				DefaultError: status.Error(codes.Aborted, fmt.Sprintf(volumeOperationAlreadyExistsFmt, "vol_1##")),
+			},
+			cleanup: func() {
+				d.volumeLocks.Release("vol_1##")
+			},
+		},
 		{
 			desc: "[Error] Not a Directory",
 			req: csi.NodeStageVolumeRequest{VolumeId: "vol_1##", StagingTargetPath: azureStagingTargetPath,
@@ -484,8 +549,10 @@ func TestNodeStageVolume(t *testing.T) {
 	}
 
 	// Setup
-	d := NewFakeDriver()
 	for _, test := range tests {
+		if test.setup != nil {
+			test.setup()
+		}
 		if test.skipOnDarwin && runtime.GOOS == "darwin" {
 			continue
 		}
@@ -522,6 +589,9 @@ func TestNodeStageVolume(t *testing.T) {
 				t.Errorf("test case: %s, \nUnexpected error: %v\nExpected error: %v", test.desc, err, test.expectedErr.GetExpectedError())
 			}
 		}
+		if test.cleanup != nil {
+			test.cleanup()
+		}
 	}
 
 	// Clean up
@@ -538,12 +608,15 @@ func TestNodeUnstageVolume(t *testing.T) {
 		errorTarget = testutil.GetWorkDirPath("error_is_likely_target", t)
 		targetFile  = testutil.GetWorkDirPath("abc.go", t)
 	)
+	d := NewFakeDriver()
 
 	tests := []struct {
 		desc         string
+		setup        func()
 		req          csi.NodeUnstageVolumeRequest
 		skipOnDarwin bool
 		expectedErr  testutil.TestError
+		cleanup      func()
 	}{
 		{
 			desc: "[Error] Volume ID missing",
@@ -559,6 +632,19 @@ func TestNodeUnstageVolume(t *testing.T) {
 				DefaultError: status.Error(codes.InvalidArgument, "Staging target not provided"),
 			},
 		},
+		{
+			desc: "[Error] Volume operation in progress",
+			setup: func() {
+				d.volumeLocks.TryAcquire("vol_1")
+			},
+			req: csi.NodeUnstageVolumeRequest{StagingTargetPath: targetFile, VolumeId: "vol_1"},
+			expectedErr: testutil.TestError{
+				DefaultError: status.Error(codes.Aborted, fmt.Sprintf(volumeOperationAlreadyExistsFmt, "vol_1")),
+			},
+			cleanup: func() {
+				d.volumeLocks.Release("vol_1")
+			},
+		},
 		{
 			desc:         "[Error] CleanupMountPoint error mocked by IsLikelyNotMountPoint",
 			req:          csi.NodeUnstageVolumeRequest{StagingTargetPath: errorTarget, VolumeId: "vol_1"},
@@ -576,7 +662,6 @@ func TestNodeUnstageVolume(t *testing.T) {
 
 	// Setup
 	_ = makeDir(errorTarget)
-	d := NewFakeDriver()
 	mounter, err := NewFakeMounter()
 	if err != nil {
 		t.Fatalf(fmt.Sprintf("failed to get fake mounter: %v", err))
@@ -587,6 +672,9 @@ func TestNodeUnstageVolume(t *testing.T) {
 	d.mounter = mounter
 
 	for _, test := range tests {
+		if test.setup != nil {
+			test.setup()
+		}
 		if test.skipOnDarwin && runtime.GOOS == "darwin" {
 			continue
 		}
@@ -594,6 +682,9 @@ func TestNodeUnstageVolume(t *testing.T) {
 		if !testutil.AssertError(err, &test.expectedErr) {
 			t.Errorf("Desc: %v\nUnexcpected error: %v\nExpected: %v", test.desc, err, test.expectedErr.GetExpectedError())
 		}
+		if test.cleanup != nil {
+			test.cleanup()
+		}
 	}
 
 	// Clean up