<p>I have the following KOPF Daemon running:</p>
<pre><code>import kopf
import kubernetes

@kopf.on.daemon(group='test.example.com', version='v1', plural='myclusters')
def worker_services(namespace, name, spec, status, stopped, logger, **kwargs):
    config = kubernetes.client.Configuration()
    client = kubernetes.client.ApiClient(config) 
    workload = kubernetes.client.CoreV1Api(client)
    watch = kubernetes.watch.Watch()
    while not stopped:
        for e in watch.stream(workload.list_service_for_all_namespaces):
            svc = e['object']
            lb = helpers.get_service_loadbalancer(name, namespace, svc, logger)
            if &quot;NodePort&quot; in svc.spec.type:
                logger.info(f&quot;Found Service of type NodePort: {svc.metadata.name}&quot;)
                do_some_work(svc)
    watch.stop()
</code></pre>
<p>When the system exits by means of <code>Ctrl + C</code> or Kubernetes killing the pod, I get the following Warning:</p>
<pre><code>INFO:kopf.reactor.running:Signal SIGINT is received. Operator is stopping.
[2020-12-11 15:07:52,107] kopf.reactor.running [INFO    ] Signal SIGINT is received. Operator is stopping.
WARNING:kopf.objects:Daemon 'worker_services' did not exit in time. Leaving it orphaned.
[2020-12-11 15:07:52,113] kopf.objects         [WARNING ] Daemon 'worker_services' did not exit in time. Leaving it orphaned.
</code></pre>
<p>This keeps the process running in the background even if I press <code>Ctrl + Z</code>.</p>
<p>I believe the <code>for loop</code> is holding up the process with the stream and does not terminate when the system exits, thus it is not hitting the <code>watch.stop()</code> on the last line of this snippet.</p>
<p>I have tried the following thus far:</p>
<ul>
<li>Adding a <code>watch.stop()</code> after the <code>do_some_work(svc)</code>, but this sends my program into a very aggressive loop consuming up to 90% of my CPU</li>
<li>Putting the whole <code>for loop</code> on a different thread, this made some components fail such as the logger</li>
<li>Implemented <code>yield e</code> to make the process non-blocking, this made the daemon complete after the first service that it watched and the watch ended</li>
<li>Implemented Signal Listeners using <code>signal</code> library to listen for <code>SIGINT</code> and then <code>watch.stop()</code> in the exit function, but the function never got called</li>
<li>Implemented <code>cancellation_timeout=3.0</code> i.e. <code>@kopf.on.daemon(group='test.example.com', version='v1', plural='myclusters', cancellation_timeout=3.0)</code> with some of the above mentioned solutions, but also with no success</li>
</ul>
<p>Any input would be appreciated, thanks in advance.</p>

Barend
<p>What I see in your example is that the code tries to watch over the resources in the cluster. However, it uses the official client library, which is synchronous. Synchronous functions (or threads) cannot be interrupted in Python, unlike asynchronous (which also require async i/o to be used). Once the function shown here is called, it never exits and has no point where it checks for the stopped flag during that long run.</p>
<p>What you can do with the current code, is to check for the <code>stopped</code> flag more often:</p>
<pre class="lang-py prettyprint-override"><code>@kopf.on.daemon(group='test.example.com', version='v1', plural='myclusters')
def worker_services(namespace, name, spec, status, stopped, logger, **kwargs):
    …
    watch = kubernetes.watch.Watch()
    for e in watch.stream(workload.list_service_for_all_namespaces):
        if stopped:  # &lt;&lt;&lt;&lt; check inside of the for-loop
            break
        svc = …
        ………
    watch.stop()
</code></pre>
<p>This will check if the daemon is stopped on every event of every service. However, it will not check for the stop-flag if there is complete silence (it happens).</p>
<p>To work around that, you can limit the watch by time (please check with the client's documentation on how this is done properly, but iirc, this way):</p>
<pre class="lang-py prettyprint-override"><code>watch = kubernetes.watch.Watch()
for e in watch.stream(workload.list_service_for_all_namespaces, timeout_seconds=123):
</code></pre>
<p>That will limit the daemon's time of no response/cancellation to 123 seconds at most — in case of no services are available in the cluster or they are not changed.</p>
<p>For that case, you have no need to check for the <code>stopped</code> condition outside of the for-loop, as the daemon function will exit with an intention to be restarted, the <code>stopped</code> will be checked by the framework, and it will not restart the function — as intended.</p>
<hr />
<p>On a side note, I should notice that watching for resources inside of the handlers might be not the best idea. Watching is complicated. Way too complicated with all the edge cases and issues it brings.</p>
<p>And since the framework already does the watching, it might be easier to utilise that, and implement the cross-resource connectivity via the operator's global state:</p>
<pre class="lang-py prettyprint-override"><code>import queue

import kopf

SERVICE_QUEUES = {}  # {(mc_namespace, mc_name) -&gt; queue.Queue}
KNOWN_SERVICES = {}  # {(svc_namespace, svc_name) -&gt; svc_body}


@kopf.on.event('v1', 'services')
def service_is_seen(type, body, meta, event, **_):

    for q in SERVICE_QUEUES.values():  # right, to all MyClusters known to the moment
        q.put(event)

    if type == 'DELETED' or meta.get('deletionTimestamp'):
        if (namespace, name) in KNOWN_SERVICES:
            del KNOWN_SERVICES[(namespace, name)]
    else:
        KNOWN_SERVICES[(namespace, name)] = body


@kopf.on.daemon(group='test.example.com', version='v1', plural='myclusters')
def worker_services(namespace, name, spec, status, stopped, logger, **kwargs):
    # Start getting the updates as soon as possible, to not miss anything while handling the &quot;known&quot; services.
    q = SERVICE_QUEUES[(namespace, name)] = queue.Queue()
    try:

        # Process the Services known before the daemon start/restart.
        for (svc_namespace, svc_name), svc in KNOWN_SERVICES.items():
            if not stopped:
                lb = helpers.get_service_loadbalancer(name, namespace, svc, logger)
                if &quot;NodePort&quot; in svc.spec['type']:
                    logger.info(f&quot;Found Service of type NodePort: {svc.metadata.name}&quot;)
                    do_some_work(svc)

        # Process the Services arriving after the daemon start/restart.
        while not stopped:
            try:
                svc_event = q.get(timeout=1.0)
            except queue.Empty:
                pass
            else:
                svc = svc_event['object']
                lb = helpers.get_service_loadbalancer(name, namespace, svc, logger)
                if &quot;NodePort&quot; in svc.spec['type']:
                    logger.info(f&quot;Found Service of type NodePort: {svc.metadata.name}&quot;)
                    do_some_work(svc)

    finally:
        del SERVICE_QUEUES[(namespace, name)]
</code></pre>
<p>It is a simplified example (but might work &quot;as is&quot; — I didn't check) — only to show an idea on how to make the resources talk to each other while using the framework's capabilities.</p>
<p>The solution depends on the use-case, and this solution might be not applicable in your intended case. Maybe I miss something on why it is done that way. It would be good if you report your use-case to Kopf's repo as a feature request so that it could be supported by the framework later.</p>
