ceph: update rgw and mds deployment for logCollector

If the CephCluster CR spec is updated to activate the logCollector, we
must reflect that change onto child CRDs, like the mds and rgw since
their configurationn would be impacted too.
Now we watch for the CephCluster object changes from the object/file
controllers and react upon the appropriate event.

Closes: https://github.com/rook/rook/issues/7022
Signed-off-by: Sébastien Han <seb@redhat.com>

pkg/operator/ceph/cluster/controller.go

@@ -165,7 +165,7 @@ func add(mgr manager.Manager, r reconcile.Reconciler, context *clusterd.Context)
 
 	// Build Handler function to return the list of ceph clusters
 	// This is used by the watchers below
-	handerFunc, err := opcontroller.ObjectToCRMapper(mgr.GetClient(), &cephv1.CephClusterList{}, mgr.GetScheme())
+	handlerFunc, err := opcontroller.ObjectToCRMapper(mgr.GetClient(), &cephv1.CephClusterList{}, mgr.GetScheme())
 	if err != nil {
 		return err
 	}
@@ -180,7 +180,7 @@ func add(mgr manager.Manager, r reconcile.Reconciler, context *clusterd.Context)
 				},
 			},
 		},
-		handler.EnqueueRequestsFromMapFunc(handerFunc),
+		handler.EnqueueRequestsFromMapFunc(handlerFunc),
 		predicateForNodeWatcher(mgr.GetClient(), context))
 	if err != nil {
 		return err
@@ -200,7 +200,7 @@ func add(mgr manager.Manager, r reconcile.Reconciler, context *clusterd.Context)
 					},
 				},
 			},
-			handler.EnqueueRequestsFromMapFunc(handerFunc),
+			handler.EnqueueRequestsFromMapFunc(handlerFunc),
 			predicateForHotPlugCMWatcher(mgr.GetClient()))
 		if err != nil {
 			return err

pkg/operator/ceph/cluster/rbd/controller.go

@@ -128,6 +128,25 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 		}
 	}
 
+	// Build Handler function to return the list of ceph rbd-mirror
+	// This is used by the watchers below
+	handlerFunc, err := opcontroller.ObjectToCRMapper(mgr.GetClient(), &cephv1.CephRBDMirrorList{}, mgr.GetScheme())
+	if err != nil {
+		return err
+	}
+
+	// Watch for CephCluster Spec changes that we want to propagate to us
+	err = c.Watch(&source.Kind{Type: &cephv1.CephCluster{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       opcontroller.ClusterResource.Kind,
+			APIVersion: opcontroller.ClusterResource.APIVersion,
+		},
+	},
+	}, handler.EnqueueRequestsFromMapFunc(handlerFunc), opcontroller.WatchCephClusterPredicate())
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 

pkg/operator/ceph/controller/predicate.go

@@ -295,6 +295,59 @@ func WatchControllerPredicate() predicate.Funcs {
 	}
 }
 
+// WatchCephClusterPredicate is a predicate used by child controllers such as Filesystem or Object
+// It watch for CR changes on the CephCluster object and reconciles if this needs to be propagated
+// For instance the logCollector option from the CephCluster spec affects the configuration of rgw pods
+// So if it changes we must update the deployment
+func WatchCephClusterPredicate() predicate.Funcs {
+	return predicate.Funcs{
+		CreateFunc: func(e event.CreateEvent) bool {
+			logger.Debug("create event from a CR")
+			return true
+		},
+		DeleteFunc: func(e event.DeleteEvent) bool {
+			logger.Debug("delete event from a CR")
+			return true
+		},
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			logger.Debug("update event from a CR")
+			// resource.Quantity has non-exportable fields, so we use its comparator method
+			resourceQtyComparer := cmp.Comparer(func(x, y resource.Quantity) bool { return x.Cmp(y) == 0 })
+
+			switch objOld := e.ObjectOld.(type) {
+			case *cephv1.CephCluster:
+				objNew := e.ObjectNew.(*cephv1.CephCluster)
+				logger.Debug("update event on CephCluster CR")
+				// If the labels "do_not_reconcile" is set on the object, let's not reconcile that request
+				isDoNotReconcile := IsDoNotReconcile(objNew.GetLabels())
+				if isDoNotReconcile {
+					logger.Debugf("object %q matched on update but %q label is set, doing nothing", DoNotReconcileLabelName, objNew.Name)
+					return false
+				}
+				diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer)
+				if diff != "" {
+					// The image change (upgrade) is being taking care by watchControllerPredicate() in the cluster package
+					if objOld.Spec.CephVersion.Image != objNew.Spec.CephVersion.Image {
+						return false
+					}
+					// If the log collector setting changes let's reconcile the child controllers
+					if !cmp.Equal(objOld.Spec.LogCollector, objNew.Spec.LogCollector) {
+						logger.Debug("log collector option changed, reconciling")
+						return true
+					}
+				} else if objOld.GetGeneration() != objNew.GetGeneration() {
+					logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name)
+				}
+			}
+
+			return false
+		},
+		GenericFunc: func(e event.GenericEvent) bool {
+			return false
+		},
+	}
+}
+
 // objectChanged checks whether the object has been updated
 func objectChanged(oldObj, newObj runtime.Object, objectName string) (bool, error) {
 	var doReconcile bool

pkg/operator/ceph/file/controller.go

@@ -120,6 +120,25 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 		}
 	}
 
+	// Build Handler function to return the list of ceph filesystems
+	// This is used by the watchers below
+	handlerFunc, err := opcontroller.ObjectToCRMapper(mgr.GetClient(), &cephv1.CephFilesystemList{}, mgr.GetScheme())
+	if err != nil {
+		return err
+	}
+
+	// Watch for CephCluster Spec changes that we want to propagate to us
+	err = c.Watch(&source.Kind{Type: &cephv1.CephCluster{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       opcontroller.ClusterResource.Kind,
+			APIVersion: opcontroller.ClusterResource.APIVersion,
+		},
+	},
+	}, handler.EnqueueRequestsFromMapFunc(handlerFunc), opcontroller.WatchCephClusterPredicate())
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 

pkg/operator/ceph/object/controller.go

@@ -136,6 +136,25 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 		}
 	}
 
+	// Build Handler function to return the list of ceph object
+	// This is used by the watchers below
+	handlerFunc, err := opcontroller.ObjectToCRMapper(mgr.GetClient(), &cephv1.CephObjectStoreList{}, mgr.GetScheme())
+	if err != nil {
+		return err
+	}
+
+	// Watch for CephCluster Spec changes that we want to propagate to us
+	err = c.Watch(&source.Kind{Type: &cephv1.CephCluster{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       opcontroller.ClusterResource.Kind,
+			APIVersion: opcontroller.ClusterResource.APIVersion,
+		},
+	},
+	}, handler.EnqueueRequestsFromMapFunc(handlerFunc), opcontroller.WatchCephClusterPredicate())
+	if err != nil {
+		return err
+	}
+
 	return nil
 }