feedc0de/my-rook-config
1
0 Fork 0
forked from rook/rook
Code Pull Requests Activity

core: typed watch handlers and predicates

Browse Source 
All existing controller runtime watches are converted to use "typed"
handlers and predicates instead of operating on `client.Object`.  The
intent is to be bug for bug equivalent with the existing logic while
replacing run time type assertions and switch statements with compile
time type constraints and type casts. In several cases, functions using
assertions were split up such that each function only handles a single
Kind at a time. It is hoped that this will improve readability and
maintainability while facilitating future refactoring such as migrating
some watches to using IndexFields.

Of particular note is that the massive switch statement in
`WatchControllerPredicate()`  from
`pkg/operator/ceph/controller/predicate.go` has been replaced with
generics, reflection, and splitting the obc logic into its own predicate
function. There are still many helper functions operating on
`client.Object`. These were not updated unless required by the compiler
in order to limit the size of this change. The type safety of these
funcs should be improved as followup work.

 It is strongly suggested that going forward, handlers and predicates
 only handle a single Kind (generic or not) and that switches / type
 assertions are heavily discouraged or forbidden. This PR removed all
 but a single switch statement in a predicate, which should be addressed
 in future work.

Signed-off-by: Joshua Hoblitt <josh@hoblitt.com>
(cherry picked from commit 9f1ed201db)
This commit is contained in:
Joshua Hoblitt authored and Mergify committed 2025-04-04 17:18:33 +00:00
1 parent ada4d97813
commit 5fb622e6d1
34 files changed
+954 -1003
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/operator/ceph/client/controller.go
+19 -13
View File
@@ -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
}
pkg/operator/ceph/cluster/controller.go
+55 -21
View File
@@ -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
}
pkg/operator/ceph/cluster/mon/predicate.go
+8 -14
View File
@@ -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")
pkg/operator/ceph/cluster/nodedaemon/add.go
+50 -59
View File
@@ -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
},
},
),
pkg/operator/ceph/cluster/predicate.go
+63 -70
View File
@@ -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
},
}
pkg/operator/ceph/cluster/predicate_test.go
-5
View File
@@ -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))
pkg/operator/ceph/cluster/rbd/controller.go
+24 -14
View File
@@ -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
}
pkg/operator/ceph/controller.go
+7 -5
View File
@@ -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
}
pkg/operator/ceph/controller/handler.go
+7 -14
View File
@@ -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
}
pkg/operator/ceph/controller/handler_test.go
+1 -1
View File
@@ -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))
}
pkg/operator/ceph/controller/predicate.go
+91 -389
View File
@@ -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()
}
pkg/operator/ceph/controller/predicate_test.go
-3
View File
@@ -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))
pkg/operator/ceph/csi/controller.go
+15 -11
View File
@@ -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
}
pkg/operator/ceph/csi/predicate.go
+71 -62
View File
@@ -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
},
}
pkg/operator/ceph/csi/predicate_test.go
+77 -75
View File
@@ -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))
})
}
pkg/operator/ceph/disruption/clusterdisruption/add.go
+90 -81
View File
@@ -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
pkg/operator/ceph/file/controller.go
+33 -12
View File
@@ -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
}
pkg/operator/ceph/file/mirror/controller.go
+24 -7
View File
@@ -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
}
pkg/operator/ceph/file/subvolumegroup/controller.go
+8 -1
View File
@@ -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
}
pkg/operator/ceph/nfs/controller.go
+24 -8
View File
@@ -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
}
pkg/operator/ceph/object/bucket/controller.go
+14 -9
View File
@@ -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
}
pkg/operator/ceph/object/bucket/predicate.go
+42 -31
View File
@@ -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
},
}
pkg/operator/ceph/object/controller.go
+44 -39
View File
@@ -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))
pkg/operator/ceph/object/cosi/controller.go
+16 -2
View File
@@ -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")
}
pkg/operator/ceph/object/notification/controller.go
+8 -1
View File
@@ -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
}
pkg/operator/ceph/object/notification/obc_label_controller.go
+52 -1
View File
@@ -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
}
pkg/operator/ceph/object/realm/controller.go
+8 -1
View File
@@ -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
}
pkg/operator/ceph/object/topic/controller.go
+8 -1
View File
@@ -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
}
pkg/operator/ceph/object/user/controller.go
+19 -9
View File
@@ -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
}
pkg/operator/ceph/object/zone/controller.go
+8 -1
View File
@@ -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
}
pkg/operator/ceph/object/zonegroup/controller.go
+8 -1
View File
@@ -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
}
pkg/operator/ceph/pool/controller.go
+21 -5
View File
@@ -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
}
pkg/operator/ceph/pool/radosnamespace/controller.go
+8 -1
View File
@@ -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
}
pkg/operator/ceph/predicate.go
+31 -36
View File
@@ -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
},
}