From 92c1696bfc55afa29cbfac610370ea01c13ecbc2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Fri, 6 Dec 2019 17:19:10 +0100 Subject: [PATCH] ceph: cleanup/trim isUpgrade variable a bit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit We now always perform ok-to-stop change when the deployment is going to be updated. This could be for various reasons like change in the spec or a new Ceph image. In any case, if Ceph resources are going to be restarted this must be done carefully. Also trimmed the isUpgrade variable, the upcoming commit will remove the need of that variable. Keeping isUpgrade from the controller code since we need it to detect whether an orchestration upgrade should be triggered or not (based on the ceph cluster status). Also, we keep it on the child controllers so we don't go into an orchestration phase if the cluster image hasn't changed. The child does not need to be updated/go into its orchestration phase because the cluster CR changed. Closes: https://github.com/rook/rook/issues/4379 Signed-off-by: Sébastien Han --- pkg/operator/ceph/cluster/cluster.go | 12 ++--- pkg/operator/ceph/cluster/controller.go | 4 +- pkg/operator/ceph/cluster/mgr/mgr.go | 26 ++++------- pkg/operator/ceph/cluster/mgr/mgr_test.go | 1 - pkg/operator/ceph/cluster/mgr/spec_test.go | 5 -- pkg/operator/ceph/cluster/mon/health_test.go | 6 +-- pkg/operator/ceph/cluster/mon/mon.go | 27 +++++------ pkg/operator/ceph/cluster/mon/mon_test.go | 14 +++--- pkg/operator/ceph/cluster/mon/node_test.go | 16 +++---- pkg/operator/ceph/cluster/mon/spec.go | 6 +-- pkg/operator/ceph/cluster/mon/spec_test.go | 3 -- pkg/operator/ceph/cluster/osd/osd.go | 49 ++++++++------------ pkg/operator/ceph/cluster/osd/osd_test.go | 10 ++-- pkg/operator/ceph/cluster/osd/spec_test.go | 8 ++-- pkg/operator/ceph/cluster/osd/status_test.go | 2 +- pkg/operator/ceph/cluster/rbd/mirror.go | 23 ++++----- pkg/operator/ceph/cluster/rbd/mirror_test.go | 1 - pkg/operator/ceph/cluster/rbd/spec_test.go | 1 - pkg/operator/ceph/file/controller.go | 12 ++--- pkg/operator/ceph/file/filesystem.go | 3 +- pkg/operator/ceph/file/filesystem_test.go | 10 ++-- pkg/operator/ceph/file/mds/mds.go | 7 +-- pkg/operator/ceph/file/mds/spec_test.go | 1 - pkg/operator/ceph/nfs/controller.go | 5 -- pkg/operator/ceph/nfs/nfs.go | 4 +- pkg/operator/ceph/object/controller.go | 7 --- pkg/operator/ceph/object/rgw.go | 9 ++-- pkg/operator/ceph/object/rgw_test.go | 4 +- pkg/operator/k8sutil/test/deployment.go | 4 +- 29 files changed, 107 insertions(+), 173 deletions(-) diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index 94114d6ad..1d80ee8ff 100755 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/pkg/operator/ceph/cluster/cluster.go @@ -86,8 +86,7 @@ func newCluster(c *cephv1.CephCluster, context *clusterd.Context, csiMutex *sync crdName: c.Name, stopCh: make(chan struct{}), ownerRef: ownerRef, - // we set isUpgrade to false since it's a new cluster - mons: mon.New(context, c.Namespace, c.Spec.DataDirHostPath, c.Spec.Network, ownerRef, csiMutex, false), + mons: mon.New(context, c.Namespace, c.Spec.DataDirHostPath, c.Spec.Network, ownerRef, csiMutex), } } @@ -261,7 +260,7 @@ func (c *cluster) doOrchestration(rookImage string, cephVersion cephver.CephVers } else { // This gets triggered on CR update so let's not run that (mon/mgr/osd daemons) // Start the mon pods - clusterInfo, err := c.mons.Start(c.Info, rookImage, cephVersion, *c.Spec, c.isUpgrade) + clusterInfo, err := c.mons.Start(c.Info, rookImage, cephVersion, *c.Spec) if err != nil { return errors.Wrapf(err, "failed to start the mons") } @@ -282,7 +281,7 @@ func (c *cluster) doOrchestration(rookImage string, cephVersion cephver.CephVers mgrs := mgr.New(c.Info, c.context, c.Namespace, rookImage, spec.CephVersion, cephv1.GetMgrPlacement(spec.Placement), cephv1.GetMgrAnnotations(c.Spec.Annotations), spec.Network, spec.Dashboard, spec.Monitoring, spec.Mgr, cephv1.GetMgrResources(spec.Resources), - cephv1.GetMgrPriorityClassName(spec.PriorityClassNames), c.ownerRef, c.Spec.DataDirHostPath, c.isUpgrade, c.Spec.SkipUpgradeChecks) + cephv1.GetMgrPriorityClassName(spec.PriorityClassNames), c.ownerRef, c.Spec.DataDirHostPath, c.Spec.SkipUpgradeChecks) err = mgrs.Start() if err != nil { return errors.Wrapf(err, "failed to start the ceph mgr") @@ -291,8 +290,7 @@ func (c *cluster) doOrchestration(rookImage string, cephVersion cephver.CephVers // Start the OSDs osds := osd.New(c.Info, c.context, c.Namespace, rookImage, spec.CephVersion, spec.Storage, spec.DataDirHostPath, cephv1.GetOSDPlacement(spec.Placement), cephv1.GetOSDAnnotations(spec.Annotations), spec.Network, - cephv1.GetOSDResources(spec.Resources), cephv1.GetPrepareOSDResources(spec.Resources), - cephv1.GetOSDPriorityClassName(spec.PriorityClassNames), c.ownerRef, c.isUpgrade, c.Spec.SkipUpgradeChecks, c.Spec.ContinueUpgradeAfterChecksEvenIfNotHealthy) + cephv1.GetOSDResources(spec.Resources), cephv1.GetPrepareOSDResources(spec.Resources), cephv1.GetOSDPriorityClassName(spec.PriorityClassNames), c.ownerRef, c.Spec.SkipUpgradeChecks, c.Spec.ContinueUpgradeAfterChecksEvenIfNotHealthy) err = osds.Start() if err != nil { return errors.Wrapf(err, "failed to start the osds") @@ -302,7 +300,7 @@ func (c *cluster) doOrchestration(rookImage string, cephVersion cephver.CephVers rbdmirror := rbd.New(c.Info, c.context, c.Namespace, rookImage, spec.CephVersion, cephv1.GetRBDMirrorPlacement(spec.Placement), cephv1.GetRBDMirrorAnnotations(spec.Annotations), spec.Network, spec.RBDMirroring, cephv1.GetRBDMirrorResources(spec.Resources), cephv1.GetRBDMirrorPriorityClassName(spec.PriorityClassNames), - c.ownerRef, c.Spec.DataDirHostPath, c.isUpgrade, c.Spec.SkipUpgradeChecks) + c.ownerRef, c.Spec.DataDirHostPath, c.Spec.SkipUpgradeChecks) err = rbdmirror.Start() if err != nil { return errors.Wrapf(err, "failed to start the rbd mirrors") diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go index b53793b3a..51f40c102 100644 --- a/pkg/operator/ceph/cluster/controller.go +++ b/pkg/operator/ceph/cluster/controller.go @@ -429,7 +429,7 @@ func (c *ClusterController) initializeCluster(cluster *cluster, clusterObj *ceph poolController.StartWatch(cluster.Namespace, cluster.stopCh) // Start object store CRD watcher - objectStoreController := object.NewObjectStoreController(cluster.Info, c.context, cluster.Namespace, c.rookImage, cluster.Spec, cluster.ownerRef, cluster.Spec.DataDirHostPath, cluster.isUpgrade) + objectStoreController := object.NewObjectStoreController(cluster.Info, c.context, cluster.Namespace, c.rookImage, cluster.Spec, cluster.ownerRef, cluster.Spec.DataDirHostPath) objectStoreController.StartWatch(cluster.Namespace, cluster.stopCh) // Start object store user CRD watcher @@ -444,7 +444,7 @@ func (c *ClusterController) initializeCluster(cluster *cluster, clusterObj *ceph go bucketController.Run(cluster.stopCh) // Start file system CRD watcher - fileController := file.NewFilesystemController(cluster.Info, c.context, cluster.Namespace, c.rookImage, cluster.Spec, cluster.ownerRef, cluster.Spec.DataDirHostPath, cluster.isUpgrade) + fileController := file.NewFilesystemController(cluster.Info, c.context, cluster.Namespace, c.rookImage, cluster.Spec, cluster.ownerRef, cluster.Spec.DataDirHostPath) fileController.StartWatch(cluster.Namespace, cluster.stopCh) // Start nfs ganesha CRD watcher diff --git a/pkg/operator/ceph/cluster/mgr/mgr.go b/pkg/operator/ceph/cluster/mgr/mgr.go index 8b0f1e2ed..67067ef4a 100644 --- a/pkg/operator/ceph/cluster/mgr/mgr.go +++ b/pkg/operator/ceph/cluster/mgr/mgr.go @@ -79,7 +79,6 @@ type Cluster struct { rookVersion string exitCode func(err error) (int, bool) dataDirHostPath string - isUpgrade bool skipUpgradeChecks bool appliedHttpBind bool } @@ -100,7 +99,6 @@ func New( priorityClassName string, ownerRef metav1.OwnerReference, dataDirHostPath string, - isUpgrade bool, skipUpgradeChecks bool, ) *Cluster { return &Cluster{ @@ -122,7 +120,6 @@ func New( ownerRef: ownerRef, exitCode: getExitCode, dataDirHostPath: dataDirHostPath, - isUpgrade: isUpgrade, skipUpgradeChecks: skipUpgradeChecks, } } @@ -179,21 +176,18 @@ func (c *Cluster) Start() error { daemon := string(config.MgrType) var cephVersionToUse cephver.CephVersion - // If this is not a Ceph upgrade there is no need to check the ceph version - if c.isUpgrade { - currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon) - if err != nil { - logger.Warningf("failed to retrieve current ceph %q version. %v", daemon, err) - logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion") - cephVersionToUse = c.clusterInfo.CephVersion - } else { - logger.Debugf("current cluster version for mgrs before upgrading is: %+v", currentCephVersion) - cephVersionToUse = currentCephVersion - } + currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon) + if err != nil { + logger.Warningf("failed to retrieve current ceph %q version. %+v", daemon, err) + logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with %+v", c.clusterInfo.CephVersion) + cephVersionToUse = c.clusterInfo.CephVersion + } else { + logger.Debugf("current cluster version for mgrs before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion } - if err := updateDeploymentAndWait(c.context, d, c.Namespace, daemon, mgrConfig.DaemonID, cephVersionToUse, c.isUpgrade, c.skipUpgradeChecks, false); err != nil { - logger.Errorf("failed to update mgr deployment %q. %v", resourceName, err) + if err := updateDeploymentAndWait(c.context, d, c.Namespace, daemon, mgrConfig.DaemonID, cephVersionToUse, c.skipUpgradeChecks, false); err != nil { + logger.Errorf("failed to update mgr deployment %q. %+v", resourceName, err) } } if existingDeployment, err := c.context.Clientset.AppsV1().Deployments(c.Namespace).Get(d.GetName(), metav1.GetOptions{}); err != nil { diff --git a/pkg/operator/ceph/cluster/mgr/mgr_test.go b/pkg/operator/ceph/cluster/mgr/mgr_test.go index 62e7b2877..22ea1cab0 100755 --- a/pkg/operator/ceph/cluster/mgr/mgr_test.go +++ b/pkg/operator/ceph/cluster/mgr/mgr_test.go @@ -72,7 +72,6 @@ func TestStartMGR(t *testing.T) { metav1.OwnerReference{}, "/var/lib/rook/", false, - false, ) defer os.RemoveAll(c.dataDir) diff --git a/pkg/operator/ceph/cluster/mgr/spec_test.go b/pkg/operator/ceph/cluster/mgr/spec_test.go index e481bb20e..0512322ac 100644 --- a/pkg/operator/ceph/cluster/mgr/spec_test.go +++ b/pkg/operator/ceph/cluster/mgr/spec_test.go @@ -61,7 +61,6 @@ func TestPodSpec(t *testing.T) { metav1.OwnerReference{}, "/var/lib/rook/", false, - false, ) mgrTestConfig := mgrConfig{ @@ -106,7 +105,6 @@ func TestServiceSpec(t *testing.T) { metav1.OwnerReference{}, "/var/lib/rook/", false, - false, ) s := c.makeMetricsService("rook-mgr") @@ -134,7 +132,6 @@ func TestHostNetwork(t *testing.T) { metav1.OwnerReference{}, "/var/lib/rook/", false, - false, ) mgrTestConfig := mgrConfig{ @@ -169,7 +166,6 @@ func TestHttpBindFix(t *testing.T) { metav1.OwnerReference{}, "/var/lib/rook/", false, - false, ) mgrTestConfig := mgrConfig{ @@ -231,7 +227,6 @@ func TestApplyPrometheusAnnotations(t *testing.T) { metav1.OwnerReference{}, "/var/lib/rook/", false, - false, ) mgrTestConfig := mgrConfig{ diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go index cb33c54b7..6d42d57c5 100644 --- a/pkg/operator/ceph/cluster/mon/health_test.go +++ b/pkg/operator/ceph/cluster/mon/health_test.go @@ -54,7 +54,7 @@ func TestCheckHealth(t *testing.T) { ConfigDir: configDir, Executor: executor, } - c := New(context, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}, false) + c := New(context, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}) setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion") logger.Infof("initial mons: %v", c.ClusterInfo.Monitors) c.waitForStart = false @@ -109,7 +109,7 @@ func TestCheckHealthNotFound(t *testing.T) { ConfigDir: configDir, Executor: executor, } - c := New(context, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}, false) + c := New(context, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}) setCommonMonProperties(c, 2, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion") c.waitForStart = false defer os.RemoveAll(c.context.ConfigDir) @@ -165,7 +165,7 @@ func TestAddRemoveMons(t *testing.T) { ConfigDir: configDir, Executor: executor, } - c := New(context, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}, false) + c := New(context, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}) setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 5, AllowMultiplePerNode: true}, "myversion") c.maxMonID = 0 // "a" is max mon id c.waitForStart = false diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index 2903a69cc..f70b14878 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/pkg/operator/ceph/cluster/mon/mon.go @@ -156,7 +156,7 @@ type SchedulingResult struct { } // New creates an instance of a mon cluster -func New(context *clusterd.Context, namespace, dataDirHostPath string, network cephv1.NetworkSpec, ownerRef metav1.OwnerReference, csiConfigMutex *sync.Mutex, isUpgrade bool) *Cluster { +func New(context *clusterd.Context, namespace, dataDirHostPath string, network cephv1.NetworkSpec, ownerRef metav1.OwnerReference, csiConfigMutex *sync.Mutex) *Cluster { return &Cluster{ context: context, dataDirHostPath: dataDirHostPath, @@ -172,12 +172,11 @@ func New(context *clusterd.Context, namespace, dataDirHostPath string, network c }, ownerRef: ownerRef, csiConfigMutex: csiConfigMutex, - isUpgrade: isUpgrade, } } // Start begins the process of running a cluster of Ceph mons. -func (c *Cluster) Start(clusterInfo *cephconfig.ClusterInfo, rookVersion string, cephVersion cephver.CephVersion, spec cephv1.ClusterSpec, isUpgrade bool) (*cephconfig.ClusterInfo, error) { +func (c *Cluster) Start(clusterInfo *cephconfig.ClusterInfo, rookVersion string, cephVersion cephver.CephVersion, spec cephv1.ClusterSpec) (*cephconfig.ClusterInfo, error) { // Only one goroutine can orchestrate the mons at a time c.acquireOrchestrationLock() @@ -186,7 +185,6 @@ func (c *Cluster) Start(clusterInfo *cephconfig.ClusterInfo, rookVersion string, c.ClusterInfo = clusterInfo c.rookVersion = rookVersion c.spec = spec - c.isUpgrade = isUpgrade // fail if we were instructed to deploy more than one mon on the same machine with host networking if c.Network.IsHost() && c.spec.Mon.AllowMultiplePerNode && c.spec.Mon.Count > 1 { @@ -793,20 +791,17 @@ func (c *Cluster) updateMon(m *monConfig, d *apps.Deployment) error { daemonType := string(config.MonType) var cephVersionToUse cephver.CephVersion - // If this is not a Ceph upgrade there is no need to check the ceph version - if c.isUpgrade { - currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.ClusterInfo.Name, daemonType) - if err != nil { - logger.Warningf("failed to retrieve current ceph %q version. %v", daemonType, err) - logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with %+v", c.ClusterInfo.CephVersion) - cephVersionToUse = c.ClusterInfo.CephVersion - } else { - logger.Debugf("current cluster version for monitors before upgrading is: %+v", currentCephVersion) - cephVersionToUse = currentCephVersion - } + currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.ClusterInfo.Name, daemonType) + if err != nil { + logger.Warningf("failed to retrieve current ceph %q version. %+v", daemonType, err) + logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with %+v", c.ClusterInfo.CephVersion) + cephVersionToUse = c.ClusterInfo.CephVersion + } else { + logger.Debugf("current cluster version for monitors before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion } - err := updateDeploymentAndWait(c.context, d, c.Namespace, daemonType, m.DaemonName, cephVersionToUse, c.isUpgrade, c.spec.SkipUpgradeChecks, false) + err = updateDeploymentAndWait(c.context, d, c.Namespace, daemonType, m.DaemonName, cephVersionToUse, c.spec.SkipUpgradeChecks, false) if err != nil { return errors.Wrapf(err, "failed to update mon deployment %s", m.ResourceName) } diff --git a/pkg/operator/ceph/cluster/mon/mon_test.go b/pkg/operator/ceph/cluster/mon/mon_test.go index 2e4466927..798faadd0 100644 --- a/pkg/operator/ceph/cluster/mon/mon_test.go +++ b/pkg/operator/ceph/cluster/mon/mon_test.go @@ -149,13 +149,13 @@ func TestStartMonPods(t *testing.T) { c := newCluster(context, namespace, cephv1.NetworkSpec{}, true, v1.ResourceRequirements{}) // start a basic cluster - _, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + _, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Nil(t, err) validateStart(t, c) // starting again should be a no-op, but still results in an error - _, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + _, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Nil(t, err) validateStart(t, c) @@ -169,7 +169,7 @@ func TestOperatorRestart(t *testing.T) { c.ClusterInfo = test.CreateConfigDir(1) // start a basic cluster - info, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + info, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Nil(t, err) assert.True(t, info.IsInitialized()) @@ -178,7 +178,7 @@ func TestOperatorRestart(t *testing.T) { c = newCluster(context, namespace, cephv1.NetworkSpec{}, true, v1.ResourceRequirements{}) // starting again should be a no-op, but will not result in an error - info, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + info, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Nil(t, err) assert.True(t, info.IsInitialized()) @@ -196,7 +196,7 @@ func TestOperatorRestartHostNetwork(t *testing.T) { c.ClusterInfo = test.CreateConfigDir(1) // start a basic cluster - info, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + info, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Nil(t, err) assert.True(t, info.IsInitialized()) @@ -206,7 +206,7 @@ func TestOperatorRestartHostNetwork(t *testing.T) { c = newCluster(context, namespace, cephv1.NetworkSpec{HostNetwork: true}, false, v1.ResourceRequirements{}) // starting again should be a no-op, but still results in an error - info, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + info, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Nil(t, err) assert.True(t, info.IsInitialized(), info) @@ -227,7 +227,7 @@ func TestSaveMonEndpoints(t *testing.T) { clientset := test.New(1) configDir, _ := ioutil.TempDir("", "") defer os.RemoveAll(configDir) - c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}, false) + c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}) setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion") // create the initial config map diff --git a/pkg/operator/ceph/cluster/mon/node_test.go b/pkg/operator/ceph/cluster/mon/node_test.go index e84a2ee33..c671847db 100644 --- a/pkg/operator/ceph/cluster/mon/node_test.go +++ b/pkg/operator/ceph/cluster/mon/node_test.go @@ -34,7 +34,7 @@ import ( func TestNodeAffinity(t *testing.T) { clientset := test.New(4) - c := New(&clusterd.Context{Clientset: clientset}, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}, false) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}) setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion") c.spec.Placement = map[rookalpha.KeyType]rookalpha.Placement{} @@ -79,7 +79,7 @@ func TestHostNetworkSameNode(t *testing.T) { c.ClusterInfo = test.CreateConfigDir(1) // start a basic cluster - _, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + _, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Error(t, err) } @@ -97,7 +97,7 @@ func TestPodMemory(t *testing.T) { c := newCluster(context, namespace, cephv1.NetworkSpec{}, true, r) c.ClusterInfo = test.CreateConfigDir(1) // start a basic cluster - _, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + _, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Error(t, err) // Test REQUEST == LIMIT @@ -113,7 +113,7 @@ func TestPodMemory(t *testing.T) { c = newCluster(context, namespace, cephv1.NetworkSpec{}, true, r) c.ClusterInfo = test.CreateConfigDir(1) // start a basic cluster - _, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + _, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Error(t, err) // Test LIMIT != REQUEST but obviously LIMIT > REQUEST @@ -129,7 +129,7 @@ func TestPodMemory(t *testing.T) { c = newCluster(context, namespace, cephv1.NetworkSpec{}, true, r) c.ClusterInfo = test.CreateConfigDir(1) // start a basic cluster - _, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + _, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Error(t, err) // Test valid case where pod resource is set approprietly @@ -145,7 +145,7 @@ func TestPodMemory(t *testing.T) { c = newCluster(context, namespace, cephv1.NetworkSpec{}, true, r) c.ClusterInfo = test.CreateConfigDir(1) // start a basic cluster - _, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + _, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Nil(t, err) // Test no resources were specified on the pod @@ -153,14 +153,14 @@ func TestPodMemory(t *testing.T) { c = newCluster(context, namespace, cephv1.NetworkSpec{}, true, r) c.ClusterInfo = test.CreateConfigDir(1) // start a basic cluster - _, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec, false) + _, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec) assert.Nil(t, err) } func TestHostNetwork(t *testing.T) { clientset := test.New(3) - c := New(&clusterd.Context{Clientset: clientset}, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}, false) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}) setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion") c.Network.HostNetwork = true diff --git a/pkg/operator/ceph/cluster/mon/spec.go b/pkg/operator/ceph/cluster/mon/spec.go index 883f6c879..cc533d4d5 100644 --- a/pkg/operator/ceph/cluster/mon/spec.go +++ b/pkg/operator/ceph/cluster/mon/spec.go @@ -331,13 +331,9 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) v1.Container { } // UpdateCephDeploymentAndWait verifies a deployment can be stopped or continued -func UpdateCephDeploymentAndWait(context *clusterd.Context, deployment *apps.Deployment, namespace, daemonType, daemonName string, cephVersion cephver.CephVersion, isUpgrade, skipUpgradeChecks, continueUpgradeAfterChecksEvenIfNotHealthy bool) error { +func UpdateCephDeploymentAndWait(context *clusterd.Context, deployment *apps.Deployment, namespace, daemonType, daemonName string, cephVersion cephver.CephVersion, skipUpgradeChecks, continueUpgradeAfterChecksEvenIfNotHealthy bool) error { callback := func(action string) error { - if !isUpgrade { - return nil - } - // At this point, we are in an upgrade if skipUpgradeChecks { logger.Warningf("this is a Ceph upgrade, not performing upgrade checks because skipUpgradeChecks is %t", skipUpgradeChecks) diff --git a/pkg/operator/ceph/cluster/mon/spec_test.go b/pkg/operator/ceph/cluster/mon/spec_test.go index d791cf4ee..ce258e8a2 100644 --- a/pkg/operator/ceph/cluster/mon/spec_test.go +++ b/pkg/operator/ceph/cluster/mon/spec_test.go @@ -49,7 +49,6 @@ func testPodSpec(t *testing.T, monID string, pvc bool) { cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}, - false, ) setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "rook/rook:myversion") c.spec.CephVersion = cephv1.CephVersionSpec{Image: "ceph/ceph:myceph"} @@ -99,7 +98,6 @@ func TestDeploymentPVCSpec(t *testing.T) { cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{}, - false, ) setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "rook/rook:myversion") c.spec.CephVersion = cephv1.CephVersionSpec{Image: "ceph/ceph:myceph"} @@ -162,7 +160,6 @@ func testPodSpecPlacement(t *testing.T, hostNet, allowMulti bool, req, pref int, cephv1.NetworkSpec{HostNetwork: hostNet}, metav1.OwnerReference{}, &sync.Mutex{}, - false, ) setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: allowMulti}, "rook/rook:myversion") diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go index 49a2ed935..e76227f37 100644 --- a/pkg/operator/ceph/cluster/osd/osd.go +++ b/pkg/operator/ceph/cluster/osd/osd.go @@ -89,7 +89,6 @@ type Cluster struct { priorityClassName string ownerRef metav1.OwnerReference kv *k8sutil.ConfigMapKVStore - isUpgrade bool skipUpgradeChecks bool continueUpgradeAfterChecksEvenIfNotHealthy bool } @@ -110,7 +109,6 @@ func New( prepareResources v1.ResourceRequirements, priorityClassName string, ownerRef metav1.OwnerReference, - isUpgrade bool, skipUpgradeChecks bool, continueUpgradeAfterChecksEvenIfNotHealthy bool, ) *Cluster { @@ -130,7 +128,6 @@ func New( priorityClassName: priorityClassName, ownerRef: ownerRef, kv: k8sutil.NewConfigMapKVStore(namespace, context.Clientset, ownerRef), - isUpgrade: isUpgrade, skipUpgradeChecks: skipUpgradeChecks, continueUpgradeAfterChecksEvenIfNotHealthy: continueUpgradeAfterChecksEvenIfNotHealthy, } @@ -486,21 +483,18 @@ func (c *Cluster) startOSDDaemonsOnPVC(pvcName string, config *provisionConfig, daemon := string(opconfig.OsdType) var cephVersionToUse cephver.CephVersion - // If this is not a Ceph upgrade there is no need to check the ceph version - if c.isUpgrade { - currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon) - if err != nil { - logger.Warningf("failed to retrieve current ceph %q version. %v", daemon, err) - logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion") - cephVersionToUse = c.clusterInfo.CephVersion - } else { - logger.Debugf("current cluster version for osds before upgrading is: %+v", currentCephVersion) - cephVersionToUse = currentCephVersion - } + currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon) + if err != nil { + logger.Warningf("failed to retrieve current ceph %q version. %+v", daemon, err) + logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with %+v", c.clusterInfo.CephVersion) + cephVersionToUse = c.clusterInfo.CephVersion + } else { + logger.Debugf("current cluster version for osds before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion } - if err = updateDeploymentAndWait(c.context, dp, c.Namespace, daemon, strconv.Itoa(osd.ID), cephVersionToUse, c.isUpgrade, c.skipUpgradeChecks, c.continueUpgradeAfterChecksEvenIfNotHealthy); err != nil { - logger.Errorf("failed to update osd deployment %d. %v", osd.ID, err) + if err = updateDeploymentAndWait(c.context, dp, c.Namespace, daemon, strconv.Itoa(osd.ID), cephVersionToUse, c.skipUpgradeChecks, c.continueUpgradeAfterChecksEvenIfNotHealthy); err != nil { + logger.Errorf("failed to update osd deployment %d. %+v", osd.ID, err) } } logger.Infof("started deployment for osd %d on pvc", osd.ID) @@ -575,21 +569,18 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig daemon := string(opconfig.OsdType) var cephVersionToUse cephver.CephVersion - // If this is not a Ceph upgrade there is no need to check the ceph version - if c.isUpgrade { - currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon) - if err != nil { - logger.Warningf("failed to retrieve current ceph %q version. %v", daemon, err) - logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion") - cephVersionToUse = c.clusterInfo.CephVersion - } else { - logger.Debugf("current cluster version for osds before upgrading is: %+v", currentCephVersion) - cephVersionToUse = currentCephVersion - } + currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon) + if err != nil { + logger.Warningf("failed to retrieve current ceph %q version. %+v", daemon, err) + logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with %+v", c.clusterInfo.CephVersion) + cephVersionToUse = c.clusterInfo.CephVersion + } else { + logger.Debugf("current cluster version for osds before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion } - if err = updateDeploymentAndWait(c.context, dp, c.Namespace, daemon, strconv.Itoa(osd.ID), cephVersionToUse, c.isUpgrade, c.skipUpgradeChecks, c.continueUpgradeAfterChecksEvenIfNotHealthy); err != nil { - logger.Errorf("failed to update osd deployment %d. %v", osd.ID, err) + if err = updateDeploymentAndWait(c.context, dp, c.Namespace, daemon, strconv.Itoa(osd.ID), cephVersionToUse, c.skipUpgradeChecks, c.continueUpgradeAfterChecksEvenIfNotHealthy); err != nil { + logger.Errorf("failed to update osd deployment %d. %+v", osd.ID, err) } } logger.Infof("started deployment for osd %d", osd.ID) diff --git a/pkg/operator/ceph/cluster/osd/osd_test.go b/pkg/operator/ceph/cluster/osd/osd_test.go index 3dd6247b2..2bbd7aa06 100644 --- a/pkg/operator/ceph/cluster/osd/osd_test.go +++ b/pkg/operator/ceph/cluster/osd/osd_test.go @@ -46,7 +46,7 @@ func TestStart(t *testing.T) { CephVersion: cephver.Nautilus, } c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{}, - rookalpha.StorageScopeSpec{}, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, false) + rookalpha.StorageScopeSpec{}, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false) // Start the first time err := c.Start() @@ -130,7 +130,7 @@ func TestAddRemoveNode(t *testing.T) { } c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: executor}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{}, - storageSpec, "/foo", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, false) + storageSpec, "/foo", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false) // kick off the start of the orchestration in a goroutine var startErr error @@ -209,7 +209,7 @@ func TestAddRemoveNode(t *testing.T) { // modify the storage spec to remove the node from the cluster storageSpec.Nodes = []rookalpha.Node{} c = New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: mockExec}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{}, - storageSpec, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, false) + storageSpec, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false) // reset the orchestration status watcher statusMapWatcher = watch.NewFake() @@ -266,7 +266,7 @@ func TestAddNodeFailure(t *testing.T) { CephVersion: cephver.Nautilus, } c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{}, - storageSpec, "/foo", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, false) + storageSpec, "/foo", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false) // kick off the start of the orchestration in a goroutine var startErr error @@ -287,7 +287,7 @@ func TestAddNodeFailure(t *testing.T) { func TestGetOSDInfo(t *testing.T) { c := New(&cephconfig.ClusterInfo{}, &clusterd.Context{}, "ns", "myversion", cephv1.CephVersionSpec{}, rookalpha.StorageScopeSpec{}, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, - v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, false) + v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false) node := "n1" location := "root=default host=myhost zone=myzone" diff --git a/pkg/operator/ceph/cluster/osd/spec_test.go b/pkg/operator/ceph/cluster/osd/spec_test.go index 736958c5f..a51b160d9 100644 --- a/pkg/operator/ceph/cluster/osd/spec_test.go +++ b/pkg/operator/ceph/cluster/osd/spec_test.go @@ -86,7 +86,7 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) { CephVersion: cephver.Nautilus, } c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephVersion, - storageSpec, dataDir, rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, false) + storageSpec, dataDir, rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false) devMountNeeded := deviceName != "" || allDevices @@ -192,7 +192,7 @@ func TestStorageSpecConfig(t *testing.T) { CephVersion: cephver.Nautilus, } c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephv1.CephVersionSpec{}, - storageSpec, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, false) + storageSpec, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false) n := c.DesiredStorage.ResolveNode(storageSpec.Nodes[0].Name) storeConfig := config.ToStoreConfig(storageSpec.Nodes[0].Config) @@ -242,7 +242,7 @@ func TestHostNetwork(t *testing.T) { CephVersion: cephver.Nautilus, } c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{}, - storageSpec, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{HostNetwork: true}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, false) + storageSpec, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{HostNetwork: true}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false) n := c.DesiredStorage.ResolveNode(storageSpec.Nodes[0].Name) osd := OSDInfo{ @@ -288,7 +288,7 @@ func TestOsdOnSDNFlag(t *testing.T) { func TestOsdPrepareResources(t *testing.T) { clientset := fake.NewSimpleClientset() c := New(&cephconfig.ClusterInfo{}, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{}, - rookalpha.StorageScopeSpec{}, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, false) + rookalpha.StorageScopeSpec{}, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false) // TEST 2: NOT running on PVC and some prepareResources are specificied rr := v1.ResourceRequirements{ diff --git a/pkg/operator/ceph/cluster/osd/status_test.go b/pkg/operator/ceph/cluster/osd/status_test.go index e853696d6..4ec5fe112 100644 --- a/pkg/operator/ceph/cluster/osd/status_test.go +++ b/pkg/operator/ceph/cluster/osd/status_test.go @@ -42,7 +42,7 @@ func TestOrchestrationStatus(t *testing.T) { CephVersion: cephver.Nautilus, } c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{}, - rookalpha.StorageScopeSpec{}, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, false) + rookalpha.StorageScopeSpec{}, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false) kv := k8sutil.NewConfigMapKVStore(c.Namespace, clientset, metav1.OwnerReference{}) nodeName := "mynode" cmName := fmt.Sprintf(orchestrationStatusMapName, nodeName) diff --git a/pkg/operator/ceph/cluster/rbd/mirror.go b/pkg/operator/ceph/cluster/rbd/mirror.go index 9be3b256b..66d0c16ae 100644 --- a/pkg/operator/ceph/cluster/rbd/mirror.go +++ b/pkg/operator/ceph/cluster/rbd/mirror.go @@ -61,7 +61,6 @@ type Mirroring struct { rookVersion string Network cephv1.NetworkSpec dataDirHostPath string - isUpgrade bool skipUpgradeChecks bool } @@ -79,7 +78,6 @@ func New( priorityClassName string, ownerRef metav1.OwnerReference, dataDirHostPath string, - isUpgrade bool, skipUpgradeChecks bool, ) *Mirroring { return &Mirroring{ @@ -96,7 +94,6 @@ func New( priorityClassName: priorityClassName, ownerRef: ownerRef, dataDirHostPath: dataDirHostPath, - isUpgrade: isUpgrade, skipUpgradeChecks: skipUpgradeChecks, } } @@ -140,20 +137,18 @@ func (m *Mirroring) Start() error { var cephVersionToUse cephver.CephVersion // If this is not a Ceph upgrade there is no need to check the ceph version - if m.isUpgrade { - currentCephVersion, err := client.LeastUptodateDaemonVersion(m.context, m.ClusterInfo.Name, daemon) - if err != nil { - logger.Warningf("failed to retrieve current ceph %q version. %v", daemon, err) - logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with m.ClusterInfo.CephVersion") - cephVersionToUse = m.ClusterInfo.CephVersion + currentCephVersion, err := client.LeastUptodateDaemonVersion(m.context, m.ClusterInfo.Name, daemon) + if err != nil { + logger.Warningf("failed to retrieve current ceph %q version. %+v", daemon, err) + logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with %+v", m.ClusterInfo.CephVersion) + cephVersionToUse = m.ClusterInfo.CephVersion - } else { - logger.Debugf("current cluster version for rbd mirrors before upgrading is: %+v", currentCephVersion) - cephVersionToUse = currentCephVersion - } + } else { + logger.Debugf("current cluster version for rbd mirrors before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion } - if err := updateDeploymentAndWait(m.context, d, m.Namespace, daemon, daemonConf.DaemonID, cephVersionToUse, m.isUpgrade, m.skipUpgradeChecks, false); err != nil { + if err := updateDeploymentAndWait(m.context, d, m.Namespace, daemon, daemonConf.DaemonID, cephVersionToUse, m.skipUpgradeChecks, false); err != nil { // fail could be an issue updating label selector (immutable), so try del and recreate logger.Debugf("updateDeploymentAndWait failed for rbd-mirror %q. Attempting del-and-recreate. %v", resourceName, err) err = m.context.Clientset.AppsV1().Deployments(m.Namespace).Delete(d.Name, &metav1.DeleteOptions{}) diff --git a/pkg/operator/ceph/cluster/rbd/mirror_test.go b/pkg/operator/ceph/cluster/rbd/mirror_test.go index 9c80a5633..e852f4121 100644 --- a/pkg/operator/ceph/cluster/rbd/mirror_test.go +++ b/pkg/operator/ceph/cluster/rbd/mirror_test.go @@ -59,7 +59,6 @@ func TestRBDMirror(t *testing.T) { metav1.OwnerReference{}, "/var/lib/rook/", false, - false, ) err := c.Start() diff --git a/pkg/operator/ceph/cluster/rbd/spec_test.go b/pkg/operator/ceph/cluster/rbd/spec_test.go index c19be683e..b2627bbd0 100644 --- a/pkg/operator/ceph/cluster/rbd/spec_test.go +++ b/pkg/operator/ceph/cluster/rbd/spec_test.go @@ -57,7 +57,6 @@ func TestPodSpec(t *testing.T) { metav1.OwnerReference{}, "/var/lib/rook/", false, - false, ) daemonConf := daemonConfig{ DaemonID: "a", diff --git a/pkg/operator/ceph/file/controller.go b/pkg/operator/ceph/file/controller.go index ac632b423..4e116247d 100644 --- a/pkg/operator/ceph/file/controller.go +++ b/pkg/operator/ceph/file/controller.go @@ -54,7 +54,6 @@ type FilesystemController struct { ownerRef metav1.OwnerReference dataDirHostPath string orchestrationMutex sync.Mutex - isUpgrade bool } // NewFilesystemController create controller for watching filesystem custom resources created @@ -66,7 +65,6 @@ func NewFilesystemController( clusterSpec *cephv1.ClusterSpec, ownerRef metav1.OwnerReference, dataDirHostPath string, - isUpgrade bool, ) *FilesystemController { return &FilesystemController{ clusterInfo: clusterInfo, @@ -76,7 +74,6 @@ func NewFilesystemController( clusterSpec: clusterSpec, ownerRef: ownerRef, dataDirHostPath: dataDirHostPath, - isUpgrade: isUpgrade, } } @@ -121,7 +118,7 @@ func (c *FilesystemController) onAdd(obj interface{}) { } updateCephFilesystemStatus(filesystem.GetName(), filesystem.GetNamespace(), k8sutil.ProcessingStatus, c.context) - err = createFilesystem(c.clusterInfo, c.context, *filesystem, c.rookVersion, c.clusterSpec, c.filesystemOwner(filesystem), c.clusterSpec.DataDirHostPath, c.isUpgrade) + err = createFilesystem(c.clusterInfo, c.context, *filesystem, c.rookVersion, c.clusterSpec, c.filesystemOwner(filesystem), c.clusterSpec.DataDirHostPath) if err != nil { logger.Errorf("failed to create filesystem %q. %v", filesystem.Name, err) updateCephFilesystemStatus(filesystem.GetName(), filesystem.GetNamespace(), k8sutil.FailedStatus, c.context) @@ -157,7 +154,7 @@ func (c *FilesystemController) onUpdate(oldObj, newObj interface{}) { // if the filesystem is modified, allow the filesystem to be created if it wasn't already logger.Infof("updating filesystem %s", newFS.Name) updateCephFilesystemStatus(newFS.GetName(), newFS.GetNamespace(), k8sutil.ProcessingStatus, c.context) - err = createFilesystem(c.clusterInfo, c.context, *newFS, c.rookVersion, c.clusterSpec, c.filesystemOwner(newFS), c.clusterSpec.DataDirHostPath, c.isUpgrade) + err = createFilesystem(c.clusterInfo, c.context, *newFS, c.rookVersion, c.clusterSpec, c.filesystemOwner(newFS), c.clusterSpec.DataDirHostPath) if err != nil { logger.Errorf("failed to create (modify) filesystem %q. %v", newFS.Name, err) updateCephFilesystemStatus(newFS.GetName(), newFS.GetNamespace(), k8sutil.FailedStatus, c.context) @@ -173,9 +170,6 @@ func (c *FilesystemController) ParentClusterChanged(cluster cephv1.ClusterSpec, return } - // This is an upgrade so let's activate the flag - c.isUpgrade = isUpgrade - c.acquireOrchestrationLock() defer c.releaseOrchestrationLock() @@ -187,7 +181,7 @@ func (c *FilesystemController) ParentClusterChanged(cluster cephv1.ClusterSpec, } for _, fs := range filesystems.Items { logger.Infof("updating the ceph version for filesystem %s to %s", fs.Name, c.clusterSpec.CephVersion.Image) - err = createFilesystem(c.clusterInfo, c.context, fs, c.rookVersion, c.clusterSpec, c.filesystemOwner(&fs), c.clusterSpec.DataDirHostPath, c.isUpgrade) + err = createFilesystem(c.clusterInfo, c.context, fs, c.rookVersion, c.clusterSpec, c.filesystemOwner(&fs), c.clusterSpec.DataDirHostPath) if err != nil { logger.Errorf("failed to update filesystem %q. %v", fs.Name, err) } else { diff --git a/pkg/operator/ceph/file/filesystem.go b/pkg/operator/ceph/file/filesystem.go index b8bf577e8..9c4d22883 100644 --- a/pkg/operator/ceph/file/filesystem.go +++ b/pkg/operator/ceph/file/filesystem.go @@ -54,7 +54,6 @@ func createFilesystem( clusterSpec *cephv1.ClusterSpec, ownerRefs metav1.OwnerReference, dataDirHostPath string, - isUpgrade bool, ) error { if err := validateFilesystem(context, fs); err != nil { return err @@ -93,7 +92,7 @@ func createFilesystem( } logger.Infof("start running mdses for filesystem %s", fs.Name) - c := mds.NewCluster(clusterInfo, context, rookVersion, clusterSpec, fs, filesystem, ownerRefs, dataDirHostPath, isUpgrade) + c := mds.NewCluster(clusterInfo, context, rookVersion, clusterSpec, fs, filesystem, ownerRefs, dataDirHostPath) if err := c.Start(); err != nil { return err } diff --git a/pkg/operator/ceph/file/filesystem_test.go b/pkg/operator/ceph/file/filesystem_test.go index ac84c7ea4..347ddb668 100644 --- a/pkg/operator/ceph/file/filesystem_test.go +++ b/pkg/operator/ceph/file/filesystem_test.go @@ -114,14 +114,14 @@ func TestCreateFilesystem(t *testing.T) { clusterInfo := &cephconfig.ClusterInfo{FSID: "myfsid"} // start a basic cluster - err := createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, "/var/lib/rook/", false) + err := createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, "/var/lib/rook/") assert.Nil(t, err) validateStart(t, context, fs) assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated)) testopk8s.ClearDeploymentsUpdated(deploymentsUpdated) // starting again should be a no-op - err = createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, "/var/lib/rook/", false) + err = createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, "/var/lib/rook/") assert.Nil(t, err) validateStart(t, context, fs) assert.ElementsMatch(t, []string{"rook-ceph-mds-myfs-a", "rook-ceph-mds-myfs-b"}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated)) @@ -142,7 +142,7 @@ func TestCreateFilesystem(t *testing.T) { Clientset: testop.New(3)} //Create another filesystem which should fail - err = createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, "/var/lib/rook/", false) + err = createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, "/var/lib/rook/") assert.Equal(t, "failed to create filesystem myfs: cannot create multiple filesystems. enable ROOK_ALLOW_MULTIPLE_FILESYSTEMS env variable to create more than one", err.Error()) } @@ -179,12 +179,12 @@ func TestCreateNopoolFilesystem(t *testing.T) { clusterInfo := &cephconfig.ClusterInfo{FSID: "myfsid"} // start a basic cluster - err := createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, "/var/lib/rook/", false) + err := createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, "/var/lib/rook/") assert.Nil(t, err) validateStart(t, context, fs) // starting again should be a no-op - err = createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, "/var/lib/rook/", false) + err = createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, "/var/lib/rook/") assert.Nil(t, err) validateStart(t, context, fs) diff --git a/pkg/operator/ceph/file/mds/mds.go b/pkg/operator/ceph/file/mds/mds.go index be0507952..9c604d34b 100644 --- a/pkg/operator/ceph/file/mds/mds.go +++ b/pkg/operator/ceph/file/mds/mds.go @@ -61,7 +61,6 @@ type Cluster struct { fsID string ownerRef metav1.OwnerReference dataDirHostPath string - isUpgrade bool } type mdsConfig struct { @@ -80,7 +79,6 @@ func NewCluster( fsdetails *client.CephFilesystemDetails, ownerRef metav1.OwnerReference, dataDirHostPath string, - isUpgrade bool, ) *Cluster { return &Cluster{ clusterInfo: clusterInfo, @@ -91,7 +89,6 @@ func NewCluster( fsID: strconv.Itoa(fsdetails.ID), ownerRef: ownerRef, dataDirHostPath: dataDirHostPath, - isUpgrade: isUpgrade, } } @@ -184,14 +181,14 @@ func (c *Cluster) Start() error { currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon) if err != nil { logger.Warningf("failed to retrieve current ceph %q version. %v", daemon, err) - logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion") + logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with %+v", c.clusterInfo.CephVersion) cephVersionToUse = c.clusterInfo.CephVersion } else { logger.Debugf("current cluster version for mdss before upgrading is: %+v", currentCephVersion) cephVersionToUse = currentCephVersion } - if err = UpdateDeploymentAndWait(c.context, d, c.fs.Namespace, daemon, daemonLetterID, cephVersionToUse, c.isUpgrade, c.clusterSpec.SkipUpgradeChecks, false); err != nil { + if err = UpdateDeploymentAndWait(c.context, d, c.fs.Namespace, daemon, daemonLetterID, cephVersionToUse, c.clusterSpec.SkipUpgradeChecks, false); err != nil { return errors.Wrapf(err, "failed to update mds deployment %s", d.Name) } } diff --git a/pkg/operator/ceph/file/mds/spec_test.go b/pkg/operator/ceph/file/mds/spec_test.go index 4f4bb4d83..bd480c840 100644 --- a/pkg/operator/ceph/file/mds/spec_test.go +++ b/pkg/operator/ceph/file/mds/spec_test.go @@ -74,7 +74,6 @@ func testDeploymentObject(network cephv1.NetworkSpec) *apps.Deployment { &client.CephFilesystemDetails{ID: 15}, metav1.OwnerReference{}, "/var/lib/rook/", - false, ) mdsTestConfig := &mdsConfig{ DaemonID: "myfs-a", diff --git a/pkg/operator/ceph/nfs/controller.go b/pkg/operator/ceph/nfs/controller.go index 832a48829..23b2f27b2 100644 --- a/pkg/operator/ceph/nfs/controller.go +++ b/pkg/operator/ceph/nfs/controller.go @@ -50,7 +50,6 @@ type CephNFSController struct { rookImage string clusterSpec *cephv1.ClusterSpec orchestrationMutex sync.Mutex - isUpgrade bool } // NewCephNFSController create controller for watching NFS custom resources created @@ -176,10 +175,6 @@ func (c *CephNFSController) ParentClusterChanged(cluster cephv1.ClusterSpec, clu return } - // This is mostly a placeholder since we don't perform any upgrade checks for nfs since it's not in Ceph's servicemap yet - // This is an upgrade so let's activate the flag - c.isUpgrade = isUpgrade - c.acquireOrchestrationLock() defer c.releaseOrchestrationLock() diff --git a/pkg/operator/ceph/nfs/nfs.go b/pkg/operator/ceph/nfs/nfs.go index 02fe8bccf..bf8e1704e 100644 --- a/pkg/operator/ceph/nfs/nfs.go +++ b/pkg/operator/ceph/nfs/nfs.go @@ -90,8 +90,8 @@ func (c *CephNFSController) upCephNFS(n cephv1.CephNFS, oldActive int) error { } logger.Infof("ganesha deployment %s already exists. updating if needed", deployment.Name) // We don't invoke ceph versions here since nfs do not show up in the service map (yet?) - if err := updateDeploymentAndWait(c.context, deployment, n.Namespace, "nfs", id, c.clusterInfo.CephVersion, c.isUpgrade, c.clusterSpec.SkipUpgradeChecks, false); err != nil { - return errors.Wrapf(err, "failed to update ganesha deployment %s", deployment.Name) + if err := updateDeploymentAndWait(c.context, deployment, n.Namespace, "nfs", id, c.clusterInfo.CephVersion, c.clusterSpec.SkipUpgradeChecks, false); err != nil { + return errors.Wrapf(err, "failed to update ganesha deployment %q", deployment.Name) } } else { logger.Infof("ganesha deployment %s started", deployment.Name) diff --git a/pkg/operator/ceph/object/controller.go b/pkg/operator/ceph/object/controller.go index 0c5ddda63..044c704c0 100644 --- a/pkg/operator/ceph/object/controller.go +++ b/pkg/operator/ceph/object/controller.go @@ -54,7 +54,6 @@ type ObjectStoreController struct { ownerRef metav1.OwnerReference dataDirHostPath string orchestrationMutex sync.Mutex - isUpgrade bool } // NewObjectStoreController create controller for watching object store custom resources created @@ -66,7 +65,6 @@ func NewObjectStoreController( clusterSpec *cephv1.ClusterSpec, ownerRef metav1.OwnerReference, dataDirHostPath string, - isUpgrade bool, ) *ObjectStoreController { return &ObjectStoreController{ clusterInfo: clusterInfo, @@ -76,7 +74,6 @@ func NewObjectStoreController( rookImage: rookImage, ownerRef: ownerRef, dataDirHostPath: dataDirHostPath, - isUpgrade: isUpgrade, } } @@ -164,7 +161,6 @@ func (c *ObjectStoreController) createOrUpdateStore(objectstore *cephv1.CephObje clusterSpec: c.clusterSpec, ownerRef: c.storeOwners(objectstore), DataPathMap: cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, objectstore.Name, c.clusterInfo.Name, c.dataDirHostPath), - isUpgrade: c.isUpgrade, } if err := cfg.createOrUpdate(); err != nil { logger.Errorf("failed to create or update object store %s. %v", objectstore.Name, err) @@ -200,9 +196,6 @@ func (c *ObjectStoreController) ParentClusterChanged(cluster cephv1.ClusterSpec, return } - // This is an upgrade so let's activate the flag - c.isUpgrade = isUpgrade - c.acquireOrchestrationLock() defer c.releaseOrchestrationLock() diff --git a/pkg/operator/ceph/object/rgw.go b/pkg/operator/ceph/object/rgw.go index d658f2ee0..c971eb055 100644 --- a/pkg/operator/ceph/object/rgw.go +++ b/pkg/operator/ceph/object/rgw.go @@ -43,7 +43,6 @@ type clusterConfig struct { clusterSpec *cephv1.ClusterSpec ownerRef metav1.OwnerReference DataPathMap *config.DataPathMap - isUpgrade bool skipUpgradeChecks bool } @@ -172,16 +171,16 @@ func (c *clusterConfig) startRGWPods() error { var cephVersionToUse cephver.CephVersion currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon) if err != nil { - logger.Warningf("failed to retrieve current ceph %s version. %v", daemon, err) - logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion") + logger.Warningf("failed to retrieve current ceph %q version. %v", daemon, err) + logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with %+v", c.clusterInfo.CephVersion) cephVersionToUse = c.clusterInfo.CephVersion } else { logger.Debugf("current cluster version for rgws before upgrading is: %+v", currentCephVersion) cephVersionToUse = currentCephVersion } - if err := updateDeploymentAndWait(c.context, deployment, c.store.Namespace, daemon, daemonLetterID, cephVersionToUse, c.isUpgrade, c.skipUpgradeChecks, false); err != nil { - return errors.Wrapf(err, "failed to update object store %s deployment %s", c.store.Name, deployment.Name) + if err := updateDeploymentAndWait(c.context, deployment, c.store.Namespace, daemon, daemonLetterID, cephVersionToUse, c.skipUpgradeChecks, false); err != nil { + return errors.Wrapf(err, "failed to update object store %q deployment %q", c.store.Name, deployment.Name) } } diff --git a/pkg/operator/ceph/object/rgw_test.go b/pkg/operator/ceph/object/rgw_test.go index c55c3fa68..0094a0329 100644 --- a/pkg/operator/ceph/object/rgw_test.go +++ b/pkg/operator/ceph/object/rgw_test.go @@ -52,7 +52,7 @@ func TestStartRGW(t *testing.T) { data := cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, "my-fs", "rook-ceph", "/var/lib/rook/") // start a basic cluster - c := &clusterConfig{info, context, store, version, &cephv1.ClusterSpec{}, metav1.OwnerReference{}, data, false, false} + c := &clusterConfig{info, context, store, version, &cephv1.ClusterSpec{}, metav1.OwnerReference{}, data, false} err := c.startRGWPods() assert.Nil(t, err) @@ -94,7 +94,7 @@ func TestCreateObjectStore(t *testing.T) { data := cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, "my-fs", "rook-ceph", "/var/lib/rook/") // create the pools - c := &clusterConfig{info, context, store, "1.2.3.4", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, data, false, false} + c := &clusterConfig{info, context, store, "1.2.3.4", &cephv1.ClusterSpec{}, metav1.OwnerReference{}, data, false} err := c.createOrUpdate() assert.Nil(t, err) } diff --git a/pkg/operator/k8sutil/test/deployment.go b/pkg/operator/k8sutil/test/deployment.go index 09cddc03e..5a0ac0e06 100644 --- a/pkg/operator/k8sutil/test/deployment.go +++ b/pkg/operator/k8sutil/test/deployment.go @@ -15,11 +15,11 @@ import ( // returns a pointer to this slice which the calling func may use to verify the expected contents of // deploymentsUpdated based on expected behavior. func UpdateDeploymentAndWaitStub() ( - stubFunc func(context *clusterd.Context, deployment *apps.Deployment, namespace, daemonType, daemonName string, cephVersion version.CephVersion, isUpgrade, skipUpgradeChecks, continueUpgradeAfterChecksEvenIfNotHealthy bool) error, + stubFunc func(context *clusterd.Context, deployment *apps.Deployment, namespace, daemonType, daemonName string, cephVersion version.CephVersion, skipUpgradeChecks, continueUpgradeAfterChecksEvenIfNotHealthy bool) error, deploymentsUpdated *[]*apps.Deployment, ) { deploymentsUpdated = &[]*apps.Deployment{} - stubFunc = func(context *clusterd.Context, deployment *apps.Deployment, namespace, daemonType, daemonName string, cephVersion version.CephVersion, isUpgrade, skipUpgradeChecks, continueUpgradeAfterChecksEvenIfNotHealthy bool) error { + stubFunc = func(context *clusterd.Context, deployment *apps.Deployment, namespace, daemonType, daemonName string, cephVersion version.CephVersion, skipUpgradeChecks, continueUpgradeAfterChecksEvenIfNotHealthy bool) error { *deploymentsUpdated = append(*deploymentsUpdated, deployment) return nil }