From 7f68d8be38604828177b599973dafc1a97b1b9de Mon Sep 17 00:00:00 2001
From: Travis Nielsen <tnielsen@redhat.com>
Date: Mon, 1 Apr 2024 16:36:15 -0600
Subject: [PATCH] osd: prevent osd reconcile when device set names duplicated

If device set names are duplicated, the OSDs will be very confused
when multiple OSDs attempt to be configured with the same
device set names. Updating the device set names does not help
get out of the error situation since the OSDs will also be
confused about which device set they belong to. The reconcile
must fail completely if the device set names are duplicated
to avoid this error scenario.

Signed-off-by: Travis Nielsen <tnielsen@redhat.com>
---
 pkg/operator/ceph/cluster/osd/osd.go      | 26 +++++++++++++-----
 pkg/operator/ceph/cluster/osd/osd_test.go | 33 +++++++++++++++++++++++
 2 files changed, 53 insertions(+), 6 deletions(-)

diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go
index 812cc5e4f..1c9b2b5c4 100644
--- a/pkg/operator/ceph/cluster/osd/osd.go
+++ b/pkg/operator/ceph/cluster/osd/osd.go
@@ -173,12 +173,7 @@ func (osdProps osdProperties) getPreparePlacement() cephv1.Placement {
 	return osdProps.placement
 }
 
-// Start the osd management
-func (c *Cluster) Start() error {
-	namespace := c.clusterInfo.Namespace
-	config := c.newProvisionConfig()
-	errs := newProvisionErrors()
-
+func (c *Cluster) validateOSDSettings() error {
 	// Validate pod's memory if specified
 	for resourceKey, resourceValue := range c.spec.Resources {
 		if strings.HasPrefix(resourceKey, cephv1.ResourcesKeyOSD) {
@@ -188,6 +183,25 @@ func (c *Cluster) Start() error {
 			}
 		}
 	}
+	deviceSetNames := map[string]bool{}
+	for _, deviceSet := range c.spec.Storage.StorageClassDeviceSets {
+		if deviceSetNames[deviceSet.Name] {
+			return errors.Errorf("device set %q name is duplicated, OSDs cannot be configured", deviceSet.Name)
+		}
+		deviceSetNames[deviceSet.Name] = true
+	}
+	return nil
+}
+
+// Start the osd management
+func (c *Cluster) Start() error {
+	namespace := c.clusterInfo.Namespace
+	config := c.newProvisionConfig()
+	errs := newProvisionErrors()
+
+	if err := c.validateOSDSettings(); err != nil {
+		return err
+	}
 	logger.Infof("start running osds in namespace %q", namespace)
 
 	if !c.spec.Storage.UseAllNodes && len(c.spec.Storage.Nodes) == 0 && len(c.spec.Storage.StorageClassDeviceSets) == 0 {
diff --git a/pkg/operator/ceph/cluster/osd/osd_test.go b/pkg/operator/ceph/cluster/osd/osd_test.go
index 2ee0d7567..043e41c7f 100644
--- a/pkg/operator/ceph/cluster/osd/osd_test.go
+++ b/pkg/operator/ceph/cluster/osd/osd_test.go
@@ -984,3 +984,36 @@ func TestGetOSDLocationFromArgs(t *testing.T) {
 	assert.Equal(t, osdLocaiton, "")
 	assert.Equal(t, locationFound, false)
 }
+
+func TestValidateOSDSettings(t *testing.T) {
+	namespace := "ns"
+	clusterInfo := &cephclient.ClusterInfo{
+		Namespace:   namespace,
+		CephVersion: cephver.Quincy,
+		Context:     context.TODO(),
+	}
+	clusterInfo.SetName("rook-ceph-test")
+	c := New(&clusterd.Context{}, clusterInfo, cephv1.ClusterSpec{}, "version")
+
+	t.Run("validate with no settings", func(t *testing.T) {
+		assert.NoError(t, c.validateOSDSettings())
+	})
+
+	t.Run("valid device sets", func(t *testing.T) {
+		c.spec.Storage.StorageClassDeviceSets = []cephv1.StorageClassDeviceSet{
+			{Name: "set1"},
+			{Name: "set2"},
+			{Name: "set3"},
+		}
+		assert.NoError(t, c.validateOSDSettings())
+	})
+
+	t.Run("duplicate device sets", func(t *testing.T) {
+		c.spec.Storage.StorageClassDeviceSets = []cephv1.StorageClassDeviceSet{
+			{Name: "set1"},
+			{Name: "set2"},
+			{Name: "set1"},
+		}
+		assert.Error(t, c.validateOSDSettings())
+	})
+}