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 1/2] 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 } From caa8382e25c72e94fe4f863899101afbe457d607 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Fri, 24 Jan 2020 10:27:04 +0100 Subject: [PATCH 2/2] ceph: do not update deployment if no changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit We know use k8s-objectmatcher to verify whether a deployment is going to change or not. This avoids calling Update() from k8s utils which takes time to run. Closes: https://github.com/rook/rook/issues/4519 and https://github.com/rook/rook/issues/4642 Signed-off-by: Sébastien Han --- Gopkg.lock | 20 +++++- Gopkg.toml | 4 ++ pkg/operator/ceph/cluster/mgr/mgr.go | 8 +++ pkg/operator/ceph/cluster/mon/mon.go | 8 +++ pkg/operator/ceph/cluster/mon/spec.go | 6 +- pkg/operator/ceph/cluster/osd/osd.go | 9 +++ pkg/operator/ceph/cluster/rbd/mirror.go | 8 +++ pkg/operator/ceph/file/mds/mds.go | 8 +++ pkg/operator/ceph/nfs/nfs.go | 11 +++- pkg/operator/ceph/object/rgw.go | 8 +++ pkg/operator/k8sutil/deployment.go | 87 +++++++++++++++---------- 11 files changed, 137 insertions(+), 40 deletions(-) diff --git a/Gopkg.lock b/Gopkg.lock index 312b0408c..f439c76bf 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -68,6 +68,14 @@ revision = "a2484497433aa110263c045df2f1d49336b38cfd" version = "v1.16.22" +[[projects]] + digest = "1:ce8e1e0bd3eb4b6d38240414955fb492cb3499a253525eb31babc4c43dda1bed" + name = "github.com/banzaicloud/k8s-objectmatcher" + packages = ["patch"] + pruneopts = "UT" + revision = "cf8ca3ded3b2b3a27a54c086e935bf228be9b416" + version = "v1.1.0" + [[projects]] branch = "master" digest = "1:d6afaeed1502aa28e80a4ed0981d570ad91b2579193404256ce672ed0a609e0d" @@ -318,6 +326,14 @@ revision = "7c663266750e7d82587642f65e60bc4083f1f84e" version = "v0.2.0" +[[projects]] + digest = "1:d8b04ac65ed598214a3fa5069577aedb5ff6a92cbc3d6650bd14e42e13e685ba" + name = "github.com/goph/emperror" + packages = ["."] + pruneopts = "UT" + revision = "9727b49dc9d66225368645198434124e02b7e355" + version = "v0.17.2" + [[projects]] digest = "1:9b7a07ac7577787a8ecc1334cb9f34df1c76ed82a917d556c5713d3ab84fbc43" name = "github.com/grpc-ecosystem/go-grpc-prometheus" @@ -899,7 +915,7 @@ version = "kubernetes-1.14.1" [[projects]] - digest = "1:12e1193fca56eef2efa8e94239f96011a2ac8f60cba50f04da1babad5029de99" + digest = "1:bea5cdc22cf065a1b485bb725697cd39c3f940190691a39a8325234035bfe42f" name = "k8s.io/apimachinery" packages = [ "pkg/api/equality", @@ -934,6 +950,7 @@ "pkg/util/framer", "pkg/util/intstr", "pkg/util/json", + "pkg/util/jsonmergepatch", "pkg/util/mergepatch", "pkg/util/naming", "pkg/util/net", @@ -1484,6 +1501,7 @@ "github.com/aws/aws-sdk-go/aws/credentials", "github.com/aws/aws-sdk-go/aws/session", "github.com/aws/aws-sdk-go/service/s3", + "github.com/banzaicloud/k8s-objectmatcher/patch", "github.com/coreos/pkg/capnslog", "github.com/coreos/prometheus-operator/pkg/apis/monitoring/v1", "github.com/coreos/prometheus-operator/pkg/client/versioned", diff --git a/Gopkg.toml b/Gopkg.toml index b75316563..4b1462b63 100644 --- a/Gopkg.toml +++ b/Gopkg.toml @@ -138,3 +138,7 @@ ignored = [ [[override]] name = "github.com/openshift/machine-api-operator" revision = "a0949226d20ea454cf08252a182a8e32054027c3" + +[[constraint]] + name = "github.com/banzaicloud/k8s-objectmatcher" + version = "=v1.1.0" \ No newline at end of file diff --git a/pkg/operator/ceph/cluster/mgr/mgr.go b/pkg/operator/ceph/cluster/mgr/mgr.go index 67067ef4a..611a9db60 100644 --- a/pkg/operator/ceph/cluster/mgr/mgr.go +++ b/pkg/operator/ceph/cluster/mgr/mgr.go @@ -24,6 +24,7 @@ import ( "strings" "sync" + "github.com/banzaicloud/k8s-objectmatcher/patch" "github.com/coreos/pkg/capnslog" "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" @@ -165,6 +166,13 @@ func (c *Cluster) Start() error { // start the deployment d := c.makeDeployment(mgrConfig) logger.Debugf("starting mgr deployment: %+v", d) + + // Set the deployment hash as an annotation + err = patch.DefaultAnnotator.SetLastAppliedAnnotation(d) + if err != nil { + return errors.Wrapf(err, "failed to set annotation for deployment %q", d.Name) + } + _, err = c.context.Clientset.AppsV1().Deployments(c.Namespace).Create(d) if err != nil { if !kerrors.IsAlreadyExists(err) { diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index f70b14878..a29ee9a0a 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/pkg/operator/ceph/cluster/mon/mon.go @@ -27,6 +27,7 @@ import ( "sync" "time" + "github.com/banzaicloud/k8s-objectmatcher/patch" "github.com/coreos/pkg/capnslog" "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" @@ -845,6 +846,13 @@ func (c *Cluster) startMon(m *monConfig, node *NodeInfo) error { pvcExists := false deploymentExists := false d := c.makeDeployment(m) + + // Set the deployment hash as an annotation + err := patch.DefaultAnnotator.SetLastAppliedAnnotation(d) + if err != nil { + return errors.Wrapf(err, "failed to set annotation for deployment %q", d.Name) + } + existingDeployment, err := c.context.Clientset.AppsV1().Deployments(c.Namespace).Get(d.Name, metav1.GetOptions{}) if err == nil { deploymentExists = true diff --git a/pkg/operator/ceph/cluster/mon/spec.go b/pkg/operator/ceph/cluster/mon/spec.go index cc533d4d5..12c10b0e0 100644 --- a/pkg/operator/ceph/cluster/mon/spec.go +++ b/pkg/operator/ceph/cluster/mon/spec.go @@ -348,9 +348,8 @@ func UpdateCephDeploymentAndWait(context *clusterd.Context, deployment *apps.Dep if continueUpgradeAfterChecksEvenIfNotHealthy { logger.Infof("The %s daemon %s is not ok-to-stop but 'continueUpgradeAfterChecksEvenIfNotHealthy' is true, so proceeding to stop...", daemonType, daemonName) return nil - } else { - return errors.Wrapf(err, "failed to check if we can %s the deployment %s", action, deployment.Name) } + return errors.Wrapf(err, "failed to check if we can %s the deployment %s", action, deployment.Name) } } @@ -360,9 +359,8 @@ func UpdateCephDeploymentAndWait(context *clusterd.Context, deployment *apps.Dep if continueUpgradeAfterChecksEvenIfNotHealthy { logger.Infof("The %s daemon %s is not ok-to-stop but 'continueUpgradeAfterChecksEvenIfNotHealthy' is true, so continuing...", daemonType, daemonName) return nil - } else { - return errors.Wrapf(err, "failed to check if we can %s the deployment %s", action, deployment.Name) } + return errors.Wrapf(err, "failed to check if we can %s the deployment %s", action, deployment.Name) } } diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go index e76227f37..7ca9f745b 100644 --- a/pkg/operator/ceph/cluster/osd/osd.go +++ b/pkg/operator/ceph/cluster/osd/osd.go @@ -25,6 +25,7 @@ import ( "k8s.io/client-go/kubernetes" + "github.com/banzaicloud/k8s-objectmatcher/patch" "github.com/coreos/pkg/capnslog" "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" @@ -544,6 +545,14 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig continue } + // Set the deployment hash as an annotation + err = patch.DefaultAnnotator.SetLastAppliedAnnotation(dp) + if err != nil { + errMsg := fmt.Sprintf("failed to set annotation for deployment %q. %v", dp.Name, err) + config.addError(errMsg) + continue + } + createdDeployment, createErr := c.context.Clientset.AppsV1().Deployments(c.Namespace).Create(dp) if createErr != nil { if !kerrors.IsAlreadyExists(createErr) { diff --git a/pkg/operator/ceph/cluster/rbd/mirror.go b/pkg/operator/ceph/cluster/rbd/mirror.go index 66d0c16ae..4662b407f 100644 --- a/pkg/operator/ceph/cluster/rbd/mirror.go +++ b/pkg/operator/ceph/cluster/rbd/mirror.go @@ -20,6 +20,7 @@ package rbd import ( "fmt" + "github.com/banzaicloud/k8s-objectmatcher/patch" "github.com/coreos/pkg/capnslog" "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" @@ -126,6 +127,13 @@ func (m *Mirroring) Start() error { // Start the deployment d := m.makeDeployment(daemonConf) + + // Set the deployment hash as an annotation + err = patch.DefaultAnnotator.SetLastAppliedAnnotation(d) + if err != nil { + return errors.Wrapf(err, "failed to set annotation for deployment %q", d.Name) + } + if _, err := m.context.Clientset.AppsV1().Deployments(m.Namespace).Create(d); err != nil { if !kerrors.IsAlreadyExists(err) { return errors.Wrapf(err, "failed to create %s deployment", resourceName) diff --git a/pkg/operator/ceph/file/mds/mds.go b/pkg/operator/ceph/file/mds/mds.go index 9c604d34b..85f990087 100644 --- a/pkg/operator/ceph/file/mds/mds.go +++ b/pkg/operator/ceph/file/mds/mds.go @@ -22,6 +22,7 @@ import ( "strconv" "time" + "github.com/banzaicloud/k8s-objectmatcher/patch" "github.com/coreos/pkg/capnslog" "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" @@ -156,6 +157,13 @@ func (c *Cluster) Start() error { // start the deployment d := c.makeDeployment(mdsConfig) logger.Debugf("starting mds: %+v", d) + + // Set the deployment hash as an annotation + err = patch.DefaultAnnotator.SetLastAppliedAnnotation(d) + if err != nil { + return errors.Wrapf(err, "failed to set annotation for deployment %q", d.Name) + } + createdDeployment, createErr := c.context.Clientset.AppsV1().Deployments(c.fs.Namespace).Create(d) if createErr != nil { if !kerrors.IsAlreadyExists(createErr) { diff --git a/pkg/operator/ceph/nfs/nfs.go b/pkg/operator/ceph/nfs/nfs.go index bf8e1704e..8f05fa06f 100644 --- a/pkg/operator/ceph/nfs/nfs.go +++ b/pkg/operator/ceph/nfs/nfs.go @@ -23,6 +23,7 @@ import ( "os/exec" "strings" + "github.com/banzaicloud/k8s-objectmatcher/patch" "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/pkg/clusterd" @@ -81,8 +82,16 @@ func (c *CephNFSController) upCephNFS(n cephv1.CephNFS, oldActive int) error { }, } - // start the deployment + // create the deployment deployment := c.makeDeployment(n, cfg) + + // Set the deployment hash as an annotation + err = patch.DefaultAnnotator.SetLastAppliedAnnotation(deployment) + if err != nil { + return errors.Wrapf(err, "failed to set annotation for deployment %q", deployment.Name) + } + + // start the deployment _, err = c.context.Clientset.AppsV1().Deployments(n.Namespace).Create(deployment) if err != nil { if !kerrors.IsAlreadyExists(err) { diff --git a/pkg/operator/ceph/object/rgw.go b/pkg/operator/ceph/object/rgw.go index c971eb055..681cbbbf0 100644 --- a/pkg/operator/ceph/object/rgw.go +++ b/pkg/operator/ceph/object/rgw.go @@ -21,6 +21,7 @@ import ( "fmt" "strings" + "github.com/banzaicloud/k8s-objectmatcher/patch" "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/pkg/clusterd" @@ -140,6 +141,13 @@ func (c *clusterConfig) startRGWPods() error { // Create deployment deployment := c.createDeployment(rgwConfig) logger.Infof("object store %s deployment %s started", c.store.Name, deployment.Name) + + // Set the deployment hash as an annotation + err = patch.DefaultAnnotator.SetLastAppliedAnnotation(deployment) + if err != nil { + return errors.Wrapf(err, "failed to set annotation for deployment %q", deployment.Name) + } + createdDeployment, createErr := c.context.Clientset.AppsV1().Deployments(c.store.Namespace).Create(deployment) if createErr != nil { if !kerrors.IsAlreadyExists(createErr) { diff --git a/pkg/operator/k8sutil/deployment.go b/pkg/operator/k8sutil/deployment.go index 7e6d2269d..01897c8b5 100644 --- a/pkg/operator/k8sutil/deployment.go +++ b/pkg/operator/k8sutil/deployment.go @@ -20,6 +20,8 @@ import ( "fmt" "time" + "github.com/banzaicloud/k8s-objectmatcher/patch" + "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/util" apps "k8s.io/api/apps/v1" @@ -58,55 +60,72 @@ func GetDeploymentSpecImage(clientset kubernetes.Interface, d apps.Deployment, c // 2. verify that we can continue the update procedure // Basically, we go one resource by one and check if we can stop and then if the resource has been successfully updated // we check if we can go ahead and move to the next one. -func UpdateDeploymentAndWait(context *clusterd.Context, deployment *apps.Deployment, namespace string, verifyCallback func(action string) error) (*v1.Deployment, error) { - original, err := context.Clientset.AppsV1().Deployments(namespace).Get(deployment.Name, metav1.GetOptions{}) +func UpdateDeploymentAndWait(context *clusterd.Context, modifiedDeployment *apps.Deployment, namespace string, verifyCallback func(action string) error) (*v1.Deployment, error) { + currentDeployment, err := context.Clientset.AppsV1().Deployments(namespace).Get(modifiedDeployment.Name, metav1.GetOptions{}) if err != nil { - return nil, fmt.Errorf("failed to get deployment %s. %+v", deployment.Name, err) + return nil, fmt.Errorf("failed to get deployment %s. %+v", modifiedDeployment.Name, err) } - // Let's verify the deployment can be stopped - // retry for 5 times, every minute - err = util.Retry(5, 60*time.Second, func() error { - return verifyCallback("stop") - }) + // Check whether the current deployement and newly generated one are identical + patchResult, err := patch.DefaultPatchMaker.Calculate(currentDeployment, modifiedDeployment) if err != nil { - return nil, fmt.Errorf("failed to check if deployment %s can be updated: %+v", deployment.Name, err) + return nil, fmt.Errorf("failed to calculate diff between current deployment %q and newly generated one. %v", currentDeployment.Name, err) } - logger.Infof("updating deployment %s", deployment.Name) - if _, err := context.Clientset.AppsV1().Deployments(namespace).Update(deployment); err != nil { - return nil, fmt.Errorf("failed to update deployment %s. %+v", deployment.Name, err) - } - - // wait for the deployment to be restarted - sleepTime := 2 - attempts := 30 - if original.Spec.ProgressDeadlineSeconds != nil { - // make the attempts double the progress deadline since the pod is both stopping and starting - attempts = 2 * (int(*original.Spec.ProgressDeadlineSeconds) / sleepTime) - } - for i := 0; i < attempts; i++ { - // check for the status of the deployment - d, err := context.Clientset.AppsV1().Deployments(namespace).Get(deployment.Name, metav1.GetOptions{}) + // If deployments are different, let's update! + if !patchResult.IsEmpty() { + // Let's verify the deployment can be stopped + // retry for 5 times, every minute + err = util.Retry(5, 60*time.Second, func() error { + return verifyCallback("stop") + }) if err != nil { - return nil, fmt.Errorf("failed to get deployment %s. %+v", deployment.Name, err) + return nil, fmt.Errorf("failed to check if deployment %q can be updated. %v", modifiedDeployment.Name, err) } - if d.Status.ObservedGeneration != original.Status.ObservedGeneration && d.Status.UpdatedReplicas > 0 && d.Status.ReadyReplicas > 0 { - logger.Infof("finished waiting for updated deployment %s", d.Name) - // Now we check if we can go to the next daemon - err = verifyCallback("continue") + // Set hash annotation to the newly generated deployment + err := patch.DefaultAnnotator.SetLastAppliedAnnotation(modifiedDeployment) + if err != nil { + return nil, fmt.Errorf("failed to set hash annotation on deployment %q. %v", modifiedDeployment.Name, err) + } + + logger.Infof("updating deployment %q", modifiedDeployment.Name) + if _, err := context.Clientset.AppsV1().Deployments(namespace).Update(modifiedDeployment); err != nil { + return nil, fmt.Errorf("failed to update deployment %q. %v", modifiedDeployment.Name, err) + } + + // wait for the deployment to be restarted + sleepTime := 2 + attempts := 30 + if currentDeployment.Spec.ProgressDeadlineSeconds != nil { + // make the attempts double the progress deadline since the pod is both stopping and starting + attempts = 2 * (int(*currentDeployment.Spec.ProgressDeadlineSeconds) / sleepTime) + } + for i := 0; i < attempts; i++ { + // check for the status of the deployment + d, err := context.Clientset.AppsV1().Deployments(namespace).Get(modifiedDeployment.Name, metav1.GetOptions{}) if err != nil { - return nil, fmt.Errorf("failed to check if deployment %s can continue: %+v", deployment.Name, err) + return nil, fmt.Errorf("failed to get deployment %q. %v", modifiedDeployment.Name, err) } + if d.Status.ObservedGeneration != currentDeployment.Status.ObservedGeneration && d.Status.UpdatedReplicas > 0 && d.Status.ReadyReplicas > 0 { + logger.Infof("finished waiting for updated deployment %q", d.Name) - return d, nil + // Now we check if we can go to the next daemon + err = verifyCallback("continue") + if err != nil { + return nil, fmt.Errorf("failed to check if deployment %q can continue: %v", modifiedDeployment.Name, err) + } + + return d, nil + } + logger.Debugf("deployment %q status=%+v", d.Name, d.Status) + time.Sleep(time.Duration(sleepTime) * time.Second) } - logger.Debugf("deployment %s status=%+v", d.Name, d.Status) - time.Sleep(time.Duration(sleepTime) * time.Second) + return nil, fmt.Errorf("gave up waiting for deployment %q to update", modifiedDeployment.Name) } - return nil, fmt.Errorf("gave up waiting for deployment %s to update", deployment.Name) + logger.Infof("deployment %q did not change, nothing to update", currentDeployment.Name) + return nil, nil } // GetDeployments returns a list of deployment names labels matching a given selector