feat(kubernetes): add post-delete pod cleanup handler registry

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,
+            )

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"]