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

Issue:
Handle concurrent operation of NodeUnstage with NodeStage and NodeUnpublish with NodePublish
**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 [...] -->

**Describe the solution you'd like in detail**
<!-- A clear and concise description of what you want to happen. -->
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 feature

**Describe alternatives you've considered**
<!-- A clear and concise description of any alternative solutions or features you've considered. -->
could refer to https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/blob/2014365ea7b3405e024be1d9552aa30d44da4a8b/pkg/gce-pd-csi-driver/node.go#L108-L110

**Additional context**
<!-- Add any other context or screenshots about the feature request here. -->


Pull request:
feat: Handle concurrency for node stage/unstage publish/unpublish
<!--  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://github.com/kubernetes/community/blob/master/contributors/devel/sig-release/release.md#issue-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://github.com/kubernetes/community/blob/master/contributors/devel/sig-testing/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 feature

<!--
Add one of the following kinds:
/kind bug
/kind test
/kind cleanup
/kind documentation
/kind feature
/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**:

**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 #656 

**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
```


Patch:
diff --git a/pkg/azuredisk/azuredisk.go b/pkg/azuredisk/azuredisk.go
index 4ec1bacda..f54972c00 100644
--- a/pkg/azuredisk/azuredisk.go
+++ b/pkg/azuredisk/azuredisk.go
@@ -37,6 +37,7 @@ import (
 
 	csicommon "sigs.k8s.io/azuredisk-csi-driver/pkg/csi-common"
 	"sigs.k8s.io/azuredisk-csi-driver/pkg/mounter"
+	volumehelper "sigs.k8s.io/azuredisk-csi-driver/pkg/util"
 	azure "sigs.k8s.io/cloud-provider-azure/pkg/provider"
 )
 
@@ -94,8 +95,9 @@ var (
 // Driver implements all interfaces of CSI drivers
 type Driver struct {
 	csicommon.CSIDriver
-	cloud   *azure.Cloud
-	mounter *mount.SafeFormatAndMount
+	cloud       *azure.Cloud
+	mounter     *mount.SafeFormatAndMount
+	volumeLocks *volumehelper.VolumeLocks
 }
 
 // NewDriver Creates a NewCSIDriver object. Assumes vendor version is equal to driver version &
@@ -105,6 +107,7 @@ func NewDriver(nodeID string) *Driver {
 	driver.Name = DriverName
 	driver.Version = driverVersion
 	driver.NodeID = nodeID
+	driver.volumeLocks = volumehelper.NewVolumeLocks()
 	return &driver
 }
 
diff --git a/pkg/azuredisk/fake_azuredisk.go b/pkg/azuredisk/fake_azuredisk.go
index eb129291a..c664500a1 100644
--- a/pkg/azuredisk/fake_azuredisk.go
+++ b/pkg/azuredisk/fake_azuredisk.go
@@ -58,6 +58,7 @@ func NewFakeDriver(t *testing.T) (*Driver, error) {
 	driver.Version = fakeDriverVersion
 	driver.NodeID = fakeNodeID
 	driver.CSIDriver = *csicommon.NewFakeCSIDriver()
+	driver.volumeLocks = volumehelper.NewVolumeLocks()
 
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
diff --git a/pkg/azuredisk/nodeserver.go b/pkg/azuredisk/nodeserver.go
index f6df8741c..b24c0dcee 100644
--- a/pkg/azuredisk/nodeserver.go
+++ b/pkg/azuredisk/nodeserver.go
@@ -42,9 +42,10 @@ import (
 )
 
 const (
-	defaultLinuxFsType      = "ext4"
-	defaultWindowsFsType    = "ntfs"
-	defaultAzureVolumeLimit = 16
+	defaultLinuxFsType              = "ext4"
+	defaultWindowsFsType            = "ntfs"
+	defaultAzureVolumeLimit         = 16
+	volumeOperationAlreadyExistsFmt = "An operation with the given Volume ID %s already exists"
 )
 
 func getDefaultFsType() string {
@@ -76,6 +77,11 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		return nil, status.Error(codes.InvalidArgument, "Volume capability not supported")
 	}
 
+	if acquired := d.volumeLocks.TryAcquire(diskURI); !acquired {
+		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, diskURI)
+	}
+	defer d.volumeLocks.Release(diskURI)
+
 	// If the access type is block, do nothing for stage
 	switch req.GetVolumeCapability().GetAccessType().(type) {
 	case *csi.VolumeCapability_Block:
@@ -161,6 +167,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: unmounting %s", stagingTargetPath)
 	err := CleanupMountPoint(stagingTargetPath, d.mounter, false)
 	if err != nil {
@@ -173,10 +184,11 @@ func (d *Driver) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolu
 
 // NodePublishVolume mount the volume from staging to target path
 func (d *Driver) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
+	volumeID := req.GetVolumeId()
 	if req.GetVolumeCapability() == nil {
 		return nil, status.Error(codes.InvalidArgument, "Volume capability missing in request")
 	}
-	if len(req.GetVolumeId()) == 0 {
+	if len(volumeID) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "Volume ID missing in request")
 	}
 
@@ -195,6 +207,11 @@ func (d *Driver) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolu
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Target path could not be prepared: %v", err))
 	}
 
+	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")
@@ -239,20 +256,27 @@ func (d *Driver) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolu
 
 // NodeUnpublishVolume unmount the volume from the target path
 func (d *Driver) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
-	if len(req.GetVolumeId()) == 0 {
+	targetPath := req.GetTargetPath()
+	volumeID := req.GetVolumeId()
+
+	if len(volumeID) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "Volume ID missing in request")
 	}
-	if len(req.GetTargetPath()) == 0 {
+	if len(targetPath) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "Target path missing in request")
 	}
-	targetPath := req.GetTargetPath()
-	volumeID := req.GetVolumeId()
 
 	klog.V(2).Infof("NodeUnpublishVolume: unmounting volume %s on %s", volumeID, targetPath)
 	err := CleanupMountPoint(targetPath, d.mounter, false)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "failed to unmount target %q: %v", targetPath, err)
 	}
+
+	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: unmount volume %s on %s successfully", volumeID, targetPath)
 
 	return &csi.NodeUnpublishVolumeResponse{}, nil
diff --git a/pkg/util/util.go b/pkg/util/util.go
index 18b2cc370..9d6df28b6 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -20,6 +20,9 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/util/sets"
 )
 
 const (
@@ -108,3 +111,30 @@ func MakeFile(pathname string) error {
 	}
 	return nil
 }
+
+type VolumeLocks struct {
+	locks sets.String
+	mux   sync.Mutex
+}
+
+func NewVolumeLocks() *VolumeLocks {
+	return &VolumeLocks{
+		locks: sets.NewString(),
+	}
+}
+
+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/azuredisk/azuredisk_test.go b/pkg/azuredisk/azuredisk_test.go
index 2d340523a..e73f89d04 100644
--- a/pkg/azuredisk/azuredisk_test.go
+++ b/pkg/azuredisk/azuredisk_test.go
@@ -31,6 +31,7 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 
+	"sigs.k8s.io/azuredisk-csi-driver/pkg/util"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/diskclient/mockdiskclient"
 	azure "sigs.k8s.io/cloud-provider-azure/pkg/provider"
 )
@@ -598,6 +599,7 @@ func TestNewDriver(t *testing.T) {
 	driver.Name = DriverName
 	driver.Version = driverVersion
 	driver.NodeID = nodeid
+	driver.volumeLocks = util.NewVolumeLocks()
 	newdriver := NewDriver(nodeid)
 	assert.Equal(t, driver, *newdriver)
 }