Merge pull request #1193 from andyzhangx/reduce-mount-lock

fix: reduce mount lock to avoid volumeID collision issue

Author: andyzhangx

pkg/blob/nodeserver.go

@@ -230,10 +230,11 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		return nil, status.Error(codes.InvalidArgument, "Volume capability not provided")
 	}
 
-	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)
 
 	mountFlags := req.GetVolumeCapability().GetMount().GetMountFlags()
 	attrib := req.GetVolumeContext()
@@ -445,10 +446,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(blobCSIDriverName, "node_unstage_volume", d.cloud.ResourceGroup, "", d.Name)
 	isOperationSucceeded := false

pkg/blob/nodeserver_test.go

@@ -435,8 +435,8 @@ func TestNodeStageVolume(t *testing.T) {
 					VolumeCapability:  &csi.VolumeCapability{AccessMode: &volumeCap},
 				}
 				d := NewFakeDriver()
-				d.volumeLocks.TryAcquire("unit-test")
-				defer d.volumeLocks.Release("unit-test")
+				d.volumeLocks.TryAcquire(fmt.Sprintf("%s-%s", "unit-test", "unit-test"))
+				defer d.volumeLocks.Release(fmt.Sprintf("%s-%s", "unit-test", "unit-test"))
 				_, err := d.NodeStageVolume(context.TODO(), req)
 				expectedErr := status.Error(codes.Aborted, fmt.Sprintf(volumeOperationAlreadyExistsFmt, "unit-test"))
 				if !reflect.DeepEqual(err, expectedErr) {
@@ -606,8 +606,8 @@ func TestNodeUnstageVolume(t *testing.T) {
 					VolumeCapability:  &csi.VolumeCapability{AccessMode: &volumeCap},
 				}
 				d := NewFakeDriver()
-				d.volumeLocks.TryAcquire("unit-test")
-				defer d.volumeLocks.Release("unit-test")
+				d.volumeLocks.TryAcquire(fmt.Sprintf("%s-%s", "unit-test", "unit-test"))
+				defer d.volumeLocks.Release(fmt.Sprintf("%s-%s", "unit-test", "unit-test"))
 				_, err := d.NodeStageVolume(context.TODO(), req)
 				expectedErr := status.Error(codes.Aborted, fmt.Sprintf(volumeOperationAlreadyExistsFmt, "unit-test"))
 				if !reflect.DeepEqual(err, expectedErr) {