如何在系统退出时优雅地停止 Kubernetes Watch on Services

Question

我正在运行以下 KOPF 守护进程：

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 "NodePort" in svc.spec.type:
                logger.info(f"Found Service of type NodePort: {svc.metadata.name}")
                do_some_work(svc)
    watch.stop()

当系统通过Ctrl + C 或 Kubernetes 杀死 pod 退出时，我收到以下警告：

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.

即使我按Ctrl + Z，这也会使进程在后台运行。

我相信for loop 正在通过流阻止进程并且在系统退出时不会终止，因此它没有在此 sn-p 的最后一行命中watch.stop()。

到目前为止，我已经尝试了以下方法：

• 在do_some_work(svc) 之后添加watch.stop()，但这会使我的程序进入一个非常激进的循环，最多消耗我 90% 的 CPU
• 将整个 for loop 放在不同的线程上，这会导致某些组件失败，例如记录器
• 实现了yield e 以使进程非阻塞，这使得守护进程在它观察到的第一个服务和观察结束后完成
• 使用 signal 库实现的信号监听器在退出函数中监听 SIGINT 和 watch.stop()，但该函数从未被调用
• 使用上述一些解决方案实现了cancellation_timeout=3.0，即@kopf.on.daemon(group='test.example.com', version='v1', plural='myclusters', cancellation_timeout=3.0)，但也没有成功

任何意见将不胜感激，在此先感谢。

Answer 1

我在您的示例中看到的是代码试图监视集群中的资源。但是，它使用的是同步的官方客户端库。与异步（也需要使用异步 i/o）不同，Python 中的同步函数（或线程）不能被中断。一旦调用了此处显示的函数，它就永远不会退出，并且在长时间运行期间没有任何检查停止标志的地方。

您可以使用当前代码执行的操作是更频繁地检查stopped 标志：

@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:  # <<<< check inside of the for-loop
            break
        svc = …
        ………
    watch.stop()

这将检查守护程序是否在每个服务的每个事件上都停止。但是，如果完全静默（发生），它不会检查停止标志。

要解决这个问题，您可以按时间限制手表（请查看客户的文档以了解如何正确完成此操作，但 iirc，这种方式）：

watch = kubernetes.watch.Watch()
for e in watch.stream(workload.list_service_for_all_namespaces, timeout_seconds=123):

这会将守护程序的无响应/取消时间限制为最多 123 秒 - 如果集群中没有可用的服务或它们没有更改。

对于这种情况，您无需在 for 循环之外检查 stopped 条件，因为守护程序函数将退出并打算重新启动，框架将检查 stopped，并且它不会按预期重新启动该功能。

---

顺便说一句，我应该注意到在处理程序内部监视资源可能不是最好的主意。观看很复杂。它带来的所有边缘情况和问题都太复杂了。

而且由于框架已经做了监视，使用它可能更容易，并通过运营商的全局状态实现跨资源连接：

import queue

import kopf

SERVICE_QUEUES = {}  # {(mc_namespace, mc_name) -> queue.Queue}
KNOWN_SERVICES = {}  # {(svc_namespace, svc_name) -> 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 "known" 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 "NodePort" in svc.spec['type']:
                    logger.info(f"Found Service of type NodePort: {svc.metadata.name}")
                    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 "NodePort" in svc.spec['type']:
                    logger.info(f"Found Service of type NodePort: {svc.metadata.name}")
                    do_some_work(svc)

    finally:
        del SERVICE_QUEUES[(namespace, name)]

这是一个简化的示例（但可能“按原样”工作——我没有检查）——只是为了展示如何在使用框架功能的同时使资源相互通信的想法。

解决方案取决于用例，并且此解决方案可能不适用于您的预期情况。也许我想念为什么会这样。如果您将您的用例作为功能请求报告给 Kopf 的 repo 会很好，以便以后框架可以支持它。