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/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..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" @@ -79,7 +80,6 @@ type Cluster struct { rookVersion string exitCode func(err error) (int, bool) dataDirHostPath string - isUpgrade bool skipUpgradeChecks bool appliedHttpBind bool } @@ -100,7 +100,6 @@ func New( priorityClassName string, ownerRef metav1.OwnerReference, dataDirHostPath string, - isUpgrade bool, skipUpgradeChecks bool, ) *Cluster { return &Cluster{ @@ -122,7 +121,6 @@ func New( ownerRef: ownerRef, exitCode: getExitCode, dataDirHostPath: dataDirHostPath, - isUpgrade: isUpgrade, skipUpgradeChecks: skipUpgradeChecks, } } @@ -168,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) { @@ -179,21 +184,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..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" @@ -156,7 +157,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 +173,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 +186,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 +792,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) } @@ -850,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/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..12c10b0e0 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) @@ -352,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) } } @@ -364,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/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..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" @@ -89,7 +90,6 @@ type Cluster struct { priorityClassName string ownerRef metav1.OwnerReference kv *k8sutil.ConfigMapKVStore - isUpgrade bool skipUpgradeChecks bool continueUpgradeAfterChecksEvenIfNotHealthy bool } @@ -110,7 +110,6 @@ func New( prepareResources v1.ResourceRequirements, priorityClassName string, ownerRef metav1.OwnerReference, - isUpgrade bool, skipUpgradeChecks bool, continueUpgradeAfterChecksEvenIfNotHealthy bool, ) *Cluster { @@ -130,7 +129,6 @@ func New( priorityClassName: priorityClassName, ownerRef: ownerRef, kv: k8sutil.NewConfigMapKVStore(namespace, context.Clientset, ownerRef), - isUpgrade: isUpgrade, skipUpgradeChecks: skipUpgradeChecks, continueUpgradeAfterChecksEvenIfNotHealthy: continueUpgradeAfterChecksEvenIfNotHealthy, } @@ -486,21 +484,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) @@ -550,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) { @@ -575,21 +578,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..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" @@ -61,7 +62,6 @@ type Mirroring struct { rookVersion string Network cephv1.NetworkSpec dataDirHostPath string - isUpgrade bool skipUpgradeChecks bool } @@ -79,7 +79,6 @@ func New( priorityClassName string, ownerRef metav1.OwnerReference, dataDirHostPath string, - isUpgrade bool, skipUpgradeChecks bool, ) *Mirroring { return &Mirroring{ @@ -96,7 +95,6 @@ func New( priorityClassName: priorityClassName, ownerRef: ownerRef, dataDirHostPath: dataDirHostPath, - isUpgrade: isUpgrade, skipUpgradeChecks: skipUpgradeChecks, } } @@ -129,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) @@ -140,20 +145,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..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" @@ -61,7 +62,6 @@ type Cluster struct { fsID string ownerRef metav1.OwnerReference dataDirHostPath string - isUpgrade bool } type mdsConfig struct { @@ -80,7 +80,6 @@ func NewCluster( fsdetails *client.CephFilesystemDetails, ownerRef metav1.OwnerReference, dataDirHostPath string, - isUpgrade bool, ) *Cluster { return &Cluster{ clusterInfo: clusterInfo, @@ -91,7 +90,6 @@ func NewCluster( fsID: strconv.Itoa(fsdetails.ID), ownerRef: ownerRef, dataDirHostPath: dataDirHostPath, - isUpgrade: isUpgrade, } } @@ -159,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) { @@ -184,14 +189,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..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) { @@ -90,8 +99,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..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" @@ -43,7 +44,6 @@ type clusterConfig struct { clusterSpec *cephv1.ClusterSpec ownerRef metav1.OwnerReference DataPathMap *config.DataPathMap - isUpgrade bool skipUpgradeChecks bool } @@ -141,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) { @@ -172,16 +179,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/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 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 }