csi: disable read affinity for ceph v20.2.0

Read affinity in Ceph v20.2.0 can cause corruption.
Disable this feature for this version of Ceph to
avoid hitting any corruption in this case. When the next
patch release is out for tentacle, this issue will be
resolved and when upgrading to it, read affinity would
automatically be re-enabled if needed.

Signed-off-by: Travis Nielsen <tnielsen@redhat.com>

Documentation/Upgrade/ceph-upgrade.md

@@ -28,6 +28,12 @@ The following Ceph versions are supported in this release of Rook:
 
 * Ceph Squid v19.2.0 or newer
 * Ceph Tentacle v20.2.0 or newer
+    * IMPORTANT: **There is a known data corruption issue in v20.2.0**, if the "read affinity" feature is enabled.
+    Read affinity is disabled by default, and is enabled by the CephCluster setting `csi.readAffinity.enabled: true`.
+    If you have enabled read affinity, we recommend waiting for v20.2.1 before upgrading to Ceph Tentacle v20.
+    If read affinity is enabled with v20.2.0, Rook automatically disables read affinity internally to help avoid this corruption.
+    After the upgrade, existing applications must be restarted to fully disable the read affinity in existing volumes.
+    See [this issue](https://github.com/rook/rook/issues/16839)) for more details.
 
 !!! important
     When an update is requested, the operator will check Ceph's status,

pkg/operator/ceph/csi/ceph_connection.go

@@ -25,6 +25,7 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 
 	csiopv1 "github.com/ceph/ceph-csi-operator/api/v1"
@@ -73,7 +74,7 @@ func CreateUpdateCephConnection(c client.Client, clusterInfo *cephclient.Cluster
 func generateCephConnSpec(c client.Client, clusterInfo *cephclient.ClusterInfo, clusterSpec cephv1.ClusterSpec) (csiopv1.CephConnectionSpec, error) {
 	csiClusterConnSpec := csiopv1.CephConnectionSpec{}
 
-	if clusterSpec.CSI.ReadAffinity.Enabled {
+	if ReadAffinityEnabled(clusterSpec.CSI.ReadAffinity.Enabled, clusterInfo.CephVersion) {
 		crushLabels := clusterSpec.CSI.ReadAffinity.CrushLocationLabels
 		if len(crushLabels) == 0 {
 			logger.Debug("using default crush topology labels")
@@ -100,3 +101,16 @@ func generateCephConnSpec(c client.Client, clusterInfo *cephclient.ClusterInfo,
 	csiClusterConnSpec.Monitors = MonEndpoints(clusterInfo.AllMonitors(), clusterSpec.RequireMsgr2())
 	return csiClusterConnSpec, nil
 }
+
+func ReadAffinityEnabled(desiredEnabled bool, version cephver.CephVersion) bool {
+	if !desiredEnabled {
+		return false
+	}
+	logger.Debugf("checking if read affinity is supported for Ceph version %q", version.String())
+	// Read affinity causes corruption on Ceph v20.2.0, so always disable it for that version
+	if version.Major == 20 && version.Minor == 2 && version.Extra == 0 {
+		logger.Warningf("read affinity disabled for Ceph version v20.2.0")
+		return false
+	}
+	return true
+}

pkg/operator/ceph/csi/ceph_connection_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
 	clienttest "github.com/rook/rook/pkg/daemon/ceph/client/test"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -149,3 +150,50 @@ func TestCephConnectionDefaultTopology(t *testing.T) {
 		assert.True(t, labels[defaultLabel], "expected label %q not found", defaultLabel)
 	}
 }
+
+func TestReadAffinityEnabled(t *testing.T) {
+	tests := []struct {
+		name         string
+		enabledField bool
+		cephVersion  cephver.CephVersion
+		expected     bool
+	}{
+		{
+			name:         "enabled field false, ceph version < 20.2.0",
+			enabledField: false,
+			cephVersion:  cephver.CephVersion{Major: 19, Minor: 2, Extra: 0},
+			expected:     false,
+		},
+		{
+			name:         "enabled field true, ceph version < 20.2.0",
+			enabledField: true,
+			cephVersion:  cephver.CephVersion{Major: 19, Minor: 2, Extra: 0},
+			expected:     true,
+		},
+		{
+			name:         "enabled field true, ceph version == 20.2.0",
+			enabledField: true,
+			cephVersion:  cephver.CephVersion{Major: 20, Minor: 2, Extra: 0},
+			expected:     false,
+		},
+		{
+			name:         "enabled field true, ceph version > 20.2.0",
+			enabledField: true,
+			cephVersion:  cephver.CephVersion{Major: 20, Minor: 2, Extra: 1},
+			expected:     true,
+		},
+		{
+			name:         "enabled field false, ceph version > 20.2.0",
+			enabledField: false,
+			cephVersion:  cephver.CephVersion{Major: 20, Minor: 2, Extra: 1},
+			expected:     false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := ReadAffinityEnabled(tt.enabledField, tt.cephVersion)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}

pkg/operator/ceph/csi/cluster_config.go

@@ -148,7 +148,7 @@ func updateNetNamespaceFilePath(clusterNamespace string, cc csiClusterConfig) {
 
 // updateCsiClusterConfig returns a json-formatted string containing
 // the cluster-to-mon mapping required to configure ceph csi.
-func updateCsiClusterConfig(curr, clusterID, clusterNamespace string, newCsiClusterConfigEntry *CSIClusterConfigEntry) (string, error) {
+func updateCsiClusterConfig(curr, clusterID string, clusterInfo *cephclient.ClusterInfo, newCsiClusterConfigEntry *CSIClusterConfigEntry) (string, error) {
 	var (
 		cc     csiClusterConfig
 		centry CSIClusterConfigEntry
@@ -165,12 +165,17 @@ func updateCsiClusterConfig(curr, clusterID, clusterNamespace string, newCsiClus
 	// wait for the other update to complete. Monitors and Subvolumegroup will be updated
 	// independently and won't collide.
 	if newCsiClusterConfigEntry != nil {
+		// Disable read affinity if needed for the ceph version
+		if !ReadAffinityEnabled(newCsiClusterConfigEntry.ReadAffinity.Enabled, clusterInfo.CephVersion) {
+			newCsiClusterConfigEntry.ReadAffinity = cephcsi.ReadAffinity{Enabled: false}
+		}
+
 		for i, centry := range cc {
 			// there is a bug with an unknown source where the csi config can have an empty
 			// namespace entry. detect and fix this scenario if it occurs
 			if centry.Namespace == "" && centry.ClusterID == clusterID {
-				logger.Infof("correcting CSI config map entry for cluster ID %q; empty namespace will be set to %q", clusterID, clusterNamespace)
-				centry.Namespace = clusterNamespace
+				logger.Infof("correcting CSI config map entry for cluster ID %q; empty namespace will be set to %q", clusterID, clusterInfo.Namespace)
+				centry.Namespace = clusterInfo.Namespace
 				cc[i] = centry
 			}
 
@@ -223,7 +228,7 @@ func updateCsiClusterConfig(curr, clusterID, clusterNamespace string, newCsiClus
 		// will fail with a dangling pointer
 		if newCsiClusterConfigEntry != nil {
 			centry.ClusterID = clusterID
-			centry.Namespace = clusterNamespace
+			centry.Namespace = clusterInfo.Namespace
 			centry.Monitors = newCsiClusterConfigEntry.Monitors
 			centry.RBD = newCsiClusterConfigEntry.RBD
 			centry.CephFS = newCsiClusterConfigEntry.CephFS
@@ -328,7 +333,7 @@ func SaveClusterConfig(clientset kubernetes.Interface, clusterID, clusterNamespa
 
 	if newCsiClusterConfigEntry != nil {
 		// set CSIDriverOptions
-		newCsiClusterConfigEntry.ReadAffinity.Enabled = clusterInfo.CSIDriverSpec.ReadAffinity.Enabled
+		newCsiClusterConfigEntry.ReadAffinity.Enabled = ReadAffinityEnabled(clusterInfo.CSIDriverSpec.ReadAffinity.Enabled, clusterInfo.CephVersion)
 		newCsiClusterConfigEntry.ReadAffinity.CrushLocationLabels = clusterInfo.CSIDriverSpec.ReadAffinity.CrushLocationLabels
 
 		newCsiClusterConfigEntry.CephFS.KernelMountOptions = clusterInfo.CSIDriverSpec.CephFS.KernelMountOptions
@@ -353,7 +358,7 @@ func SaveClusterConfig(clientset kubernetes.Interface, clusterID, clusterNamespa
 		currData = "[]"
 	}
 
-	newData, err := updateCsiClusterConfig(currData, clusterID, clusterNamespace, newCsiClusterConfigEntry)
+	newData, err := updateCsiClusterConfig(currData, clusterID, clusterInfo, newCsiClusterConfigEntry)
 	if err != nil {
 		return errors.Wrap(err, "failed to update csi config map data")
 	}
@@ -369,7 +374,7 @@ func SaveClusterConfig(clientset kubernetes.Interface, clusterID, clusterNamespa
 
 // updateCSIDriverOptions updates the CSI driver options, including read affinity, kernel mount options
 // and fuse mount options, for all entries belonging to the same cluster.
-func updateCSIDriverOptions(curr, clusterKey string,
+func updateCSIDriverOptions(curr string, clusterInfo *cephclient.ClusterInfo,
 	csiDriverOptions *cephv1.CSIDriverSpec,
 ) (string, error) {
 	cc, err := parseCsiClusterConfig(curr)
@@ -379,8 +384,8 @@ func updateCSIDriverOptions(curr, clusterKey string,
 
 	for i := range cc {
 		// If the clusterID belongs to the same cluster, update the entry.
-		if clusterKey == cc[i].Namespace {
-			cc[i].ReadAffinity.Enabled = csiDriverOptions.ReadAffinity.Enabled
+		if clusterInfo.Namespace == cc[i].Namespace {
+			cc[i].ReadAffinity.Enabled = ReadAffinityEnabled(csiDriverOptions.ReadAffinity.Enabled, clusterInfo.CephVersion)
 			cc[i].ReadAffinity.CrushLocationLabels = csiDriverOptions.ReadAffinity.CrushLocationLabels
 
 			cc[i].CephFS.KernelMountOptions = csiDriverOptions.CephFS.KernelMountOptions
@@ -388,7 +393,7 @@ func updateCSIDriverOptions(curr, clusterKey string,
 		}
 	}
 
-	updateNetNamespaceFilePath(clusterKey, cc)
+	updateNetNamespaceFilePath(clusterInfo.Namespace, cc)
 	return formatCsiClusterConfig(cc)
 }
 
@@ -423,7 +428,7 @@ func SaveCSIDriverOptions(clientset kubernetes.Interface, clusterNamespace strin
 		currData = "[]"
 	}
 
-	newData, err := updateCSIDriverOptions(currData, clusterNamespace, &clusterInfo.CSIDriverSpec)
+	newData, err := updateCSIDriverOptions(currData, clusterInfo, &clusterInfo.CSIDriverSpec)
 	if err != nil {
 		return errors.Wrap(err, "failed to update csi config map data")
 	}

pkg/operator/ceph/csi/cluster_config_test.go

@@ -118,9 +118,10 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 
 	var want, s string
 	var err error
+	clusterInfo := &cephclient.ClusterInfo{Namespace: "rook-ceph-1"}
 
 	t.Run("add a simple mons list", func(t *testing.T) {
-		s, err = updateCsiClusterConfig("[]", "rook-ceph-1", "rook-ceph-1", &csiClusterConfigEntry)
+		s, err = updateCsiClusterConfig("[]", "rook-ceph-1", clusterInfo, &csiClusterConfigEntry)
 		assert.NoError(t, err)
 		want = `[{"clusterID":"rook-ceph-1","monitors":["1.2.3.4:5000"],"namespace":"rook-ceph-1"}]`
 		compareJSON(t, want, s)
@@ -128,7 +129,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 
 	t.Run("add a 2nd mon to the current cluster", func(t *testing.T) {
 		csiClusterConfigEntry.Monitors = append(csiClusterConfigEntry.Monitors, "10.11.12.13:5000")
-		s, err = updateCsiClusterConfig(s, "rook-ceph-1", "rook-ceph-1", &csiClusterConfigEntry)
+		s, err = updateCsiClusterConfig(s, "rook-ceph-1", clusterInfo, &csiClusterConfigEntry)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -140,7 +141,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("add mount options to the current cluster", func(t *testing.T) {
-		configWithMountOptions, err := updateCsiClusterConfig(s, "rook-ceph-1", "rook-ceph-1", &csiClusterConfigEntryMountOptions)
+		configWithMountOptions, err := updateCsiClusterConfig(s, "rook-ceph-1", clusterInfo, &csiClusterConfigEntryMountOptions)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(configWithMountOptions)
 		assert.NoError(t, err)
@@ -151,7 +152,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("add a 2nd cluster with 3 mons", func(t *testing.T) {
-		s, err = updateCsiClusterConfig(s, "beta", "beta", &csiClusterConfigEntry2)
+		clusterInfo.Namespace = "beta"
+		s, err = updateCsiClusterConfig(s, "beta", clusterInfo, &csiClusterConfigEntry2)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -174,8 +176,9 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	t.Run("remove a mon from the 2nd cluster", func(t *testing.T) {
 		i := 2
 		// Remove last element of the slice
+		clusterInfo.Namespace = "beta"
 		csiClusterConfigEntry2.Monitors = append(csiClusterConfigEntry2.Monitors[:i], csiClusterConfigEntry2.Monitors[i+1:]...)
-		s, err = updateCsiClusterConfig(s, "beta", "beta", &csiClusterConfigEntry2)
+		s, err = updateCsiClusterConfig(s, "beta", clusterInfo, &csiClusterConfigEntry2)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -191,7 +194,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("add a 3rd cluster with subvolumegroup", func(t *testing.T) {
-		s, err = updateCsiClusterConfig(s, "baba", "baba", &csiClusterConfigEntry3)
+		clusterInfo.Namespace = "baba"
+		s, err = updateCsiClusterConfig(s, "baba", clusterInfo, &csiClusterConfigEntry3)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -206,7 +210,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("update mount options in presence of subvolumegroup", func(t *testing.T) {
-		sMntOptionUpdate, err := updateCsiClusterConfig(s, "baba", "baba", &csiClusterConfigEntryMountOptions)
+		clusterInfo.Namespace = "baba"
+		sMntOptionUpdate, err := updateCsiClusterConfig(s, "baba", clusterInfo, &csiClusterConfigEntryMountOptions)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(sMntOptionUpdate)
 		assert.NoError(t, err)
@@ -218,8 +223,9 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("add a 4th mon to the 3rd cluster and subvolumegroup is preserved", func(t *testing.T) {
+		clusterInfo.Namespace = "baba"
 		csiClusterConfigEntry3.Monitors = append(csiClusterConfigEntry3.Monitors, "10.11.12.13:5000")
-		s, err = updateCsiClusterConfig(s, "baba", "baba", &csiClusterConfigEntry3)
+		s, err = updateCsiClusterConfig(s, "baba", clusterInfo, &csiClusterConfigEntry3)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -230,7 +236,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 
 	t.Run("remove subvolumegroup", func(t *testing.T) {
 		csiClusterConfigEntry3.CephFS.SubvolumeGroup = ""
-		s, err = updateCsiClusterConfig(s, "baba", "baba", nil)
+		clusterInfo.Namespace = "baba"
+		s, err = updateCsiClusterConfig(s, "baba", clusterInfo, nil)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -245,7 +252,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 				},
 			},
 		}
-		s, err = updateCsiClusterConfig(s, "quatre", "quatre", &csiClusterConfigEntry4)
+		clusterInfo.Namespace = "quatre"
+		s, err = updateCsiClusterConfig(s, "quatre", clusterInfo, &csiClusterConfigEntry4)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -254,7 +262,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		assert.Equal(t, "my-group2", cc[2].CephFS.SubvolumeGroup, cc)
 
 		csiClusterConfigEntry4.Monitors = []string{"10.1.1.1:5000", "10.1.1.2:5000", "10.1.1.3:5000"}
-		s, err = updateCsiClusterConfig(s, "quatre", "quatre", &csiClusterConfigEntry4)
+		s, err = updateCsiClusterConfig(s, "quatre", clusterInfo, &csiClusterConfigEntry4)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -263,7 +271,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("does it return error on garbage input?", func(t *testing.T) {
-		_, err = updateCsiClusterConfig("qqq", "beta", "beta", &csiClusterConfigEntry2)
+		clusterInfo.Namespace = "beta"
+		_, err = updateCsiClusterConfig("qqq", "beta", clusterInfo, &csiClusterConfigEntry2)
 		assert.Error(t, err)
 	})
 
@@ -278,7 +287,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 				Monitors: []string{"1.2.3.4:5000"},
 			},
 		}
-		s, err := updateCsiClusterConfig("[]", clusterIDofCluster1, clusterIDofCluster1, &csiCluster1ConfigEntry)
+		clusterInfo.Namespace = clusterIDofCluster1
+		s, err := updateCsiClusterConfig("[]", clusterIDofCluster1, clusterInfo, &csiCluster1ConfigEntry)
 		assert.NoError(t, err)
 		want = `[{"clusterID":"rook-ceph","monitors":["1.2.3.4:5000"],"namespace":"rook-ceph"}]`
 		compareJSON(t, want, s)
@@ -292,7 +302,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 				},
 			},
 		}
-		s, err = updateCsiClusterConfig(s, subvolGrpNameofCluster1, clusterIDofCluster1, &subVolCsiCluster1Config)
+		s, err = updateCsiClusterConfig(s, subvolGrpNameofCluster1, clusterInfo, &subVolCsiCluster1Config)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -310,7 +320,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 				},
 			},
 		}
-		s, err = updateCsiClusterConfig(s, radosNSofCluster1, clusterIDofCluster1, &radosNsCsiCluster1Config)
+		s, err = updateCsiClusterConfig(s, radosNSofCluster1, clusterInfo, &radosNsCsiCluster1Config)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -320,7 +330,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 
 		// update mon IP's and check is it updating for all clusterID's
 		csiCluster1ConfigEntry.Monitors = append(csiCluster1ConfigEntry.Monitors, "1.2.3.10:5000")
-		s, err = updateCsiClusterConfig(s, clusterIDofCluster1, clusterIDofCluster1, &csiCluster1ConfigEntry)
+		s, err = updateCsiClusterConfig(s, clusterIDofCluster1, clusterInfo, &csiCluster1ConfigEntry)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -359,11 +369,12 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 				},
 			},
 		}
-		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterIDofCluster2, &csiCluster2ConfigEntry)
+		clusterInfo.Namespace = clusterIDofCluster2
+		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterInfo, &csiCluster2ConfigEntry)
 		assert.NoError(t, err)
-		s, err = updateCsiClusterConfig(s, subvolGrpNameofCluster2, clusterIDofCluster2, &subVolCsiCluster2Config)
+		s, err = updateCsiClusterConfig(s, subvolGrpNameofCluster2, clusterInfo, &subVolCsiCluster2Config)
 		assert.NoError(t, err)
-		s, err = updateCsiClusterConfig(s, radosNSofCluster2, clusterIDofCluster2, &radosNsCsiCluster2Config)
+		s, err = updateCsiClusterConfig(s, radosNSofCluster2, clusterInfo, &radosNsCsiCluster2Config)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -382,7 +393,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		// update mon on 2nd cluster and check is it updating for all clusterID's
 		// of 2nd cluster
 		csiCluster2ConfigEntry.Monitors = append(csiCluster2ConfigEntry.Monitors, "192.168.0.3:5000")
-		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterIDofCluster2, &csiCluster2ConfigEntry)
+		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterInfo, &csiCluster2ConfigEntry)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -404,7 +415,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		// clusterID's of 2nd cluster
 		i := 1
 		csiCluster2ConfigEntry.Monitors = append(csiCluster2ConfigEntry.Monitors[:i], csiCluster2ConfigEntry.Monitors[i+1:]...)
-		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterIDofCluster2, &csiCluster2ConfigEntry)
+		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterInfo, &csiCluster2ConfigEntry)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -423,13 +434,15 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("test multus cluster", func(t *testing.T) {
-		s, err = updateCsiClusterConfig("[]", "rook-ceph-1", "rook-ceph-1", &csiClusterConfigEntryMultus)
+		clusterInfo.Namespace = "rook-ceph-1"
+		s, err = updateCsiClusterConfig("[]", "rook-ceph-1", clusterInfo, &csiClusterConfigEntryMultus)
 		assert.NoError(t, err)
 		compareJSON(t, `[{"clusterID":"rook-ceph-1","monitors":["1.2.3.4:5000"],"rbd":{"netNamespaceFilePath":"/var/run/netns/rook-ceph-1","radosNamespace":"rook-ceph-1"},"namespace":"rook-ceph-1"}]`, s)
 	})
 
 	t.Run("test crush location labels are set", func(t *testing.T) {
-		s, err = updateCsiClusterConfig("[]", "rook-ceph-4", "rook-ceph-4", &csiClusterConfigEntry4)
+		clusterInfo.Namespace = "rook-ceph-4"
+		s, err = updateCsiClusterConfig("[]", "rook-ceph-4", clusterInfo, &csiClusterConfigEntry4)
 		assert.NoError(t, err)
 		compareJSON(t, `[{"clusterID":"rook-ceph-4","monitors":["10.1.1.1:5000"],"readAffinity": {"enabled": true, "crushLocationLabels":["kubernetes.io/hostname",
 		"topology.kubernetes.io/region","topology.kubernetes.io/zone","topology.rook.io/chassis","topology.rook.io/rack","topology.rook.io/row","topology.rook.io/pdu",
@@ -450,7 +463,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 			},
 		}
 		currentConfigWithoutNamespace := fmt.Sprintf(currentConfigFormatString, "")
-		out, err := updateCsiClusterConfig(currentConfigWithoutNamespace, idAndNs, idAndNs, csiConfigEntry)
+		clusterInfo.Namespace = idAndNs
+		out, err := updateCsiClusterConfig(currentConfigWithoutNamespace, idAndNs, clusterInfo, csiConfigEntry)
 		assert.NoError(t, err)
 		expectedOutput := fmt.Sprintf(currentConfigFormatString, idAndNs)
 		assert.Equal(t, expectedOutput, out)
@@ -473,7 +487,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 			},
 		}
 		currentConfigWithoutNamespace := fmt.Sprintf(currentConfigFormatString, cephFsNetNsFilePath, rbdNetNsFilePath, "")
-		out, err := updateCsiClusterConfig(currentConfigWithoutNamespace, idAndNs, idAndNs, csiConfigEntry)
+		clusterInfo.Namespace = idAndNs
+		out, err := updateCsiClusterConfig(currentConfigWithoutNamespace, idAndNs, clusterInfo, csiConfigEntry)
 		assert.NoError(t, err)
 		expectedOutput := fmt.Sprintf(currentConfigFormatString, "", "", idAndNs)
 		assert.Equal(t, expectedOutput, out)
@@ -700,7 +715,7 @@ func TestUpdateCSIDriverOptions(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			dataString, err := formatCsiClusterConfig(tt.args.clusterConfig)
 			assert.NoError(t, err)
-			got, err := updateCSIDriverOptions(dataString, tt.args.clusterKey, tt.args.csiDriverOptions)
+			got, err := updateCSIDriverOptions(dataString, &cephclient.ClusterInfo{Namespace: tt.args.clusterKey}, tt.args.csiDriverOptions)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("updateCSIDriverOptions() error = %v, wantErr %v", err, tt.wantErr)
 				return

pkg/operator/ceph/file/subvolumegroup/controller.go

@@ -40,6 +40,7 @@ import (
 	"github.com/coreos/pkg/capnslog"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/operator/ceph/config"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/ceph/file"
@@ -261,6 +262,13 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 		return reconcile.Result{}, nil
 	}
 
+	// Detect running Ceph version
+	runningCephVersion, err := cephclient.LeastUptodateDaemonVersion(r.context, r.clusterInfo, config.OsdType)
+	if err != nil {
+		return reconcile.Result{}, errors.Wrapf(err, "failed to retrieve current ceph %q version", config.OsdType)
+	}
+	r.clusterInfo.CephVersion = runningCephVersion
+
 	cephFilesystemSubVolumeGroupName := cephFilesystemSubVolumeGroup.Name
 	if cephFilesystemSubVolumeGroup.Spec.Name != "" {
 		cephFilesystemSubVolumeGroupName = cephFilesystemSubVolumeGroup.Spec.Name
@@ -358,7 +366,7 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) updateClusterConfig(cephFilesyst
 				FuseMountOptions:   r.clusterInfo.CSIDriverSpec.CephFS.FuseMountOptions,
 			},
 			ReadAffinity: cephcsi.ReadAffinity{
-				Enabled:             r.clusterInfo.CSIDriverSpec.ReadAffinity.Enabled,
+				Enabled:             csi.ReadAffinityEnabled(r.clusterInfo.CSIDriverSpec.ReadAffinity.Enabled, r.clusterInfo.CephVersion),
 				CrushLocationLabels: r.clusterInfo.CSIDriverSpec.ReadAffinity.CrushLocationLabels,
 			},
 		},

pkg/operator/ceph/file/subvolumegroup/controller_test.go

@@ -77,11 +77,22 @@ func TestFilesystemSubvolumeGroupController(t *testing.T) {
 		cephFilesystemSubVolumeGroup,
 	}
 
+	cephDaemonVersions := `{
+	"mon": {
+		"ceph version 20.2.1 (0000000000000000) tentacle (stable)": 3
+	},
+	"osd": {
+		"ceph version 20.2.1 (0000000000000000) tentacle (stable)": 3
+	}
+}`
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
 			if args[0] == "status" {
 				return `{"fsid":"c47cac40-9bee-4d52-823b-ccd803ba5bfe","health":{"checks":{},"status":"HEALTH_ERR"},"pgmap":{"num_pgs":100,"pgs_by_state":[{"state_name":"active+clean","count":100}]}}`, nil
 			}
+			if args[0] == "versions" {
+				return cephDaemonVersions, nil
+			}
 
 			return "", nil
 		},
@@ -203,6 +214,9 @@ func TestFilesystemSubvolumeGroupController(t *testing.T) {
 				} else if args[0] == "fs" && args[1] == "subvolumegroup" && args[2] == "pin" {
 					return "", nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 
 				return "", errors.Errorf("unknown command. %v", args)
 			},

pkg/operator/ceph/pool/radosnamespace/controller.go

@@ -28,6 +28,7 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/ceph/config"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
@@ -224,6 +225,13 @@ func (r *ReconcileCephBlockPoolRadosNamespace) reconcile(request reconcile.Reque
 	}
 	r.clusterInfo.Context = r.opManagerContext
 
+	// Detect running Ceph version
+	runningCephVersion, err := cephclient.LeastUptodateDaemonVersion(r.context, r.clusterInfo, config.OsdType)
+	if err != nil {
+		return reconcile.Result{}, radosNamespace, errors.Wrapf(err, "failed to retrieve current ceph %q version", config.OsdType)
+	}
+	r.clusterInfo.CephVersion = runningCephVersion
+
 	// DELETE: the CR was deleted
 	if !radosNamespace.GetDeletionTimestamp().IsZero() {
 		cephRNSList := &cephv1.CephBlockPoolRadosNamespaceList{}
@@ -380,7 +388,7 @@ func (r *ReconcileCephBlockPoolRadosNamespace) updateClusterConfig(cephBlockPool
 				FuseMountOptions:   r.clusterInfo.CSIDriverSpec.CephFS.FuseMountOptions,
 			},
 			ReadAffinity: cephcsi.ReadAffinity{
-				Enabled:             r.clusterInfo.CSIDriverSpec.ReadAffinity.Enabled,
+				Enabled:             csi.ReadAffinityEnabled(r.clusterInfo.CSIDriverSpec.ReadAffinity.Enabled, r.clusterInfo.CephVersion),
 				CrushLocationLabels: r.clusterInfo.CSIDriverSpec.ReadAffinity.CrushLocationLabels,
 			},
 		},

pkg/operator/ceph/pool/radosnamespace/controller_test.go

@@ -77,11 +77,23 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 		cephBlockPoolRadosNamespace,
 	}
 
+	cephDaemonVersions := `{
+	"mon": {
+		"ceph version 20.2.1 (0000000000000000) tentacle (stable)": 3
+	},
+	"osd": {
+		"ceph version 20.2.1 (0000000000000000) tentacle (stable)": 3
+	}
+}`
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("COMMAND: %s %v", command, args)
 			if args[0] == "status" {
 				return `{"fsid":"c47cac40-9bee-4d52-823b-ccd803ba5bfe","health":{"checks":{},"status":"HEALTH_ERR"},"pgmap":{"num_pgs":100,"pgs_by_state":[{"state_name":"active+clean","count":100}]}}`, nil
 			}
+			if args[0] == "versions" {
+				return cephDaemonVersions, nil
+			}
 
 			return "", nil
 		},
@@ -220,6 +232,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" {
 					return `{"mode":"disabled"}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 
 				return "", nil
 			},
@@ -341,6 +356,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" {
 					return `{"mode":""}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 
 				return "", nil
 			},
@@ -388,6 +406,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" {
 					return `{"mode":""}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 
 				return "", nil
 			},
@@ -440,6 +461,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" && args[2] == "info" {
 					return `{}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 				return "", nil
 			},
 		}
@@ -491,6 +515,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" && args[2] == "info" {
 					return `{}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 				return "", nil
 			},
 		}
@@ -544,6 +571,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" && args[2] == "info" {
 					return `{}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 				return "", nil
 			},
 		}
@@ -601,6 +631,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 					assert.Equal(t, cephBlockPool.Name+"/"+cephBlockPoolRadosNamespace.Name, args[3])
 					return `{}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 				return "", nil
 			},
 		}