feat(operator): pod management

sentry_streams_k8s/sentry_streams_k8s/operator/__init__.py

@@ -1,13 +1,15 @@
 from sentry_streams_k8s.operator.streaming_pipeline import (
     StreamingPipelineSpec,
     from_crd_spec,
-    render,
+    render_deployments,
+    render_pods,
     validate,
 )
 
 __all__ = [
     "StreamingPipelineSpec",
     "from_crd_spec",
-    "render",
+    "render_deployments",
+    "render_pods",
     "validate",
 ]

sentry_streams_k8s/sentry_streams_k8s/operator/operator.py

@@ -2,26 +2,36 @@
 
 import asyncio
 import copy
+import json
 import os
 from collections.abc import Mapping
+from datetime import datetime, timezone
+from types import SimpleNamespace
 from typing import Any, cast
 
 import kopf
 from kubernetes import client
+from kubernetes.client import V1Pod
+from kubernetes.client.exceptions import ApiException
 
 from sentry_streams_k8s.k8s_types import V1ConditionDict
 from sentry_streams_k8s.operator.constants import (
+    FIELD_MANAGER,
     GROUP,
     HEALTH_SCAN_INTERVAL_SECONDS,
+    MANAGED_BY_LABEL,
     MAX_CONCURRENT_RECONCILES,
+    OWNER_UID_LABEL,
     PLURAL,
     VERSION,
     WORKLOAD_NAMESPACE_ENV,
     Logger,
 )
+from sentry_streams_k8s.operator.pod_health import pod_health
+from sentry_streams_k8s.operator.pod_resources import delete_owned_pods
 from sentry_streams_k8s.operator.reconcile import (
     PipelineStatusPatch,
-    _prune_stale_resources,
+    prune_stale_configmaps,
     reconcile_pipeline,
 )
 
@@ -85,16 +95,39 @@ def _workload_namespace() -> str:
     return namespace
 
 
-def _previous_conditions(body: kopf.Body) -> list[V1ConditionDict] | None:
-    status = body.get("status")
-    if not isinstance(status, Mapping):
-        return None
+def _published_conditions(status: Mapping[str, object]) -> list[V1ConditionDict] | None:
     conditions = status.get("conditions")
     if not isinstance(conditions, list):
         return None
     return cast(list[V1ConditionDict], conditions)
 
 
+def _deserialize_pod(body: kopf.Body) -> V1Pod:
+    # kopf provides the event's raw JSON body so we need to deserialize.
+    # Use a simple namespace since ApiClient expects a RESTResponse:
+    json_response = SimpleNamespace(data=json.dumps(dict(body)))
+    return cast(V1Pod, client.ApiClient().deserialize(json_response, "V1Pod"))
+
+
+def _get_pipeline_status(name: str, namespace: str) -> Mapping[str, object]:
+    api = client.CustomObjectsApi()
+    try:
+        obj = api.get_namespaced_custom_object(
+            group=GROUP,
+            version=VERSION,
+            namespace=namespace,
+            plural=PLURAL,
+            name=name,
+        )
+    except ApiException as e:
+        if e.status == 404:
+            return {}
+        raise
+    obj = cast(Mapping[str, object], obj)
+    status = obj.get("status")
+    return cast(Mapping[str, object], status) if isinstance(status, Mapping) else {}
+
+
 def _patch_pipeline_status(name: str, namespace: str, status: PipelineStatusPatch) -> None:
     api = client.CustomObjectsApi()
     api.patch_namespaced_custom_object_status(
@@ -147,7 +180,6 @@ async def _reconcile_once(
     logger: Logger,
     scheduler: ReconcileScheduler,
     stopped: kopf.DaemonStopped,
-    previous_conditions: list[V1ConditionDict] | None = None,
 ) -> float | None:
     status_patch: PipelineStatusPatch = {}
     try:
@@ -156,6 +188,7 @@ async def _reconcile_once(
                 if stopped:
                     return None
                 try:
+                    published = await asyncio.to_thread(_get_pipeline_status, name, namespace)
                     await asyncio.to_thread(
                         reconcile_pipeline,
                         spec=copy.deepcopy(dict(spec)),
@@ -165,7 +198,8 @@ async def _reconcile_once(
                         workload_namespace=_workload_namespace(),
                         logger=logger,
                         status=status_patch,
-                        previous_conditions=previous_conditions,
+                        previous_conditions=_published_conditions(published),
+                        previous_generations=published.get("generations"),
                     )
                 except kopf.PermanentError as e:
                     if status_patch:
@@ -188,7 +222,6 @@ async def _reconcile_once(
 async def reconcile_pipeline_daemon(
     stopped: kopf.DaemonStopped,
     spec: kopf.Spec,
-    body: kopf.Body,
     name: str,
     namespace: str | None,
     uid: str,
@@ -212,7 +245,6 @@ async def reconcile_pipeline_daemon(
                 logger=logger,
                 scheduler=scheduler,
                 stopped=stopped,
-                previous_conditions=_previous_conditions(body),
             )
             if stopped:
                 break
@@ -221,16 +253,52 @@ async def reconcile_pipeline_daemon(
         scheduler.unregister(uid, event)
 
 
+@kopf.on.event("", "v1", "pods", labels={MANAGED_BY_LABEL: FIELD_MANAGER})
+async def handle_pipeline_pod_event(
+    type: str | None,
+    body: kopf.Body,
+    meta: kopf.Meta,
+    labels: kopf.Labels,
+    name: str | None,
+    namespace: str | None,
+    memo: kopf.Memo,
+    logger: Logger,
+    **_: Any,
+) -> None:
+    if type not in {"DELETED", "MODIFIED"}:
+        return
+
+    if type == "MODIFIED" and meta.deletion_timestamp is None:
+        health = pod_health(_deserialize_pod(body), datetime.now(timezone.utc))
+        if not health.delete:
+            return
+
+    owner_uid = labels.get(OWNER_UID_LABEL)
+    if not owner_uid:
+        logger.warning(
+            "managed Pod %s/%s is missing its owner UID label; cannot reconcile",
+            namespace,
+            name,
+        )
+        return
+
+    if _scheduler(memo).notify(owner_uid):
+        logger.info("requested reconciliation after Pod %s event=%s", name, type)
+
+
 @kopf.on.delete(GROUP, VERSION, PLURAL)
 async def cleanup(uid: str, memo: kopf.Memo, logger: Logger, **_: Any) -> None:
     scheduler = _scheduler(memo)
     async with scheduler.limit:
         async with scheduler.lock(uid):
+            workload_namespace = _workload_namespace()
+            core = client.CoreV1Api()
+            await asyncio.to_thread(delete_owned_pods, core, workload_namespace, uid, logger)
             await asyncio.to_thread(
-                _prune_stale_resources,
-                workload_namespace=_workload_namespace(),
+                prune_stale_configmaps,
+                core=core,
+                workload_namespace=workload_namespace,
                 owner_uid=uid,
-                desired_deployments=set(),
                 desired_configmaps=set(),
                 logger=logger,
             )

sentry_streams_k8s/sentry_streams_k8s/operator/pod_health.py

@@ -82,14 +82,6 @@ def _first_waiting_reason(
     return None
 
 
-def _first_unhealthy_waiting(statuses: list[V1ContainerStatus] | None) -> str | None:
-    return _first_waiting_reason(statuses, UNHEALTHY_WAITING_REASONS)
-
-
-def _first_permanent_waiting(statuses: list[V1ContainerStatus] | None) -> str | None:
-    return _first_waiting_reason(statuses, PERMANENT_WAITING_REASONS)
-
-
 def _container_failed_terminated_reason(status: V1ContainerStatus) -> str | None:
     terminated = status.state.terminated if status.state else None
     if terminated is None:
@@ -140,27 +132,6 @@ class PodHealth:
     permanent: bool = False
 
 
-def _verdict(
-    pod_name: str,
-    *,
-    ready: bool = False,
-    reason: str | None = None,
-    unhealthy: bool = False,
-    delete: bool = False,
-    force: bool = False,
-    permanent: bool = False,
-) -> PodHealth:
-    return PodHealth(
-        name=pod_name,
-        ready=ready,
-        reason=reason,
-        unhealthy=unhealthy,
-        delete=delete,
-        force=force,
-        permanent=permanent,
-    )
-
-
 def _container_statuses_verdict(
     pod_name: str,
     statuses: list[V1ContainerStatus] | None,
@@ -169,9 +140,9 @@ def _container_statuses_verdict(
     *,
     reason_prefix: str = "",
 ) -> PodHealth | None:
-    permanent_waiting = _first_permanent_waiting(statuses)
+    permanent_waiting = _first_waiting_reason(statuses, PERMANENT_WAITING_REASONS)
     if permanent_waiting is not None:
-        return _verdict(
+        return PodHealth(
             pod_name,
             reason=f"{reason_prefix}{permanent_waiting}",
             unhealthy=True,
@@ -180,16 +151,16 @@ def _container_statuses_verdict(
 
     terminated_reason = _first_failed_terminated_reason(statuses)
     if terminated_reason is not None:
-        return _verdict(
+        return PodHealth(
             pod_name,
             reason=f"{reason_prefix}{terminated_reason}",
             unhealthy=True,
             delete=True,
         )
 
-    unhealthy_waiting = _first_unhealthy_waiting(statuses)
+    unhealthy_waiting = _first_waiting_reason(statuses, UNHEALTHY_WAITING_REASONS)
     if unhealthy_waiting is not None:
-        return _verdict(
+        return PodHealth(
             pod_name,
             reason=f"{reason_prefix}{unhealthy_waiting}",
             unhealthy=True,
@@ -198,7 +169,7 @@ def _container_statuses_verdict(
 
     waiting_reason = _first_waiting_reason(statuses)
     if waiting_reason is not None:
-        return _verdict(
+        return PodHealth(
             pod_name,
             reason=f"{reason_prefix}{waiting_reason}",
         )
@@ -228,7 +199,7 @@ def pod_health(pod: V1Pod, now: datetime) -> PodHealth:
     if terminating_age is not None:
         stuck = terminating_age >= POD_TERMINATING_GRACE_SECONDS
         reason = "StuckTerminating" if stuck else "Terminating"
-        return _verdict(
+        return PodHealth(
             pod_name,
             reason=reason,
             unhealthy=stuck,
@@ -238,10 +209,10 @@ def pod_health(pod: V1Pod, now: datetime) -> PodHealth:
 
     status = pod.status
     if status is None:
-        return _verdict(pod_name)
+        return PodHealth(pod_name)
 
     if _pod_unschedulable(status.conditions):
-        return _verdict(pod_name, reason="Unschedulable", unhealthy=True)
+        return PodHealth(pod_name, reason="Unschedulable", unhealthy=True)
 
     init_verdict = _container_statuses_verdict(
         pod_name,
@@ -256,7 +227,7 @@ def pod_health(pod: V1Pod, now: datetime) -> PodHealth:
     phase = status.phase
 
     if phase == "Succeeded":
-        return _verdict(pod_name, reason="Succeeded", delete=True)
+        return PodHealth(pod_name, reason="Succeeded", delete=True)
 
     app_verdict = _container_statuses_verdict(
         pod_name,
@@ -269,6 +240,6 @@ def pod_health(pod: V1Pod, now: datetime) -> PodHealth:
 
     if phase == "Failed":
         reason = status.reason or phase or "Terminated"
-        return _verdict(pod_name, reason=reason, unhealthy=True, delete=True)
+        return PodHealth(pod_name, reason=reason, unhealthy=True, delete=True)
 
-    return _verdict(pod_name, ready=pod_is_ready(pod))
+    return PodHealth(pod_name, ready=pod_is_ready(pod))

sentry_streams_k8s/sentry_streams_k8s/operator/pod_status.py

@@ -35,10 +35,6 @@ class ReportedPodStatus:
     reason: str | None = None
     permanent: bool = False
 
-    @property
-    def is_unhealthy(self) -> bool:
-        return self.unhealthy
-
     def to_status_dict(self) -> PodStatusEntry:
         entry: PodStatusEntry = {
             "name": self.name,