Merge pull request #5160 from leseb/upgrade-child

ceph: trigger update on child controllers

pkg/operator/ceph/cluster/cluster.go

@@ -20,6 +20,7 @@ package cluster
 import (
 	"reflect"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -66,12 +67,6 @@ type cluster struct {
 	isUpgrade            bool
 }
 
-// ChildController is implemented by CRs that are owned by the CephCluster
-type childController interface {
-	// ParentClusterChanged is called when the CephCluster CR is updated, for example for a newer ceph version
-	ParentClusterChanged(cluster cephv1.ClusterSpec, clusterInfo *cephconfig.ClusterInfo, isUpgrade bool)
-}
-
 func newCluster(c *cephv1.CephCluster, context *clusterd.Context, csiMutex *sync.Mutex) *cluster {
 	ownerRef := ClusterOwnerRef(c.Name, string(c.UID))
 	return &cluster{
@@ -277,6 +272,15 @@ func (c *cluster) doOrchestration(rookImage string, cephVersion cephver.CephVers
 			return errors.Wrapf(err, "failed to execute post actions after all the monitors started")
 		}
 
+		// If this is an upgrade, notify all the child controllers
+		if c.isUpgrade {
+			logger.Info("upgrade in progress, notifying child CRs")
+			err := c.notifyChildControllerOfUpgrade()
+			if err != nil {
+				return errors.Wrap(err, "failed to notify child CRs of upgrade")
+			}
+		}
+
 		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),
@@ -437,3 +441,55 @@ func (c *cluster) postMonStartupActions() error {
 
 	return nil
 }
+
+func (c *cluster) notifyChildControllerOfUpgrade() error {
+	version := strings.Replace(c.Info.CephVersion.String(), " ", "-", -1)
+
+	// List all child controllers
+	cephFilesystems, err := c.context.RookClientset.CephV1().CephFilesystems(c.Namespace).List(metav1.ListOptions{})
+	if err != nil {
+		return errors.Wrap(err, "failed to list ceph filesystem CRs")
+	}
+	for _, cephFilesystem := range cephFilesystems.Items {
+		if cephFilesystem.Labels == nil {
+			cephFilesystem.Labels = map[string]string{}
+		}
+		cephFilesystem.Labels["ceph_version"] = version
+		_, err := c.context.RookClientset.CephV1().CephFilesystems(c.Namespace).Update(&cephFilesystem)
+		if err != nil {
+			return errors.Wrapf(err, "failed to update ceph filesystem CR %q with new label", cephFilesystem.Name)
+		}
+	}
+
+	cephObjectStores, err := c.context.RookClientset.CephV1().CephObjectStores(c.Namespace).List(metav1.ListOptions{})
+	if err != nil {
+		return errors.Wrap(err, "failed to list ceph object store CRs")
+	}
+	for _, cephObjectStore := range cephObjectStores.Items {
+		if cephObjectStore.Labels == nil {
+			cephObjectStore.Labels = map[string]string{}
+		}
+		cephObjectStore.Labels["ceph_version"] = version
+		_, err := c.context.RookClientset.CephV1().CephObjectStores(c.Namespace).Update(&cephObjectStore)
+		if err != nil {
+			return errors.Wrapf(err, "failed to update ceph object store CR %q with new label", cephObjectStore.Name)
+		}
+	}
+
+	cephNFSes, err := c.context.RookClientset.CephV1().CephNFSes(c.Namespace).List(metav1.ListOptions{})
+	if err != nil {
+		return errors.Wrap(err, "failed to list ceph nfs CRs")
+	}
+	for _, cephNFS := range cephNFSes.Items {
+		if cephNFS.Labels == nil {
+			cephNFS.Labels = map[string]string{}
+		}
+		cephNFS.Labels["ceph_version"] = version
+		_, err := c.context.RookClientset.CephV1().CephNFSes(c.Namespace).Update(&cephNFS)
+		if err != nil {
+			return errors.Wrapf(err, "failed to update ceph nfs CR %q with new label", cephNFS.Name)
+		}
+	}
+
+	return nil
+}

pkg/operator/ceph/controller/predicate.go

@@ -31,6 +31,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
 
+const (
+	cephVersionLabelKey = "ceph_version"
+)
+
 // WatchControllerPredicate is a special update filter for update events
 // do not reconcile if the the status changes, this avoids a reconcile storm loop
 //
@@ -65,6 +69,11 @@ func WatchControllerPredicate() predicate.Funcs {
 					}
 					return true
 				}
+				// Handling upgrades
+				isUpgrade := isUpgrade(objOld.GetLabels(), objNew.GetLabels())
+				if isUpgrade {
+					return true
+				}
 
 			case *cephv1.CephObjectStoreUser:
 				objNew := e.ObjectNew.(*cephv1.CephObjectStoreUser)
@@ -107,6 +116,11 @@ func WatchControllerPredicate() predicate.Funcs {
 					}
 					return true
 				}
+				// Handling upgrades
+				isUpgrade := isUpgrade(objOld.GetLabels(), objNew.GetLabels())
+				if isUpgrade {
+					return true
+				}
 
 			case *cephv1.CephNFS:
 				objNew := e.ObjectNew.(*cephv1.CephNFS)
@@ -121,7 +135,11 @@ func WatchControllerPredicate() predicate.Funcs {
 					}
 					return true
 				}
-
+				// Handling upgrades
+				isUpgrade := isUpgrade(objOld.GetLabels(), objNew.GetLabels())
+				if isUpgrade {
+					return true
+				}
 			}
 
 			logger.Debug("wont update unknown object")
@@ -236,3 +254,25 @@ func isValidEvent(patch []byte) bool {
 	logger.Debugf("will reconcile based on patch %s", patchString)
 	return true
 }
+
+func isUpgrade(oldLabels, newLabels map[string]string) bool {
+	oldLabelVal, oldLabelKeyExist := oldLabels[cephVersionLabelKey]
+	newLabelVal, newLabelKeyExist := newLabels[cephVersionLabelKey]
+
+	// Nothing exists
+	if !oldLabelKeyExist && !newLabelKeyExist {
+		return false
+	}
+
+	// The new object has the label key so we reconcile
+	if !oldLabelKeyExist && newLabelKeyExist {
+		return true
+	}
+
+	// Both objects have the label and values are different so we reconcile
+	if (oldLabelKeyExist && newLabelKeyExist) && oldLabelVal != newLabelVal {
+		return true
+	}
+
+	return false
+}

pkg/operator/ceph/controller/predicate_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package controller
 
 import (
+	"fmt"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -74,3 +75,30 @@ func TestObjectChanged(t *testing.T) {
 	assert.NoError(t, err)
 	assert.True(t, changed)
 }
+
+func TestIsUpgrade(t *testing.T) {
+	oldLabel := make(map[string]string)
+	newLabel := map[string]string{
+		"foo": "bar",
+	}
+
+	// no value do nothing
+	b := isUpgrade(oldLabel, newLabel)
+	assert.False(t, b)
+
+	// different value do something
+	newLabel["ceph_version"] = "15.2.0-octopus"
+	b = isUpgrade(oldLabel, newLabel)
+	assert.True(t, b, fmt.Sprintf("%v,%v", oldLabel, newLabel))
+
+	// same value do nothing
+	oldLabel["ceph_version"] = "15.2.0-octopus"
+	newLabel["ceph_version"] = "15.2.0-octopus"
+	b = isUpgrade(oldLabel, newLabel)
+	assert.False(t, b, fmt.Sprintf("%v,%v", oldLabel, newLabel))
+
+	// different value do something
+	newLabel["ceph_version"] = "15.2.1-octopus"
+	b = isUpgrade(oldLabel, newLabel)
+	assert.True(t, b, fmt.Sprintf("%v,%v", oldLabel, newLabel))
+}