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