Concurrency rules between handlers

[kwohlfahrt]

What are the concurrency rules between handlers? For example, for my CRD, I have these async handlers:

1. A create/update handler, watching spec.replicas on my CRD, and creating that many worker pods.
2. A timer, which polls an external API that specifies the number of pods that should be spun up/down. It modifies spec.replicas, and if the number of active pods should be reduced, it calls another external API, that causes some worker pods to gracefully exit. This handler also records on the CRD the number of pods that are expected to gracefully exit (status.completing).
3. A completion handler on the pods, that deletes pods which have gracefully exited.
4. A deletion handler on the pods, that decrements on the CRD the number of pods we are expecting to exit (status.completing). If the field is 0, then a pod has unexpectedly exited, and should be recreated.

The problem is that (2) and (5)(4) both write to the same field, the number of pods that are expected to gracefully shutdown. They also perform a read/modify/write update, because they read the current value, increment/decrement it, and then write it back. Is this safe to do, or could (2) and (5)(4) run concurrrently, and I need to have a mutex on the field? What about multiple instances of (5)(4)?

I haven't been able to find any description in the documentation about what things can run concurrently or not, mostly I'm concerned about:

• Can a single handler watching be invoked several times concurrently?
• If not, what about multiple different handlers?
• If I read a field in a handler, calculate a new value, and then write it back, can I assume the value hasn't been mutated (by kopf at least) in the meantime? Or do I need to lock it in all cases?

I'm also happy to consider alternate patterns that don't require two handlers writing to the same field, but I can't think of a way to avoid it easily...

[nolar]

It was somewhere is the docs, but the general rule is: one resource object is always sequential within its individual stream/lifecycle, multiple objects and kinds are concurrent to each other. This rule applies to short-running handlers only (creation/update/deletion/raw events).

Timers & daemons run in parallel by design. You can consider them as side tools running elsewhere doing the updates.

In neither case you can assume that the field is not patched by anything else while processing the change. It can be changed by any other application, such as kubectl, other operators, or the same operator accidentally running the 2nd instance.

Kopf does not mutate domain-specific fields because it does not understand the domain. It only mutates its own annotations, finalizers, and whatever you code in the operator.

Sorry, I didn’t get the pattern with 2 handlers, so as what is (5) — there are only 4 points. So I cannot hint on a better approach.

But you can consider using Kopf’s indexes to keep track of pods belonging to a resource in a daemon/timer. It will add some inertia to the operator (e.g., 0.1-1 seconds) but might make it easier to track child pods.

[kwohlfahrt]

It was somewhere is the docs, but the general rule is: one resource object is always sequential within its individual stream/lifecycle, multiple objects and kinds are concurrent to each other.

OK, thank you! This answers the core of my question. And thanks as well for the quick response.

Sorry, I didn’t get the pattern with 2 handlers, so as what is (5) — there are only 4 points. So I cannot hint on a better approach.

Sorry, this should have been (4) - I reworded some things and didn't proofread well enough. I have updated the original question to refer to the correct numbers. In case it helps, I'll try to reword it:

I think I need at least three handlers, because I have three sources of events:

1. Updates to my CRD, which changes the expected number of worker pods. This is done via a create/update handler on the CRD.
2. The external API indicating that the number of required workers has changed. This is polled via a timer, which updates the CRD, triggering the update handler. On scale-down, it also commands some pods to gracefully exit.
3. A pod gracefully exiting or being unexpectedly deleted.

Both the timer and the pod deletion handler need to trigger the update handler, to create the correct number of worker pods. I could put all my logic into a timer/daemon, with an index, but this would incur either a large delay, or a lot of unnecessary execution of the operator, so I'd prefer the event-based approach.

My plan was to have both the timer and the pod deletion handler modify the CRD's status.completing field, with the timer incrementing it when commanding pods to exit and the deletion handler decrementing it, and the update handler watching it. But it seems this cannot work without careful locking because multiple pod deletion handlers can run concurrently, and they are all concurrent with the timer.

Instead, maybe I should just write some meaningless value (e.g. a random number, or timestamp) to a "last worker update" field, and just use it to trigger the update handler... or is there a cleaner way to trigger one handler from multiple event sources?

[nolar]

You can have multiple decorators on one function.

Meaningless values are never a good idea. In the worst case, you can do that in memory without touching the kubernetes cluster.

The typical approach, to my knowledge, is to store the “current” state in the status, the “desired” state in spec, and sync when anything changes. As far as I understand, this is exactly what you describe.

Also, you can create a per-resource lock/queue and store it to “memo” (a kwarg) for all handlers/timers/daemons to share, thus syncing the background timers/daemons with handlers.

In this case, however, you should call the patching api yourself (under the lock) and not delegate it to kopf, since there will be a small pause between the release of the lock in your handlers and the actual patch by kopf.

[kwohlfahrt]

You can have multiple decorators on one function.

Yes, but this doesn't affect kopf's internal synchronization, right? The function will still be called concurrently for the different pods/CRDs it manages if I understand correctly.

The typical approach, to my knowledge, is to store the “current” state in the status, the “desired” state in spec, and sync when anything changes. As far as I understand, this is exactly what you describe.

Yes, the problem is that there are multiple handlers that need to write to the CRD's status (the pod deletion handler, and the timer on the CRDs), so I need to be cautious of concurrency here.

Also, you can create a per-resource lock/queue and store it to “memo” (a kwarg) for all handlers/timers/daemons to share, thus syncing the background timers/daemons with handlers.

OK, yes, I can see how this would work.

In this case, however, you should call the patching api yourself (under the lock) and not delegate it to kopf, since there will be a small pause between the release of the lock in your handlers and the actual patch by kopf.

Yes, that makes sense. I also don't think it's even possible to patch my CRD from the pod deletion handler via kopf - you can patch the object the handler is attached to via the patch kwarg, or the status via the return value, but not from different resources.

Thank you very much for the detailed explanation, you're really going above and beyond as mantainer here 🙏

[nolar]

Correct, it will run concurrently across objects. Do you want to sync within one custom resource object, or across multiple objects too?

Yes, the problem is that there are multiple handlers that need to write to the CRD's status (the pod deletion handler, and the timer on the CRDs), so I need to be cautious of concurrency here.

If you mean different handlers of different kinds, then you are right, the memo is local to a single resource object, so not shared between your resource & e.g. pods.

However, there is the so-called "global memo" when populated at the operator startup handler. It is shared (shallow-copied) to each and every resource & handler. Because of shallow coping, you can add mutable containers there: dicts, lists, etc. As such, you can e.g. make a global dict of locks, invent some "key" using the same logic in all handlers, and lock it. It would look like this (not tested):

import kopf
import collections
import threading


@kopf.on.startup()
def setup_locks(memo, **_):
    memo.locks = collections.defaultdict(threading.Lock)


@kopf.on.create('myresource')
@kopf.on.resume('myresource')
def res_changed(memo, namespace, name, **_):
    key = f"{namespace}/{name}"
    with memo.locks[key]:
        do_something_syncronous()


@kopf.on.update('pods', labels={'parent-res-name': kopf.PRESENT})
def pod_changed(memo, namespace, labels, **_):
    parent_resource_name = labels['parent-res-name']
    key = f"{namespace}/{parent_resource_name}"
    with memo.locks[key]:
        patch_parent_resource_safely()

It will have some minor memory leakage for created-by-abandoned locks for deleted resources, but this does not seem like a big deal. For a pure solution, that should be an index on 'myresources' with locks in it — in that case, it will be garbage collected on resource deletion — but it required more code.

PS: In the pure approach, you might also want a single global lock for the dict modification. But in Python, that role is performed by GIL, so seems safe enough.

[kwohlfahrt]

Correct, it will run concurrently across objects. Do you want to sync within one custom resource object, or across multiple objects too?

Just across one instance of my CRD, but across all pods created from it. So across both kinds (my CRD & pods) and multiple instances for pods.

For a pure solution, that should be an index on 'myresources' with locks in it

Oh, that's a clever solution.

I think I have all of the pieces I need, thanks again for your help!