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>
(cherry picked from commit 7f68d8b)

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 {

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())
+	})
+}