core: check for duplicate ceph fs pool names

Only single pool will get created if there are multiple
data pool entries with same name. This PR just adds a check
to fail if duplicate pools are present.

Signed-off-by: sp98 <[email protected]>
(cherry picked from commit 8fa8612)

pkg/operator/ceph/file/filesystem.go

@@ -144,6 +144,14 @@ func validateFilesystem(context *clusterd.Context, clusterInfo *cephclient.Clust
 	if len(f.Spec.DataPools) == 0 {
 		return nil
 	}
+
+	// Ensure duplicate pool names are not present in the spec.
+	if len(f.Spec.DataPools) > 1 {
+		if hasDuplicatePoolNames(f.Spec.DataPools) {
+			return errors.New("duplicate pool names in the data pool spec")
+		}
+	}
+
 	if err := cephpool.ValidatePoolSpec(context, clusterInfo, clusterSpec, &f.Spec.MetadataPool); err != nil {
 		return errors.Wrap(err, "invalid metadata pool")
 	}
@@ -157,6 +165,21 @@ func validateFilesystem(context *clusterd.Context, clusterInfo *cephclient.Clust
 	return nil
 }
 
+func hasDuplicatePoolNames(poolSpecList []cephv1.NamedPoolSpec) bool {
+	poolNames := make(map[string]struct{})
+	for _, poolSpec := range poolSpecList {
+		if poolSpec.Name != "" {
+			if _, has := poolNames[poolSpec.Name]; has {
+				logger.Errorf("duplicate pool name %q in the data pool spec", poolSpec.Name)
+				return true
+			}
+			poolNames[poolSpec.Name] = struct{}{}
+		}
+	}
+
+	return false
+}
+
 // newFS creates a new instance of the file (MDS) service
 func newFS(name, namespace string) *Filesystem {
 	return &Filesystem{

pkg/operator/ceph/file/filesystem_test.go

@@ -73,6 +73,26 @@ func TestValidateSpec(t *testing.T) {
 	assert.Nil(t, validateFilesystem(context, clusterInfo, clusterSpec, fs))
 }
 
+func TestHasDuplicatePoolNames(t *testing.T) {
+	// PoolSpec with no duplicates
+	fs := &cephv1.CephFilesystem{
+		Spec: cephv1.FilesystemSpec{
+			DataPools: []cephv1.NamedPoolSpec{
+				{Name: "pool1"},
+				{Name: "pool2"},
+			},
+		},
+	}
+
+	result := hasDuplicatePoolNames(fs.Spec.DataPools)
+	assert.False(t, result)
+
+	// add duplicate pool name in the spec.
+	fs.Spec.DataPools = append(fs.Spec.DataPools, cephv1.NamedPoolSpec{Name: "pool1"})
+	result = hasDuplicatePoolNames(fs.Spec.DataPools)
+	assert.True(t, result)
+}
+
 func TestGenerateDataPoolNames(t *testing.T) {
 	fs := &Filesystem{Name: "fake", Namespace: "fake"}
 	fsSpec := cephv1.FilesystemSpec{