From c5075d55d16549de717216644f429769f997347e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Fri, 3 Apr 2020 12:10:06 +0200 Subject: [PATCH] ceph: trigger update on child controllers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the CephCluster CR gets updated with a new image version, we need to notify the child controllers of that upgrade. For this, we set a 'ceph_version' label on the CR itself which will effectively trigger a reconcile based on an update event. This needs to be revisited and see how we can make use of `EnqueueRequestsFromMapFunc` which might be a better approach using controller-runtime built-in. Closes: https://github.com/rook/rook/issues/5153 Signed-off-by: Sébastien Han --- pkg/operator/ceph/cluster/cluster.go | 68 +++++++++++++++++-- pkg/operator/ceph/controller/predicate.go | 42 +++++++++++- .../ceph/controller/predicate_test.go | 28 ++++++++ 3 files changed, 131 insertions(+), 7 deletions(-) diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index ce157e58c..ca3336bbe 100755 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/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 +} diff --git a/pkg/operator/ceph/controller/predicate.go b/pkg/operator/ceph/controller/predicate.go index 4686b3f92..a8894151d 100644 --- a/pkg/operator/ceph/controller/predicate.go +++ b/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 +} diff --git a/pkg/operator/ceph/controller/predicate_test.go b/pkg/operator/ceph/controller/predicate_test.go index 4b74934eb..26a436cc8 100644 --- a/pkg/operator/ceph/controller/predicate_test.go +++ b/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)) +}