Merge pull request #1194 from k8s-infra-cherrypick-robot/cherry-pick-1193-to-release-1.23

[release-1.23] fix: reduce mount lock to avoid volumeID collision issue

Author: k8s-ci-robot

pkg/blob/nodeserver.go

@@ -229,10 +229,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()
@@ -437,10 +438,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)
 
 	klog.V(2).Infof("NodeUnstageVolume: volume %s unmounting on %s", volumeID, stagingTargetPath)
 	err := mount.CleanupMountPoint(stagingTargetPath, d.mounter, true /*extensiveMountPointCheck*/)

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) {