core: add check for pv access modes' multinode support

During faster node loss recovery,
Rook needs to check and blocklist only rbd pvs
that do not support multinode access modes such as
ReadOnlyMany(ROX) and ReadWriteMany(RWX).
This commit adds check for the same.

Signed-off-by: Rakshith R <rar@redhat.com>
(cherry picked from commit 12e2cd550f)

pkg/operator/ceph/cluster/watcher.go

@@ -262,6 +262,10 @@ func listRBDPV(listPVs *corev1.PersistentVolumeList, cluster *cephv1.CephCluster
 	var listRbdPV []corev1.PersistentVolume
 
 	for _, pv := range listPVs.Items {
+		// Ignore PVs that support multinode access (RWX, ROX), since they can be mounted on multiple nodes.
+		if pvSupportsMultiNodeAccess(pv.Spec.AccessModes) {
+			continue
+		}
 		if pv.Spec.CSI.Driver == fmt.Sprintf("%s.rbd.csi.ceph.com", cluster.Namespace) {
 			for _, rbdVolume := range rbdVolumesInUse {
 				if pv.Spec.CSI.VolumeHandle == rbdVolume {
@@ -273,6 +277,18 @@ func listRBDPV(listPVs *corev1.PersistentVolumeList, cluster *cephv1.CephCluster
 	return listRbdPV
 }
 
+// pvSupportsMultiNodeAccess returns true if the PV access modes contain ReadWriteMany or ReadOnlyMany.
+func pvSupportsMultiNodeAccess(accessModes []corev1.PersistentVolumeAccessMode) bool {
+	for _, accessMode := range accessModes {
+		switch accessMode {
+		case corev1.ReadOnlyMany, corev1.ReadWriteMany:
+			return true
+		}
+	}
+
+	return false
+}
+
 func (c *clientCluster) fenceRbdImage(
 	ctx context.Context, node *corev1.Node, cluster *cephv1.CephCluster,
 	clusterInfo *cephclient.ClusterInfo, rbdPV corev1.PersistentVolume) error {

pkg/operator/ceph/cluster/watcher_test.go

@@ -358,3 +358,77 @@ func TestOnDeviceCMUpdate(t *testing.T) {
 	b = clientCluster.onDeviceCMUpdate(oldCM, newCM)
 	assert.True(t, b)
 }
+
+func Test_pvSupportMultiNodeAccess(t *testing.T) {
+	type args struct {
+		accessModes []corev1.PersistentVolumeAccessMode
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			name: "single access mode, contains RWX",
+			args: args{
+				accessModes: []corev1.PersistentVolumeAccessMode{
+					corev1.ReadWriteMany,
+				},
+			},
+			want: true,
+		},
+		{
+			name: "single access mode, contains ROX",
+			args: args{
+				accessModes: []corev1.PersistentVolumeAccessMode{
+					corev1.ReadOnlyMany,
+				},
+			},
+			want: true,
+		},
+		{
+			name: "single access mode, contains only RWO",
+			args: args{
+				accessModes: []corev1.PersistentVolumeAccessMode{
+					corev1.ReadWriteOnce,
+				},
+			},
+			want: false,
+		},
+		{
+			name: "single access mode, contains only RWOP",
+			args: args{
+				accessModes: []corev1.PersistentVolumeAccessMode{
+					corev1.ReadWriteOncePod,
+				},
+			},
+			want: false,
+		},
+		{
+			name: "multiple access mode, contains RWX",
+			args: args{
+				accessModes: []corev1.PersistentVolumeAccessMode{
+					corev1.ReadWriteOnce,
+					corev1.ReadWriteMany,
+				},
+			},
+			want: true,
+		},
+		{
+			name: "multiple access mode, contains RWX and ROX",
+			args: args{
+				accessModes: []corev1.PersistentVolumeAccessMode{
+					corev1.ReadWriteMany,
+					corev1.ReadOnlyMany,
+				},
+			},
+			want: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := pvSupportsMultiNodeAccess(tt.args.accessModes)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}