From 76e9c04b360e1db170b4a87b5bbd8dc7a150e652 Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Fri, 9 Jan 2026 15:54:24 -0700 Subject: [PATCH] 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 --- Documentation/Upgrade/ceph-upgrade.md | 6 ++ pkg/operator/ceph/csi/ceph_connection.go | 16 ++++- pkg/operator/ceph/csi/ceph_connection_test.go | 48 +++++++++++++ pkg/operator/ceph/csi/cluster_config.go | 27 +++++--- pkg/operator/ceph/csi/cluster_config_test.go | 67 ++++++++++++------- .../ceph/file/subvolumegroup/controller.go | 10 ++- .../file/subvolumegroup/controller_test.go | 14 ++++ .../ceph/pool/radosnamespace/controller.go | 10 ++- .../pool/radosnamespace/controller_test.go | 33 +++++++++ 9 files changed, 191 insertions(+), 40 deletions(-) diff --git a/Documentation/Upgrade/ceph-upgrade.md b/Documentation/Upgrade/ceph-upgrade.md index 1efa5535f..1cca31cad 100644 --- a/Documentation/Upgrade/ceph-upgrade.md +++ b/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, diff --git a/pkg/operator/ceph/csi/ceph_connection.go b/pkg/operator/ceph/csi/ceph_connection.go index 30de5cb86..be7c068cd 100644 --- a/pkg/operator/ceph/csi/ceph_connection.go +++ b/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 +} diff --git a/pkg/operator/ceph/csi/ceph_connection_test.go b/pkg/operator/ceph/csi/ceph_connection_test.go index 2255be4e4..c4ad1a2c8 100644 --- a/pkg/operator/ceph/csi/ceph_connection_test.go +++ b/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) + }) + } +} diff --git a/pkg/operator/ceph/csi/cluster_config.go b/pkg/operator/ceph/csi/cluster_config.go index b4531ff6f..855a77d05 100644 --- a/pkg/operator/ceph/csi/cluster_config.go +++ b/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") } diff --git a/pkg/operator/ceph/csi/cluster_config_test.go b/pkg/operator/ceph/csi/cluster_config_test.go index 23c78d85a..0954a970b 100644 --- a/pkg/operator/ceph/csi/cluster_config_test.go +++ b/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 diff --git a/pkg/operator/ceph/file/subvolumegroup/controller.go b/pkg/operator/ceph/file/subvolumegroup/controller.go index 0731fc9d8..4f7b00f49 100644 --- a/pkg/operator/ceph/file/subvolumegroup/controller.go +++ b/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, }, }, diff --git a/pkg/operator/ceph/file/subvolumegroup/controller_test.go b/pkg/operator/ceph/file/subvolumegroup/controller_test.go index 674ffe0a7..78524c48f 100644 --- a/pkg/operator/ceph/file/subvolumegroup/controller_test.go +++ b/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) }, diff --git a/pkg/operator/ceph/pool/radosnamespace/controller.go b/pkg/operator/ceph/pool/radosnamespace/controller.go index 7f855c9e7..e73aa81d4 100644 --- a/pkg/operator/ceph/pool/radosnamespace/controller.go +++ b/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, }, }, diff --git a/pkg/operator/ceph/pool/radosnamespace/controller_test.go b/pkg/operator/ceph/pool/radosnamespace/controller_test.go index 66049a815..946f6b197 100644 --- a/pkg/operator/ceph/pool/radosnamespace/controller_test.go +++ b/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 }, }