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