Merge pull request #4442 from leseb/refact-is-upgrade-2

ceph: improve kubernetes deployment upgrade detection

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",

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"

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

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

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 {

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)
 

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{

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

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

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

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

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

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

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)

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"

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{

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)

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

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

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",

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 {

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
 	}

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)
 

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

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",

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

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)

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

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

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

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

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
 	}