util: use csi objectuuid for rados locks

CSI vol ids > 100bytes fail with error `ENAMETOOLONG`
when acquiring rados lock as ceph sets default value
for `osd_max_attr_name_len` to 100.

This patch fixes the issue by decomposing the CSI ID
and using the ObjectUUID to ensure the lock names
are always < 100 bytes ("lock." + UUID + "-mutexlock"" = 51bytes).

"lock."(5 bytes) is a prefix applied by Ceph internally for every
lock name so we can only pass lock names that are <= 95bytes.

The hash based approach is not used to keep things
debuggable.

Fixes: #5419

Signed-off-by: Niraj Yadav <niryadav@redhat.com>

Author: black-dragon74

internal/cephfs/nodeserver.go

@@ -138,9 +138,19 @@ func maybeUnlockFileEncryption(
 		return nil
 	}
 
-	// Define Mutex Lock variables
+	// Extract the ObjectUUID from the volume ID to use as the RADOS lock name.
+	// Using the full ID would exceed Ceph's default osd_max_attr_name_len i.e. 100bytes.
+	// Ceph also prefixes "lock." (5 bytes) internally to every lock name.
 	volIDStr := string(volID)
-	lockName := volIDStr + "-mutexLock"
+	var csiID util.CSIIdentifier
+	if err := csiID.DecomposeCSIID(volIDStr); err != nil {
+		log.ErrorLog(ctx, "failed to decompose volume ID %s: %v", volID, err)
+
+		return err
+	}
+	objectUUID := csiID.ObjectUUID
+	// 5(lock.) + 36(uuid) + 10(suffix) = 51 bytes < 100 bytes limit.
+	lockName := objectUUID + "-mutexLock"
 	lockDesc := "Lock for " + volIDStr
 	lockDuration := 150 * time.Second
 	// Generate a consistent lock cookie for the client using hostname and process ID
@@ -156,7 +166,7 @@ func maybeUnlockFileEncryption(
 	}
 	defer ioctx.Destroy()
 
-	lock := iolock.NewLock(ioctx, volIDStr, lockName, lockCookie, lockDesc, lockDuration)
+	lock := iolock.NewLock(ioctx, objectUUID, lockName, lockCookie, lockDesc, lockDuration)
 	err = lock.LockExclusive(ctx)
 	if err != nil {
 		log.ErrorLog(ctx, "failed to create lock for volume ID %s: %v", volID, err)

internal/rbd/encryption.go

@@ -549,10 +549,18 @@ func (rv *rbdVolume) RotateEncryptionKey(ctx context.Context) error {
 		return fmt.Errorf("failed to open ioctx, err: %w", err)
 	}
 
-	// Lock params
-	lockName := rv.VolID + "-mutexlock"
+	// Extract the ObjectUUID from the volume ID to use as the RADOS lock name.
+	// Using the full ID would exceed Ceph's default osd_max_attr_name_len i.e. 100bytes.
+	// Ceph also prefixes "lock." (5 bytes) internally to every lock name.
+	var csiID util.CSIIdentifier
+	if err = csiID.DecomposeCSIID(rv.VolID); err != nil {
+		return fmt.Errorf("failed to decompose volume ID %s: %w", rv.VolID, err)
+	}
+	objectUUID := csiID.ObjectUUID
+	// 5(lock.) + 36(uuid) + 10(suffix) = 51 bytes < 100 bytes limit.
+	lockName := objectUUID + "-mutexLock"
 	lockDesc := "Key rotation mutex lock for " + rv.VolID
-	lockCookie := rv.VolID + "-enc-key-rotate"
+	lockCookie := objectUUID + "-enc-key-rotate"
 
 	// Keep this a little more than ExecutionTimeout to have some buffer
 	// for cleanup. If this lock is a part of some gRPC call, the client
@@ -562,7 +570,7 @@ func (rv *rbdVolume) RotateEncryptionKey(ctx context.Context) error {
 	defer cancel()
 
 	// Acquire the exclusive lock based on vol id
-	lck := lock.NewLock(rv.ioctx, rv.VolID, lockName, lockCookie, lockDesc, lockDuration)
+	lck := lock.NewLock(rv.ioctx, objectUUID, lockName, lockCookie, lockDesc, lockDuration)
 	err = lck.LockExclusive(ctx)
 	if err != nil {
 		return err