From b32ec01625d0ccacfe1fe655d355234c0f6de882 Mon Sep 17 00:00:00 2001 From: Morgan Wowk Date: Mon, 10 Aug 2026 13:35:09 -0700 Subject: [PATCH] feat(kubernetes): add post-delete pod cleanup handler registry Add a generic registry of handlers invoked after a launched pod is deleted (during cleanup or terminate). Handlers receive the deleted pod's identity and the cluster ApiClient and are best-effort -- an exception in one is logged and stops neither the other handlers nor the delete. This is an extension point for out-of-tree operators: one that stamps a hold finalizer on its task pods to keep them readable until their terminal status is observed can register a handler here to remove that finalizer inline the instant the pod is deleted, rather than reconciling it out of band. --- .../launchers/kubernetes_launchers.py | 62 ++++++++++++++ tests/test_pod_cleanup_handlers.py | 81 +++++++++++++++++++ 2 files changed, 143 insertions(+) create mode 100644 tests/test_pod_cleanup_handlers.py diff --git a/cloud_pipelines_backend/launchers/kubernetes_launchers.py b/cloud_pipelines_backend/launchers/kubernetes_launchers.py index 2aef66d..e20e5c0 100644 --- a/cloud_pipelines_backend/launchers/kubernetes_launchers.py +++ b/cloud_pipelines_backend/launchers/kubernetes_launchers.py @@ -980,6 +980,12 @@ def _delete_pod(self): namespace=self._namespace, grace_period_seconds=10, ) + _run_pod_cleanup_handlers( + pod_name=self._pod_name, + namespace=self._namespace, + api_client=launcher._api_client, + pod=self._debug_pod, + ) def terminate(self): self._delete_pod() @@ -1840,3 +1846,59 @@ def _remove_keys_with_none_values(d: dict): del d[k] if isinstance(v, dict): _remove_keys_with_none_values(v) + + +# Internal/experimental: this post-delete cleanup-handler extension point is not +# a stable public API. It is intentionally leading-underscore so out-of-tree +# consumers do not build on it before the Tangle team commits to it; the names +# (and shape) may change or be removed without a deprecation cycle. +class _PodCleanupHandler(typing.Protocol): + def __call__( + self, + *, + pod_name: str, + namespace: str, + api_client: k8s_client_lib.ApiClient, + pod: k8s_client_lib.V1Pod | None = None, + ) -> None: ... + + +# Callbacks invoked after a launched pod is deleted (during cleanup or +# terminate), in registration order. Each is best-effort: an exception in one +# handler is logged and stops neither the other handlers nor the delete. This is +# a generic extension point -- e.g. an operator that stamps a finalizer on its +# pods to hold them can register a handler here to remove that finalizer inline +# the instant the pod is deleted, rather than reconciling it out of band. +_pod_cleanup_handlers: list[_PodCleanupHandler] = [] + + +def _register_pod_cleanup_handler(handler: _PodCleanupHandler) -> None: + """Register a callback run after a launched pod is deleted. See _PodCleanupHandler. + + Internal/experimental -- see the note on _PodCleanupHandler. Not a stable API. + """ + _pod_cleanup_handlers.append(handler) + + +def _run_pod_cleanup_handlers( + *, + pod_name: str, + namespace: str, + api_client: k8s_client_lib.ApiClient, + pod: k8s_client_lib.V1Pod | None = None, +) -> None: + for handler in _pod_cleanup_handlers: + try: + handler( + pod_name=pod_name, + namespace=namespace, + api_client=api_client, + pod=pod, + ) + except Exception: + _logger.exception( + "Pod cleanup handler %r failed for pod %s in namespace %s.", + getattr(handler, "__name__", handler), + pod_name, + namespace, + ) diff --git a/tests/test_pod_cleanup_handlers.py b/tests/test_pod_cleanup_handlers.py new file mode 100644 index 0000000..b399a72 --- /dev/null +++ b/tests/test_pod_cleanup_handlers.py @@ -0,0 +1,81 @@ +"""Tests for the launcher's post-delete pod cleanup handler registry. + +The registry is a generic extension point: an operator that stamps a finalizer +on its pods can register a handler to remove it inline the moment the pod is +deleted, rather than reconciling it out of band. These pin that handlers run on +every pod delete, receive the pod's identity, and are best-effort. +""" + +from __future__ import annotations + +from unittest import mock + +import pytest + +from cloud_pipelines_backend.launchers import kubernetes_launchers as k8sL + + +@pytest.fixture(autouse=True) +def _clear_handlers(): + k8sL._pod_cleanup_handlers.clear() + yield + k8sL._pod_cleanup_handlers.clear() + + +def test_a_registered_handler_runs_with_the_pod_identity() -> None: + calls = [] + + def handler(*, pod_name, namespace, api_client, pod=None) -> None: + calls.append((pod_name, namespace, api_client, pod)) + + k8sL._register_pod_cleanup_handler(handler) + k8sL._run_pod_cleanup_handlers( + pod_name="task-1", + namespace="ns", + api_client=mock.sentinel.api_client, + pod=mock.sentinel.pod, + ) + + assert calls == [("task-1", "ns", mock.sentinel.api_client, mock.sentinel.pod)] + + +def test_a_failing_handler_does_not_stop_the_others() -> None: + ran = [] + + def boom(*, pod_name, namespace, api_client, pod=None) -> None: + raise RuntimeError("handler blew up") + + def ok(*, pod_name, namespace, api_client, pod=None) -> None: + ran.append(pod_name) + + k8sL._register_pod_cleanup_handler(boom) + k8sL._register_pod_cleanup_handler(ok) + + # Best-effort: the exception is swallowed and the delete/other handlers proceed. + k8sL._run_pod_cleanup_handlers(pod_name="task-2", namespace="ns", api_client=None) + + assert ran == ["task-2"] + + +def test_delete_pod_runs_the_handlers_after_deleting() -> None: + seen = [] + k8sL._register_pod_cleanup_handler( + lambda *, pod_name, namespace, api_client, pod=None: seen.append(pod_name) + ) + + launcher = mock.MagicMock() + launcher._api_client = mock.sentinel.api_client + container = k8sL.LaunchedKubernetesContainer( + pod_name="task-9", + namespace="ns", + output_uris={}, + log_uri="", + debug_pod=mock.sentinel.debug_pod, + launcher=launcher, + ) + + with mock.patch.object(k8sL.k8s_client_lib, "CoreV1Api") as core_api_cls: + container._delete_pod() + core_api_cls.return_value.delete_namespaced_pod.assert_called_once() + + assert seen == ["task-9"]