register new handlers during run (dynamically generating handlers)

[smokaleksander]

Question

Is there a possibility to register new handlers while kopf is already running, like creating dynamically handlers ? I only want to specify new resources as the handler method will be always the same.

Checklist

• I have read the documentation and searched there for the problem
• I have searched in the GitHub Issues for similar questions

Keywords

dynamic handlers

[nolar]

Hello. Can you please briefly describe the use-case behind this supposed solution?

[smokaleksander]

The scenario I have in my head is to apply CRD with resource type specified in it, e.g. Services and to trigger some method, defined earlier. Then I would like to apply new CRD with specified another type of resource and call that method again.

[nolar]

Related: #378

[nolar]

Currently, the operator is assumed to be statically formed. Dynamic registration/removal of handlers brings some complications to the API/DSL (not so much to the implementation). However, it might be implemented indefinitely later.

Specifically, if handlers are dynamically registered, they must be dynamically viewable/iterable and dynamically removable. For that, a handler token must be returned and stored in memory by the operator for reference to the handler in the operator's internals (the handler itself can be a token though, as it is immutable).

As of now, two ways are possible to implement this scenario:

---

Way 1: watch for everything, react only to what is expected.

import kopf

@kopf.on.startup()
def init(memo, **_):
    memo.handled_resources = []  # list[tuple(group, plural), ...]

# For the resource that controls what to serve:

@kopf.on.create('metaresource')
@kopf.on.resume('metaresource')
def register(spec, memo, **_):
    t = (spec['selector']['group'], spec['selector']['plural'])
    memo.handled_resources.append(t)

@kopf.on.delete('metaresource')
def unregister(spec, memo, **_):
    t = (spec['selector']['group'], spec['selector']['plural'])
    if t in memo.handled_resources:
        memo.handled_resources.remove(t)

# For regular resources being served:

def is_of_interest(resource, memo, **_):
    t = (resource.group, resource.plural)
    return t in memo.handled_resources

@kopf.on.event(kopf.EVERYTHING, when=is_of_interest)
def fn(resource, name, **_):
    print(resource, name)

The downside is that the operator will listen for everything (really everything). This adds extra load on the API servers (extra 10-20-30 connections — depends on how many resources you have). And pollutes logs with the watch-stream starting. Unrelated resources are not mentioned in the logs, they are filtered out silently — so it is not a big problem, just some hassle.

---

Way 2: Register specialised handlers in the meta-resource creation.

import kopf

@kopf.on.create('metaresource')
@kopf.on.resume('metaresource')
def register(spec, memo, **_):
    @kopf.on.event(spec['selector']['group'], spec['selector']['plural'])
    def fn(resource, name, **_):
        print(resource, name)

The upside: It only listens/watches for what was registered at least once, not for everything.

The downside 1: beware of the closures. Here, spec & memo are of the meta-resource, not of the served resource. You need to add spec & memo to fn() to get the proper values.

The downside 2: there is no de-registration (removal). So, the handlers remain forever. When the meta-resource is removed, the fn() remains there. Some memory leaks are possible over time (when meta-resources are added/removed often). Double-registration is also possible (when meta-resource are removed and then re-added for the same target).

To partially mitigate that, a bool-container can be used to toggle handlers off later (e.g. an Event, but anything would work):

import kopf
import threading

@kopf.on.create('metaresource')
@kopf.on.resume('metaresource')
def register(spec, memo, **_):
    memo.disabled = disabled = threading.Event()  # initially False

    @kopf.on.event(spec['selector']['group'], spec['selector']['plural'], when=lambda **_: not disabled.is_set())
    def fn(resource, name, **_):
        print(resource, name)

@kopf.on.delete('metaresource')
def unregister(spec, memo, **_):
    memo.disabled.set()  # deactivate the fn-handler (but it stays in memory)

Still, the watch-streams of un-registered resources will remain there — even if the handlers are disabled. Restarting the operator from time to time (e.g. daily) can help with cleanups.

---

Both ways are hacky. I will think about a nicer way with an extra API of the framework.

[brianthelion]

@nolar Any updates on this issue since 2021?