mirrors/ceph-csi
1
0
Fork 0
mirror of https://github.com/ceph/ceph-csi.git synced 2024-11-21 22:00:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

rbd: take lock on targetpath during node operation

Browse Source 
We should not be dependent on the CO to ensure
that it will serialize the request instead of
that we need to have own internal locks to ensure
that we dont do concurrent operations for same
request.

Signed-off-by: Madhu Rajanna <madhupr007@gmail.com>
This commit is contained in:
Madhu Rajanna 2024-11-20 10:07:38 +01:00 committed by mergify[bot]
parent 23c1d7c3e2
commit b6bd8ca71a
2 changed files with 17 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
internal/rbd/nodeserver.go
View File

@ -709,8 +709,12 @@ func (ns *NodeServer) NodePublishVolume(
volID := req.GetVolumeId() volID := req.GetVolumeId()
stagingPath += "/" + volID stagingPath += "/" + volID
// Considering kubelet make sure the stage and publish operations if acquired := ns.VolumeLocks.TryAcquire(targetPath); !acquired {
// are serialized, we dont need any extra locking in nodePublish log.ErrorLog(ctx, util.TargetPathOperationAlreadyExistsFmt, targetPath)
return nil, status.Errorf(codes.Aborted, util.TargetPathOperationAlreadyExistsFmt, targetPath)
}
defer ns.VolumeLocks.Release(targetPath)
// Check if that target path exists properly // Check if that target path exists properly
notMnt, err := ns.createTargetMountPath(ctx, targetPath, isBlock) notMnt, err := ns.createTargetMountPath(ctx, targetPath, isBlock)
@ -914,8 +918,14 @@ func (ns *NodeServer) NodeUnpublishVolume(
} }
targetPath := req.GetTargetPath() targetPath := req.GetTargetPath()
// considering kubelet make sure node operations like unpublish/unstage...etc can not be called
// at same time, an explicit locking at time of nodeunpublish is not required. if acquired := ns.VolumeLocks.TryAcquire(targetPath); !acquired {
log.ErrorLog(ctx, util.TargetPathOperationAlreadyExistsFmt, targetPath)
return nil, status.Errorf(codes.Aborted, util.TargetPathOperationAlreadyExistsFmt, targetPath)
}
defer ns.VolumeLocks.Release(targetPath)
isMnt, err := ns.Mounter.IsMountPoint(targetPath) isMnt, err := ns.Mounter.IsMountPoint(targetPath)
if err != nil { if err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {

3
internal/util/idlocker.go
View File

@ -28,6 +28,9 @@ const (
// SnapshotOperationAlreadyExistsFmt string format to return for concurrent operation. // SnapshotOperationAlreadyExistsFmt string format to return for concurrent operation.
SnapshotOperationAlreadyExistsFmt = "an operation with the given Snapshot ID %s already exists" SnapshotOperationAlreadyExistsFmt = "an operation with the given Snapshot ID %s already exists"
// TargetPathOperationAlreadyExistsFmt string format to return for concurrent operation on target path.
TargetPathOperationAlreadyExistsFmt = "an operation with the given target path %s already exists"
) )
// VolumeLocks implements a map with atomic operations. It stores a set of all volume IDs // VolumeLocks implements a map with atomic operations. It stores a set of all volume IDs