Question: In the reconciliation of an object, is it possible to check if this object has been enqueued again?

[matheuscscp]

Context: In Flux we have just shipped a feature to fail faster when kustomize-controller is performing health checks on the manifests it has applied to the cluster. The implementation is essentially a Goroutine fired during the Flux Kustomization reconciliation that continuously checks if the revision of the source consumed by the Flux Kustomization has changed. If it did, then we bail out earlier from the health checks to let the new enqueued reconciliation request of the object take place and apply the new source revision. We want to go beyond and let this happen for any other things we watch beyond the Flux sources. For example, we also watch ConfigMaps and Secrets that have an effect on the final manifests kustomize-controller applies to the cluster. Instead of probing all the things that are relevant for a particular Flux Kustomization object, we would like to probe the reconciliation queue instead to take advantage of the fact that our watches are already in place and will enqueue reconciliation requests. This would be the ideal solution for us.

Is there any function/method of the manager or something else that can answer queries like IsEnqueued(reconcile.Request) bool? Would the project consider accepting this contribution in case it's not already available? Or in the worst case would it be possible/simple for us on our side to build a thin wrapper around the default queue implementation to add this functionality? I saw in another issue that now controller-runtime supports injecting your own queue implementation.

@stefanprodan

[matheuscscp]

I have started a thread in #sig-api-machinery to ask if a contribution for adding .Has(item) to the workqueue would be considered:

https://kubernetes.slack.com/archives/C0EG7JC6T/p1759596923301419

[alvaroaleman]

As mentioned in the sig-apimachinery thread, I don't think a Has is a good solution because what you really want is to get notified and needing to poll is inefficient. And to answer the original question no, there is no direct way to do that at the moment.

It might be nice to have this to cancel active reconciliations but that would also be a bit of a footgun, because if the reconciliation updates the object and that then causes a cancellation, the reconciliation has effectively just cancelled itself. We may be able to work around that in turn by adding a ShouldCancel paramter to AddWithOpts that for example only gets set to true if the generation has changed.

All of this is to say: I do see value in the feature, but there are some details we'd have to figure out. If you want to volunteer to look into that, that would be highly appreciated. Luckily, we can make it all happen downstream since we already have our own workqueue in the form of the priorityqueue.

[stefanprodan]

Hey @alvaroaleman we don't want to cancel the reconciliation, what we need is a way to be notified via a Go chan that the current object UID has been added to the queue. Based on this signal we want to make decisions which are very specific to each reconciler logic. Our primary use case is not reacting to just the generation change, but to various other events that we watch for to add an object to the queue.

[alvaroaleman]

Hey @alvaroaleman we don't want to cancel the reconciliation, what we need is a way to be notified via a Go chan that the current object UID has been added to the queue.

Re UIDs, the queue doesn't know anything about those, as it just holds a key. An easy way to get notified if the item gets added again is to write a wrapper around the queue.

based on this signal we want to make decisions which are very specific to each reconciler logic.

Any chance you could provide a specific example for that? I'd be curious to understand this use-case.

Our primary use case is not reacting to just the generation change, but to various other events that we watch for to add an object to the queue.

Do you mean you want to get notified not only if the object generation changes, but also if other objects change that causes the main object to get enqueued? The workqueue only holds object keys of the primary object and doesn't know anything at all about what triggered an addition, that is all in the sources. Meaning a wrapper should work if that is indeed what you meant.

[matheuscscp]

I don't think a Has is a good solution

Yep, very well noted. I started a PoC and ended up with something else entirely, much more aligned with what you said, I want to be notified when it happens and avoid polling. (By the way, polling is what we are doing right now, .Get()ing the Flux source object every 5 seconds to check if a new revision is available.)

This seems to be working-ish in my staging cluster, please check this out:

fluxcd/kustomize-controller@main...cancel-hc-requeue

I'm observing some races between the first reconciliations when the controller starts up and the queue being created, please see the comment I wrote about this in the diff

[matheuscscp]

Hmm I'm not sure these are really races, the error is happening on every reconciliation. I'm pretty sure it would be impossible to have any reconciliations before the queue is created right? Probably some mistake in the code. Will report back when I find it.

[matheuscscp]

Okay, it's fixed now. Seems to be working for real this time! Please check if this is making sense 🙏

In Flux we could definitely move QueueFactory and NotifyingQueue to our reusable packages in github.com/fluxcd/pkg and reuse this across our controllers, but I think there's nothing specific to Flux in this code, so maybe a contribution is worth it?

[matheuscscp]

I can confirm this is working pretty well after testing several times. When the watch event enqueues a reconciliation I see the ongoing reconciliation give up immediately and the new one take place (since my cluster is super chill and the queue is mostly empty all the time).

[matheuscscp]

One issue that I see with this implementation is that the reconciler will miss all the enqueue events since the reconciliation started up until calling .ContextWithCancelOnObjectEnqueued(), so I think the right place to call this method is on the very beginning of the reconciliation. In this case, how about making controller-runtime itself create this context and inject it as a value in the context passed to .Reconcile(ctx, req)?

import (
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/controller"
)

func (r *KustomizationReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {

	notifyingCtx := controller.GetNotifyingContext(ctx)
	// use notifyingCtx where needed

}

The calling code of .Reconcile() in controller-runtime would look somewhat like this (very confusing but correct):

type notifyingContextContextKey struct{}

var ctx context.Context // this is the context currently being passed directly to .Reconcile(ctx, req)

notifyingCtx, cancel := notifyingQueue.ContextWithCancelOnObjectEnqueued(ctx, req)
defer cancel()

ctxWithNotifyingCtx := context.WithValue(ctx, notifyingContextContextKey{}, notifyingCtx)

res, err := r.Reconcile(ctxWithNotifyingCtx, req)

And the helpers:

func GetNotifyingContext(ctx context.Context) context.Context {
	v := ctx.Value(notifyingContextContextKey{})
	if v != nil {
		return v.(context.Context)
	}
	return nil
}

func IsObjectEnqueued(ctx context.context) bool {
	notifyingCtx := GetNotifyingContext(ctx)
	return ctx.Err() == nil && notifyingCtx.Err() != nil
}

[stefanprodan]

Any chance you could provide a specific example for that? I'd be curious to understand this use-case.

For example, in Flux source-controller one of the reconcilers verifies the integrity of the fetched content, the content can have 1M files, so while we calculate the checksums we would like to abort this process if meanwhile the end-user changed the spec to point to a diffrent upstream revision (in this case, the generation will change).

Another example where it's not a generation change, in Flux kustomize-controller we run health checks for the deployed workloads with a timeout set by the user, this can be let's say 5m. While running the health checks, we would like to abort this process if meanwhile source-controller produced a new revision. The revision is not something set in spec, we have watchers of Flux sources which would add to the kustomize-controller queue all objects referencing that source.

In both cases, we don't want to "cancel" the reconciliation, instead we want to detect that the current object we reconcile was added to the queue. Having this information, we then can then decide if and when to gracefully stop some long running process.

[matheuscscp]

I brought my PoC a bit closer to a real implementation to make it easier to see what controller-runtime improvements could really help the Flux project.

Here's a controller-agnostic wrapper for the queue and for the reconciler adding the context notification functionality:

fluxcd/pkg@main...queue

Here's how this wrapper can be used in a real controller (this code is currently deployed in an EKS cluster and it's working pretty seamlessly):

fluxcd/kustomize-controller@main...cancel-hc-requeue

This code being erased in the diff is exactly Stefan's second example above:

Another example where it's not a generation change, in Flux kustomize-controller we run health checks for the deployed workloads with a timeout set by the user, this can be let's say 5m. While running the health checks, we would like to abort this process if meanwhile source-controller produced a new revision. The revision is not something set in spec, we have watchers of Flux sources which would add to the kustomize-controller queue all objects referencing that source.

How can controller-runtime help the Flux project in this improvement:

• While this kustomize-controller integration is already pretty seamless, the manual queue setup in .SetupWithManager() could be completely avoided if controller-runtime implemented this (opt-in, non-breaking) functionality by default. Controller authors would opt in by calling the helpers GetQueueContext(ctx) and IsObjectEnqueued(ctx).
• Due to our lack of control over when the queue is created, our implementation ends up with a dead code path to avoid swallowing this impossible error. See how scary this comment in the link is, and how it could be completely avoided if controller-runtime, as the owner of the initialization process, provided this functionality.
• Creating the queue ourselves is a bit scary. I wrote this code by going along the creation chain [rate-limiting queue -> delaying queue -> queue], copying the defaults and the options we configure in the Flux controllers (pretty much just the rateLimiter). It's not a ton of code to be copying, but if controller-runtime changes the default queue creation logic we need to keep up.
• This whole reconciler wrapper preparing and injecting the queue context looks ok, but not great. Really feels like something that should be part of controller-runtime.

@alvaroaleman I'd be more than happy to open a PR for this feature.

[csviri]

Just adding this, since might help, in Java Operator SDK method for this to check if there will be a reconiliation instantly after the current one:
https://github.com/operator-framework/java-operator-sdk/blob/148dcb68aa977cb0b9bc09aa95fbcf4394529c39/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Context.java#L74

But unfortunatelly we did not got much feadback on this for now.

[alvaroaleman]

In both cases, we don't want to "cancel" the reconciliation, instead we want to detect that the current object we reconcile was added to the queue. Having this information, we then can then decide if and when to gracefully stop some long running process.

Okay, so to describe this a but more abstract, if you want to cancel something depends on what you are currently doing? It is not something that you would want per reconciler based on the event? It is worth pointing out that you could use context.WithoutCancel to achieve the inverse, i.E. ignore cancellations under specific circumstances.

Creating the queue ourselves is a bit scary. I

Yeah, you already missed using the priorityqueue which from personal experience would be very useful for flux 🙃 I also just realized this won't work correctly with AddAfter and AddRatelimited.

This whole reconciler wrapper preparing and injecting the queue context looks ok, but not great. Really feels like something that should be part of controller-runtime.

As mentioned, I am not opposed to that idea, but if we do that I'd like it to be something that will work by default and correct most of the time, which is why I ended up suggesting to cancel the context. The use-case here appears to be "We want to get notified in the reconciler and then decide if we do something about it" and that seems like something that can not be made to work by default, as any reconciler that wants to use this would need to actively start dealing with it.

[matheuscscp]

I also just realized this won't work correctly with AddAfter and AddRatelimited.

Well, that is concerning! Can you please ellaborate🙏? I read the code of the rate-limiting and delaying queues, and their Add* methods seem to be ultimately calling .Add() at some point. In my implementation what I did was to wrap the queue that provides .Add(), pass it to the DelayingQueue, and then pass this one to the RateLimitingQueue. This should make it work for AddAfter and AddRatelimited, no? I did it like that aiming specifically to solve this problem. I'd love to understand where it fails! 🙏

Yeah, you already missed using the priorityqueue which from personal experience would be very useful for flux 🙃

Interesting! The comment says this:

// PriorityQueue is a priority queue for a controller. It
// internally de-duplicates all items that are added to
// it. ...

This property indeed sounds very important for Flux. I thought the default queue already did that. Is this feature available only in the priority queue?

It is not something that you would want per reconciler based on the event?

In our case, it's not the type of event that influences if an ongoing reconciliation must be cancelled or not. We essentially watch just updates, so our watch events have a simple outcome: "What's the set of managed objects that are affected by this update? Enqueue all of them." We do look at specific properties of the watched objects in our predicates, but once an object passes the predicate it's just a matter of computing what managed objects are affected by this update and enqueuing them.

Okay, so to describe this a but more abstract, if you want to cancel something depends on what you are currently doing?

The use-case here appears to be "We want to get notified in the reconciler and then decide if we do something about it" and that seems like something that can not be made to work by default, as any reconciler that wants to use this would need to actively start dealing with it.

It is worth pointing out that you could use context.WithoutCancel to achieve the inverse, i.E. ignore cancellations under specific circumstances.

Yeah, I guess you could put it like that. The kustomize-controller has to do several tasks before it reaches the point of performing health checks, some involving querying k8s objects through the controller-runtime client, some involving making network calls, some involving file system operations. Many of those operations receive a context. I'm not sure if it's okay for each and every one of them to be cancelled because a new reconciliation request for the same object was enqueued, but I'm sure it's okay for them to be cancelled if the process is under termination because this is the current behavior for years. The operation we are sure that can be interrupted and that has a strong reason for deserving this optimization is the health check, which is a read-only operation that just waits for state convergence, a process that in many cases locks the Goroutine for minutes.

I'd imagine that many other controllers in the ecosystem would be similar, i.e. it's not immediately clear that is okay to cancel most of the operations they perform in a reconciliation due to a requeue event of the same object, but there could be a few operations worth cancelling in this scenario. So I'm not sure addressing this use case in a binary way would be very helpful for most controller-runtime users. Flexibility seems to matter in this case. If we use this idea of inverting the problem with context.WithoutCancel we would need to do that for most of our operations, and I understand that this would force us to lose the ability of cancelling those operations when the process is under termination, which is our current behavior.

@csviri How was it in your case? Did you want to cancel any ongoing operation of the reconciliation, or just in case the operation was a specific one?

@stefanprodan Would it be okay to cancel any of the operations the kustomize-controller performs to reconcile an object when this object is requeued? I understand in Flux we strive to make an apply transactional despite the challenges imposed by Kubernetes, so I imagine it could be pretty bad for an apply to be cancelled halfway, having deployed only some of the workloads, as this would hurt an important design goal for us.

Currently, Flux users enable this feature through a binary flag in the controller process, so they can only enable this for all their workloads or for none of them. How I imagine the future of this feature in Flux, is allowing this behavior to be configured at the object level, through a .spec field. I think it depends on the type of workload being deployed if it makes sense to bail out early in the health checks or not. So even if it was okay to cancel the ongoing reconciliation during any of the underlying operations, we would still need to deal with enabling this for some objects but not for others.