fix: reduce mount lock to avoid volumeID collision issue

Author: andyzhangx

pkg/azurefile/nodeserver.go

@@ -240,10 +240,11 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		return nil, status.Errorf(codes.InvalidArgument, "fsGroupChangePolicy(%s) is not supported, supported fsGroupChangePolicy list: %v", fsGroupChangePolicy, supportedFSGroupChangePolicyList)
 	}
 
-	if acquired := d.volumeLocks.TryAcquire(volumeID); !acquired {
+	lockKey := fmt.Sprintf("%s-%s", volumeID, targetPath)
+	if acquired := d.volumeLocks.TryAcquire(lockKey); !acquired {
 		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, volumeID)
 	}
-	defer d.volumeLocks.Release(volumeID)
+	defer d.volumeLocks.Release(lockKey)
 
 	if strings.TrimSpace(storageEndpointSuffix) == "" {
 		if d.cloud.Environment.StorageEndpointSuffix != "" {
@@ -393,10 +394,11 @@ func (d *Driver) NodeUnstageVolume(_ context.Context, req *csi.NodeUnstageVolume
 		return nil, status.Error(codes.InvalidArgument, "Staging target not provided")
 	}
 
-	if acquired := d.volumeLocks.TryAcquire(volumeID); !acquired {
+	lockKey := fmt.Sprintf("%s-%s", volumeID, stagingTargetPath)
+	if acquired := d.volumeLocks.TryAcquire(lockKey); !acquired {
 		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, volumeID)
 	}
-	defer d.volumeLocks.Release(volumeID)
+	defer d.volumeLocks.Release(lockKey)
 
 	mc := metrics.NewMetricContext(azureFileCSIDriverName, "node_unstage_volume", d.cloud.ResourceGroup, "", d.Name)
 	isOperationSucceeded := false

pkg/azurefile/nodeserver_test.go

@@ -525,7 +525,7 @@ func TestNodeStageVolume(t *testing.T) {
 		{
 			desc: "[Error] Volume operation in progress",
 			setup: func() {
-				d.volumeLocks.TryAcquire("vol_1##")
+				d.volumeLocks.TryAcquire(fmt.Sprintf("%s-%s", "vol_1##", sourceTest))
 			},
 			req: csi.NodeStageVolumeRequest{VolumeId: "vol_1##", StagingTargetPath: sourceTest,
 				VolumeCapability: &stdVolCap,
@@ -535,7 +535,7 @@ func TestNodeStageVolume(t *testing.T) {
 				DefaultError: status.Error(codes.Aborted, fmt.Sprintf(volumeOperationAlreadyExistsFmt, "vol_1##")),
 			},
 			cleanup: func() {
-				d.volumeLocks.Release("vol_1##")
+				d.volumeLocks.Release(fmt.Sprintf("%s-%s", "vol_1##", sourceTest))
 			},
 		},
 		{
@@ -784,14 +784,14 @@ func TestNodeUnstageVolume(t *testing.T) {
 		{
 			desc: "[Error] Volume operation in progress",
 			setup: func() {
-				d.volumeLocks.TryAcquire("vol_1")
+				d.volumeLocks.TryAcquire(fmt.Sprintf("%s-%s", "vol_1", targetFile))
 			},
 			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")
+				d.volumeLocks.Release(fmt.Sprintf("%s-%s", "vol_1", targetFile))
 			},
 		},
 		{