From 5fb622e6d134aedf120f1d3390846cd4f36011f6 Mon Sep 17 00:00:00 2001 From: Joshua Hoblitt Date: Mon, 31 Mar 2025 08:46:53 -0700 Subject: [PATCH] core: typed watch handlers and predicates All existing controller runtime watches are converted to use "typed" handlers and predicates instead of operating on `client.Object`. The intent is to be bug for bug equivalent with the existing logic while replacing run time type assertions and switch statements with compile time type constraints and type casts. In several cases, functions using assertions were split up such that each function only handles a single Kind at a time. It is hoped that this will improve readability and maintainability while facilitating future refactoring such as migrating some watches to using IndexFields. Of particular note is that the massive switch statement in `WatchControllerPredicate()` from `pkg/operator/ceph/controller/predicate.go` has been replaced with generics, reflection, and splitting the obc logic into its own predicate function. There are still many helper functions operating on `client.Object`. These were not updated unless required by the compiler in order to limit the size of this change. The type safety of these funcs should be improved as followup work. It is strongly suggested that going forward, handlers and predicates only handle a single Kind (generic or not) and that switches / type assertions are heavily discouraged or forbidden. This PR removed all but a single switch statement in a predicate, which should be addressed in future work. Signed-off-by: Joshua Hoblitt (cherry picked from commit 9f1ed201db3f0f0a0c9a2477a1837f78c952a7b4) --- pkg/operator/ceph/client/controller.go | 32 +- pkg/operator/ceph/cluster/controller.go | 76 ++- pkg/operator/ceph/cluster/mon/predicate.go | 22 +- pkg/operator/ceph/cluster/nodedaemon/add.go | 109 ++-- pkg/operator/ceph/cluster/predicate.go | 133 +++-- pkg/operator/ceph/cluster/predicate_test.go | 5 - pkg/operator/ceph/cluster/rbd/controller.go | 38 +- pkg/operator/ceph/controller.go | 12 +- pkg/operator/ceph/controller/handler.go | 21 +- pkg/operator/ceph/controller/handler_test.go | 2 +- pkg/operator/ceph/controller/predicate.go | 480 ++++-------------- .../ceph/controller/predicate_test.go | 3 - pkg/operator/ceph/csi/controller.go | 26 +- pkg/operator/ceph/csi/predicate.go | 133 ++--- pkg/operator/ceph/csi/predicate_test.go | 152 +++--- .../ceph/disruption/clusterdisruption/add.go | 171 ++++--- pkg/operator/ceph/file/controller.go | 45 +- pkg/operator/ceph/file/mirror/controller.go | 31 +- .../ceph/file/subvolumegroup/controller.go | 9 +- pkg/operator/ceph/nfs/controller.go | 32 +- pkg/operator/ceph/object/bucket/controller.go | 23 +- pkg/operator/ceph/object/bucket/predicate.go | 73 +-- pkg/operator/ceph/object/controller.go | 83 +-- pkg/operator/ceph/object/cosi/controller.go | 18 +- .../ceph/object/notification/controller.go | 9 +- .../notification/obc_label_controller.go | 53 +- pkg/operator/ceph/object/realm/controller.go | 9 +- pkg/operator/ceph/object/topic/controller.go | 9 +- pkg/operator/ceph/object/user/controller.go | 28 +- pkg/operator/ceph/object/zone/controller.go | 9 +- .../ceph/object/zonegroup/controller.go | 9 +- pkg/operator/ceph/pool/controller.go | 26 +- .../ceph/pool/radosnamespace/controller.go | 9 +- pkg/operator/ceph/predicate.go | 67 ++- 34 files changed, 954 insertions(+), 1003 deletions(-) diff --git a/pkg/operator/ceph/client/controller.go b/pkg/operator/ceph/client/controller.go index 2543764f8..fcf54bf3a 100644 --- a/pkg/operator/ceph/client/controller.go +++ b/pkg/operator/ceph/client/controller.go @@ -98,25 +98,31 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the CephClient CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephClient{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephClient{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephClient]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephClient](mgr.GetScheme()), + ), + ) if err != nil { return err } // Watch secrets - - ownerRequest := handler.EnqueueRequestForOwner( - mgr.GetScheme(), - mgr.GetRESTMapper(), - &cephv1.CephClient{}, + err = c.Watch( + source.Kind( + mgr.GetCache(), + &v1.Secret{TypeMeta: metav1.TypeMeta{Kind: "Secret", APIVersion: v1.SchemeGroupVersion.String()}}, + handler.TypedEnqueueRequestForOwner[*v1.Secret]( + mgr.GetScheme(), + mgr.GetRESTMapper(), + &cephv1.CephClient{}, + ), + opcontroller.WatchPredicateForNonCRDObject[*v1.Secret](&cephv1.CephClient{TypeMeta: controllerTypeMeta}, mgr.GetScheme()), + ), ) - secretSource := source.Kind[client.Object]( - mgr.GetCache(), - &v1.Secret{TypeMeta: metav1.TypeMeta{Kind: "Secret", APIVersion: v1.SchemeGroupVersion.String()}}, - ownerRequest, - opcontroller.WatchPredicateForNonCRDObject(&cephv1.CephClient{TypeMeta: controllerTypeMeta}, mgr.GetScheme()), - ) - err = c.Watch(secretSource) if err != nil { return err } diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go index 4d3797aaa..75b310a58 100644 --- a/pkg/operator/ceph/cluster/controller.go +++ b/pkg/operator/ceph/cluster/controller.go @@ -127,6 +127,21 @@ func newReconciler(mgr manager.Manager, ctx *clusterd.Context, clusterController } } +func watchOwnedCoreObject[T client.Object](c controller.Controller, mgr manager.Manager, obj T) error { + return c.Watch( + source.Kind( + mgr.GetCache(), + obj, + handler.TypedEnqueueRequestForOwner[T]( + mgr.GetScheme(), + mgr.GetRESTMapper(), + &cephv1.CephCluster{}, + ), + opcontroller.WatchPredicateForNonCRDObject[T](&cephv1.CephCluster{TypeMeta: ControllerTypeMeta}, mgr.GetScheme()), + ), + ) +} + func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reconciler, context *clusterd.Context, opConfig opcontroller.OperatorConfig) error { // Create a new controller c, err := controller.New(controllerName, mgr, controller.Options{Reconciler: r}) @@ -146,22 +161,21 @@ func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reco } // Watch for changes on the CephCluster CR object - s := source.Kind[client.Object]( - mgr.GetCache(), &cephv1.CephCluster{TypeMeta: ControllerTypeMeta}, - &handler.EnqueueRequestForObject{}, watchControllerPredicate(opManagerContext, mgr.GetClient()), + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephCluster{TypeMeta: ControllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephCluster]{}, + watchControllerPredicate(opManagerContext, mgr.GetClient()), + ), ) - err = c.Watch(s) if err != nil { return err } // Watch all other resources of the Ceph Cluster for _, t := range objectsToWatch { - err = c.Watch( - source.Kind[client.Object]( - mgr.GetCache(), t, handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &cephv1.CephCluster{}), - opcontroller.WatchPredicateForNonCRDObject(&cephv1.CephCluster{TypeMeta: ControllerTypeMeta}, mgr.GetScheme())), - ) + err = watchOwnedCoreObject(c, mgr, t) if err != nil { return err } @@ -169,18 +183,35 @@ func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reco // Build Handler function to return the list of ceph clusters // This is used by the watchers below - handlerFunc, err := opcontroller.ObjectToCRMapper(opManagerContext, mgr.GetClient(), &cephv1.CephClusterList{}, mgr.GetScheme()) + nodeHandler, err := opcontroller.ObjectToCRMapper[*cephv1.CephClusterList, *corev1.Node]( + opManagerContext, + mgr.GetClient(), + &cephv1.CephClusterList{}, + mgr.GetScheme(), + ) if err != nil { return err } // Watch for nodes additions and updates - nodeKind := source.Kind[client.Object]( - mgr.GetCache(), - &corev1.Node{TypeMeta: metav1.TypeMeta{Kind: "Node", APIVersion: corev1.SchemeGroupVersion.String()}}, - handler.EnqueueRequestsFromMapFunc(handlerFunc), - predicateForNodeWatcher(opManagerContext, mgr.GetClient(), context, opConfig.OperatorNamespace)) - err = c.Watch(nodeKind) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &corev1.Node{TypeMeta: metav1.TypeMeta{Kind: "Node", APIVersion: corev1.SchemeGroupVersion.String()}}, + handler.TypedEnqueueRequestsFromMapFunc(nodeHandler), + predicateForNodeWatcher(opManagerContext, mgr.GetClient(), context, opConfig.OperatorNamespace), + ), + ) + if err != nil { + return err + } + + cmHandler, err := opcontroller.ObjectToCRMapper[*cephv1.CephClusterList, *corev1.ConfigMap]( + opManagerContext, + mgr.GetClient(), + &cephv1.CephClusterList{}, + mgr.GetScheme(), + ) if err != nil { return err } @@ -190,11 +221,14 @@ func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reco disableVal := k8sutil.GetOperatorSetting(disableHotplugEnv, "false") if disableVal != "true" { logger.Info("enabling hotplug orchestration") - s := source.Kind[client.Object]( - mgr.GetCache(), - &corev1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: corev1.SchemeGroupVersion.String()}}, - handler.EnqueueRequestsFromMapFunc(handlerFunc), predicateForHotPlugCMWatcher(mgr.GetClient())) - err = c.Watch(s) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &corev1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: corev1.SchemeGroupVersion.String()}}, + handler.TypedEnqueueRequestsFromMapFunc(cmHandler), + predicateForHotPlugCMWatcher(mgr.GetClient()), + ), + ) if err != nil { return err } diff --git a/pkg/operator/ceph/cluster/mon/predicate.go b/pkg/operator/ceph/cluster/mon/predicate.go index 023e92ec9..c3438f0ce 100644 --- a/pkg/operator/ceph/cluster/mon/predicate.go +++ b/pkg/operator/ceph/cluster/mon/predicate.go @@ -27,26 +27,20 @@ import ( "sigs.k8s.io/controller-runtime/pkg/predicate" ) -func PredicateMonEndpointChanges() predicate.Funcs { - return predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { +func PredicateMonEndpointChanges[T *corev1.ConfigMap]() predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { return false }, - DeleteFunc: func(e event.DeleteEvent) bool { + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { return false }, - GenericFunc: func(e event.GenericEvent) bool { + GenericFunc: func(e event.TypedGenericEvent[T]) bool { return false }, - UpdateFunc: func(e event.UpdateEvent) bool { - cmNew, ok := e.ObjectNew.(*corev1.ConfigMap) - if !ok { - return false - } - cmOld, ok := e.ObjectOld.(*corev1.ConfigMap) - if !ok { - return false - } + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + cmOld := (*corev1.ConfigMap)(e.ObjectOld) + cmNew := (*corev1.ConfigMap)(e.ObjectNew) if cmNew.GetName() == EndpointConfigMapName { if wereMonEndpointsUpdated(cmOld.Data, cmNew.Data) { logger.Info("monitor endpoints changed, updating the bootstrap peer token") diff --git a/pkg/operator/ceph/cluster/nodedaemon/add.go b/pkg/operator/ceph/cluster/nodedaemon/add.go index 325114cb9..b2b8c8886 100644 --- a/pkg/operator/ceph/cluster/nodedaemon/add.go +++ b/pkg/operator/ceph/cluster/nodedaemon/add.go @@ -29,7 +29,6 @@ import ( appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/types" - "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" @@ -74,21 +73,23 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes to the nodes - specChangePredicate := predicate.Funcs{ - UpdateFunc: func(e event.UpdateEvent) bool { - nodeOld, ok := e.ObjectOld.DeepCopyObject().(*corev1.Node) - if !ok { - return false - } - nodeNew, ok := e.ObjectNew.DeepCopyObject().(*corev1.Node) - if !ok { - return false - } + specChangePredicate := predicate.TypedFuncs[*corev1.Node]{ + UpdateFunc: func(e event.TypedUpdateEvent[*corev1.Node]) bool { + nodeOld := (*corev1.Node)(e.ObjectOld) + nodeNew := (*corev1.Node)(e.ObjectNew) + return !reflect.DeepEqual(nodeOld.Spec, nodeNew.Spec) }, } logger.Debugf("watch for changes to the nodes") - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &corev1.Node{}, &handler.EnqueueRequestForObject{}, specChangePredicate)) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &corev1.Node{}, + &handler.TypedEnqueueRequestForObject[*corev1.Node]{}, + specChangePredicate, + ), + ) if err != nil { return errors.Wrap(err, "failed to watch for node changes") } @@ -96,24 +97,23 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { // Watch for changes to the ceph-crash deployments logger.Debugf("watch for changes to the ceph-crash deployments") err = c.Watch( - source.Kind[client.Object](mgr.GetCache(), &appsv1.Deployment{}, - handler.EnqueueRequestsFromMapFunc(handler.MapFunc(func(context context.Context, obj client.Object) []reconcile.Request { - deployment, ok := obj.(*appsv1.Deployment) - if !ok { - return []reconcile.Request{} - } - labels := deployment.GetLabels() - appName, ok := labels[k8sutil.AppAttr] - if !ok || appName != CrashCollectorAppName { - return []reconcile.Request{} - } - nodeName, ok := deployment.Spec.Template.ObjectMeta.Labels[NodeNameLabel] - if !ok { - return []reconcile.Request{} - } - req := reconcile.Request{NamespacedName: types.NamespacedName{Name: nodeName}} - return []reconcile.Request{req} - }), + source.Kind( + mgr.GetCache(), + &appsv1.Deployment{}, + handler.TypedEnqueueRequestsFromMapFunc( + func(context context.Context, deployment *appsv1.Deployment) []reconcile.Request { + labels := deployment.GetLabels() + appName, ok := labels[k8sutil.AppAttr] + if !ok || appName != CrashCollectorAppName { + return []reconcile.Request{} + } + nodeName, ok := deployment.Spec.Template.ObjectMeta.Labels[NodeNameLabel] + if !ok { + return []reconcile.Request{} + } + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: nodeName}} + return []reconcile.Request{req} + }, ), ), ) @@ -124,39 +124,30 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { // Watch for changes to the ceph pods and enqueue their nodes logger.Debugf("watch for changes to the ceph pods and enqueue their nodes") err = c.Watch( - source.Kind[client.Object](mgr.GetCache(), &corev1.Pod{}, - handler.EnqueueRequestsFromMapFunc(handler.MapFunc(func(context context.Context, obj client.Object) []reconcile.Request { - pod, ok := obj.(*corev1.Pod) - if !ok { + source.Kind( + mgr.GetCache(), + &corev1.Pod{}, + handler.TypedEnqueueRequestsFromMapFunc[*corev1.Pod]( + func(context context.Context, pod *corev1.Pod) []reconcile.Request { + nodeName := pod.Spec.NodeName + if nodeName == "" { + return []reconcile.Request{} + } + if isCephPod(pod.Labels, pod.Name) { + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: nodeName}} + return []reconcile.Request{req} + } return []reconcile.Request{} - } - nodeName := pod.Spec.NodeName - if nodeName == "" { - return []reconcile.Request{} - } - if isCephPod(pod.Labels, pod.Name) { - req := reconcile.Request{NamespacedName: types.NamespacedName{Name: nodeName}} - return []reconcile.Request{req} - } - return []reconcile.Request{} - }), + }, ), // only enqueue the update event if the pod moved nodes - predicate.Funcs{ - UpdateFunc: func(event event.UpdateEvent) bool { - oldPod, ok := event.ObjectOld.(*corev1.Pod) - if !ok { - return false - } - newPod, ok := event.ObjectNew.(*corev1.Pod) - if !ok { - return false - } + predicate.TypedFuncs[*corev1.Pod]{ + UpdateFunc: func(event event.TypedUpdateEvent[*corev1.Pod]) bool { + oldPod := (*corev1.Pod)(event.ObjectOld) + newPod := (*corev1.Pod)(event.ObjectNew) + // only enqueue if the nodename has changed - if oldPod.Spec.NodeName == newPod.Spec.NodeName { - return false - } - return true + return oldPod.Spec.NodeName != newPod.Spec.NodeName }, }, ), diff --git a/pkg/operator/ceph/cluster/predicate.go b/pkg/operator/ceph/cluster/predicate.go index 709b5b6bf..2da2439df 100644 --- a/pkg/operator/ceph/cluster/predicate.go +++ b/pkg/operator/ceph/cluster/predicate.go @@ -30,7 +30,6 @@ import ( corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -52,75 +51,71 @@ func shouldReconcileChangedNode(objOld, objNew *corev1.Node) bool { } // predicateForNodeWatcher is the predicate function to trigger reconcile on Node events -func predicateForNodeWatcher(ctx context.Context, client client.Client, context *clusterd.Context, opNamespace string) predicate.Funcs { - return predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { - clientCluster := newClientCluster(client, e.Object.GetNamespace(), context) - return clientCluster.onK8sNode(ctx, e.Object, opNamespace) +func predicateForNodeWatcher[T *corev1.Node](ctx context.Context, client client.Client, context *clusterd.Context, opNamespace string) predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { + obj := (*corev1.Node)(e.Object) + + clientCluster := newClientCluster(client, obj.GetNamespace(), context) + return clientCluster.onK8sNode(ctx, obj, opNamespace) }, - UpdateFunc: func(e event.UpdateEvent) bool { - if objOld, ok := e.ObjectOld.(*corev1.Node); ok { - if objNew, ok := e.ObjectNew.(*corev1.Node); ok { - if !shouldReconcileChangedNode(objOld, objNew) { - return false - } + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + objOld := (*corev1.Node)(e.ObjectOld) + objNew := (*corev1.Node)(e.ObjectNew) - clientCluster := newClientCluster(client, e.ObjectNew.GetNamespace(), context) - return clientCluster.onK8sNode(ctx, e.ObjectNew, opNamespace) - } + if !shouldReconcileChangedNode(objOld, objNew) { + return false } + + clientCluster := newClientCluster(client, objNew.GetNamespace(), context) + return clientCluster.onK8sNode(ctx, objNew, opNamespace) + }, + + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { return false }, - DeleteFunc: func(e event.DeleteEvent) bool { - return false - }, - - GenericFunc: func(e event.GenericEvent) bool { + GenericFunc: func(e event.TypedGenericEvent[T]) bool { return false }, } } // predicateForHotPlugCMWatcher is the predicate function to trigger reconcile on ConfigMap events (hot-plug) -func predicateForHotPlugCMWatcher(client client.Client) predicate.Funcs { - return predicate.Funcs{ - UpdateFunc: func(e event.UpdateEvent) bool { - isHotPlugCM := isHotPlugCM(e.ObjectNew) - if !isHotPlugCM { +func predicateForHotPlugCMWatcher[T *corev1.ConfigMap](client client.Client) predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + objOld := (*corev1.ConfigMap)(e.ObjectOld) + objNew := (*corev1.ConfigMap)(e.ObjectNew) + + if !isHotPlugCM(objNew) { return false } - clientCluster := newClientCluster(client, e.ObjectNew.GetNamespace(), &clusterd.Context{}) - return clientCluster.onDeviceCMUpdate(e.ObjectOld, e.ObjectNew) + clientCluster := newClientCluster(client, objNew.GetNamespace(), &clusterd.Context{}) + return clientCluster.onDeviceCMUpdate(objOld, objNew) }, - DeleteFunc: func(e event.DeleteEvent) bool { + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { // TODO: if the configmap goes away we could retrigger rook-discover DS // However at this point the returned bool can only trigger a reconcile of the CephCluster object // Definitely non-trivial but nice to have in the future return false }, - CreateFunc: func(e event.CreateEvent) bool { + CreateFunc: func(e event.TypedCreateEvent[T]) bool { return false }, - GenericFunc: func(e event.GenericEvent) bool { + GenericFunc: func(e event.TypedGenericEvent[T]) bool { return false }, } } // isHotPlugCM informs whether the object is the cm for hot-plug disk -func isHotPlugCM(obj runtime.Object) bool { - // If not a ConfigMap, let's not reconcile - cm, ok := obj.(*corev1.ConfigMap) - if !ok { - return false - } - +func isHotPlugCM(cm *corev1.ConfigMap) bool { // Get the labels labels := cm.GetLabels() @@ -132,20 +127,20 @@ func isHotPlugCM(obj runtime.Object) bool { return false } -func watchControllerPredicate(ctx context.Context, c client.Client) predicate.Funcs { - return predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { +func watchControllerPredicate[T *cephv1.CephCluster](ctx context.Context, c client.Client) predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { if controller.DuplicateCephClusters(ctx, c, e.Object, true) { return false } logger.Debug("create event from a CR") return true }, - DeleteFunc: func(e event.DeleteEvent) bool { + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { logger.Debug("delete event from a CR") return true }, - UpdateFunc: func(e event.UpdateEvent) bool { + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { // We still need to check on update event since the user must delete the additional CR // Until this is done, the user can still update the CR and the operator will reconcile // This should not happen @@ -156,46 +151,44 @@ func watchControllerPredicate(ctx context.Context, c client.Client) predicate.Fu // 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 := controller.IsDoNotReconcile(objNew.GetLabels()) - if isDoNotReconcile { - logger.Debugf("object %q matched on update but %q label is set, doing nothing", controller.DoNotReconcileLabelName, objNew.Name) + objOld := (*cephv1.CephCluster)(e.ObjectOld) + objNew := (*cephv1.CephCluster)(e.ObjectNew) + + logger.Debug("update event on CephCluster CR") + // If the labels "do_not_reconcile" is set on the object, let's not reconcile that request + if controller.IsDoNotReconcile(objNew.GetLabels()) { + logger.Debugf("object %q matched on update but %q label is set, doing nothing", controller.DoNotReconcileLabelName, objNew.Name) + return false + } + diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer) + if diff != "" { + logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) + + if objNew.Spec.CleanupPolicy.HasDataDirCleanPolicy() { + logger.Infof("skipping orchestration for cluster object %q in namespace %q because its cleanup policy is set. not reloading the manager", objNew.GetName(), objNew.GetNamespace()) return false } - diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer) - if diff != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - if objNew.Spec.CleanupPolicy.HasDataDirCleanPolicy() { - logger.Infof("skipping orchestration for cluster object %q in namespace %q because its cleanup policy is set. not reloading the manager", objNew.GetName(), objNew.GetNamespace()) - return false - } + // Stop any ongoing orchestration + controller.ReloadManager() - // Stop any ongoing orchestration - controller.ReloadManager() + return false - return false + } else if !objOld.GetDeletionTimestamp().Equal(objNew.GetDeletionTimestamp()) { + logger.Infof("CR %q is going be deleted, cancelling any ongoing orchestration", objNew.Name) - } else if !objOld.GetDeletionTimestamp().Equal(objNew.GetDeletionTimestamp()) { - logger.Infof("CR %q is going be deleted, cancelling any ongoing orchestration", objNew.Name) + // Stop any ongoing orchestration + controller.ReloadManager() - // Stop any ongoing orchestration - controller.ReloadManager() + return false - return false - - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } + } 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 { + GenericFunc: func(e event.TypedGenericEvent[T]) bool { return false }, } diff --git a/pkg/operator/ceph/cluster/predicate_test.go b/pkg/operator/ceph/cluster/predicate_test.go index ce760d42a..dfbbf8c7d 100644 --- a/pkg/operator/ceph/cluster/predicate_test.go +++ b/pkg/operator/ceph/cluster/predicate_test.go @@ -20,17 +20,12 @@ import ( "testing" "time" - cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/stretchr/testify/assert" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) func TestIsHotPlugCM(t *testing.T) { - blockPool := &cephv1.CephBlockPool{} - - assert.False(t, isHotPlugCM(blockPool)) - cm := &corev1.ConfigMap{} assert.False(t, isHotPlugCM(cm)) diff --git a/pkg/operator/ceph/cluster/rbd/controller.go b/pkg/operator/ceph/cluster/rbd/controller.go index 5e2d14d85..051b223bb 100644 --- a/pkg/operator/ceph/cluster/rbd/controller.go +++ b/pkg/operator/ceph/cluster/rbd/controller.go @@ -107,6 +107,21 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont } } +func watchOwnedCoreObject[T client.Object](c controller.Controller, mgr manager.Manager, obj T) error { + return c.Watch( + source.Kind( + mgr.GetCache(), + obj, + handler.TypedEnqueueRequestForOwner[T]( + mgr.GetScheme(), + mgr.GetRESTMapper(), + &cephv1.CephRBDMirror{}, + ), + opcontroller.WatchPredicateForNonCRDObject[T](&cephv1.CephRBDMirror{TypeMeta: controllerTypeMeta}, mgr.GetScheme()), + ), + ) +} + func add(mgr manager.Manager, r reconcile.Reconciler) error { // Create a new controller c, err := controller.New(controllerName, mgr, controller.Options{Reconciler: r}) @@ -116,26 +131,21 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the cephRBDMirror CRD object - s := source.Kind[client.Object]( - mgr.GetCache(), &cephv1.CephRBDMirror{TypeMeta: controllerTypeMeta}, - &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate()) - err = c.Watch(s) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephRBDMirror{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephRBDMirror]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephRBDMirror](mgr.GetScheme()), + ), + ) if err != nil { return err } // Watch all other resources for _, t := range objectsToWatch { - ownerRequest := handler.EnqueueRequestForOwner( - mgr.GetScheme(), - mgr.GetRESTMapper(), - &cephv1.CephRBDMirror{}, - ) - err = c.Watch( - source.Kind[client.Object]( - mgr.GetCache(), t, ownerRequest, - opcontroller.WatchPredicateForNonCRDObject(&cephv1.CephRBDMirror{TypeMeta: controllerTypeMeta}, mgr.GetScheme())), - ) + err = watchOwnedCoreObject(c, mgr, t) if err != nil { return err } diff --git a/pkg/operator/ceph/controller.go b/pkg/operator/ceph/controller.go index 5aa86f336..19d1f206d 100644 --- a/pkg/operator/ceph/controller.go +++ b/pkg/operator/ceph/controller.go @@ -73,12 +73,14 @@ func add(ctx context.Context, context *clusterd.Context, mgr manager.Manager, r logger.Infof("%s successfully started", controllerName) // Watch for ConfigMap (operator config) - s := source.Kind[client.Object]( - mgr.GetCache(), - &v1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: v1.SchemeGroupVersion.String()}}, - &handler.EnqueueRequestForObject{}, predicateController(ctx, mgr.GetClient()), + err = c.Watch( + source.Kind( + mgr.GetCache(), + &v1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: v1.SchemeGroupVersion.String()}}, + &handler.TypedEnqueueRequestForObject[*v1.ConfigMap]{}, + operatorSettingConfigMapPredicate(), + ), ) - err = c.Watch(s) if err != nil { return err } diff --git a/pkg/operator/ceph/controller/handler.go b/pkg/operator/ceph/controller/handler.go index 17eff08c0..462bd4177 100644 --- a/pkg/operator/ceph/controller/handler.go +++ b/pkg/operator/ceph/controller/handler.go @@ -19,32 +19,25 @@ package controller import ( "context" - "github.com/pkg/errors" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" - ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/apiutil" "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/reconcile" ) // ObjectToCRMapper returns the list of a given object type metadata // It is used to trigger a reconcile object Kind A when watching object Kind B // So we reconcile Kind A instead of Kind B // For instance, we watch for CephCluster CR changes but want to reconcile CephFilesystem based on a Spec change -func ObjectToCRMapper(ctx context.Context, c client.Client, ro runtime.Object, scheme *runtime.Scheme) (handler.MapFunc, error) { - if _, ok := ro.(metav1.ListInterface); !ok { - return nil, errors.Errorf("expected a metav1.ListInterface, got %T instead", ro) - } - - gvk, err := apiutil.GVKForObject(ro, scheme) +func ObjectToCRMapper[List client.ObjectList, T runtime.Object](ctx context.Context, c client.Client, list List, scheme *runtime.Scheme) (handler.TypedMapFunc[T, reconcile.Request], error) { + gvk, err := apiutil.GVKForObject(list, scheme) if err != nil { return nil, err } - // return handler.EnqueueRequestsFromMapFunc(func(o client.Object) []ctrl.Request { - return handler.MapFunc(func(ctx context.Context, o client.Object) []ctrl.Request { + return func(ctx context.Context, obj T) []reconcile.Request { list := &unstructured.UnstructuredList{} list.SetGroupVersionKind(gvk) err := c.List(ctx, list) @@ -52,9 +45,9 @@ func ObjectToCRMapper(ctx context.Context, c client.Client, ro runtime.Object, s return nil } - results := []ctrl.Request{} + results := []reconcile.Request{} for _, obj := range list.Items { - results = append(results, ctrl.Request{ + results = append(results, reconcile.Request{ NamespacedName: client.ObjectKey{ Namespace: obj.GetNamespace(), Name: obj.GetName(), @@ -62,5 +55,5 @@ func ObjectToCRMapper(ctx context.Context, c client.Client, ro runtime.Object, s }) } return results - }), nil + }, nil } diff --git a/pkg/operator/ceph/controller/handler_test.go b/pkg/operator/ceph/controller/handler_test.go index 830201914..0701b9e6b 100644 --- a/pkg/operator/ceph/controller/handler_test.go +++ b/pkg/operator/ceph/controller/handler_test.go @@ -62,7 +62,7 @@ func TestObjectToCRMapper(t *testing.T) { {NamespacedName: client.ObjectKey{Name: "my-pool", Namespace: "rook-ceph"}}, } - handlerFunc, err := ObjectToCRMapper(context.TODO(), cl, objects[0], s) + handlerFunc, err := ObjectToCRMapper[*cephv1.CephFilesystemList, runtime.Object](context.TODO(), cl, &cephv1.CephFilesystemList{}, s) assert.NoError(t, err) assert.ElementsMatch(t, fakeRequest, handlerFunc(context.TODO(), fs)) } diff --git a/pkg/operator/ceph/controller/predicate.go b/pkg/operator/ceph/controller/predicate.go index 3ee7ac115..cf51f2332 100644 --- a/pkg/operator/ceph/controller/predicate.go +++ b/pkg/operator/ceph/controller/predicate.go @@ -19,7 +19,6 @@ package controller import ( "context" "encoding/json" - "fmt" "os" "reflect" "strings" @@ -27,7 +26,6 @@ import ( "github.com/banzaicloud/k8s-objectmatcher/patch" "github.com/google/go-cmp/cmp" - bktv1alpha1 "github.com/kube-object-storage/lib-bucket-provisioner/pkg/apis/objectbucket.io/v1alpha1" "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/pkg/operator/ceph/config" @@ -37,7 +35,9 @@ import ( "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/api/resource" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/apiutil" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/predicate" ) @@ -46,372 +46,71 @@ const ( DoNotReconcileLabelName = "do_not_reconcile" ) +// Retrieve the GVK (GroupVersionKind) and NamespacedName for the given object. +// Primarily used for logging. +func objectInfo(scheme *runtime.Scheme, obj client.Object) (string, types.NamespacedName) { + gvk, err := apiutil.GVKForObject(obj, scheme) + if err != nil { + logger.Debugf("Unable to get GVK for object %+v", obj) + } + + nsName := types.NamespacedName{ + Namespace: obj.GetNamespace(), + Name: obj.GetName(), + } + + return gvk.Kind, nsName +} + // WatchControllerPredicate is a special update filter for update events // do not reconcile if the status changes, this avoids a reconcile storm loop // // returning 'true' means triggering a reconciliation // returning 'false' means do NOT trigger a reconciliation -func WatchControllerPredicate() predicate.Funcs { - return predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { - logger.Debugf("create event from a CR: %q", e.Object.GetName()) +func WatchControllerPredicate[T client.Object](scheme *runtime.Scheme) predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { + kind, nsName := objectInfo(scheme, e.Object) + logger.Debugf("create event for %q: %q", kind, nsName) return true }, - DeleteFunc: func(e event.DeleteEvent) bool { - logger.Debugf("delete event from a CR: %q", e.Object.GetName()) + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { + kind, nsName := objectInfo(scheme, e.Object) + logger.Debugf("delete event for %q: %q", kind, nsName) return true }, - UpdateFunc: func(e event.UpdateEvent) bool { - logger.Debugf("update event from a CR: %q", e.ObjectOld.GetName()) + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + objOld := e.ObjectOld + objNew := e.ObjectNew + kind, nsName := objectInfo(scheme, objNew) + + logger.Debugf("update event for %q: %q", kind, nsName) + + // 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("resource %q: %q had update event but %q label is set, doing nothing", kind, nsName, DoNotReconcileLabelName) + return false + } + // 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.CephObjectStore: - objNew := e.ObjectNew.(*cephv1.CephObjectStore) - logger.Debug("update event on CephObjectStore 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 != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephObjectStoreUser: - objNew := e.ObjectNew.(*cephv1.CephObjectStoreUser) - logger.Debug("update event on CephObjectStoreUser 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 != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephObjectRealm: - objNew := e.ObjectNew.(*cephv1.CephObjectRealm) - logger.Debug("update event on CephObjectRealm") - // 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 != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephObjectZoneGroup: - objNew := e.ObjectNew.(*cephv1.CephObjectZoneGroup) - logger.Debug("update event on CephObjectZoneGroup") - // 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 != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephObjectZone: - objNew := e.ObjectNew.(*cephv1.CephObjectZone) - logger.Debug("update event on CephObjectZone") - // 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 != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephBlockPool: - objNew := e.ObjectNew.(*cephv1.CephBlockPool) - logger.Debug("update event on CephBlockPool 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 != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephFilesystem: - objNew := e.ObjectNew.(*cephv1.CephFilesystem) - logger.Debug("update event on CephFilesystem 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 != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephNFS: - objNew := e.ObjectNew.(*cephv1.CephNFS) - logger.Debug("update event on CephNFS 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 != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephRBDMirror: - objNew := e.ObjectNew.(*cephv1.CephRBDMirror) - logger.Debug("update event on CephRBDMirror 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 != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephClient: - objNew := e.ObjectNew.(*cephv1.CephClient) - logger.Debug("update event on CephClient 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 != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephFilesystemMirror: - objNew := e.ObjectNew.(*cephv1.CephFilesystemMirror) - logger.Debug("update event on CephFilesystemMirror 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 != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephBucketTopic: - objNew := e.ObjectNew.(*cephv1.CephBucketTopic) - logger.Debug("update event on CephBucketTopic 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", objNew.Name, DoNotReconcileLabelName) - return false - } - diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer) - if diff != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephBucketNotification: - objNew := e.ObjectNew.(*cephv1.CephBucketNotification) - logger.Debug("update event on CephBucketNotification 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", objNew.Name, DoNotReconcileLabelName) - return false - } - diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer) - if diff != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *cephv1.CephFilesystemSubVolumeGroup: - objNew := e.ObjectNew.(*cephv1.CephFilesystemSubVolumeGroup) - logger.Debug("update event on CephFilesystemSubVolumeGroup 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", objNew.Name, DoNotReconcileLabelName) - return false - } - diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer) - if diff != "" { - logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CR %q is going be deleted", objNew.Name) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name) - } - - case *bktv1alpha1.ObjectBucketClaim: - objNew := e.ObjectNew.(*bktv1alpha1.ObjectBucketClaim) - logger.Debug("update event on ObjectBucketClaim 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", objNew.Name, DoNotReconcileLabelName) - return false - } - if !reflect.DeepEqual(objOld.Labels, objNew.Labels) { - logger.Infof("CR labels has changed for %q", objNew.Name) - return true - } else if objOld.Spec.ObjectBucketName != objNew.Spec.ObjectBucketName { - logger.Infof("CR %q bucket name changed from %q to %q", objNew.Name, objOld.Spec.ObjectBucketName, objNew.Spec.ObjectBucketName) - return true - } - logger.Debugf("no change in CR %q", objNew.Name) - - case *cephv1.CephBlockPoolRadosNamespace: - objNew := e.ObjectNew.(*cephv1.CephBlockPoolRadosNamespace) - namespacedName := fmt.Sprintf("%s/%s", objNew.Namespace, objNew.Name) - logger.Debugf("update event on CephBlockPoolRadosNamespace %q CR", namespacedName) - // 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", namespacedName, DoNotReconcileLabelName) - return false - } - diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer) - if diff != "" { - logger.Infof("CephBlockPoolRadosNamespace CR has changed for %q. diff=%s", namespacedName, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CephBlockPoolRadosNamespace CR %q is going be deleted", namespacedName) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping CephBlockPoolRadosNamespace resource %q update with unchanged spec", namespacedName) - } - - case *cephv1.CephCOSIDriver: - objNew := e.ObjectNew.(*cephv1.CephCOSIDriver) - namespacedName := fmt.Sprintf("%s/%s", objNew.Namespace, objNew.Name) - logger.Debugf("update event on CephCOSIDriver %q CR", namespacedName) - // 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", - namespacedName, DoNotReconcileLabelName) - return false - } - diff := cmp.Diff(objOld.Spec, objNew.Spec) - if diff != "" { - logger.Infof("CephCOSIDriver CR has changed for %q. diff=%s", namespacedName, diff) - return true - } else if objectToBeDeleted(objOld, objNew) { - logger.Debugf("CephCOSIDriver CR %q is going be deleted", namespacedName) - return true - } else if objOld.GetGeneration() != objNew.GetGeneration() { - logger.Debugf("skipping CephCOSIDriver resource %q update with unchanged spec", namespacedName) - } - + diff := cmp.Diff(GetSpec(objOld), GetSpec(objNew), resourceQtyComparer) + if diff != "" { + logger.Infof("resource %q: %q spec has changed. diff=%s", kind, nsName, diff) + return true + } else if objectToBeDeleted(objOld, objNew) { + logger.Debugf("resource %q: %q is going to be deleted", kind, nsName) + return true + } else if objOld.GetGeneration() != objNew.GetGeneration() { + logger.Debugf("skipping resource %q: %q, update event with unchanged spec", kind, nsName) } return false }, - GenericFunc: func(e event.GenericEvent) bool { + GenericFunc: func(e event.TypedGenericEvent[T]) bool { + kind, nsName := objectInfo(scheme, e.Object) + logger.Debugf("generic event for %q: %q", kind, nsName) return false }, } @@ -468,19 +167,19 @@ func objectChanged(oldObj, newObj runtime.Object, objectName string) (bool, erro // // We return 'false' on a create event so we don't overstep with the main watcher on cephv1.CephBlockPool{} // This avoids a double reconcile when the secret gets deleted. -func WatchPredicateForNonCRDObject(owner runtime.Object, scheme *runtime.Scheme) predicate.Funcs { +func WatchPredicateForNonCRDObject[T client.Object](owner runtime.Object, scheme *runtime.Scheme) predicate.TypedFuncs[T] { // Initialize the Owner Matcher, which is the main controller object: e.g. cephv1.CephBlockPool{} ownerMatcher, err := NewOwnerReferenceMatcher(owner, scheme) if err != nil { logger.Errorf("failed to initialize owner matcher. %v", err) } - return predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { return false }, - DeleteFunc: func(e event.DeleteEvent) bool { + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { match, object, err := ownerMatcher.Match(e.Object) if err != nil { logger.Errorf("failed to check if object kind %q matched. %v", e.Object.GetObjectKind(), err) @@ -506,7 +205,7 @@ func WatchPredicateForNonCRDObject(owner runtime.Object, scheme *runtime.Scheme) return false }, - UpdateFunc: func(e event.UpdateEvent) bool { + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { match, object, err := ownerMatcher.Match(e.ObjectNew) if err != nil { logger.Errorf("failed to check if object matched. %v", err) @@ -530,24 +229,22 @@ func WatchPredicateForNonCRDObject(owner runtime.Object, scheme *runtime.Scheme) return true } - // If the resource is a ConfigMap we don't reconcile - _, ok := e.ObjectNew.(*corev1.ConfigMap) - if ok { + switch newObjCopy := any(e.ObjectNew).(type) { + case *corev1.ConfigMap: + // If the resource is a ConfigMap we don't reconcile logger.Debugf("do not reconcile on configmap %q", objectName) return false - } - // SECRETS BLACKLIST - // If the resource is a Secret, we might want to ignore it - // We want to reconcile Secrets in case their content gets altered - isSecretToIgnoreOnUpdate := isSecretToIgnoreOnUpdate(e.ObjectNew) - if isSecretToIgnoreOnUpdate { - return false - } + case *corev1.Secret: + // SECRETS BLACKLIST + // If the resource is a Secret, we might want to ignore it + // We want to reconcile Secrets in case their content gets altered + if isSecretToIgnoreOnUpdate(newObjCopy) { + return false + } - // If the resource is a deployment we don't reconcile - _, ok = e.ObjectNew.(*appsv1.Deployment) - if ok { + case *appsv1.Deployment: + // If the resource is a deployment we don't reconcile logger.Debug("do not reconcile deployments updates") return false } @@ -564,7 +261,7 @@ func WatchPredicateForNonCRDObject(owner runtime.Object, scheme *runtime.Scheme) return false }, - GenericFunc: func(e event.GenericEvent) bool { + GenericFunc: func(e event.TypedGenericEvent[T]) bool { return false }, } @@ -709,16 +406,10 @@ func isCMToIgnoreOnDelete(obj runtime.Object) bool { return false } -func isSecretToIgnoreOnUpdate(obj runtime.Object) bool { - s, ok := obj.(*corev1.Secret) - if !ok { - return false - } - - objectName := s.GetName() - switch objectName { +func isSecretToIgnoreOnUpdate(secret *corev1.Secret) bool { + switch secret.GetName() { case config.StoreName: - logger.Debugf("do not reconcile on %q secret changes", objectName) + logger.Debugf("do not reconcile on %q secret changes", secret.GetName()) return true } @@ -743,13 +434,7 @@ func ReloadManager() { // DuplicateCephClusters determine whether a similar object exists in the same namespace // mainly used for the CephCluster which we only support a single instance per namespace -func DuplicateCephClusters(ctx context.Context, c client.Client, object client.Object, log bool) bool { - objectType, ok := object.(*cephv1.CephCluster) - if !ok { - logger.Errorf("expected type CephCluster but found %T", objectType) - return false - } - +func DuplicateCephClusters(ctx context.Context, c client.Client, object *cephv1.CephCluster, log bool) bool { cephClusterList := &cephv1.CephClusterList{} listOpts := []client.ListOption{ client.InNamespace(object.GetNamespace()), @@ -777,3 +462,20 @@ func DuplicateCephClusters(ctx context.Context, c client.Client, object client.O return false } + +func GetSpec(obj client.Object) interface{} { + val := reflect.ValueOf(obj) + + // If obj is a pointer, get the element + if val.Kind() == reflect.Ptr { + val = val.Elem() + } + + spec := val.FieldByName("Spec") + if !spec.IsValid() { + logger.Warningf("No Spec field found for object %+v. This should not happen.", obj) + return nil + } + + return spec.Interface() +} diff --git a/pkg/operator/ceph/controller/predicate_test.go b/pkg/operator/ceph/controller/predicate_test.go index 5e20900ce..509e206a1 100644 --- a/pkg/operator/ceph/controller/predicate_test.go +++ b/pkg/operator/ceph/controller/predicate_test.go @@ -185,9 +185,6 @@ func TestIsCMToIgnoreOnDelete(t *testing.T) { } func TestIsSecretToIgnoreOnUpdate(t *testing.T) { - blockPool := &cephv1.CephBlockPool{} - assert.False(t, isSecretToIgnoreOnUpdate(blockPool)) - s := &corev1.Secret{} assert.False(t, isSecretToIgnoreOnUpdate(s)) diff --git a/pkg/operator/ceph/csi/controller.go b/pkg/operator/ceph/csi/controller.go index 809b2da4d..ffebd3464 100644 --- a/pkg/operator/ceph/csi/controller.go +++ b/pkg/operator/ceph/csi/controller.go @@ -38,7 +38,7 @@ import ( opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" "github.com/rook/rook/pkg/operator/ceph/csi/peermap" "github.com/rook/rook/pkg/operator/k8sutil" - v1 "k8s.io/api/core/v1" + corev1 "k8s.io/api/core/v1" kerrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -91,23 +91,27 @@ func add(ctx context.Context, mgr manager.Manager, r reconcile.Reconciler, opCon } // Watch for ConfigMap (operator config) - configmapKind := source.Kind[client.Object]( - mgr.GetCache(), - &v1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: v1.SchemeGroupVersion.String()}}, - &handler.EnqueueRequestForObject{}, predicateController(ctx, mgr.GetClient(), opConfig.OperatorNamespace), + err = c.Watch( + source.Kind( + mgr.GetCache(), + &corev1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: corev1.SchemeGroupVersion.String()}}, + &handler.TypedEnqueueRequestForObject[*corev1.ConfigMap]{}, + cmPredicate(), + ), ) - err = c.Watch(configmapKind) if err != nil { return err } // Watch for CephCluster - clusterKind := source.Kind[client.Object]( - mgr.GetCache(), - &cephv1.CephCluster{TypeMeta: metav1.TypeMeta{Kind: "CephCluster", APIVersion: v1.SchemeGroupVersion.String()}}, - &handler.EnqueueRequestForObject{}, predicateController(ctx, mgr.GetClient(), opConfig.OperatorNamespace), + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephCluster{TypeMeta: metav1.TypeMeta{Kind: "CephCluster", APIVersion: corev1.SchemeGroupVersion.String()}}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephCluster]{}, + cephClusterPredicate(ctx, mgr.GetClient(), opConfig.OperatorNamespace), + ), ) - err = c.Watch(clusterKind) if err != nil { return err } diff --git a/pkg/operator/ceph/csi/predicate.go b/pkg/operator/ceph/csi/predicate.go index d04a8ac5f..ebe6fa9e2 100644 --- a/pkg/operator/ceph/csi/predicate.go +++ b/pkg/operator/ceph/csi/predicate.go @@ -22,7 +22,7 @@ import ( "github.com/google/go-cmp/cmp" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" - v1 "k8s.io/api/core/v1" + corev1 "k8s.io/api/core/v1" kerrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" "k8s.io/apimachinery/pkg/types" @@ -31,81 +31,90 @@ import ( "sigs.k8s.io/controller-runtime/pkg/predicate" ) -// predicateController is the predicate function to trigger reconcile on operator configuration cm change -func predicateController(ctx context.Context, c client.Client, opNamespace string) predicate.Funcs { - return predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { - // if the operator configuration file is created we want to reconcile - if cm, ok := e.Object.(*v1.ConfigMap); ok { - // We don't want to use cm.Generation here, it case the operator was stopped and the - // ConfigMap was created - return cm.Name == opcontroller.OperatorSettingConfigMapName - } +// reconcile on operator settings cm change +func cmPredicate[T *corev1.ConfigMap]() predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { + cm := (*corev1.ConfigMap)(e.Object) - // If a Ceph Cluster is created we want to reconcile the csi driver - if cephCluster, ok := e.Object.(*cephv1.CephCluster); ok { - // If there are more than one ceph cluster in the same namespace do not reconcile - if opcontroller.DuplicateCephClusters(ctx, c, e.Object, false) { - return false - } + // We don't want to use cm.Generation here, it case the operator was stopped and the + // ConfigMap was created + return cm.Name == opcontroller.OperatorSettingConfigMapName + }, - // We still have users that don't use the ConfigMap to configure the - // operator/csi and still rely on environment variables from the operator pod's - // spec. So we still want to reconcile the controller if the ConfigMap cannot be found. - err := c.Get(ctx, types.NamespacedName{Name: opcontroller.OperatorSettingConfigMapName, Namespace: opNamespace}, &v1.ConfigMap{}) - if err != nil && kerrors.IsNotFound(err) { - logger.Debugf("could not find operator configuration ConfigMap, will reconcile the csi controller") + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + cmOld := (*corev1.ConfigMap)(e.ObjectOld) + cmNew := (*corev1.ConfigMap)(e.ObjectNew) + + resourceQtyComparer := cmp.Comparer(func(x, y resource.Quantity) bool { return x.Cmp(y) == 0 }) + if cmOld.GetName() == opcontroller.OperatorSettingConfigMapName && cmNew.GetName() == opcontroller.OperatorSettingConfigMapName { + diff := cmp.Diff(cmOld.Data, cmNew.Data, resourceQtyComparer) + logger.Debugf("operator configmap diff:\n %s", diff) + if diff != "" { return true } - - // This allows us to avoid a double reconcile of the CSI controller if this is not - // the first generation of the CephCluster. So only return true if this is the very - // first instance of the CephCluster - // Corner case is when the cluster is created but the operator is down AND the cm - // does not exist... However, these days the operator config map is part of the - // operator.yaml so it's probably acceptable? - // This does not account for the case where the cephcluster is already deployed and - // the upgrade the operator or restart it. However, the CM check above should catch that - return cephCluster.Generation == 1 } return false }, - UpdateFunc: func(e event.UpdateEvent) bool { - resourceQtyComparer := cmp.Comparer(func(x, y resource.Quantity) bool { return x.Cmp(y) == 0 }) - if old, ok := e.ObjectOld.(*v1.ConfigMap); ok { - if new, ok := e.ObjectNew.(*v1.ConfigMap); ok { - if old.Name == opcontroller.OperatorSettingConfigMapName && new.Name == opcontroller.OperatorSettingConfigMapName { - diff := cmp.Diff(old.Data, new.Data, resourceQtyComparer) - logger.Debugf("operator configmap diff:\n %s", diff) - if diff != "" { - return true - } - } - } - } + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { + cm := (*corev1.ConfigMap)(e.Object) - return false - }, - - DeleteFunc: func(e event.DeleteEvent) bool { // if the operator configuration file is deleted we want to reconcile to apply the // configuration based on environment variables present in the operator's pod spec - if cm, ok := e.Object.(*v1.ConfigMap); ok { - return cm.Name == opcontroller.OperatorSettingConfigMapName - } - - // if cephCluster is deleted, trigger reconcile to cleanup the csi driver resources - // if zero cephClusters exist. - if _, ok := e.Object.(*cephv1.CephCluster); ok { - return true - } - - return false + return cm.Name == opcontroller.OperatorSettingConfigMapName }, - GenericFunc: func(e event.GenericEvent) bool { + GenericFunc: func(e event.TypedGenericEvent[T]) bool { + return false + }, + } +} + +// predicateController is the predicate function to trigger reconcile on operator configuration cm change +func cephClusterPredicate[T *cephv1.CephCluster](ctx context.Context, c client.Client, opNamespace string) predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { + cephCluster := (*cephv1.CephCluster)(e.Object) + + // If a Ceph Cluster is created we want to reconcile the csi driver + // If there are more than one ceph cluster in the same namespace do not reconcile + if opcontroller.DuplicateCephClusters(ctx, c, cephCluster, false) { + return false + } + + // We still have users that don't use the ConfigMap to configure the + // operator/csi and still rely on environment variables from the operator pod's + // spec. So we still want to reconcile the controller if the ConfigMap cannot be found. + err := c.Get(ctx, types.NamespacedName{Name: opcontroller.OperatorSettingConfigMapName, Namespace: opNamespace}, &corev1.ConfigMap{}) + if err != nil && kerrors.IsNotFound(err) { + logger.Debugf("could not find operator configuration ConfigMap, will reconcile the csi controller") + return true + } + + // This allows us to avoid a double reconcile of the CSI controller if this is not + // the first generation of the CephCluster. So only return true if this is the very + // first instance of the CephCluster + // Corner case is when the cluster is created but the operator is down AND the cm + // does not exist... However, these days the operator config map is part of the + // operator.yaml so it's probably acceptable? + // This does not account for the case where the cephcluster is already deployed and + // the upgrade the operator or restart it. However, the CM check above should catch that + return cephCluster.Generation == 1 + }, + + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + return false + }, + + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { + // if cephCluster is deleted, trigger reconcile to cleanup the csi driver resources + // if zero cephClusters exist. + return true + }, + + GenericFunc: func(e event.TypedGenericEvent[T]) bool { return false }, } diff --git a/pkg/operator/ceph/csi/predicate_test.go b/pkg/operator/ceph/csi/predicate_test.go index ae3ff720b..d16f3689b 100644 --- a/pkg/operator/ceph/csi/predicate_test.go +++ b/pkg/operator/ceph/csi/predicate_test.go @@ -24,116 +24,118 @@ import ( cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/pkg/client/clientset/versioned/scheme" "github.com/stretchr/testify/assert" - v1 "k8s.io/api/core/v1" - "sigs.k8s.io/controller-runtime/pkg/client" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "sigs.k8s.io/controller-runtime/pkg/client/fake" "sigs.k8s.io/controller-runtime/pkg/event" - "sigs.k8s.io/controller-runtime/pkg/predicate" ) -func Test_predicateController(t *testing.T) { +func Test_cmPredicate(t *testing.T) { capnslog.SetGlobalLogLevel(capnslog.DEBUG) - var ( - client client.WithWatch - namespace v1.Namespace - p predicate.Funcs - c event.CreateEvent - d event.DeleteEvent - u event.UpdateEvent - cm v1.ConfigMap - cm2 v1.ConfigMap - cluster cephv1.CephCluster - cluster2 cephv1.CephCluster - ) - - s := scheme.Scheme - s.AddKnownTypes(cephv1.SchemeGroupVersion, &v1.ConfigMap{}, &v1.ConfigMapList{}, &cephv1.CephCluster{}, &cephv1.CephClusterList{}) - client = fake.NewClientBuilder().WithScheme(s).Build() - - t.Run("create event is not the one we are looking for", func(t *testing.T) { - c = event.CreateEvent{Object: &namespace} - p = predicateController(context.TODO(), client, "rook-ceph") - assert.False(t, p.Create(c)) - }) t.Run("create event is a CM but not the operator config", func(t *testing.T) { - c = event.CreateEvent{Object: &cm} - p = predicateController(context.TODO(), client, "rook-ceph") - assert.False(t, p.Create(c)) + cm := &corev1.ConfigMap{} + e := event.TypedCreateEvent[*corev1.ConfigMap]{Object: cm} + p := cmPredicate() + assert.False(t, p.Create(e)) }) t.Run("delete event is a CM and it's not the operator config", func(t *testing.T) { - d = event.DeleteEvent{Object: &cm} - p = predicateController(context.TODO(), client, "rook-ceph") - assert.False(t, p.Delete(d)) + cm := &corev1.ConfigMap{} + e := event.TypedDeleteEvent[*corev1.ConfigMap]{Object: cm} + p := cmPredicate() + assert.False(t, p.Delete(e)) }) t.Run("create event is a CM and it's the operator config", func(t *testing.T) { - cm.Name = "rook-ceph-operator-config" - c = event.CreateEvent{Object: &cm} - p = predicateController(context.TODO(), client, "rook-ceph") - assert.True(t, p.Create(c)) + cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "rook-ceph-operator-config"}} + e := event.TypedCreateEvent[*corev1.ConfigMap]{Object: cm} + p := cmPredicate() + assert.True(t, p.Create(e)) }) t.Run("delete event is a CM and it's the operator config", func(t *testing.T) { - d = event.DeleteEvent{Object: &cm} - p = predicateController(context.TODO(), client, "rook-ceph") - assert.True(t, p.Delete(d)) + cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "rook-ceph-operator-config"}} + e := event.TypedDeleteEvent[*corev1.ConfigMap]{Object: cm} + p := cmPredicate() + assert.True(t, p.Delete(e)) }) t.Run("update event is a CM and it's the operator config but nothing changed", func(t *testing.T) { - u = event.UpdateEvent{ObjectOld: &cm, ObjectNew: &cm} - p = predicateController(context.TODO(), client, "rook-ceph") - assert.False(t, p.Update(u)) + cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "rook-ceph-operator-config"}} + e := event.TypedUpdateEvent[*corev1.ConfigMap]{ObjectOld: cm, ObjectNew: cm} + p := cmPredicate() + assert.False(t, p.Update(e)) }) t.Run("update event is a CM and the content changed with correct setting", func(t *testing.T) { + cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "rook-ceph-operator-config"}} + cm2 := cm.DeepCopy() cm.Data = map[string]string{"foo": "bar"} - cm2.Name = "rook-ceph-operator-config" cm2.Data = map[string]string{"CSI_PROVISIONER_REPLICAS": "2"} - u = event.UpdateEvent{ObjectOld: &cm, ObjectNew: &cm2} - p = predicateController(context.TODO(), client, "rook-ceph") - assert.True(t, p.Update(u)) + e := event.TypedUpdateEvent[*corev1.ConfigMap]{ObjectOld: cm, ObjectNew: cm2} + p := cmPredicate() + assert.True(t, p.Update(e)) }) +} + +func Test_cephClusterPredicate(t *testing.T) { + capnslog.SetGlobalLogLevel(capnslog.DEBUG) + + s := scheme.Scheme + s.AddKnownTypes(cephv1.SchemeGroupVersion, &corev1.ConfigMap{}, &corev1.ConfigMapList{}, &cephv1.CephCluster{}, &cephv1.CephClusterList{}) t.Run("create event is a CephCluster and it's the first instance and a cm is present", func(t *testing.T) { - cm.Namespace = "rook-ceph" - err := client.Create(context.TODO(), &cm) - assert.NoError(t, err) - cluster.Generation = 1 - c = event.CreateEvent{Object: &cluster} - p = predicateController(context.TODO(), client, "rook-ceph") - assert.True(t, p.Create(c)) + client := fake.NewClientBuilder().WithScheme(s).Build() + + cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "rook-ceph-operator-config", Namespace: "rook-ceph"}} + err := client.Create(context.TODO(), cm) + require.NoError(t, err) + + cluster := &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Namespace: "rook-ceph", Generation: 1}} + e := event.TypedCreateEvent[*cephv1.CephCluster]{Object: cluster} + p := cephClusterPredicate(context.TODO(), client, "rook-ceph") + assert.True(t, p.Create(e)) }) t.Run("create event is a CephCluster and it's not the first instance", func(t *testing.T) { - cluster.Generation = 2 - c = event.CreateEvent{Object: &cluster} - p = predicateController(context.TODO(), client, "rook-ceph") - assert.False(t, p.Create(c)) + client := fake.NewClientBuilder().WithScheme(s).Build() + + cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "rook-ceph-operator-config", Namespace: "rook-ceph"}} + err := client.Create(context.TODO(), cm) + require.NoError(t, err) + + cluster := &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Namespace: "rook-ceph", Generation: 2}} + e := event.TypedCreateEvent[*cephv1.CephCluster]{Object: cluster} + p := cephClusterPredicate(context.TODO(), client, "rook-ceph") + assert.False(t, p.Create(e)) }) t.Run("create event is a CephCluster and it's not the first instance but no cm so we reconcile", func(t *testing.T) { - err := client.Delete(context.TODO(), &cm) - assert.NoError(t, err) - cluster.Generation = 2 - c = event.CreateEvent{Object: &cluster} - p = predicateController(context.TODO(), client, "rook-ceph") - assert.True(t, p.Create(c)) + client := fake.NewClientBuilder().WithScheme(s).Build() + + cluster := &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Namespace: "rook-ceph", Generation: 2}} + e := event.TypedCreateEvent[*cephv1.CephCluster]{Object: cluster} + p := cephClusterPredicate(context.TODO(), client, "rook-ceph") + assert.True(t, p.Create(e)) }) t.Run("create event is a CephCluster and a cluster already exists - not reconciling", func(t *testing.T) { - cluster.Name = "ceph1" - cluster2.Name = "ceph2" - cluster.Namespace = "rook-ceph" - cluster2.Namespace = "rook-ceph" - err := client.Create(context.TODO(), &cluster) - assert.NoError(t, err) - err = client.Create(context.TODO(), &cluster2) - assert.NoError(t, err) - c = event.CreateEvent{Object: &cluster2} - assert.Equal(t, "rook-ceph", c.Object.GetNamespace()) - p = predicateController(context.TODO(), client, "rook-ceph") - assert.False(t, p.Create(c)) + client := fake.NewClientBuilder().WithScheme(s).Build() + + { + cluster := &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Name: "ceph1", Namespace: "rook-ceph", Generation: 2}} + err := client.Create(context.TODO(), cluster) + require.NoError(t, err) + } + + cluster := &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Name: "ceph2", Namespace: "rook-ceph"}} + err := client.Create(context.TODO(), cluster) + require.NoError(t, err) + + e := event.TypedCreateEvent[*cephv1.CephCluster]{Object: cluster} + p := cephClusterPredicate(context.TODO(), client, "rook-ceph") + assert.False(t, p.Create(e)) }) } diff --git a/pkg/operator/ceph/disruption/clusterdisruption/add.go b/pkg/operator/ceph/disruption/clusterdisruption/add.go index 0faf21e8e..08bed3174 100644 --- a/pkg/operator/ceph/disruption/clusterdisruption/add.go +++ b/pkg/operator/ceph/disruption/clusterdisruption/add.go @@ -36,6 +36,72 @@ import ( "k8s.io/apimachinery/pkg/types" ) +// Watch for CephBlockPools and enqueue the CephCluster in the namespace +// Watch for CephFileSystems and enqueue the CephCluster in the namespace +// Watch for CephObjectStores and enqueue the CephCluster in the namespace +var objectsToWatch = []client.Object{ + &cephv1.CephBlockPool{}, + &cephv1.CephFilesystem{}, + &cephv1.CephObjectStore{}, +} + +func cephClusterPredicate[T *cephv1.CephCluster]() predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { + logger.Debug("create event from ceph cluster CR") + return true + }, + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + objOld := (*cephv1.CephCluster)(e.ObjectOld) + objNew := (*cephv1.CephCluster)(e.ObjectNew) + + return !reflect.DeepEqual(objOld.Spec, objNew.Spec) + }, + } +} + +func pdbPredicate[T *policyv1.PodDisruptionBudget]() predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { + // Do not reconcile when PDB is created + return false + }, + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + pdb := (*policyv1.PodDisruptionBudget)(e.ObjectNew) + + // reconcile for the main PDB update event when first OSD goes down, that is, when `DisruptionsAllowed` gets updated to 0. + return pdb.Name == osdPDBAppName && pdb.Spec.MaxUnavailable.IntVal == 1 && pdb.Status.DisruptionsAllowed == 0 + }, + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { + // Do not reconcile when PDB is deleted + return false + }, + } +} + +func watchNamespacedObject[T client.Object](c controller.Controller, mgr manager.Manager, obj T) error { + // enqueues with an empty name that is populated by the reconciler. + // There is a one-per-namespace limit on CephClusters + return c.Watch( + source.Kind( + mgr.GetCache(), + obj, + handler.TypedEnqueueRequestsFromMapFunc( + func(context ctx.Context, obj T) []reconcile.Request { + // The name will be populated in the reconcile + namespace := obj.GetNamespace() + if len(namespace) == 0 { + logger.Errorf("enqueueByNamespace received an obj without a namespace. %+v", obj) + return []reconcile.Request{} + } + req := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}} + return []reconcile.Request{req} + }, + ), + ), + ) +} + // Add adds a new Controller to the Manager based on clusterdisruption.ReconcileClusterDisruption and registers the relevant watches and handlers. // Read more about how Managers, Controllers, and their Watches, Handlers, Predicates, etc work here: // https://godoc.org/github.com/kubernetes-sigs/controller-runtime/pkg @@ -56,100 +122,43 @@ func Add(mgr manager.Manager, context *controllerconfig.Context) error { return err } - cephClusterPredicate := predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { - logger.Debug("create event from ceph cluster CR") - return true - }, - UpdateFunc: func(e event.UpdateEvent) bool { - oldCluster, ok := e.ObjectOld.DeepCopyObject().(*cephv1.CephCluster) - if !ok { - return false - } - newCluster, ok := e.ObjectNew.DeepCopyObject().(*cephv1.CephCluster) - if !ok { - return false - } - return !reflect.DeepEqual(oldCluster.Spec, newCluster.Spec) - }, - } - // Watch for CephClusters - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephCluster{}, &handler.EnqueueRequestForObject{}, cephClusterPredicate)) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephCluster{}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephCluster]{}, + cephClusterPredicate(), + ), + ) if err != nil { return err } - pdbPredicate := predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { - // Do not reconcile when PDB is created - return false - }, - UpdateFunc: func(e event.UpdateEvent) bool { - pdb, ok := e.ObjectNew.DeepCopyObject().(*policyv1.PodDisruptionBudget) - if !ok { - return false - } - // reconcile for the main PDB update event when first OSD goes down, that is, when `DisruptionsAllowed` gets updated to 0. - return pdb.Name == osdPDBAppName && pdb.Spec.MaxUnavailable.IntVal == 1 && pdb.Status.DisruptionsAllowed == 0 - }, - DeleteFunc: func(e event.DeleteEvent) bool { - // Do not reconcile when PDB is deleted - return false - }, - } - // Watch for main PodDisruptionBudget and enqueue the CephCluster in the namespace err = c.Watch( - source.Kind[client.Object](mgr.GetCache(), &policyv1.PodDisruptionBudget{}, - handler.EnqueueRequestsFromMapFunc(handler.MapFunc(func(context ctx.Context, obj client.Object) []reconcile.Request { - pdb, ok := obj.(*policyv1.PodDisruptionBudget) - if !ok { - // Not a pdb, returning empty - logger.Error("PDB handler received non-PDB") - return []reconcile.Request{} - } - namespace := pdb.GetNamespace() - req := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}} - return []reconcile.Request{req} - }), + source.Kind( + mgr.GetCache(), + &policyv1.PodDisruptionBudget{}, + handler.TypedEnqueueRequestsFromMapFunc( + func(context ctx.Context, pdb *policyv1.PodDisruptionBudget) []reconcile.Request { + namespace := pdb.GetNamespace() + req := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}} + return []reconcile.Request{req} + }, ), - pdbPredicate, - )) - if err != nil { - return err - } - - // enqueues with an empty name that is populated by the reconciler. - // There is a one-per-namespace limit on CephClusters - enqueueByNamespace := handler.EnqueueRequestsFromMapFunc(handler.MapFunc(func(context ctx.Context, obj client.Object) []reconcile.Request { - // The name will be populated in the reconcile - namespace := obj.GetNamespace() - if len(namespace) == 0 { - logger.Errorf("enqueueByNamespace received an obj without a namespace. %+v", obj) - return []reconcile.Request{} - } - req := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}} - return []reconcile.Request{req} - }), + pdbPredicate(), + ), ) - - // Watch for CephBlockPools and enqueue the CephCluster in the namespace - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephBlockPool{}, enqueueByNamespace)) if err != nil { return err } - // Watch for CephFileSystems and enqueue the CephCluster in the namespace - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephFilesystem{}, enqueueByNamespace)) - if err != nil { - return err - } - - // Watch for CephObjectStores and enqueue the CephCluster in the namespace - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephObjectStore{}, enqueueByNamespace)) - if err != nil { - return err + for _, t := range objectsToWatch { + err = watchNamespacedObject(c, mgr, t) + if err != nil { + return err + } } return nil diff --git a/pkg/operator/ceph/file/controller.go b/pkg/operator/ceph/file/controller.go index fa2a80e14..51ffecee5 100644 --- a/pkg/operator/ceph/file/controller.go +++ b/pkg/operator/ceph/file/controller.go @@ -108,6 +108,21 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont } } +func watchOwnedCoreObject[T client.Object](c controller.Controller, mgr manager.Manager, obj T) error { + return c.Watch( + source.Kind( + mgr.GetCache(), + obj, + handler.TypedEnqueueRequestForOwner[T]( + mgr.GetScheme(), + mgr.GetRESTMapper(), + &cephv1.CephFilesystem{}, + ), + opcontroller.WatchPredicateForNonCRDObject[T](&cephv1.CephFilesystem{TypeMeta: controllerTypeMeta}, mgr.GetScheme()), + ), + ) +} + func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reconciler) error { // Create a new controller c, err := controller.New(controllerName, mgr, controller.Options{Reconciler: r}) @@ -117,20 +132,21 @@ func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reco logger.Info("successfully started") // Watch for changes on the CephFilesystem CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephFilesystem{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephFilesystem{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephFilesystem]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephFilesystem](mgr.GetScheme()), + ), + ) if err != nil { return err } // Watch all other resources for _, t := range objectsToWatch { - ownerRequest := handler.EnqueueRequestForOwner( - mgr.GetScheme(), - mgr.GetRESTMapper(), - &cephv1.CephFilesystem{}, - ) - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), t, ownerRequest, - opcontroller.WatchPredicateForNonCRDObject(&cephv1.CephFilesystem{TypeMeta: controllerTypeMeta}, mgr.GetScheme()))) + err = watchOwnedCoreObject(c, mgr, t) if err != nil { return err } @@ -138,15 +154,20 @@ func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reco // Build Handler function to return the list of ceph filesystems // This is used by the watchers below - handlerFunc, err := opcontroller.ObjectToCRMapper(opManagerContext, mgr.GetClient(), &cephv1.CephFilesystemList{}, mgr.GetScheme()) + handlerFunc, err := opcontroller.ObjectToCRMapper[*cephv1.CephFilesystemList, *corev1.ConfigMap](opManagerContext, mgr.GetClient(), &cephv1.CephFilesystemList{}, mgr.GetScheme()) if err != nil { return err } // Watch for ConfigMap "rook-ceph-mon-endpoints" update and reconcile, which will reconcile update the bootstrap peer token - err = c.Watch(source.Kind[client.Object]( - mgr.GetCache(), &corev1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: corev1.SchemeGroupVersion.String()}}, - handler.EnqueueRequestsFromMapFunc(handlerFunc), mon.PredicateMonEndpointChanges())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &corev1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: corev1.SchemeGroupVersion.String()}}, + handler.TypedEnqueueRequestsFromMapFunc(handlerFunc), + mon.PredicateMonEndpointChanges(), + ), + ) if err != nil { return err } diff --git a/pkg/operator/ceph/file/mirror/controller.go b/pkg/operator/ceph/file/mirror/controller.go index 2857d39ce..2afc008ef 100644 --- a/pkg/operator/ceph/file/mirror/controller.go +++ b/pkg/operator/ceph/file/mirror/controller.go @@ -98,6 +98,21 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont } } +func watchOwnedCoreObject[T client.Object](c controller.Controller, mgr manager.Manager, obj T) error { + return c.Watch( + source.Kind( + mgr.GetCache(), + obj, + handler.TypedEnqueueRequestForOwner[T]( + mgr.GetScheme(), + mgr.GetRESTMapper(), + &cephv1.CephFilesystemMirror{}, + ), + opcontroller.WatchPredicateForNonCRDObject[T](&cephv1.CephFilesystemMirror{TypeMeta: controllerTypeMeta}, mgr.GetScheme()), + ), + ) +} + func add(mgr manager.Manager, r reconcile.Reconciler) error { // Create a new controller c, err := controller.New(controllerName, mgr, controller.Options{Reconciler: r}) @@ -107,19 +122,21 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the CephFilesystemMirror CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephFilesystemMirror{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephFilesystemMirror{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephFilesystemMirror]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephFilesystemMirror](mgr.GetScheme()), + ), + ) if err != nil { return err } // Watch all other resources for _, t := range objectsToWatch { - ownerRequest := handler.EnqueueRequestForOwner( - mgr.GetScheme(), - mgr.GetRESTMapper(), - &cephv1.CephFilesystemMirror{}, - ) - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), t, ownerRequest, opcontroller.WatchPredicateForNonCRDObject(&cephv1.CephFilesystemMirror{TypeMeta: controllerTypeMeta}, mgr.GetScheme()))) + err = watchOwnedCoreObject(c, mgr, t) if err != nil { return err } diff --git a/pkg/operator/ceph/file/subvolumegroup/controller.go b/pkg/operator/ceph/file/subvolumegroup/controller.go index 4ccbd2635..ae387dc31 100644 --- a/pkg/operator/ceph/file/subvolumegroup/controller.go +++ b/pkg/operator/ceph/file/subvolumegroup/controller.go @@ -102,7 +102,14 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the CephFilesystemSubVolumeGroup CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephFilesystemSubVolumeGroup{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephFilesystemSubVolumeGroup{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephFilesystemSubVolumeGroup]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephFilesystemSubVolumeGroup](mgr.GetScheme()), + ), + ) if err != nil { return err } diff --git a/pkg/operator/ceph/nfs/controller.go b/pkg/operator/ceph/nfs/controller.go index 487724150..27c4be1e6 100644 --- a/pkg/operator/ceph/nfs/controller.go +++ b/pkg/operator/ceph/nfs/controller.go @@ -100,6 +100,21 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont } } +func watchOwnedCoreObject[T client.Object](c controller.Controller, mgr manager.Manager, obj T) error { + return c.Watch( + source.Kind( + mgr.GetCache(), + obj, + handler.TypedEnqueueRequestForOwner[T]( + mgr.GetScheme(), + mgr.GetRESTMapper(), + &cephv1.CephNFS{}, + ), + opcontroller.WatchPredicateForNonCRDObject[T](&cephv1.CephNFS{TypeMeta: controllerTypeMeta}, mgr.GetScheme()), + ), + ) +} + func add(mgr manager.Manager, r reconcile.Reconciler) error { // Create a new controller c, err := controller.New(controllerName, mgr, controller.Options{Reconciler: r}) @@ -109,20 +124,21 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the cephNFS CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephNFS{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephNFS{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephNFS]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephNFS](mgr.GetScheme()), + ), + ) if err != nil { return err } // Watch all other resources for _, t := range objectsToWatch { - ownerRequest := handler.EnqueueRequestForOwner( - mgr.GetScheme(), - mgr.GetRESTMapper(), - &cephv1.CephNFS{}, - ) - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), t, ownerRequest, - opcontroller.WatchPredicateForNonCRDObject(&cephv1.CephNFS{TypeMeta: controllerTypeMeta}, mgr.GetScheme()))) + err = watchOwnedCoreObject(c, mgr, t) if err != nil { return err } diff --git a/pkg/operator/ceph/object/bucket/controller.go b/pkg/operator/ceph/object/bucket/controller.go index b9a4007f7..6cdd3365e 100644 --- a/pkg/operator/ceph/object/bucket/controller.go +++ b/pkg/operator/ceph/object/bucket/controller.go @@ -80,22 +80,27 @@ func add(ctx context.Context, mgr manager.Manager, r reconcile.Reconciler) error logger.Infof("%s successfully started", controllerName) // Watch for ConfigMap (operator config) - cmKind := source.Kind[client.Object]( - mgr.GetCache(), - &v1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: v1.SchemeGroupVersion.String()}}, - &handler.EnqueueRequestForObject{}, predicateController(ctx, mgr.GetClient()), + err = c.Watch( + source.Kind( + mgr.GetCache(), + &v1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: v1.SchemeGroupVersion.String()}}, + &handler.TypedEnqueueRequestForObject[*v1.ConfigMap]{}, + cmPredicate(), + ), ) - err = c.Watch(cmKind) if err != nil { return err } // Watch for CephCluster - clusterKind := source.Kind[client.Object](mgr.GetCache(), - &cephv1.CephCluster{TypeMeta: metav1.TypeMeta{Kind: "CephCluster", APIVersion: v1.SchemeGroupVersion.String()}}, - &handler.EnqueueRequestForObject{}, predicateController(ctx, mgr.GetClient()), + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephCluster{TypeMeta: metav1.TypeMeta{Kind: "CephCluster", APIVersion: v1.SchemeGroupVersion.String()}}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephCluster]{}, + cephClusterPredicate(ctx, mgr.GetClient()), + ), ) - err = c.Watch(clusterKind) if err != nil { return err } diff --git a/pkg/operator/ceph/object/bucket/predicate.go b/pkg/operator/ceph/object/bucket/predicate.go index 4e481055e..686a2816a 100644 --- a/pkg/operator/ceph/object/bucket/predicate.go +++ b/pkg/operator/ceph/object/bucket/predicate.go @@ -21,7 +21,7 @@ import ( cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/pkg/operator/ceph/controller" - v1 "k8s.io/api/core/v1" + corev1 "k8s.io/api/core/v1" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -29,49 +29,60 @@ import ( const rookOBCWatchOperatorNamespace = "ROOK_OBC_WATCH_OPERATOR_NAMESPACE" -// predicateController is the predicate function to trigger reconcile on operator configuration cm change -func predicateController(ctx context.Context, c client.Client) predicate.Funcs { - return predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { - // if the operator configuration file is created we want to reconcile - if cm, ok := e.Object.(*v1.ConfigMap); ok { - // It's probably fine to use the Generation value here. The case where the operator was stopped and the - // ConfigMap was created is low since the cm is always present these days - return cm.Name == controller.OperatorSettingConfigMapName && cm.Generation == 1 - } +// reconcile on ConfigMap change +func cmPredicate[T *corev1.ConfigMap]() predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { + cm := (*corev1.ConfigMap)(e.Object) - // If a Ceph Cluster is created we want to reconcile the bucket provisioner - if _, ok := e.Object.(*cephv1.CephCluster); ok { - // If there are more than one ceph cluster in the same namespace do not reconcile - return !controller.DuplicateCephClusters(ctx, c, e.Object, false) - } - - return false + return cm.Name == controller.OperatorSettingConfigMapName && cm.Generation == 1 }, - UpdateFunc: func(e event.UpdateEvent) bool { - if old, ok := e.ObjectOld.(*v1.ConfigMap); ok { - if new, ok := e.ObjectNew.(*v1.ConfigMap); ok { - if old.Name == controller.OperatorSettingConfigMapName && new.Name == controller.OperatorSettingConfigMapName { - if old.Data[rookOBCWatchOperatorNamespace] != new.Data[rookOBCWatchOperatorNamespace] { - logger.Infof("%s changed. reconciling bucket controller", rookOBCWatchOperatorNamespace) + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + cmOld := (*corev1.ConfigMap)(e.ObjectOld) + cmNew := (*corev1.ConfigMap)(e.ObjectNew) - // We reload the manager so that the controller restarts and goes - // through the CreateFunc again. Then the CephCluster watcher will be triggered - controller.ReloadManager() - } - } + if cmOld.GetName() == controller.OperatorSettingConfigMapName && cmNew.GetName() == controller.OperatorSettingConfigMapName { + if cmOld.Data[rookOBCWatchOperatorNamespace] != cmNew.Data[rookOBCWatchOperatorNamespace] { + logger.Infof("%s changed. reconciling bucket controller", rookOBCWatchOperatorNamespace) + + // We reload the manager so that the controller restarts and goes + // through the CreateFunc again. Then the CephCluster watcher will be triggered + controller.ReloadManager() } } return false }, - DeleteFunc: func(e event.DeleteEvent) bool { + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { return false }, - GenericFunc: func(e event.GenericEvent) bool { + GenericFunc: func(e event.TypedGenericEvent[T]) bool { + return false + }, + } +} + +// trigger reconcile on CephCluster change +func cephClusterPredicate[T *cephv1.CephCluster](ctx context.Context, c client.Client) predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { + // If a Ceph Cluster is created we want to reconcile the bucket provisioner + // If there are more than one ceph cluster in the same namespace do not reconcile + return !controller.DuplicateCephClusters(ctx, c, e.Object, false) + }, + + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + return false + }, + + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { + return false + }, + + GenericFunc: func(e event.TypedGenericEvent[T]) bool { return false }, } diff --git a/pkg/operator/ceph/object/controller.go b/pkg/operator/ceph/object/controller.go index 59a11f315..e1b010dc4 100644 --- a/pkg/operator/ceph/object/controller.go +++ b/pkg/operator/ceph/object/controller.go @@ -116,6 +116,21 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont } } +func watchOwnedCoreObject[T client.Object](c controller.Controller, mgr manager.Manager, obj T) error { + return c.Watch( + source.Kind( + mgr.GetCache(), + obj, + handler.TypedEnqueueRequestForOwner[T]( + mgr.GetScheme(), + mgr.GetRESTMapper(), + &cephv1.CephObjectStore{}, + ), + opcontroller.WatchPredicateForNonCRDObject[T](&cephv1.CephObjectStore{TypeMeta: controllerTypeMeta}, mgr.GetScheme()), + ), + ) +} + func add(mgr manager.Manager, r reconcile.Reconciler) error { // Create a new controller c, err := controller.New(controllerName, mgr, controller.Options{Reconciler: r}) @@ -125,30 +140,35 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the cephObjectStore CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephObjectStore{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephObjectStore{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephObjectStore]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephObjectStore](mgr.GetScheme()), + ), + ) if err != nil { return err } // Watch all other resources for _, t := range objectsToWatch { - ownerRequest := handler.EnqueueRequestForOwner( - mgr.GetScheme(), - mgr.GetRESTMapper(), - &cephv1.CephObjectStore{}, - ) - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), t, ownerRequest, - opcontroller.WatchPredicateForNonCRDObject(&cephv1.CephObjectStore{TypeMeta: controllerTypeMeta}, mgr.GetScheme()))) + err = watchOwnedCoreObject(c, mgr, t) if err != nil { return err } } // Watch Secrets secrets annotated for the object store - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), - &corev1.Secret{TypeMeta: metav1.TypeMeta{Kind: "Secret", APIVersion: corev1.SchemeGroupVersion.String()}}, - handler.EnqueueRequestsFromMapFunc(mapSecretToCR(mgr.GetClient())), - secretPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &corev1.Secret{TypeMeta: metav1.TypeMeta{Kind: "Secret", APIVersion: corev1.SchemeGroupVersion.String()}}, + handler.TypedEnqueueRequestsFromMapFunc(mapSecretToCR(mgr.GetClient())), + secretPredicate(), + ), + ) if err != nil { return err } @@ -157,15 +177,12 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { } // Watch all secrets not owned by Rook -func secretPredicate() predicate.Predicate { +func secretPredicate[T *corev1.Secret]() predicate.TypedPredicate[T] { rookGV := cephv1.SchemeGroupVersion.String() - return predicate.Funcs{ - UpdateFunc: func(e event.UpdateEvent) bool { - secret, ok := e.ObjectNew.(*corev1.Secret) - if !ok { - // not a secret - return false - } + return predicate.TypedFuncs[T]{ + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + secret := (*corev1.Secret)(e.ObjectNew) + // check if secret already owned by Rook: for _, owner := range secret.OwnerReferences { if owner.APIVersion == rookGV { @@ -175,12 +192,9 @@ func secretPredicate() predicate.Predicate { } return true }, - CreateFunc: func(e event.CreateEvent) bool { - secret, ok := e.Object.(*corev1.Secret) - if !ok { - // not a secret - return false - } + CreateFunc: func(e event.TypedCreateEvent[T]) bool { + secret := (*corev1.Secret)(e.Object) + // check if secret already owned by Rook: for _, owner := range secret.OwnerReferences { if owner.APIVersion == rookGV { @@ -190,12 +204,8 @@ func secretPredicate() predicate.Predicate { } return true }, - DeleteFunc: func(e event.DeleteEvent) bool { - secret, ok := e.Object.(*corev1.Secret) - if !ok { - // not a secret - return false - } + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { + secret := (*corev1.Secret)(e.Object) // check if secret already owned by Rook: for _, owner := range secret.OwnerReferences { if owner.APIVersion == rookGV { @@ -209,13 +219,8 @@ func secretPredicate() predicate.Predicate { } // Maps secret referenced by object store to the object store CR -func mapSecretToCR(k8sClient client.Client) func(context.Context, client.Object) []reconcile.Request { - return func(ctx context.Context, obj client.Object) []reconcile.Request { - secret, ok := obj.(*corev1.Secret) - if !ok { - return nil - } - +func mapSecretToCR(k8sClient client.Client) handler.TypedMapFunc[*corev1.Secret, reconcile.Request] { + return func(ctx context.Context, secret *corev1.Secret) []reconcile.Request { // lookup object store CRs by name objStores := cephv1.CephObjectStoreList{} err := k8sClient.List(ctx, &objStores, client.InNamespace(secret.Namespace)) diff --git a/pkg/operator/ceph/object/cosi/controller.go b/pkg/operator/ceph/object/cosi/controller.go index 22a400e44..175fd4254 100644 --- a/pkg/operator/ceph/object/cosi/controller.go +++ b/pkg/operator/ceph/object/cosi/controller.go @@ -94,13 +94,27 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes to CephCOSIDriver - err = controller.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephCOSIDriver{}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = controller.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephCOSIDriver{}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephCOSIDriver]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephCOSIDriver](mgr.GetScheme()), + ), + ) if err != nil { return errors.Wrap(err, "failed to watch for CephCOSIDriver object changes") } // Watch for changes to CephObjectStore as arbitrary resource and predicate functions - err = controller.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephObjectStore{}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = controller.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephObjectStore{}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephObjectStore]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephObjectStore](mgr.GetScheme()), + ), + ) if err != nil { return errors.Wrap(err, "failed to watch for CephObjectStore object changes") } diff --git a/pkg/operator/ceph/object/notification/controller.go b/pkg/operator/ceph/object/notification/controller.go index 722cf2a64..ec398d366 100644 --- a/pkg/operator/ceph/object/notification/controller.go +++ b/pkg/operator/ceph/object/notification/controller.go @@ -102,7 +102,14 @@ func addNotificationReconciler(mgr manager.Manager, r reconcile.Reconciler) erro logger.Info("successfully started") // Watch for changes on the OBC CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephBucketNotification{}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephBucketNotification{}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephBucketNotification]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephBucketNotification](mgr.GetScheme()), + ), + ) if err != nil { return err } diff --git a/pkg/operator/ceph/object/notification/obc_label_controller.go b/pkg/operator/ceph/object/notification/obc_label_controller.go index 226144d63..4b849c3e1 100644 --- a/pkg/operator/ceph/object/notification/obc_label_controller.go +++ b/pkg/operator/ceph/object/notification/obc_label_controller.go @@ -19,6 +19,7 @@ package notification import ( "context" + "reflect" "strings" bktv1alpha1 "github.com/kube-object-storage/lib-bucket-provisioner/pkg/apis/objectbucket.io/v1alpha1" @@ -37,8 +38,10 @@ import ( "k8s.io/client-go/tools/record" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" "sigs.k8s.io/controller-runtime/pkg/source" ) @@ -57,6 +60,47 @@ type ReconcileOBCLabels struct { recorder record.EventRecorder } +func obcPredicate[T *bktv1alpha1.ObjectBucketClaim]() predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { + obj := (*bktv1alpha1.ObjectBucketClaim)(e.Object) + + logger.Debugf("create event from a CR: %q", obj.GetName()) + return true + }, + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { + obj := (*bktv1alpha1.ObjectBucketClaim)(e.Object) + + logger.Debugf("delete event from a CR: %q", obj.GetName()) + return true + }, + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + // break generic-ness in order to access .Spec.ObjectBucketName + objOld := (*bktv1alpha1.ObjectBucketClaim)(e.ObjectOld) + objNew := (*bktv1alpha1.ObjectBucketClaim)(e.ObjectNew) + + logger.Debug("update event on ObjectBucketClaim CR") + // If the labels "do_not_reconcile" is set on the object, let's not reconcile that request + if opcontroller.IsDoNotReconcile(objNew.GetLabels()) { + logger.Debugf("object %q matched on update but %q label is set, doing nothing", objNew.GetName(), opcontroller.DoNotReconcileLabelName) + return false + } + if !reflect.DeepEqual(objOld.GetLabels(), objNew.GetLabels()) { + logger.Infof("CR labels has changed for %q", objNew.GetName()) + return true + } else if objOld.Spec.ObjectBucketName != objNew.Spec.ObjectBucketName { + logger.Infof("CR %q bucket name changed from %q to %q", objNew.GetName(), objOld.Spec.ObjectBucketName, objNew.Spec.ObjectBucketName) + return true + } + logger.Debugf("no change in CR %q", objNew.GetName()) + return false + }, + GenericFunc: func(e event.TypedGenericEvent[T]) bool { + return false + }, + } +} + func addOBCLabelReconciler(mgr manager.Manager, r reconcile.Reconciler) error { // Create a new controller c, err := controller.New(controllerName, mgr, controller.Options{Reconciler: r}) @@ -66,7 +110,14 @@ func addOBCLabelReconciler(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the OBC CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &bktv1alpha1.ObjectBucketClaim{}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &bktv1alpha1.ObjectBucketClaim{}, + &handler.TypedEnqueueRequestForObject[*bktv1alpha1.ObjectBucketClaim]{}, + obcPredicate(), + ), + ) if err != nil { return err } diff --git a/pkg/operator/ceph/object/realm/controller.go b/pkg/operator/ceph/object/realm/controller.go index 4f8b3fd1f..4e59c99b9 100644 --- a/pkg/operator/ceph/object/realm/controller.go +++ b/pkg/operator/ceph/object/realm/controller.go @@ -106,7 +106,14 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the CephObjectRealm CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephObjectRealm{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephObjectRealm{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephObjectRealm]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephObjectRealm](mgr.GetScheme()), + ), + ) if err != nil { return err } diff --git a/pkg/operator/ceph/object/topic/controller.go b/pkg/operator/ceph/object/topic/controller.go index 71734a675..0250baf15 100644 --- a/pkg/operator/ceph/object/topic/controller.go +++ b/pkg/operator/ceph/object/topic/controller.go @@ -73,7 +73,14 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the CephBucketTopic CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephBucketTopic{}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephBucketTopic{}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephBucketTopic]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephBucketTopic](mgr.GetScheme()), + ), + ) if err != nil { return err } diff --git a/pkg/operator/ceph/object/user/controller.go b/pkg/operator/ceph/object/user/controller.go index bf3aef339..8488d47f2 100644 --- a/pkg/operator/ceph/object/user/controller.go +++ b/pkg/operator/ceph/object/user/controller.go @@ -110,21 +110,31 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the CephObjectStoreUser CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephObjectStoreUser{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephObjectStoreUser{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephObjectStoreUser]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephObjectStoreUser](mgr.GetScheme()), + ), + ) if err != nil { return err } // Watch secrets - ownerRequest := handler.EnqueueRequestForOwner( - mgr.GetScheme(), - mgr.GetRESTMapper(), - &cephv1.CephObjectStoreUser{}, + err = c.Watch( + source.Kind( + mgr.GetCache(), + &corev1.Secret{TypeMeta: metav1.TypeMeta{Kind: "Secret", APIVersion: corev1.SchemeGroupVersion.String()}}, + handler.TypedEnqueueRequestForOwner[*corev1.Secret]( + mgr.GetScheme(), + mgr.GetRESTMapper(), + &cephv1.CephObjectStoreUser{}, + ), + opcontroller.WatchPredicateForNonCRDObject[*corev1.Secret](&cephv1.CephObjectStoreUser{TypeMeta: controllerTypeMeta}, mgr.GetScheme()), + ), ) - secretSource := source.Kind[client.Object](mgr.GetCache(), &corev1.Secret{TypeMeta: metav1.TypeMeta{Kind: "Secret", APIVersion: corev1.SchemeGroupVersion.String()}}, ownerRequest, - opcontroller.WatchPredicateForNonCRDObject(&cephv1.CephObjectStoreUser{TypeMeta: controllerTypeMeta}, mgr.GetScheme()), - ) - err = c.Watch(secretSource) if err != nil { return err } diff --git a/pkg/operator/ceph/object/zone/controller.go b/pkg/operator/ceph/object/zone/controller.go index ef51bd712..105265130 100644 --- a/pkg/operator/ceph/object/zone/controller.go +++ b/pkg/operator/ceph/object/zone/controller.go @@ -114,7 +114,14 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the CephObjectZone CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephObjectZone{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephObjectZone{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephObjectZone]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephObjectZone](mgr.GetScheme()), + ), + ) if err != nil { return err } diff --git a/pkg/operator/ceph/object/zonegroup/controller.go b/pkg/operator/ceph/object/zonegroup/controller.go index 111d52138..60420c956 100644 --- a/pkg/operator/ceph/object/zonegroup/controller.go +++ b/pkg/operator/ceph/object/zonegroup/controller.go @@ -97,7 +97,14 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the CephObjectZoneGroup CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephObjectZoneGroup{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephObjectZoneGroup{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephObjectZoneGroup]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephObjectZoneGroup](mgr.GetScheme()), + ), + ) if err != nil { return err } diff --git a/pkg/operator/ceph/pool/controller.go b/pkg/operator/ceph/pool/controller.go index e1ccf71e2..b04af10dc 100644 --- a/pkg/operator/ceph/pool/controller.go +++ b/pkg/operator/ceph/pool/controller.go @@ -114,23 +114,39 @@ func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reco logger.Info("successfully started") // Watch for changes on the CephBlockPool CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephBlockPool{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephBlockPool{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephBlockPool]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephBlockPool](mgr.GetScheme()), + ), + ) if err != nil { return err } // Build Handler function to return the list of ceph block pool // This is used by the watchers below - handlerFunc, err := opcontroller.ObjectToCRMapper(opManagerContext, mgr.GetClient(), &cephv1.CephBlockPoolList{}, mgr.GetScheme()) + handlerFunc, err := opcontroller.ObjectToCRMapper[*cephv1.CephBlockPoolList, *corev1.ConfigMap]( + opManagerContext, + mgr.GetClient(), + &cephv1.CephBlockPoolList{}, + mgr.GetScheme(), + ) if err != nil { return err } // Watch for ConfigMap "rook-ceph-mon-endpoints" update and reconcile, which will reconcile update the bootstrap peer token - cmSource := source.Kind[client.Object](mgr.GetCache(), &corev1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: corev1.SchemeGroupVersion.String()}}, - handler.EnqueueRequestsFromMapFunc(handlerFunc), mon.PredicateMonEndpointChanges(), + err = c.Watch( + source.Kind( + mgr.GetCache(), + &corev1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: corev1.SchemeGroupVersion.String()}}, + handler.TypedEnqueueRequestsFromMapFunc(handlerFunc), + mon.PredicateMonEndpointChanges(), + ), ) - err = c.Watch(cmSource) if err != nil { return err } diff --git a/pkg/operator/ceph/pool/radosnamespace/controller.go b/pkg/operator/ceph/pool/radosnamespace/controller.go index d03f35638..18a4d5de7 100644 --- a/pkg/operator/ceph/pool/radosnamespace/controller.go +++ b/pkg/operator/ceph/pool/radosnamespace/controller.go @@ -108,7 +108,14 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { logger.Info("successfully started") // Watch for changes on the CephBlockPoolRadosNamespace CRD object - err = c.Watch(source.Kind[client.Object](mgr.GetCache(), &cephv1.CephBlockPoolRadosNamespace{TypeMeta: controllerTypeMeta}, &handler.EnqueueRequestForObject{}, opcontroller.WatchControllerPredicate())) + err = c.Watch( + source.Kind( + mgr.GetCache(), + &cephv1.CephBlockPoolRadosNamespace{TypeMeta: controllerTypeMeta}, + &handler.TypedEnqueueRequestForObject[*cephv1.CephBlockPoolRadosNamespace]{}, + opcontroller.WatchControllerPredicate[*cephv1.CephBlockPoolRadosNamespace](mgr.GetScheme()), + ), + ) if err != nil { return err } diff --git a/pkg/operator/ceph/predicate.go b/pkg/operator/ceph/predicate.go index 578ba264b..2b353ef21 100644 --- a/pkg/operator/ceph/predicate.go +++ b/pkg/operator/ceph/predicate.go @@ -18,59 +18,54 @@ limitations under the License. package operator import ( - "context" - "github.com/rook/rook/pkg/operator/ceph/controller" - v1 "k8s.io/api/core/v1" - "sigs.k8s.io/controller-runtime/pkg/client" + corev1 "k8s.io/api/core/v1" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/predicate" ) // predicateController is the predicate function to trigger reconcile on operator configuration cm change -func predicateController(ctx context.Context, client client.Client) predicate.Funcs { - return predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { - if cm, ok := e.Object.(*v1.ConfigMap); ok { - return cm.Name == controller.OperatorSettingConfigMapName - } - return false +func operatorSettingConfigMapPredicate[T *corev1.ConfigMap]() predicate.TypedFuncs[T] { + return predicate.TypedFuncs[T]{ + CreateFunc: func(e event.TypedCreateEvent[T]) bool { + obj := (*corev1.ConfigMap)(e.Object) + + return obj.GetName() == controller.OperatorSettingConfigMapName }, - UpdateFunc: func(e event.UpdateEvent) bool { - if old, ok := e.ObjectOld.(*v1.ConfigMap); ok { - if new, ok := e.ObjectNew.(*v1.ConfigMap); ok { - if old.Name == controller.OperatorSettingConfigMapName && new.Name == controller.OperatorSettingConfigMapName { - if old.Data["ROOK_CURRENT_NAMESPACE_ONLY"] != new.Data["ROOK_CURRENT_NAMESPACE_ONLY"] { - logger.Debug("ROOK_CURRENT_NAMESPACE_ONLY config updated, reloading the manager") - controller.ReloadManager() + UpdateFunc: func(e event.TypedUpdateEvent[T]) bool { + objOld := (*corev1.ConfigMap)(e.ObjectOld) + objNew := (*corev1.ConfigMap)(e.ObjectNew) - // No need to ask for reconciliation since the context is going to be terminated when - // the signal is caught and the reconcile will run when the controller starts. - return false - } + if objOld.GetName() == controller.OperatorSettingConfigMapName && objNew.GetName() == controller.OperatorSettingConfigMapName { + if objOld.Data["ROOK_CURRENT_NAMESPACE_ONLY"] != objNew.Data["ROOK_CURRENT_NAMESPACE_ONLY"] { + logger.Debug("ROOK_CURRENT_NAMESPACE_ONLY config updated, reloading the manager") + controller.ReloadManager() - // We still want to reconcile the operator manager if the configmap is updated - return true - } - } - } - - return false - }, - - DeleteFunc: func(e event.DeleteEvent) bool { - if cm, ok := e.Object.(*v1.ConfigMap); ok { - if cm.Name == controller.OperatorSettingConfigMapName { - logger.Debug("operator configmap deleted, not reconciling") + // No need to ask for reconciliation since the context is going to be terminated when + // the signal is caught and the reconcile will run when the controller starts. return false } + + // We still want to reconcile the operator manager if the configmap is updated + return true } return false }, - GenericFunc: func(e event.GenericEvent) bool { + DeleteFunc: func(e event.TypedDeleteEvent[T]) bool { + obj := (*corev1.ConfigMap)(e.Object) + + if obj.GetName() == controller.OperatorSettingConfigMapName { + logger.Debug("operator configmap deleted, not reconciling") + return false + } + + return false + }, + + GenericFunc: func(e event.TypedGenericEvent[T]) bool { return false }, }