From e72b69f8e09a30a58d0ecea6250ea007ee3afa71 Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Mon, 26 Oct 2020 19:16:18 -0600 Subject: [PATCH] ceph: validate pools for stretch clusters Pools in stretch clusters will all use the same crush rule that is generated by the operator when configuring the cluster for stretch mode, with replica 4 and two replicas per failure domain. Pools cannot create new crush rules, so we require that replica be 4 when the pools is created, EC is not allowed, and no new crush rule will be created. Signed-off-by: Travis Nielsen --- pkg/daemon/ceph/client/mon.go | 6 +-- pkg/daemon/ceph/client/pool.go | 53 +++++++++++++-------- pkg/daemon/ceph/client/pool_test.go | 3 +- pkg/operator/ceph/cluster/mon/mon.go | 2 +- pkg/operator/ceph/file/controller.go | 2 +- pkg/operator/ceph/file/filesystem.go | 22 ++++----- pkg/operator/ceph/file/filesystem_test.go | 13 ++--- pkg/operator/ceph/object/controller.go | 18 +++---- pkg/operator/ceph/object/objectstore.go | 10 ++-- pkg/operator/ceph/object/rgw.go | 6 +-- pkg/operator/ceph/object/spec_test.go | 4 +- pkg/operator/ceph/object/zone/controller.go | 16 ++++--- pkg/operator/ceph/pool/controller.go | 12 ++--- pkg/operator/ceph/pool/controller_test.go | 5 +- pkg/operator/ceph/pool/validate.go | 18 +++++-- pkg/operator/ceph/pool/validate_test.go | 45 +++++++++-------- 16 files changed, 133 insertions(+), 102 deletions(-) diff --git a/pkg/daemon/ceph/client/mon.go b/pkg/daemon/ceph/client/mon.go index aa706e013..c9e4b8d3a 100644 --- a/pkg/daemon/ceph/client/mon.go +++ b/pkg/daemon/ceph/client/mon.go @@ -87,12 +87,12 @@ func EnableStretchElectionStrategy(context *clusterd.Context, clusterInfo *Clust } // CreateDefaultStretchCrushRule creates the default CRUSH rule for the stretch cluster -func CreateDefaultStretchCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, failureDomain, subFailureDomain string) error { +func CreateDefaultStretchCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, clusterSpec *cephv1.ClusterSpec, failureDomain string) error { pool := cephv1.PoolSpec{ FailureDomain: failureDomain, - Replicated: cephv1.ReplicatedSpec{SubFailureDomain: subFailureDomain}, + Replicated: cephv1.ReplicatedSpec{SubFailureDomain: clusterSpec.Mon.StretchCluster.SubFailureDomain}, } - if err := createTwoStepCrushRule(context, clusterInfo, defaultStretchCrushRuleName, pool); err != nil { + if err := createTwoStepCrushRule(context, clusterInfo, clusterSpec, defaultStretchCrushRuleName, pool); err != nil { return errors.Wrap(err, "failed to create default stretch crush rule") } logger.Info("successfully created the default stretch crush rule") diff --git a/pkg/daemon/ceph/client/pool.go b/pkg/daemon/ceph/client/pool.go index 42228d3ae..531535d5b 100644 --- a/pkg/daemon/ceph/client/pool.go +++ b/pkg/daemon/ceph/client/pool.go @@ -149,9 +149,9 @@ func GetPoolDetails(context *clusterd.Context, clusterInfo *ClusterInfo, name st return poolDetails, nil } -func CreatePoolWithProfile(context *clusterd.Context, clusterInfo *ClusterInfo, poolName string, pool cephv1.PoolSpec, appName string, cluster cephv1.ClusterSpec) error { +func CreatePoolWithProfile(context *clusterd.Context, clusterInfo *ClusterInfo, clusterSpec *cephv1.ClusterSpec, poolName string, pool cephv1.PoolSpec, appName string) error { if pool.IsReplicated() { - return CreateReplicatedPoolForApp(context, clusterInfo, poolName, pool, DefaultPGCount, appName, cluster) + return CreateReplicatedPoolForApp(context, clusterInfo, clusterSpec, poolName, pool, DefaultPGCount, appName) } if !pool.IsErasureCoded() { @@ -303,29 +303,40 @@ func CreateECPoolForApp(context *clusterd.Context, clusterInfo *ClusterInfo, poo return nil } -func CreateReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterInfo, poolName string, pool cephv1.PoolSpec, pgCount, appName string, cluster cephv1.ClusterSpec) error { - // Create a CRUSH rule for stretched clusters - if pool.Replicated.ReplicasPerFailureDomain != 0 { - err := createStretchedReplicationCrushRule(context, clusterInfo, poolName, pool, cluster) - if err != nil { - return errors.Wrap(err, "failed to create stretched replicated crush rule") - } +func CreateReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterInfo, clusterSpec *cephv1.ClusterSpec, poolName string, pool cephv1.PoolSpec, pgCount, appName string) error { + // The crush rule name is the same as the pool unless we have a stretch cluster. + crushRuleName := poolName + if clusterSpec.IsStretchCluster() { + // A stretch cluster enforces using the same crush rule for all pools. + // The stretch cluster rule is created initially by the operator when the stretch cluster is configured + // so there is no need to create a new crush rule for the pools here. + crushRuleName = defaultStretchCrushRuleName } else { - // create a crush rule for a replicated pool, if a failure domain is specified - if err := createReplicationCrushRule(context, clusterInfo, poolName, pool, cluster); err != nil { - return errors.Wrap(err, "failed to create replicated crush rule") + if pool.Replicated.ReplicasPerFailureDomain != 0 { + // Create a two-step CRUSH rule for pools other than stretch clusters + err := createTwoStepCrushRule(context, clusterInfo, clusterSpec, crushRuleName, pool) + if err != nil { + return errors.Wrapf(err, "failed to create two-step crush rule %q", crushRuleName) + } + } else { + // create a crush rule for a replicated pool, if a failure domain is specified + if err := createReplicationCrushRule(context, clusterInfo, clusterSpec, crushRuleName, pool); err != nil { + return errors.Wrapf(err, "failed to create replicated crush rule %q", crushRuleName) + } } } - args := []string{"osd", "pool", "create", poolName, pgCount, "replicated", poolName, "--size", strconv.FormatUint(uint64(pool.Replicated.Size), 10)} + args := []string{"osd", "pool", "create", poolName, pgCount, "replicated", crushRuleName, "--size", strconv.FormatUint(uint64(pool.Replicated.Size), 10)} output, err := NewCephCommand(context, clusterInfo, args).Run() if err != nil { return errors.Wrapf(err, "failed to create replicated pool %s. %s", poolName, string(output)) } - // the pool is type replicated, set the size for the pool now that it's been created - if err := SetPoolReplicatedSizeProperty(context, clusterInfo, poolName, strconv.FormatUint(uint64(pool.Replicated.Size), 10)); err != nil { - return errors.Wrapf(err, "failed to set size property to replicated pool %q to %d", poolName, pool.Replicated.Size) + if !clusterSpec.IsStretchCluster() { + // the pool is type replicated, set the size for the pool now that it's been created + if err := SetPoolReplicatedSizeProperty(context, clusterInfo, poolName, strconv.FormatUint(uint64(pool.Replicated.Size), 10)); err != nil { + return errors.Wrapf(err, "failed to set size property to replicated pool %q to %d", poolName, pool.Replicated.Size) + } } if err = setCommonPoolProperties(context, clusterInfo, pool, poolName, appName); err != nil { @@ -336,7 +347,7 @@ func CreateReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterI return nil } -func createStretchedReplicationCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, ruleName string, pool cephv1.PoolSpec, cluster cephv1.ClusterSpec) error { +func createTwoStepCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, clusterSpec *cephv1.ClusterSpec, ruleName string, pool cephv1.PoolSpec) error { // set the crush failure domain to the "host" if not already specified if pool.FailureDomain == "" { pool.FailureDomain = cephv1.DefaultFailureDomain @@ -347,7 +358,7 @@ func createStretchedReplicationCrushRule(context *clusterd.Context, clusterInfo } // set the crush root to the default if not already specified if pool.CrushRoot == "" { - pool.CrushRoot = GetCrushRootFromSpec(&cluster) + pool.CrushRoot = GetCrushRootFromSpec(clusterSpec) } // Get the current CRUSH map @@ -391,7 +402,7 @@ func createStretchedReplicationCrushRule(context *clusterd.Context, clusterInfo }() // Build plain text rule - plainRule := buildStretchClusterPlainCrushRule(crushMap, ruleName, pool) + plainRule := buildTwoStepPlainCrushRule(crushMap, ruleName, pool) // Append plain rule to the decompiled crush map f, err := os.OpenFile(filepath.Clean(decompiledCRUSHMapFilePath), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0400) @@ -431,7 +442,7 @@ func createStretchedReplicationCrushRule(context *clusterd.Context, clusterInfo return nil } -func createReplicationCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, ruleName string, pool cephv1.PoolSpec, cluster cephv1.ClusterSpec) error { +func createReplicationCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, clusterSpec *cephv1.ClusterSpec, ruleName string, pool cephv1.PoolSpec) error { failureDomain := pool.FailureDomain if failureDomain == "" { failureDomain = cephv1.DefaultFailureDomain @@ -439,7 +450,7 @@ func createReplicationCrushRule(context *clusterd.Context, clusterInfo *ClusterI // set the crush root to the default if not already specified crushRoot := pool.CrushRoot if pool.CrushRoot == "" { - crushRoot = GetCrushRootFromSpec(&cluster) + crushRoot = GetCrushRootFromSpec(clusterSpec) } args := []string{"osd", "crush", "rule", "create-replicated", ruleName, crushRoot, failureDomain} diff --git a/pkg/daemon/ceph/client/pool_test.go b/pkg/daemon/ceph/client/pool_test.go index 16328229b..2436872ac 100644 --- a/pkg/daemon/ceph/client/pool_test.go +++ b/pkg/daemon/ceph/client/pool_test.go @@ -172,7 +172,8 @@ func testCreateReplicaPool(t *testing.T, failureDomain, crushRoot, deviceClass, if compressionMode != "" { p.CompressionMode = compressionMode } - err := CreateReplicatedPoolForApp(context, AdminClusterInfo("mycluster"), "mypool", p, DefaultPGCount, "myapp", cephv1.ClusterSpec{Storage: rookv1.StorageScopeSpec{Config: map[string]string{CrushRootConfigKey: "cluster-crush-root"}}}) + clusterSpec := &cephv1.ClusterSpec{Storage: rookv1.StorageScopeSpec{Config: map[string]string{CrushRootConfigKey: "cluster-crush-root"}}} + err := CreateReplicatedPoolForApp(context, AdminClusterInfo("mycluster"), clusterSpec, "mypool", p, DefaultPGCount, "myapp") assert.Nil(t, err) assert.True(t, crushRuleCreated) if compressionMode != "" { diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index d8f1bfaaa..26f00c3d3 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/pkg/operator/ceph/cluster/mon/mon.go @@ -311,7 +311,7 @@ func (c *Cluster) configureStretchCluster(mons []*monConfig) error { } // Create the default crush rule for stretch clusters, that by default will also apply to all pools - if err := client.CreateDefaultStretchCrushRule(c.context, c.ClusterInfo, c.stretchFailureDomainName(), c.spec.Mon.StretchCluster.SubFailureDomain); err != nil { + if err := client.CreateDefaultStretchCrushRule(c.context, c.ClusterInfo, &c.spec, c.stretchFailureDomainName()); err != nil { return errors.Wrap(err, "failed to create default stretch rule") } diff --git a/pkg/operator/ceph/file/controller.go b/pkg/operator/ceph/file/controller.go index 3d0133b7b..a542adba2 100644 --- a/pkg/operator/ceph/file/controller.go +++ b/pkg/operator/ceph/file/controller.go @@ -218,7 +218,7 @@ func (r *ReconcileCephFilesystem) reconcile(request reconcile.Request) (reconcil } // validate the filesystem settings - if err := validateFilesystem(r.context, r.clusterInfo, cephFilesystem); err != nil { + if err := validateFilesystem(r.context, r.clusterInfo, r.cephClusterSpec, cephFilesystem); err != nil { return reconcile.Result{}, errors.Wrapf(err, "invalid object filesystem %q arguments", cephFilesystem.Name) } diff --git a/pkg/operator/ceph/file/filesystem.go b/pkg/operator/ceph/file/filesystem.go index bd1b69c05..e210cd0a0 100644 --- a/pkg/operator/ceph/file/filesystem.go +++ b/pkg/operator/ceph/file/filesystem.go @@ -57,7 +57,7 @@ func createFilesystem( ) error { if len(fs.Spec.DataPools) != 0 { f := newFS(fs.Name, fs.Namespace) - if err := f.doFilesystemCreate(context, clusterInfo, fs.Spec, *clusterSpec); err != nil { + if err := f.doFilesystemCreate(context, clusterInfo, clusterSpec, fs.Spec); err != nil { return errors.Wrapf(err, "failed to create filesystem %q", fs.Name) } } @@ -137,7 +137,7 @@ func deleteFilesystem( return nil } -func validateFilesystem(context *clusterd.Context, clusterInfo *client.ClusterInfo, f *cephv1.CephFilesystem) error { +func validateFilesystem(context *clusterd.Context, clusterInfo *client.ClusterInfo, clusterSpec *cephv1.ClusterSpec, f *cephv1.CephFilesystem) error { if f.Name == "" { return errors.New("missing name") } @@ -151,12 +151,12 @@ func validateFilesystem(context *clusterd.Context, clusterInfo *client.ClusterIn if len(f.Spec.DataPools) == 0 { return nil } - if err := pool.ValidatePoolSpec(context, clusterInfo, &f.Spec.MetadataPool); err != nil { + if err := pool.ValidatePoolSpec(context, clusterInfo, clusterSpec, &f.Spec.MetadataPool); err != nil { return errors.Wrap(err, "invalid metadata pool") } for _, p := range f.Spec.DataPools { localpoolSpec := p - if err := pool.ValidatePoolSpec(context, clusterInfo, &localpoolSpec); err != nil { + if err := pool.ValidatePoolSpec(context, clusterInfo, clusterSpec, &localpoolSpec); err != nil { return errors.Wrap(err, "Invalid data pool") } } @@ -173,10 +173,10 @@ func newFS(name, namespace string) *Filesystem { } // SetPoolSize function sets the sizes for MetadataPool and dataPool -func SetPoolSize(f *Filesystem, context *clusterd.Context, clusterInfo *client.ClusterInfo, spec cephv1.FilesystemSpec, cluster cephv1.ClusterSpec) error { +func SetPoolSize(f *Filesystem, context *clusterd.Context, clusterInfo *client.ClusterInfo, clusterSpec *cephv1.ClusterSpec, spec cephv1.FilesystemSpec) error { // generating the metadata pool's name metadataPoolName := generateMetaDataPoolName(f) - err := client.CreatePoolWithProfile(context, clusterInfo, metadataPoolName, spec.MetadataPool, "", cluster) + err := client.CreatePoolWithProfile(context, clusterInfo, clusterSpec, metadataPoolName, spec.MetadataPool, "") if err != nil { return errors.Wrapf(err, "failed to update metadata pool %q", metadataPoolName) } @@ -184,7 +184,7 @@ func SetPoolSize(f *Filesystem, context *clusterd.Context, clusterInfo *client.C dataPoolNames := generateDataPoolNames(f, spec) for i, pool := range spec.DataPools { poolName := dataPoolNames[i] - err := client.CreatePoolWithProfile(context, clusterInfo, poolName, pool, "", cluster) + err := client.CreatePoolWithProfile(context, clusterInfo, clusterSpec, poolName, pool, "") if err != nil { return errors.Wrapf(err, "failed to update datapool %q", poolName) } @@ -193,7 +193,7 @@ func SetPoolSize(f *Filesystem, context *clusterd.Context, clusterInfo *client.C } // doFilesystemCreate starts the Ceph file daemons and creates the filesystem in Ceph. -func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo *client.ClusterInfo, spec cephv1.FilesystemSpec, cluster cephv1.ClusterSpec) error { +func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo *client.ClusterInfo, clusterSpec *cephv1.ClusterSpec, spec cephv1.FilesystemSpec) error { _, err := client.GetFilesystem(context, clusterInfo, f.Name) if err == nil { @@ -208,7 +208,7 @@ func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo * fmt.Sprintf(". %v", err), ) } - if err := SetPoolSize(f, context, clusterInfo, spec, cluster); err != nil { + if err := SetPoolSize(f, context, clusterInfo, clusterSpec, spec); err != nil { return errors.Wrap(err, "failed to set pools size") } return nil @@ -242,7 +242,7 @@ func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo * metadataPoolName := generateMetaDataPoolName(f) if _, poolFound := reversedPoolMap[metadataPoolName]; !poolFound { poolsCreated = true - err = client.CreatePoolWithProfile(context, clusterInfo, metadataPoolName, spec.MetadataPool, "", cluster) + err = client.CreatePoolWithProfile(context, clusterInfo, clusterSpec, metadataPoolName, spec.MetadataPool, "") if err != nil { return errors.Wrapf(err, "failed to create metadata pool %q", metadataPoolName) } @@ -253,7 +253,7 @@ func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo * poolName := dataPoolNames[i] if _, poolFound := reversedPoolMap[poolName]; !poolFound { poolsCreated = true - err = client.CreatePoolWithProfile(context, clusterInfo, poolName, pool, "", cluster) + err = client.CreatePoolWithProfile(context, clusterInfo, clusterSpec, poolName, pool, "") if err != nil { return errors.Wrapf(err, "failed to create data pool %q", poolName) } diff --git a/pkg/operator/ceph/file/filesystem_test.go b/pkg/operator/ceph/file/filesystem_test.go index ebd2d8be7..2fd8c2c41 100644 --- a/pkg/operator/ceph/file/filesystem_test.go +++ b/pkg/operator/ceph/file/filesystem_test.go @@ -44,30 +44,31 @@ func TestValidateSpec(t *testing.T) { context := &clusterd.Context{Executor: &exectest.MockExecutor{}} clusterInfo := &client.ClusterInfo{Namespace: "ns"} fs := &cephv1.CephFilesystem{} + clusterSpec := &cephv1.ClusterSpec{} // missing name - assert.NotNil(t, validateFilesystem(context, clusterInfo, fs)) + assert.NotNil(t, validateFilesystem(context, clusterInfo, clusterSpec, fs)) fs.Name = "myfs" // missing namespace - assert.NotNil(t, validateFilesystem(context, clusterInfo, fs)) + assert.NotNil(t, validateFilesystem(context, clusterInfo, clusterSpec, fs)) fs.Namespace = "myns" // missing data pools - assert.NotNil(t, validateFilesystem(context, clusterInfo, fs)) + assert.NotNil(t, validateFilesystem(context, clusterInfo, clusterSpec, fs)) p := cephv1.PoolSpec{Replicated: cephv1.ReplicatedSpec{Size: 1, RequireSafeReplicaSize: false}} fs.Spec.DataPools = append(fs.Spec.DataPools, p) // missing metadata pool - assert.NotNil(t, validateFilesystem(context, clusterInfo, fs)) + assert.NotNil(t, validateFilesystem(context, clusterInfo, clusterSpec, fs)) fs.Spec.MetadataPool = p // missing mds count - assert.NotNil(t, validateFilesystem(context, clusterInfo, fs)) + assert.NotNil(t, validateFilesystem(context, clusterInfo, clusterSpec, fs)) fs.Spec.MetadataServer.ActiveCount = 1 // valid! - assert.Nil(t, validateFilesystem(context, clusterInfo, fs)) + assert.Nil(t, validateFilesystem(context, clusterInfo, clusterSpec, fs)) } func TestCreateFilesystem(t *testing.T) { diff --git a/pkg/operator/ceph/object/controller.go b/pkg/operator/ceph/object/controller.go index 72ae74818..9c2497b90 100644 --- a/pkg/operator/ceph/object/controller.go +++ b/pkg/operator/ceph/object/controller.go @@ -79,7 +79,7 @@ type ReconcileCephObjectStore struct { bktclient bktclient.Interface scheme *runtime.Scheme context *clusterd.Context - cephClusterSpec *cephv1.ClusterSpec + clusterSpec *cephv1.ClusterSpec clusterInfo *cephclient.ClusterInfo objectStoreChannels map[string]*objectStoreHealth } @@ -194,7 +194,7 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci return reconcileResponse, nil } - r.cephClusterSpec = &cephCluster.Spec + r.clusterSpec = &cephCluster.Spec // Initialize the channel for this object store // This allows us to track multiple ObjectStores in the same namespace @@ -240,7 +240,7 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci cfg := clusterConfig{ context: r.context, store: cephObjectStore, - clusterSpec: r.cephClusterSpec, + clusterSpec: r.clusterSpec, clusterInfo: r.clusterInfo, } err = cfg.deleteStore() @@ -289,9 +289,9 @@ func (r *ReconcileCephObjectStore) reconcileCreateObjectStore(cephObjectStore *c context: r.context, clusterInfo: r.clusterInfo, store: cephObjectStore, - rookVersion: r.cephClusterSpec.CephVersion.Image, - clusterSpec: r.cephClusterSpec, - DataPathMap: opconfig.NewStatelessDaemonDataPathMap(opconfig.RgwType, cephObjectStore.Name, cephObjectStore.Namespace, r.cephClusterSpec.DataDirHostPath), + rookVersion: r.clusterSpec.CephVersion.Image, + clusterSpec: r.clusterSpec, + DataPathMap: opconfig.NewStatelessDaemonDataPathMap(opconfig.RgwType, cephObjectStore.Name, cephObjectStore.Namespace, r.clusterSpec.DataDirHostPath), client: r.client, scheme: r.scheme, } @@ -301,7 +301,7 @@ func (r *ReconcileCephObjectStore) reconcileCreateObjectStore(cephObjectStore *c var serviceIP string var err error - if r.cephClusterSpec.External.Enable { + if r.clusterSpec.External.Enable { logger.Info("reconciling external object store") // RECONCILE SERVICE @@ -351,7 +351,7 @@ func (r *ReconcileCephObjectStore) reconcileCreateObjectStore(cephObjectStore *c // Reconcile Pool Creation if !cephObjectStore.Spec.IsMultisite() { logger.Info("reconciling object store pools") - err = CreatePools(objContext, cephObjectStore.Spec.MetadataPool, cephObjectStore.Spec.DataPool, cluster) + err = CreatePools(objContext, r.clusterSpec, cephObjectStore.Spec.MetadataPool, cephObjectStore.Spec.DataPool) if err != nil { return r.setFailedStatus(namespacedName, "failed to create object pools", err) } @@ -482,7 +482,7 @@ func (r *ReconcileCephObjectStore) startMonitoring(objectstore *cephv1.CephObjec return } - rgwChecker := newBucketChecker(r.context, objContext, serviceIP, port, r.client, namespacedName, &objectstore.Spec.HealthCheck, r.cephClusterSpec.External.Enable) + rgwChecker := newBucketChecker(r.context, objContext, serviceIP, port, r.client, namespacedName, &objectstore.Spec.HealthCheck, r.clusterSpec.External.Enable) logger.Info("starting rgw healthcheck") go rgwChecker.checkObjectStore(r.objectStoreChannels[objectstore.Name].stopChan) } diff --git a/pkg/operator/ceph/object/objectstore.go b/pkg/operator/ceph/object/objectstore.go index 04a4e6627..6127eecbc 100644 --- a/pkg/operator/ceph/object/objectstore.go +++ b/pkg/operator/ceph/object/objectstore.go @@ -605,7 +605,7 @@ func deletePools(context *Context, spec cephv1.ObjectStoreSpec, lastStore bool) return nil } -func CreatePools(context *Context, metadataPool, dataPool cephv1.PoolSpec, cluster cephv1.ClusterSpec) error { +func CreatePools(context *Context, clusterSpec *cephv1.ClusterSpec, metadataPool, dataPool cephv1.PoolSpec) error { if emptyPool(dataPool) && emptyPool(metadataPool) { logger.Info("no pools specified for the CR, checking for their existence...") pools := append(metadataPools, dataPoolName) @@ -631,7 +631,7 @@ func CreatePools(context *Context, metadataPool, dataPool cephv1.PoolSpec, clust metadataPoolPGs = ceph.DefaultPGCount } - if err := createSimilarPools(context, append(metadataPools, rootPool), metadataPool, metadataPoolPGs, "", cluster); err != nil { + if err := createSimilarPools(context, append(metadataPools, rootPool), clusterSpec, metadataPool, metadataPoolPGs, ""); err != nil { return errors.Wrap(err, "failed to create metadata pools") } @@ -644,14 +644,14 @@ func CreatePools(context *Context, metadataPool, dataPool cephv1.PoolSpec, clust } } - if err := createSimilarPools(context, []string{dataPoolName}, dataPool, ceph.DefaultPGCount, ecProfileName, cluster); err != nil { + if err := createSimilarPools(context, []string{dataPoolName}, clusterSpec, dataPool, ceph.DefaultPGCount, ecProfileName); err != nil { return errors.Wrap(err, "failed to create data pool") } return nil } -func createSimilarPools(context *Context, pools []string, poolSpec cephv1.PoolSpec, pgCount, ecProfileName string, cluster cephv1.ClusterSpec) error { +func createSimilarPools(context *Context, pools []string, clusterSpec *cephv1.ClusterSpec, poolSpec cephv1.PoolSpec, pgCount, ecProfileName string) error { for _, pool := range pools { // create the pool if it doesn't exist yet name := poolName(context.Name, pool) @@ -664,7 +664,7 @@ func createSimilarPools(context *Context, pools []string, poolSpec cephv1.PoolSp // created with that property disabled to avoid unnecessary performance impact. err = ceph.CreateECPoolForApp(context.Context, context.clusterInfo, name, ecProfileName, poolSpec, pgCount, AppName, false /* enableECOverwrite */) } else { - err = ceph.CreateReplicatedPoolForApp(context.Context, context.clusterInfo, name, poolSpec, pgCount, AppName, cluster) + err = ceph.CreateReplicatedPoolForApp(context.Context, context.clusterInfo, clusterSpec, name, poolSpec, pgCount, AppName) } if err != nil { return errors.Wrapf(err, "failed to create pool %s for object store %s.", name, context.Name) diff --git a/pkg/operator/ceph/object/rgw.go b/pkg/operator/ceph/object/rgw.go index 8046a2028..c0fe39560 100644 --- a/pkg/operator/ceph/object/rgw.go +++ b/pkg/operator/ceph/object/rgw.go @@ -344,18 +344,18 @@ func (r *ReconcileCephObjectStore) validateStore(s *cephv1.CephObjectStore) erro // Validate the pool settings, but allow for empty pools specs in case they have already been created // such as by the ceph mgr if !emptyPool(s.Spec.MetadataPool) { - if err := pool.ValidatePoolSpec(r.context, r.clusterInfo, &s.Spec.MetadataPool); err != nil { + if err := pool.ValidatePoolSpec(r.context, r.clusterInfo, r.clusterSpec, &s.Spec.MetadataPool); err != nil { return errors.Wrap(err, "invalid metadata pool spec") } } if !emptyPool(s.Spec.DataPool) { - if err := pool.ValidatePoolSpec(r.context, r.clusterInfo, &s.Spec.DataPool); err != nil { + if err := pool.ValidatePoolSpec(r.context, r.clusterInfo, r.clusterSpec, &s.Spec.DataPool); err != nil { return errors.Wrap(err, "invalid data pool spec") } } // Fail if we detected an external CephCluster CR and the list of endpoints is empty - if r.cephClusterSpec.External.Enable && r.clusterInfo.CephCred.Username != cephclient.AdminUsername { + if r.clusterSpec.External.Enable && r.clusterInfo.CephCred.Username != cephclient.AdminUsername { if len(s.Spec.Gateway.ExternalRgwEndpoints) == 0 { return errors.New("ceph cluster is external but externalRgwEndpoints list is empty") } diff --git a/pkg/operator/ceph/object/spec_test.go b/pkg/operator/ceph/object/spec_test.go index 3efe1b251..e6ca79f41 100644 --- a/pkg/operator/ceph/object/spec_test.go +++ b/pkg/operator/ceph/object/spec_test.go @@ -148,7 +148,7 @@ func TestValidateSpec(t *testing.T) { r := &ReconcileCephObjectStore{ context: context, - cephClusterSpec: &cephv1.ClusterSpec{ + clusterSpec: &cephv1.ClusterSpec{ External: cephv1.ExternalSpec{ Enable: false, }, @@ -190,7 +190,7 @@ func TestValidateSpec(t *testing.T) { assert.Nil(t, err) // external with no endpoints but ok since client.admin is used - r.cephClusterSpec.External.Enable = true + r.clusterSpec.External.Enable = true err = r.validateStore(s) assert.NoError(t, err) diff --git a/pkg/operator/ceph/object/zone/controller.go b/pkg/operator/ceph/object/zone/controller.go index 6770d4b67..340fa7e31 100644 --- a/pkg/operator/ceph/object/zone/controller.go +++ b/pkg/operator/ceph/object/zone/controller.go @@ -70,6 +70,7 @@ type ReconcileObjectZone struct { scheme *runtime.Scheme context *clusterd.Context clusterInfo *cephclient.ClusterInfo + clusterSpec *cephv1.ClusterSpec } // Add creates a new CephObjectZone Controller and adds it to the Manager. The Manager will set fields on the Controller @@ -152,6 +153,7 @@ func (r *ReconcileObjectZone) reconcile(request reconcile.Request) (reconcile.Re } return reconcileResponse, nil } + r.clusterSpec = &cephCluster.Spec // DELETE: the CR was deleted if !cephObjectZone.GetDeletionTimestamp().IsZero() { @@ -190,7 +192,7 @@ func (r *ReconcileObjectZone) reconcile(request reconcile.Request) (reconcile.Re } // Create Ceph Zone - _, err = r.createCephZone(cephObjectZone, realmName, cephCluster.Spec) + _, err = r.createCephZone(cephObjectZone, realmName) if err != nil { return r.setFailedStatus(request.NamespacedName, "failed to create ceph zone", err) } @@ -203,7 +205,7 @@ func (r *ReconcileObjectZone) reconcile(request reconcile.Request) (reconcile.Re return reconcile.Result{}, nil } -func (r *ReconcileObjectZone) createCephZone(zone *cephv1.CephObjectZone, realmName string, cluster cephv1.ClusterSpec) (reconcile.Result, error) { +func (r *ReconcileObjectZone) createCephZone(zone *cephv1.CephObjectZone, realmName string) (reconcile.Result, error) { logger.Infof("creating object zone %q in zonegroup %q in realm %q", zone.Name, zone.Spec.ZoneGroup, realmName) realmArg := fmt.Sprintf("--rgw-realm=%s", realmName) @@ -242,7 +244,7 @@ func (r *ReconcileObjectZone) createCephZone(zone *cephv1.CephObjectZone, realmN zoneIsMaster = true } - err = r.createPoolsAndZone(objContext, zone, realmName, zoneIsMaster, cluster) + err = r.createPoolsAndZone(objContext, zone, realmName, zoneIsMaster) if err != nil { return reconcile.Result{}, err } @@ -253,14 +255,14 @@ func (r *ReconcileObjectZone) createCephZone(zone *cephv1.CephObjectZone, realmN return reconcile.Result{}, nil } -func (r *ReconcileObjectZone) createPoolsAndZone(objContext *object.Context, zone *cephv1.CephObjectZone, realmName string, zoneIsMaster bool, cluster cephv1.ClusterSpec) error { +func (r *ReconcileObjectZone) createPoolsAndZone(objContext *object.Context, zone *cephv1.CephObjectZone, realmName string, zoneIsMaster bool) error { // create pools for zone logger.Debugf("creating pools ceph zone %q", zone.Name) realmArg := fmt.Sprintf("--rgw-realm=%s", realmName) zoneGroupArg := fmt.Sprintf("--rgw-zonegroup=%s", zone.Spec.ZoneGroup) zoneArg := fmt.Sprintf("--rgw-zone=%s", zone.Name) - err := object.CreatePools(objContext, zone.Spec.MetadataPool, zone.Spec.DataPool, cluster) + err := object.CreatePools(objContext, r.clusterSpec, zone.Spec.MetadataPool, zone.Spec.DataPool) if err != nil { return errors.Wrapf(err, "failed to create pools for zone %v", zone.Name) } @@ -330,10 +332,10 @@ func (r *ReconcileObjectZone) validateZoneCR(z *cephv1.CephObjectZone) error { if z.Spec.ZoneGroup == "" { return errors.New("missing zonegroup") } - if err := pool.ValidatePoolSpec(r.context, r.clusterInfo, &z.Spec.MetadataPool); err != nil { + if err := pool.ValidatePoolSpec(r.context, r.clusterInfo, r.clusterSpec, &z.Spec.MetadataPool); err != nil { return errors.Wrap(err, "invalid metadata pool spec") } - if err := pool.ValidatePoolSpec(r.context, r.clusterInfo, &z.Spec.DataPool); err != nil { + if err := pool.ValidatePoolSpec(r.context, r.clusterInfo, r.clusterSpec, &z.Spec.DataPool); err != nil { return errors.Wrap(err, "invalid data pool spec") } return nil diff --git a/pkg/operator/ceph/pool/controller.go b/pkg/operator/ceph/pool/controller.go index 3e5c4d38f..365bbbe0e 100644 --- a/pkg/operator/ceph/pool/controller.go +++ b/pkg/operator/ceph/pool/controller.go @@ -223,7 +223,7 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile } // validate the pool settings - if err := ValidatePool(r.context, clusterInfo, cephBlockPool); err != nil { + if err := ValidatePool(r.context, clusterInfo, &cephCluster.Spec, cephBlockPool); err != nil { return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "invalid pool CR %q spec", cephBlockPool.Name) } @@ -246,7 +246,7 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile } // CREATE/UPDATE - reconcileResponse, err = r.reconcileCreatePool(clusterInfo, cephBlockPool, cephCluster.Spec) + reconcileResponse, err = r.reconcileCreatePool(clusterInfo, &cephCluster.Spec, cephBlockPool) if err != nil { updateStatus(r.client, request.NamespacedName, cephv1.ConditionFailure, nil) return reconcileResponse, errors.Wrapf(err, "failed to create pool %q.", cephBlockPool.GetName()) @@ -294,8 +294,8 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile return reconcile.Result{}, nil } -func (r *ReconcileCephBlockPool) reconcileCreatePool(clusterInfo *cephclient.ClusterInfo, cephBlockPool *cephv1.CephBlockPool, cluster cephv1.ClusterSpec) (reconcile.Result, error) { - err := createPool(r.context, clusterInfo, cephBlockPool, cluster) +func (r *ReconcileCephBlockPool) reconcileCreatePool(clusterInfo *cephclient.ClusterInfo, cephCluster *cephv1.ClusterSpec, cephBlockPool *cephv1.CephBlockPool) (reconcile.Result, error) { + err := createPool(r.context, clusterInfo, cephCluster, cephBlockPool) if err != nil { return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "failed to create pool %q.", cephBlockPool.GetName()) } @@ -305,10 +305,10 @@ func (r *ReconcileCephBlockPool) reconcileCreatePool(clusterInfo *cephclient.Clu } // Create the pool -func createPool(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, p *cephv1.CephBlockPool, cluster cephv1.ClusterSpec) error { +func createPool(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, clusterSpec *cephv1.ClusterSpec, p *cephv1.CephBlockPool) error { // create the pool logger.Infof("creating pool %q in namespace %q", p.Name, p.Namespace) - if err := cephclient.CreatePoolWithProfile(context, clusterInfo, p.Name, p.Spec, poolApplicationNameRBD, cluster); err != nil { + if err := cephclient.CreatePoolWithProfile(context, clusterInfo, clusterSpec, p.Name, p.Spec, poolApplicationNameRBD); err != nil { return errors.Wrapf(err, "failed to create pool %q", p.Name) } diff --git a/pkg/operator/ceph/pool/controller_test.go b/pkg/operator/ceph/pool/controller_test.go index 57f8e35ab..30e144286 100644 --- a/pkg/operator/ceph/pool/controller_test.go +++ b/pkg/operator/ceph/pool/controller_test.go @@ -58,14 +58,15 @@ func TestCreatePool(t *testing.T) { p.Spec.Replicated.Size = 1 p.Spec.Replicated.RequireSafeReplicaSize = false - err := createPool(context, clusterInfo, p, cephv1.ClusterSpec{Storage: rookv1.StorageScopeSpec{Config: map[string]string{cephclient.CrushRootConfigKey: "cluster-crush-root"}}}) + clusterSpec := &cephv1.ClusterSpec{Storage: rookv1.StorageScopeSpec{Config: map[string]string{cephclient.CrushRootConfigKey: "cluster-crush-root"}}} + err := createPool(context, clusterInfo, clusterSpec, p) assert.Nil(t, err) // succeed with EC p.Spec.Replicated.Size = 0 p.Spec.ErasureCoded.CodingChunks = 1 p.Spec.ErasureCoded.DataChunks = 2 - err = createPool(context, clusterInfo, p, cephv1.ClusterSpec{Storage: rookv1.StorageScopeSpec{Config: map[string]string{cephclient.CrushRootConfigKey: "cluster-crush-root"}}}) + err = createPool(context, clusterInfo, clusterSpec, p) assert.Nil(t, err) } diff --git a/pkg/operator/ceph/pool/validate.go b/pkg/operator/ceph/pool/validate.go index 591fe50aa..8a46276bd 100644 --- a/pkg/operator/ceph/pool/validate.go +++ b/pkg/operator/ceph/pool/validate.go @@ -27,25 +27,37 @@ import ( ) // ValidatePool Validate the pool arguments -func ValidatePool(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, p *cephv1.CephBlockPool) error { +func ValidatePool(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, clusterSpec *cephv1.ClusterSpec, p *cephv1.CephBlockPool) error { if p.Name == "" { return errors.New("missing name") } if p.Namespace == "" { return errors.New("missing namespace") } - if err := ValidatePoolSpec(context, clusterInfo, &p.Spec); err != nil { + if err := ValidatePoolSpec(context, clusterInfo, clusterSpec, &p.Spec); err != nil { return err } return nil } // ValidatePoolSpec validates the Ceph block pool spec CR -func ValidatePoolSpec(context *clusterd.Context, clusterInfo *client.ClusterInfo, p *cephv1.PoolSpec) error { +func ValidatePoolSpec(context *clusterd.Context, clusterInfo *client.ClusterInfo, clusterSpec *cephv1.ClusterSpec, p *cephv1.PoolSpec) error { if p.IsReplicated() && p.IsErasureCoded() { return errors.New("both replication and erasure code settings cannot be specified") } + // validate pools for stretch clusters + if clusterSpec.IsStretchCluster() { + if p.IsReplicated() { + if p.Replicated.Size != 4 { + return errors.New("pools in a stretch cluster must have replication size 4") + } + } + if p.IsErasureCoded() { + return errors.New("erasure coded pools are not supported in stretch clusters") + } + } + var crush cephclient.CrushMap var err error if p.FailureDomain != "" || p.CrushRoot != "" { diff --git a/pkg/operator/ceph/pool/validate_test.go b/pkg/operator/ceph/pool/validate_test.go index 4a06ba9d8..882bcbd8e 100644 --- a/pkg/operator/ceph/pool/validate_test.go +++ b/pkg/operator/ceph/pool/validate_test.go @@ -32,20 +32,21 @@ import ( func TestValidatePool(t *testing.T) { context := &clusterd.Context{Executor: &exectest.MockExecutor{}} clusterInfo := &cephclient.ClusterInfo{Namespace: "myns"} + clusterSpec := &cephv1.ClusterSpec{} // not specifying some replication or EC settings is fine p := cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}} - err := ValidatePool(context, clusterInfo, &p) + err := ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Nil(t, err) // must specify name p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Namespace: clusterInfo.Namespace}} - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.NotNil(t, err) // must specify namespace p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool"}} - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.NotNil(t, err) // must not specify both replication and EC settings @@ -54,28 +55,28 @@ func TestValidatePool(t *testing.T) { p.Spec.Replicated.RequireSafeReplicaSize = false p.Spec.ErasureCoded.CodingChunks = 2 p.Spec.ErasureCoded.DataChunks = 3 - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.NotNil(t, err) // succeed with replication settings p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}} p.Spec.Replicated.Size = 1 p.Spec.Replicated.RequireSafeReplicaSize = false - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Nil(t, err) // size is 1 and RequireSafeReplicaSize is true p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}} p.Spec.Replicated.Size = 1 p.Spec.Replicated.RequireSafeReplicaSize = true - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Error(t, err) // succeed with ec settings p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}} p.Spec.ErasureCoded.CodingChunks = 1 p.Spec.ErasureCoded.DataChunks = 2 - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Nil(t, err) // Tests with various compression modes @@ -84,7 +85,7 @@ func TestValidatePool(t *testing.T) { p.Spec.Replicated.Size = 1 p.Spec.Replicated.RequireSafeReplicaSize = false p.Spec.CompressionMode = "none" - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Nil(t, err) // succeed with compression mode "aggressive" @@ -92,7 +93,7 @@ func TestValidatePool(t *testing.T) { p.Spec.Replicated.Size = 1 p.Spec.Replicated.RequireSafeReplicaSize = false p.Spec.CompressionMode = "aggressive" - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Nil(t, err) // fail with compression mode "unsupported" @@ -100,35 +101,35 @@ func TestValidatePool(t *testing.T) { p.Spec.Replicated.Size = 1 p.Spec.Replicated.RequireSafeReplicaSize = false p.Spec.CompressionMode = "unsupported" - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Error(t, err) // fail since replica size is lower than ReplicasPerFailureDomain p.Spec.Replicated.ReplicasPerFailureDomain = 2 - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Error(t, err) // fail since replica size is equal than ReplicasPerFailureDomain p.Spec.Replicated.Size = 2 p.Spec.Replicated.ReplicasPerFailureDomain = 2 - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Error(t, err) // fail since ReplicasPerFailureDomain is not a power of 2 p.Spec.Replicated.Size = 4 p.Spec.Replicated.ReplicasPerFailureDomain = 3 - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Error(t, err) // fail since ReplicasPerFailureDomain is not a power of 2 p.Spec.Replicated.Size = 4 p.Spec.Replicated.ReplicasPerFailureDomain = 5 - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Error(t, err) // Failure the sub domain does not exist p.Spec.Replicated.SubFailureDomain = "dummy" - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Error(t, err) // succeed with ec pool and valid compression mode @@ -136,7 +137,7 @@ func TestValidatePool(t *testing.T) { p.Spec.ErasureCoded.CodingChunks = 1 p.Spec.ErasureCoded.DataChunks = 2 p.Spec.CompressionMode = "passive" - err = ValidatePool(context, clusterInfo, &p) + err = ValidatePool(context, clusterInfo, clusterSpec, &p) assert.Nil(t, err) } @@ -159,28 +160,30 @@ func TestValidateCrushProperties(t *testing.T) { Replicated: cephv1.ReplicatedSpec{Size: 1, RequireSafeReplicaSize: false}, }, } - err := ValidatePool(context, clusterInfo, p) + clusterSpec := &cephv1.ClusterSpec{} + + err := ValidatePool(context, clusterInfo, clusterSpec, p) assert.Nil(t, err) // fail with a failure domain that doesn't exist p.Spec.FailureDomain = "doesntexist" - err = ValidatePool(context, clusterInfo, p) + err = ValidatePool(context, clusterInfo, clusterSpec, p) assert.NotNil(t, err) // fail with a crush root that doesn't exist p.Spec.FailureDomain = "osd" p.Spec.CrushRoot = "bad" - err = ValidatePool(context, clusterInfo, p) + err = ValidatePool(context, clusterInfo, clusterSpec, p) assert.NotNil(t, err) // fail with a crush root that does exist p.Spec.CrushRoot = "good" - err = ValidatePool(context, clusterInfo, p) + err = ValidatePool(context, clusterInfo, clusterSpec, p) assert.Nil(t, err) // Success replica size is 4 and replicasPerFailureDomain is 2 p.Spec.Replicated.Size = 4 p.Spec.Replicated.ReplicasPerFailureDomain = 2 - err = ValidatePool(context, clusterInfo, p) + err = ValidatePool(context, clusterInfo, clusterSpec, p) assert.NoError(t, err) }