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?

[nolar]

(Ouch… The conversion to a discussion did not go well.)

Hello.

Sorry, there are no plans to implement dynamic handlers in the foreseeable future, probably at all. The reason is, as described, the enormous complexity (including the further maintenance of the codebase) with little benefit in rare and very specialized use-cases.

Since there is a workaround mentioned above ("Way 1"), this scenario can be implemented now too. Actually, with in-memory indexes, it can be made even simpler (I do not know why I did not use them in the response above):

import kopf

# For the resource that controls what to serve:

@kopf.index('metaresource')
def metaplurals(spec, **_):
    return {(spec['selector']['group'], spec['selector']['plural']): None}

# For regular resources being served:

def is_of_interest(resource, metaplurals, **_):
    return (resource.group, resource.plural) in metaplurals

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

This is short enough and should solve the described problem.

If you have some other scenario that cannot be implemented with the current features, please describe it here.

[brianthelion]

@nolar

Assuming I want to implement "Way 3" as follows, do you have any guidance on how the registry.handler_setup(...) and registry.handler_teardown(...) methods should be implemented?

LOCAL_REFS = {}

@kopf.on.create('metaresource')
@kopf.on.resume('metaresource')
def register(spec, memo, **_):
    registry = kopf.get_default_registry()
    def dynamic_handler(resource, name, **_):
        print(resource, name)
    handler_ref = registry.handler_setup(dynamic_handler, ...) # <== proposed addition to OperatorRegistry class
    LOCAL_REFS[...] = handler_ref

@kopf.on.delete('metaresource')
def unregister(spec, memo, **_):
    registry = kopf.get_default_registry()
    handler_ref = LOCAL_REFS.pop(...)
    registry.handler_teardown(handler_ref) # <== proposed addition to OperatorRegistry class

[maxnasonov]

I started calling kopf.on.event decorators as methods to register new watchers. This works as expected.

def manifest_created(**kwargs):
   print("A new manifest is created")

@kop.on.create('group','v1', 'crd')
def create(**kwargs):
   # this is acquired from the watched manifest
   r = ('group', 'version', 'kind') 
   kopf.on.create(*r)(manifest_created)

Is it still expected to work? I'm trying to implement it with kopf==1.36.1:

import kopf

def manifest_created(logger, **_):
    logger.info("A new manifest is created")

@kopf.on.resume('', 'v1', 'ConfigMap')
@kopf.on.create('', 'v1', 'ConfigMap')
def on_create_clusterobject(spec, namespace, logger, **_):
    kopf.on.create('', 'v1', 'Secret')(manifest_created)
    logger.info(kopf.get_default_registry()._changing.__dict__)

and I see the handler in the registry:

ChangingHandler(id='manifest_created', fn=<function manifest_created at 0x107641c60>, param=None, errors=None, timeout=None, retries=None, backoff=None, selector=Selector(group='', version='v1', any_name='Secret'), labels=None, annotations=None, when=None, field=None, value=None, reason=<Reason.CREATE: 'create'>, initial=None, deleted=None, requires_finalizer=None, field_needs_change=False, old=None, new=None)`

but a Secret creation event does not trigger it (no logs generated by kopf run -v -A test.py). Adding the decorator directly to the function works as expected.