feedc0de/my-rook-config
1
0
Fork 0
forked from rook/rook
Code Pull Requests Activity

Merge pull request #7567 from alrs/ceph-file-imports

Browse Source 
ceph: `pkg/operator/ceph/file`: multiple imports
This commit is contained in:
Travis Nielsen
2021-04-09 12:51:46 -06:00
committed by GitHub
parent 194bd9ab95 017590c920
commit 8b96114e80
4 changed files with 34 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/operator/ceph/file/filesystem.go
+23 -24
View File
@@ -26,7 +26,6 @@ import (
"github.com/pkg/errors"
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
"github.com/rook/rook/pkg/clusterd"
"github.com/rook/rook/pkg/daemon/ceph/client"
cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
"github.com/rook/rook/pkg/operator/ceph/file/mds"
"github.com/rook/rook/pkg/operator/ceph/pool"
@@ -59,20 +58,20 @@ func createFilesystem(
}
}
filesystem, err := client.GetFilesystem(context, clusterInfo, fs.Name)
filesystem, err := cephclient.GetFilesystem(context, clusterInfo, fs.Name)
if err != nil {
return errors.Wrapf(err, "failed to get filesystem %q", fs.Name)
}
if fs.Spec.MetadataServer.ActiveStandby {
if err = client.AllowStandbyReplay(context, clusterInfo, fs.Name, fs.Spec.MetadataServer.ActiveStandby); err != nil {
if err = cephclient.AllowStandbyReplay(context, clusterInfo, fs.Name, fs.Spec.MetadataServer.ActiveStandby); err != nil {
return errors.Wrapf(err, "failed to set allow_standby_replay to filesystem %q", fs.Name)
}
}
// set the number of active mds instances
if fs.Spec.MetadataServer.ActiveCount > 1 {
if err = client.SetNumMDSRanks(context, clusterInfo, fs.Name, fs.Spec.MetadataServer.ActiveCount); err != nil {
if err = cephclient.SetNumMDSRanks(context, clusterInfo, fs.Name, fs.Spec.MetadataServer.ActiveCount); err != nil {
logger.Warningf("failed setting active mds count to %d. %v", fs.Spec.MetadataServer.ActiveCount, err)
}
}
@@ -95,7 +94,7 @@ func deleteFilesystem(
ownerInfo *k8sutil.OwnerInfo,
dataDirHostPath string,
) error {
filesystem, err := client.GetFilesystem(context, clusterInfo, fs.Name)
filesystem, err := cephclient.GetFilesystem(context, clusterInfo, fs.Name)
if err != nil {
if code, ok := exec.ExitStatus(err); ok && code == int(syscall.ENOENT) {
// If we're deleting the filesystem anyway, ignore the error that the filesystem doesn't exist
@@ -126,14 +125,14 @@ func deleteFilesystem(
// Permanently remove the filesystem if it was created by rook and the spec does not prevent it.
if len(fs.Spec.DataPools) != 0 && !fs.Spec.PreserveFilesystemOnDelete {
if err := client.RemoveFilesystem(context, clusterInfo, fs.Name, fs.Spec.PreservePoolsOnDelete); err != nil {
if err := cephclient.RemoveFilesystem(context, clusterInfo, fs.Name, fs.Spec.PreservePoolsOnDelete); err != nil {
return errors.Wrapf(err, "failed to remove filesystem %q", fs.Name)
}
}
return nil
}
func validateFilesystem(context *clusterd.Context, clusterInfo *client.ClusterInfo, clusterSpec *cephv1.ClusterSpec, f *cephv1.CephFilesystem) error {
func validateFilesystem(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, clusterSpec *cephv1.ClusterSpec, f *cephv1.CephFilesystem) error {
if f.Name == "" {
return errors.New("missing name")
}
@@ -169,10 +168,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, clusterSpec *cephv1.ClusterSpec, spec cephv1.FilesystemSpec) error {
func SetPoolSize(f *Filesystem, context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, clusterSpec *cephv1.ClusterSpec, spec cephv1.FilesystemSpec) error {
// generating the metadata pool's name
metadataPoolName := generateMetaDataPoolName(f)
err := client.CreatePoolWithProfile(context, clusterInfo, clusterSpec, metadataPoolName, spec.MetadataPool, "")
err := cephclient.CreatePoolWithProfile(context, clusterInfo, clusterSpec, metadataPoolName, spec.MetadataPool, "")
if err != nil {
return errors.Wrapf(err, "failed to update metadata pool %q", metadataPoolName)
}
@@ -180,7 +179,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, clusterSpec, poolName, pool, "")
err := cephclient.CreatePoolWithProfile(context, clusterInfo, clusterSpec, poolName, pool, "")
if err != nil {
return errors.Wrapf(err, "failed to update datapool %q", poolName)
}
@@ -189,9 +188,9 @@ func SetPoolSize(f *Filesystem, context *clusterd.Context, clusterInfo *client.C
}
// updateFilesystem ensures that a filesystem which already exists matches the provided spec.
func (f *Filesystem) updateFilesystem(context *clusterd.Context, clusterInfo *client.ClusterInfo, clusterSpec *cephv1.ClusterSpec, spec cephv1.FilesystemSpec) error {
func (f *Filesystem) updateFilesystem(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, clusterSpec *cephv1.ClusterSpec, spec cephv1.FilesystemSpec) error {
// Even if the fs already exists, the num active mdses may have changed
if err := client.SetNumMDSRanks(context, clusterInfo, f.Name, spec.MetadataServer.ActiveCount); err != nil {
if err := cephclient.SetNumMDSRanks(context, clusterInfo, f.Name, spec.MetadataServer.ActiveCount); err != nil {
logger.Errorf(
fmt.Sprintf("failed to set num mds ranks (max_mds) to %d for filesystem %s, still continuing. ", spec.MetadataServer.ActiveCount, f.Name) +
"this error is not critical, but mdses may not be as failure tolerant as desired. " +
@@ -206,7 +205,7 @@ func (f *Filesystem) updateFilesystem(context *clusterd.Context, clusterInfo *cl
dataPoolNames := generateDataPoolNames(f, spec)
for i := range spec.DataPools {
if err := client.AddDataPoolToFilesystem(context, clusterInfo, f.Name, dataPoolNames[i]); err != nil {
if err := cephclient.AddDataPoolToFilesystem(context, clusterInfo, f.Name, dataPoolNames[i]); err != nil {
return err
}
}
@@ -214,9 +213,9 @@ func (f *Filesystem) updateFilesystem(context *clusterd.Context, clusterInfo *cl
}
// doFilesystemCreate starts the Ceph file daemons and creates the filesystem in Ceph.
func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo *client.ClusterInfo, clusterSpec *cephv1.ClusterSpec, spec cephv1.FilesystemSpec) error {
func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, clusterSpec *cephv1.ClusterSpec, spec cephv1.FilesystemSpec) error {
_, err := client.GetFilesystem(context, clusterInfo, f.Name)
_, err := cephclient.GetFilesystem(context, clusterInfo, f.Name)
if err == nil {
logger.Infof("filesystem %q already exists", f.Name)
return f.updateFilesystem(context, clusterInfo, clusterSpec, spec)
@@ -225,17 +224,17 @@ func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo *
return errors.New("at least one data pool must be specified")
}
fslist, err := client.ListFilesystems(context, clusterInfo)
fslist, err := cephclient.ListFilesystems(context, clusterInfo)
if err != nil {
return errors.Wrap(err, "failed to list existing filesystem(s)")
}
// This check prevents from concurrent CephFilesystem CRD trying to create a filesystem
// Whoever gets to create the Filesystem first wins the race, then we fail if that cluster is not Ceph Pacific and one Filesystem is present
if len(fslist) > 0 && !clusterInfo.CephVersion.IsAtLeastPacific() && !client.IsMultiFSEnabled() {
if len(fslist) > 0 && !clusterInfo.CephVersion.IsAtLeastPacific() && !cephclient.IsMultiFSEnabled() {
return errors.New("multiple filesystems are only supported as of ceph pacific")
}
poolNames, err := client.GetPoolNamesByID(context, clusterInfo)
poolNames, err := cephclient.GetPoolNamesByID(context, clusterInfo)
if err != nil {
return errors.Wrap(err, "failed to get pool names")
}
@@ -252,7 +251,7 @@ func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo *
metadataPoolName := generateMetaDataPoolName(f)
if _, poolFound := reversedPoolMap[metadataPoolName]; !poolFound {
poolsCreated = true
err = client.CreatePoolWithProfile(context, clusterInfo, clusterSpec, metadataPoolName, spec.MetadataPool, "")
err = cephclient.CreatePoolWithProfile(context, clusterInfo, clusterSpec, metadataPoolName, spec.MetadataPool, "")
if err != nil {
return errors.Wrapf(err, "failed to create metadata pool %q", metadataPoolName)
}
@@ -263,13 +262,13 @@ func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo *
poolName := dataPoolNames[i]
if _, poolFound := reversedPoolMap[poolName]; !poolFound {
poolsCreated = true
err = client.CreatePoolWithProfile(context, clusterInfo, clusterSpec, poolName, pool, "")
err = cephclient.CreatePoolWithProfile(context, clusterInfo, clusterSpec, poolName, pool, "")
if err != nil {
return errors.Wrapf(err, "failed to create data pool %q", poolName)
}
if pool.IsErasureCoded() {
// An erasure coded data pool used for a filesystem must allow overwrites
if err := client.SetPoolProperty(context, clusterInfo, poolName, "allow_ec_overwrites", "true"); err != nil {
if err := cephclient.SetPoolProperty(context, clusterInfo, poolName, "allow_ec_overwrites", "true"); err != nil {
logger.Warningf("failed to set ec pool property. %v", err)
}
}
@@ -278,7 +277,7 @@ func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo *
// create the filesystem ('fs new' needs to be forced in order to reuse pre-existing pools)
// if only one pool is created new it won't work (to avoid inconsistencies).
if err := client.CreateFilesystem(context, clusterInfo, f.Name, metadataPoolName, dataPoolNames, !poolsCreated); err != nil {
if err := cephclient.CreateFilesystem(context, clusterInfo, f.Name, metadataPoolName, dataPoolNames, !poolsCreated); err != nil {
return err
}
@@ -287,10 +286,10 @@ func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo *
}
// downFilesystem marks the filesystem as down and the MDS' as failed
func downFilesystem(context *clusterd.Context, clusterInfo *client.ClusterInfo, filesystemName string) error {
func downFilesystem(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, filesystemName string) error {
logger.Infof("downing filesystem %q", filesystemName)
if err := client.FailFilesystem(context, clusterInfo, filesystemName); err != nil {
if err := cephclient.FailFilesystem(context, clusterInfo, filesystemName); err != nil {
return err
}
logger.Infof("downed filesystem %q", filesystemName)
pkg/operator/ceph/file/filesystem_test.go
+5 -6
View File
@@ -29,7 +29,6 @@ import (
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
"github.com/rook/rook/pkg/clusterd"
"github.com/rook/rook/pkg/daemon/ceph/client"
cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
clienttest "github.com/rook/rook/pkg/daemon/ceph/client/test"
"github.com/rook/rook/pkg/operator/ceph/file/mds"
@@ -47,7 +46,7 @@ import (
func TestValidateSpec(t *testing.T) {
context := &clusterd.Context{Executor: &exectest.MockExecutor{}}
clusterInfo := &client.ClusterInfo{Namespace: "ns"}
clusterInfo := &cephclient.ClusterInfo{Namespace: "ns"}
fs := &cephv1.CephFilesystem{}
clusterSpec := &cephv1.ClusterSpec{}
@@ -216,7 +215,7 @@ func TestCreateFilesystem(t *testing.T) {
ConfigDir: configDir,
Clientset: clientset}
fs := fsTest(fsName)
clusterInfo := &client.ClusterInfo{FSID: "myfsid", CephVersion: version.Octopus}
clusterInfo := &cephclient.ClusterInfo{FSID: "myfsid", CephVersion: version.Octopus}
// start a basic cluster
ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
@@ -299,7 +298,7 @@ func TestCreateFilesystem(t *testing.T) {
assert.Equal(t, fmt.Sprintf("failed to create filesystem %q: multiple filesystems are only supported as of ceph pacific", fsName), err.Error())
// It works since the op env var is specified even if we don't run on pacific
os.Setenv(client.MultiFsEnv, "true")
os.Setenv(cephclient.MultiFsEnv, "true")
fsName = "myfs2"
fs = fsTest(fsName)
executor = fsExecutor(t, fsName, configDir, true)
@@ -311,7 +310,7 @@ func TestCreateFilesystem(t *testing.T) {
}
err = createFilesystem(context, clusterInfo, fs, &cephv1.ClusterSpec{}, ownerInfo, "/var/lib/rook/")
assert.NoError(t, err)
os.Unsetenv(client.MultiFsEnv)
os.Unsetenv(cephclient.MultiFsEnv)
// It works since the Ceph version is Pacific
fsName = "myfs3"
@@ -357,7 +356,7 @@ func TestCreateNopoolFilesystem(t *testing.T) {
},
},
}
clusterInfo := &client.ClusterInfo{FSID: "myfsid"}
clusterInfo := &cephclient.ClusterInfo{FSID: "myfsid"}
// start a basic cluster
ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
pkg/operator/ceph/file/mds/mds.go
+5 -6
View File
@@ -29,7 +29,6 @@ import (
"github.com/pkg/errors"
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
"github.com/rook/rook/pkg/clusterd"
"github.com/rook/rook/pkg/daemon/ceph/client"
cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
"github.com/rook/rook/pkg/operator/ceph/config"
@@ -73,7 +72,7 @@ func NewCluster(
context *clusterd.Context,
clusterSpec *cephv1.ClusterSpec,
fs cephv1.CephFilesystem,
fsdetails *client.CephFilesystemDetails,
fsdetails *cephclient.CephFilesystemDetails,
ownerInfo *k8sutil.OwnerInfo,
dataDirHostPath string,
) *Cluster {
@@ -219,7 +218,7 @@ func (c *Cluster) scaleDownDeployments(replicas int32, desiredDeployments map[st
logger.Infof("Deleting extraneous mds deployment %s", d.GetName())
// if the extraneous mdses are the only ones active, Ceph may experience fs downtime
// if deleting them too quickly; therefore, wait until number of active mdses is desired
if err := client.WaitForActiveRanks(c.context, c.clusterInfo, c.fs.Name,
if err := cephclient.WaitForActiveRanks(c.context, c.clusterInfo, c.fs.Name,
c.fs.Spec.MetadataServer.ActiveCount, true, fsWaitForActiveTimeout); err != nil {
errCount++
logger.Errorf(
@@ -260,7 +259,7 @@ func (c *Cluster) DeleteMdsCephObjects(mdsID string) error {
}
logger.Infof("successfully deleted mds config for %q in mon configuration database", who)
err = client.AuthDelete(c.context, c.clusterInfo, who)
err = cephclient.AuthDelete(c.context, c.clusterInfo, who)
if err != nil {
return err
}
@@ -270,12 +269,12 @@ func (c *Cluster) DeleteMdsCephObjects(mdsID string) error {
// finishedWithDaemonUpgrade performs all actions necessary to bring the filesystem back to its
// ideal state following an upgrade of its daemon(s).
func finishedWithDaemonUpgrade(context *clusterd.Context, clusterInfo *client.ClusterInfo, fsName string, activeMDSCount int32) error {
func finishedWithDaemonUpgrade(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, fsName string, activeMDSCount int32) error {
logger.Debugf("restoring filesystem %s from daemon upgrade", fsName)
logger.Debugf("bringing num active MDS daemons for fs %s back to %d", fsName, activeMDSCount)
// TODO: Unknown (Apr 2020) if this can be removed once Rook no longer supports Nautilus.
// upgrade guide according to nautilus https://docs.ceph.com/docs/nautilus/cephfs/upgrading/#upgrading-the-mds-cluster
if err := client.SetNumMDSRanks(context, clusterInfo, fsName, activeMDSCount); err != nil {
if err := cephclient.SetNumMDSRanks(context, clusterInfo, fsName, activeMDSCount); err != nil {
return errors.Wrapf(err, "Failed to restore filesystem %s following daemon upgrade", fsName)
}
return nil
pkg/operator/ceph/file/mds/spec_test.go
+1 -2
View File
@@ -26,7 +26,6 @@ import (
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
"github.com/rook/rook/pkg/clusterd"
"github.com/rook/rook/pkg/daemon/ceph/client"
cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
cephver "github.com/rook/rook/pkg/operator/ceph/version"
@@ -73,7 +72,7 @@ func testDeploymentObject(t *testing.T, network cephv1.NetworkSpec) (*apps.Deplo
Network: network,
},
fs,
&client.CephFilesystemDetails{ID: 15},
&cephclient.CephFilesystemDetails{ID: 15},
&k8sutil.OwnerInfo{},
"/var/lib/rook/",
)