diff --git a/sentry_streams_k8s/chart/streaming-operator/templates/rbac.yaml b/sentry_streams_k8s/chart/streaming-operator/templates/rbac.yaml deleted file mode 100644 index dfa6be89..00000000 --- a/sentry_streams_k8s/chart/streaming-operator/templates/rbac.yaml +++ /dev/null @@ -1,68 +0,0 @@ -apiVersion: rbac.authorization.k8s.io/v1 -kind: ClusterRole -metadata: - name: {{ include "streaming-operator.fullname" . }} - labels: - {{- include "streaming-operator.labels" . | nindent 4 }} -rules: - - apiGroups: [apiextensions.k8s.io] - resources: [customresourcedefinitions] - verbs: [list, watch] - - apiGroups: [""] - resources: [namespaces] - verbs: [list, watch] - - apiGroups: [streams.sentry.io] - resources: [streamingpipelines] - verbs: [list, watch, patch] - - apiGroups: [streams.sentry.io] - resources: [streamingpipelines/status] - verbs: [patch] - - apiGroups: [""] - resources: [events] - verbs: [create] ---- -apiVersion: rbac.authorization.k8s.io/v1 -kind: ClusterRoleBinding -metadata: - name: {{ include "streaming-operator.fullname" . }} - labels: - {{- include "streaming-operator.labels" . | nindent 4 }} -roleRef: - apiGroup: rbac.authorization.k8s.io - kind: ClusterRole - name: {{ include "streaming-operator.fullname" . }} -subjects: - - kind: ServiceAccount - name: {{ include "streaming-operator.serviceAccountName" . }} - namespace: {{ .Release.Namespace }} ---- -apiVersion: rbac.authorization.k8s.io/v1 -kind: Role -metadata: - name: {{ include "streaming-operator.fullname" . }} - namespace: {{ include "streaming-operator.workloadNamespace" . }} - labels: - {{- include "streaming-operator.labels" . | nindent 4 }} -rules: - - apiGroups: [""] - resources: [configmaps] - verbs: [get, list, create, patch, delete] - - apiGroups: [apps] - resources: [deployments] - verbs: [get, list, create, patch, delete] ---- -apiVersion: rbac.authorization.k8s.io/v1 -kind: RoleBinding -metadata: - name: {{ include "streaming-operator.fullname" . }} - namespace: {{ include "streaming-operator.workloadNamespace" . }} - labels: - {{- include "streaming-operator.labels" . | nindent 4 }} -roleRef: - apiGroup: rbac.authorization.k8s.io - kind: Role - name: {{ include "streaming-operator.fullname" . }} -subjects: - - kind: ServiceAccount - name: {{ include "streaming-operator.serviceAccountName" . }} - namespace: {{ .Release.Namespace }} diff --git a/sentry_streams_k8s/sentry_streams_k8s/operator/constants.py b/sentry_streams_k8s/sentry_streams_k8s/operator/constants.py new file mode 100644 index 00000000..b5199180 --- /dev/null +++ b/sentry_streams_k8s/sentry_streams_k8s/operator/constants.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +from typing import Any + +GROUP = "streams.sentry.io" +VERSION = "v1alpha1" +PLURAL = "streamingpipelines" + +FIELD_MANAGER = "streaming-operator" +WORKLOAD_NAMESPACE_ENV = "WORKLOAD_NAMESPACE" + +# The CR and its workloads may live in different namespaces, so we cannot use +# ownerReferences (Kubernetes forbids cross-namespace owning) or kopf.adopt. +# Instead every workload carries the owning CR's UID as a label with +# name/namespace annotations so the operator can delete resources itself: + +OWNER_UID_LABEL = "streams.sentry.io/owner-uid" +OWNER_NAME_ANNOTATION = "streams.sentry.io/owner-name" +OWNER_NAMESPACE_ANNOTATION = "streams.sentry.io/owner-namespace" + +# Marks every Pod the operator manages so the cluster-wide Pod event watcher can +# select operator-owned Pods without knowing their individual UIDs: + +MANAGED_BY_LABEL = "streams.sentry.io/managed-by" + +# Identifies independently reconciled Pod sets owned by the same pipeline: + +WORKLOAD_SET_LABEL = "streams.sentry.io/workload-set" + +PRIMARY_WORKLOAD_SET = "primary" +CANARY_WORKLOAD_SET = "canary" + +ALL_WORKLOAD_SETS = (PRIMARY_WORKLOAD_SET, CANARY_WORKLOAD_SET) + +# The ordinal is the stable replica number. The generation increments each time a replica's +# Pod is replaced so the new Pod's name never collides with a still-terminating old one: + +ORDINAL_LABEL = "streams.sentry.io/ordinal" +GENERATION_LABEL = "streams.sentry.io/generation" + +# Pod names are cannot exceed 63 characters, so we reserve space for the +# largest possible replicas + generation and give the rest to the base name: + +MAX_POD_NAME_LENGTH = 63 + +MAX_REPLICAS = 999 +MAX_GENERATION = 9999 + +MAX_BASE_NAME_LENGTH = MAX_POD_NAME_LENGTH - len(f"-{MAX_REPLICAS}-{MAX_GENERATION}") + +# Hash of the desired Pod state. Used to detect config/spec changes: + +SPEC_HASH_ANNOTATION = "streams.sentry.io/spec-hash" + +# Waiting states that may recover when the operator creates a new Pod: + +UNHEALTHY_WAITING_REASONS = frozenset({"ErrImagePull", "ImagePullBackOff"}) + +# Waiting states caused by an invalid workload specification. +# Should raise a PermanentError and not trigger replacement: + +PERMANENT_WAITING_REASONS = frozenset({"InvalidImageName"}) + +# WAITING_GRACE: A Pod stuck in a bad waiting state is terminated after this long. +# TERMINATING_GRACE: A Pod stuck in a terminating state for this long is force-deleted. + +POD_WAITING_GRACE_SECONDS = 300 +POD_TERMINATING_GRACE_SECONDS = 600 + +# How often the daemon re-runs a full reconcile as a safety net behind the watch: + +HEALTH_SCAN_INTERVAL_SECONDS = 60 + +# Bound full reconciliations across pipelines so changes +# to many CRs cannot overload Kubernetes API requests: + +MAX_CONCURRENT_RECONCILES = 4 + +Logger = Any diff --git a/sentry_streams_k8s/sentry_streams_k8s/operator/operator.py b/sentry_streams_k8s/sentry_streams_k8s/operator/operator.py index 3f929519..8dc4fdce 100644 --- a/sentry_streams_k8s/sentry_streams_k8s/operator/operator.py +++ b/sentry_streams_k8s/sentry_streams_k8s/operator/operator.py @@ -1,30 +1,80 @@ from __future__ import annotations -import logging +import asyncio +import copy +import json import os -from typing import Any +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, dynamic +from kubernetes import client +from kubernetes.client import V1Pod from kubernetes.client.exceptions import ApiException -from sentry_streams_k8s.consumer_builder import compute_config_version -from sentry_streams_k8s.operator.streaming_pipeline import ( - from_crd_spec, - render, - validate, +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_configmaps, + reconcile_pipeline, +) + + +class ReconcileScheduler: + def __init__(self, max_concurrent: int = MAX_CONCURRENT_RECONCILES) -> None: + self._events: dict[str, asyncio.Event] = {} + self._locks: dict[str, asyncio.Lock] = {} + self.limit = asyncio.Semaphore(max_concurrent) + + def register(self, uid: str) -> asyncio.Event: + event = self._events.setdefault(uid, asyncio.Event()) + self._locks.setdefault(uid, asyncio.Lock()) + return event + + def unregister(self, uid: str, event: asyncio.Event) -> None: + if self._events.get(uid) is event: + self._events.pop(uid, None) + + def notify(self, uid: str) -> bool: + event = self._events.get(uid) + if event is None: + return False + event.set() + return True + + def lock(self, uid: str) -> asyncio.Lock: + return self._locks.setdefault(uid, asyncio.Lock()) -logger = logging.getLogger(__name__) + def discard(self, uid: str) -> None: + self._events.pop(uid, None) + self._locks.pop(uid, None) -GROUP = "streams.sentry.io" -VERSION = "v1alpha1" -PLURAL = "streamingpipelines" -FIELD_MANAGER = "streaming-operator" -WORKLOAD_NAMESPACE_ENV = "WORKLOAD_NAMESPACE" -OWNER_UID_LABEL = "streams.sentry.io/owner-uid" -OWNER_NAME_ANNOTATION = "streams.sentry.io/owner-name" -OWNER_NAMESPACE_ANNOTATION = "streams.sentry.io/owner-namespace" + +def _scheduler(memo: kopf.Memo) -> ReconcileScheduler: + scheduler = getattr(memo, "reconcile_scheduler", None) + if not isinstance(scheduler, ReconcileScheduler): + raise RuntimeError("Reconcile scheduler is not initialized.") + return scheduler + + +@kopf.on.startup() +async def configure_reconciliation(memo: kopf.Memo, **_: Any) -> None: + memo.reconcile_scheduler = ReconcileScheduler() def _workload_namespace() -> str: @@ -34,188 +84,203 @@ def _workload_namespace() -> str: return namespace -def _prepare_manifest( - manifest: dict[str, Any], - *, - workload_namespace: str, - owner_uid: str, - owner_name: str, - owner_namespace: str, -) -> None: - metadata = manifest.setdefault("metadata", {}) - metadata["namespace"] = workload_namespace - metadata["labels"] = { - **metadata.get("labels", {}), - OWNER_UID_LABEL: owner_uid, - } - metadata["annotations"] = { - **metadata.get("annotations", {}), - OWNER_NAME_ANNOTATION: owner_name, - OWNER_NAMESPACE_ANNOTATION: owner_namespace, - } - - -def _apply( - dyn: dynamic.DynamicClient, - manifest: dict[str, Any], - *, - workload_namespace: str, - owner_uid: str, -) -> None: - resource = dyn.resources.get(api_version=manifest["apiVersion"], kind=manifest["kind"]) - name = manifest["metadata"]["name"] - try: - existing = resource.get(name=name, namespace=workload_namespace) - except ApiException as e: - if e.status != 404: - raise - else: - labels = existing.metadata.labels or {} - existing_owner_uid = labels.get(OWNER_UID_LABEL) - if existing_owner_uid != owner_uid: - raise kopf.PermanentError( - f"{manifest['kind']} {workload_namespace}/{name} is already present and is not " - "managed by this StreamingPipeline." - ) +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")) + - dyn.server_side_apply( - resource, - body=manifest, - namespace=workload_namespace, - field_manager=FIELD_MANAGER, - force_conflicts=True, +def _patch_pipeline_status(name: str, namespace: str, status: PipelineStatusPatch) -> None: + api = client.CustomObjectsApi() + api.patch_namespaced_custom_object_status( + group=GROUP, + version=VERSION, + namespace=namespace, + plural=PLURAL, + name=name, + body={"status": status}, ) -def _prune_stale_resources( - *, - workload_namespace: str, - owner_uid: str, - desired_deployments: set[str], - desired_configmaps: set[str], +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 {} + + +@kopf.on.update(GROUP, VERSION, PLURAL) +async def request_pipeline_reconcile( + uid: str, + memo: kopf.Memo, + logger: Logger, + **_: Any, ) -> None: - selector = f"{OWNER_UID_LABEL}={owner_uid}" + if _scheduler(memo).notify(uid): + logger.debug("requested pipeline reconciliation uid=%s", uid) - apps = client.AppsV1Api() - deployments = apps.list_namespaced_deployment( - namespace=workload_namespace, - label_selector=selector, - ) - for deployment in deployments.items: - if deployment.metadata.name not in desired_deployments: - logger.info( - "Pruning stale deployment %s/%s", - workload_namespace, - deployment.metadata.name, - ) - apps.delete_namespaced_deployment( - name=deployment.metadata.name, - namespace=workload_namespace, - ) - core = client.CoreV1Api() - configmaps = core.list_namespaced_config_map( - namespace=workload_namespace, - label_selector=selector, +async def _wait_for_reconcile( + event: asyncio.Event, + stopped: kopf.DaemonStopped, + timeout: float | None, +) -> None: + event_waiter: asyncio.Future[Any] = asyncio.ensure_future(event.wait()) + stop_waiter: asyncio.Future[Any] = asyncio.ensure_future(stopped.wait(timeout)) + _, pending = await asyncio.wait( + {event_waiter, stop_waiter}, + return_when=asyncio.FIRST_COMPLETED, ) - for configmap in configmaps.items: - if configmap.metadata.name not in desired_configmaps: - logger.info( - "Pruning stale configmap %s/%s", - workload_namespace, - configmap.metadata.name, - ) - core.delete_namespaced_config_map( - name=configmap.metadata.name, - namespace=workload_namespace, - ) + for task in pending: + task.cancel() + if pending: + await asyncio.gather( + *cast(tuple[asyncio.Future[Any], ...], tuple(pending)), + return_exceptions=True, + ) -def _condition(type_: str, status: bool, reason: str, message: str = "") -> dict[str, Any]: - return { - "type": type_, - "status": "True" if status else "False", - "reason": reason, - "message": message, - } +async def _reconcile_once( + *, + spec: kopf.Spec, + name: str, + namespace: str, + uid: str, + logger: Logger, + scheduler: ReconcileScheduler, + stopped: kopf.DaemonStopped, +) -> float | None: + status: PipelineStatusPatch = {} + try: + async with scheduler.limit: + async with scheduler.lock(uid): + if stopped: + return None + try: + previous_status = await asyncio.to_thread(_get_pipeline_status, name, namespace) + await asyncio.to_thread( + reconcile_pipeline, + spec=copy.deepcopy(dict(spec)), + name=name, + namespace=namespace, + uid=uid, + workload_namespace=_workload_namespace(), + logger=logger, + status=status, + previous_generations=previous_status.get("generations"), + ) + except kopf.PermanentError as e: + if status: + await asyncio.to_thread(_patch_pipeline_status, name, namespace, status) + logger.error("%s", e) + return None + if status: + await asyncio.to_thread(_patch_pipeline_status, name, namespace, status) + except Exception: + logger.exception("pipeline reconciliation failed; retrying") + return 5.0 + logger.info("reconciled pipeline Pods %s/%s", namespace, name) + return float(HEALTH_SCAN_INTERVAL_SECONDS) -@kopf.on.create(GROUP, VERSION, PLURAL) -@kopf.on.update(GROUP, VERSION, PLURAL) -@kopf.on.resume(GROUP, VERSION, PLURAL) -def reconcile( + +@kopf.daemon(GROUP, VERSION, PLURAL) +async def reconcile_pipeline_daemon( + stopped: kopf.DaemonStopped, spec: kopf.Spec, name: str, namespace: str | None, uid: str, - patch: kopf.Patch, + memo: kopf.Memo, + logger: Logger, **_: Any, ) -> None: - assert namespace is not None - workload_namespace = _workload_namespace() + if namespace is None: + raise kopf.PermanentError("Missing namespace!") - consumer = from_crd_spec(dict(spec), name=name) + scheduler = _scheduler(memo) + event = scheduler.register(uid) try: - validate(consumer) - result = render(consumer) - except Exception as e: - patch.status["conditions"] = [_condition("Rendered", False, type(e).__name__, str(e))] - raise kopf.PermanentError(f"StreamingPipeline {namespace}/{name} failed to render: {e}") - - manifests = [result["configmap"], result["deployment"]] - if "canary_deployment" in result: - manifests.append(result["canary_deployment"]) - - dyn = dynamic.DynamicClient(client.ApiClient()) - for manifest in manifests: - _prepare_manifest( - manifest, - workload_namespace=workload_namespace, - owner_uid=uid, - owner_name=name, - owner_namespace=namespace, - ) - _apply( - dyn, - manifest, - workload_namespace=workload_namespace, - owner_uid=uid, - ) + while not stopped: + event.clear() + timeout = await _reconcile_once( + spec=spec, + name=name, + namespace=namespace, + uid=uid, + logger=logger, + scheduler=scheduler, + stopped=stopped, + ) + if stopped: + break + await _wait_for_reconcile(event, stopped, timeout) + finally: + scheduler.unregister(uid, event) - _prune_stale_resources( - workload_namespace=workload_namespace, - owner_uid=uid, - desired_deployments={ - manifest["metadata"]["name"] - for manifest in manifests - if manifest["kind"] == "Deployment" - }, - desired_configmaps={ - manifest["metadata"]["name"] - for manifest in manifests - if manifest["kind"] == "ConfigMap" - }, - ) - replicas = consumer.get("replicas", 1) - canary = 1 if "canary_deployment" in result else 0 - patch.status["conditions"] = [ - _condition("Rendered", True, "Rendered"), - _condition("Applied", True, "Applied"), - ] - patch.status["config_version"] = compute_config_version(consumer["pipeline_config"]) - patch.status["replicas"] = {"primary": replicas - canary, "canary": canary} - patch.status["workload_namespace"] = workload_namespace +@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) -def cleanup(uid: str, **_: Any) -> None: - _prune_stale_resources( - workload_namespace=_workload_namespace(), - owner_uid=uid, - desired_deployments=set(), - desired_configmaps=set(), - ) +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_configmaps, + workload_namespace=workload_namespace, + owner_uid=uid, + desired_configmaps=set(), + logger=logger, + ) + scheduler.discard(uid) def main() -> None: diff --git a/sentry_streams_k8s/sentry_streams_k8s/operator/pod_health.py b/sentry_streams_k8s/sentry_streams_k8s/operator/pod_health.py new file mode 100644 index 00000000..d357e51c --- /dev/null +++ b/sentry_streams_k8s/sentry_streams_k8s/operator/pod_health.py @@ -0,0 +1,235 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone + +from kubernetes.client import V1ContainerStatus, V1Pod + +from sentry_streams_k8s.operator.constants import ( + PERMANENT_WAITING_REASONS, + POD_TERMINATING_GRACE_SECONDS, + POD_WAITING_GRACE_SECONDS, + SPEC_HASH_ANNOTATION, + UNHEALTHY_WAITING_REASONS, +) +from sentry_streams_k8s.operator.pod_types import PodManifest + + +def is_deleting(pod: V1Pod) -> bool: + metadata = pod.metadata + return metadata is not None and metadata.deletion_timestamp is not None + + +def age_seconds(timestamp: datetime | None, now: datetime) -> float | None: + if timestamp is None: + return None + if timestamp.tzinfo is None: + timestamp = timestamp.replace(tzinfo=timezone.utc) + return (now - timestamp).total_seconds() + + +def pod_is_ready(pod: V1Pod) -> bool: + status = pod.status + if status is None: + return False + conditions = status.conditions or [] + return status.phase == "Running" and any( + condition.type == "Ready" and condition.status == "True" for condition in conditions + ) + + +def pod_spec_changed(current: V1Pod, desired: PodManifest) -> bool: + metadata = current.metadata + current_hash = ((metadata.annotations or {}) if metadata else {}).get(SPEC_HASH_ANNOTATION) + desired_annotations = desired["metadata"].get("annotations") or {} + desired_hash = desired_annotations.get(SPEC_HASH_ANNOTATION) + return desired_hash is not None and current_hash != desired_hash + + +def _container_waiting_reason(status: V1ContainerStatus) -> str | None: + waiting = status.state.waiting if status.state else None + return waiting.reason if waiting else None + + +def _first_waiting_reason( + statuses: list[V1ContainerStatus] | None, reasons: frozenset[str] +) -> str | None: + for status in statuses or []: + reason = _container_waiting_reason(status) + if reason in reasons: + return 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: + return None + + exit_code = terminated.exit_code + reason = terminated.reason + + if exit_code == 0 or reason == "Completed": + return None + + if isinstance(reason, str) and reason: + return reason + + if isinstance(exit_code, int): + return f"ExitCode{exit_code}" + + return "Terminated" + + +def _first_failed_terminated_reason(statuses: list[V1ContainerStatus] | None) -> str | None: + for status in statuses or []: + reason = _container_failed_terminated_reason(status) + if reason is not None: + return reason + return None + + +def _waiting_grace_elapsed(pod: V1Pod, now: datetime) -> bool: + metadata = pod.metadata + status = pod.status + if metadata is None or status is None: + return False + age = age_seconds(status.start_time, now) + if age is None: + age = age_seconds(metadata.creation_timestamp, now) + return age is not None and age >= POD_WAITING_GRACE_SECONDS + + +@dataclass(frozen=True) +class PodHealth: + name: str + ready: bool = False + reason: str | None = None + unhealthy: bool = False + delete: bool = False + force: bool = False + 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, + pod: V1Pod, + now: datetime, + *, + reason_prefix: str = "", +) -> PodHealth | None: + permanent_waiting = _first_permanent_waiting(statuses) + if permanent_waiting is not None: + return _verdict( + pod_name, + reason=f"{reason_prefix}{permanent_waiting}", + unhealthy=True, + permanent=True, + ) + + terminated_reason = _first_failed_terminated_reason(statuses) + if terminated_reason is not None: + return _verdict( + pod_name, + reason=f"{reason_prefix}{terminated_reason}", + unhealthy=True, + delete=True, + ) + + unhealthy_waiting = _first_unhealthy_waiting(statuses) + if unhealthy_waiting is not None: + return _verdict( + pod_name, + reason=f"{reason_prefix}{unhealthy_waiting}", + unhealthy=True, + delete=_waiting_grace_elapsed(pod, now), + ) + + return None + + +def pod_health(pod: V1Pod, now: datetime) -> PodHealth: + """Classify a Pod into a PodHealth verdict.""" + + metadata = pod.metadata + pod_name = metadata.name if metadata and metadata.name else "" + + # If the pod is terminating, let it finish terminating by itself. + # Only force-delete it if it has been stuck terminating for too long. + + terminating_age = age_seconds(metadata.deletion_timestamp if metadata else None, now) + + if terminating_age is not None: + stuck = terminating_age >= POD_TERMINATING_GRACE_SECONDS + reason = "StuckTerminating" if stuck else "Terminating" + return _verdict( + pod_name, + reason=reason, + unhealthy=stuck, + delete=stuck, + force=stuck, + ) + + status = pod.status + if status is None: + return _verdict(pod_name) + + init_verdict = _container_statuses_verdict( + pod_name, + status.init_container_statuses, + pod, + now, + reason_prefix="InitContainer", + ) + if init_verdict is not None: + return init_verdict + + phase = status.phase + + if phase == "Succeeded": + return _verdict(pod_name, reason="Succeeded", delete=True) + + app_verdict = _container_statuses_verdict( + pod_name, + status.container_statuses, + pod, + now, + ) + if app_verdict is not None: + return app_verdict + + if phase == "Failed": + reason = status.reason or phase or "Terminated" + return _verdict(pod_name, reason=reason, unhealthy=True, delete=True) + + return _verdict(pod_name, ready=pod_is_ready(pod)) diff --git a/sentry_streams_k8s/sentry_streams_k8s/operator/pod_resources.py b/sentry_streams_k8s/sentry_streams_k8s/operator/pod_resources.py new file mode 100644 index 00000000..fc8bec8a --- /dev/null +++ b/sentry_streams_k8s/sentry_streams_k8s/operator/pod_resources.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import copy +import hashlib +import json +from collections.abc import Mapping +from typing import Any, cast + +from kubernetes import client +from kubernetes.client import V1Pod, V1PodList + +from sentry_streams_k8s.operator.constants import ( + FIELD_MANAGER, + GENERATION_LABEL, + MANAGED_BY_LABEL, + ORDINAL_LABEL, + OWNER_NAME_ANNOTATION, + OWNER_NAMESPACE_ANNOTATION, + OWNER_UID_LABEL, + SPEC_HASH_ANNOTATION, + WORKLOAD_SET_LABEL, + Logger, +) +from sentry_streams_k8s.operator.pod_health import PodHealth, is_deleting +from sentry_streams_k8s.operator.pod_types import PodManifest + + +def list_owned_pods( + core: client.CoreV1Api, + namespace: str, + owner_uid: str, + workload_set: str | None = None, +) -> list[V1Pod]: + selector = f"{OWNER_UID_LABEL}={owner_uid}" + if workload_set is not None: + selector = f"{selector},{WORKLOAD_SET_LABEL}={workload_set}" + listing = cast( + V1PodList, core.list_namespaced_pod(namespace=namespace, label_selector=selector) + ) + return cast(list[V1Pod], listing.items) + + +def apply_pod(core: client.CoreV1Api, pod: PodManifest, namespace: str) -> None: + core.patch_namespaced_pod( + name=pod["metadata"]["name"], + namespace=namespace, + body=dict(pod), + field_manager=FIELD_MANAGER, + force=True, + _content_type="application/apply-patch+yaml", + ) + + +def delete_pod(core: client.CoreV1Api, name: str, namespace: str, force: bool = False) -> None: + if force: + # Only use this for a Pod that has been stuck terminating: + core.delete_namespaced_pod( + name=name, + namespace=namespace, + body=client.V1DeleteOptions(grace_period_seconds=0), + ) + else: + core.delete_namespaced_pod(name=name, namespace=namespace) + + +def pod_name(pod: V1Pod) -> str: + metadata = pod.metadata + return metadata.name if metadata and metadata.name else "" + + +def consumer_pod_name(base_name: str, ordinal: int, generation: int) -> str: + return f"{base_name}-{ordinal}-{generation}" + + +def pod_ordinal(pod: V1Pod) -> int | None: + metadata = pod.metadata + label = (metadata.labels or {}).get(ORDINAL_LABEL) if metadata else None + if label is None: + return None + try: + return int(label) + except ValueError: + return None + + +def pod_generation(pod: V1Pod) -> int: + metadata = pod.metadata + label = (metadata.labels or {}).get(GENERATION_LABEL) if metadata else None + if label is None: + return 0 + try: + return int(label) + except ValueError: + return 0 + + +def pod_workload_set(pod: V1Pod) -> str | None: + metadata = pod.metadata + return (metadata.labels or {}).get(WORKLOAD_SET_LABEL) if metadata else None + + +def pod_keep_key(pod: V1Pod, health: PodHealth) -> tuple[bool, int, str]: + return (health.ready, pod_generation(pod), pod_name(pod)) + + +def pod_template_from_deployment( + deployment: Mapping[str, Any], +) -> tuple[Mapping[str, Any], Mapping[str, Any]]: + template = deployment["spec"]["template"] + return template.get("metadata", {}) or {}, template["spec"] + + +def _pod_spec_hash(pod: PodManifest) -> str: + metadata = pod["metadata"] + labels = { + key: value + for key, value in (metadata.get("labels", {}) or {}).items() + if key != GENERATION_LABEL + } + annotations = { + key: value + for key, value in (metadata.get("annotations", {}) or {}).items() + if key != SPEC_HASH_ANNOTATION + } + desired = { + "metadata": {"labels": labels, "annotations": annotations}, + "spec": pod["spec"], + } + encoded = json.dumps(desired, sort_keys=True, separators=(",", ":")).encode() + return hashlib.sha256(encoded).hexdigest() + + +def build_pipeline_pod( + *, + base_name: str, + template_metadata: Mapping[str, Any], + template_spec: Mapping[str, Any], + ordinal: int, + generation: int, + owner_uid: str, + owner_name: str, + owner_namespace: str, + workload_set: str, +) -> PodManifest: + pod_spec = copy.deepcopy(dict(template_spec)) + + # The operator replaces unhealthy Pods rather than letting k8s restart them: + + pod_spec["restartPolicy"] = "Never" + + # Pods can live in a different namespace from their StreamingPipeline, so they + # cannot use owner references. The operator finds them by owner label instead. + + labels = { + **(template_metadata.get("labels", {}) or {}), + MANAGED_BY_LABEL: FIELD_MANAGER, + OWNER_UID_LABEL: owner_uid, + WORKLOAD_SET_LABEL: workload_set, + ORDINAL_LABEL: str(ordinal), + GENERATION_LABEL: str(generation), + } + + annotations = { + **(template_metadata.get("annotations", {}) or {}), + OWNER_NAME_ANNOTATION: owner_name, + OWNER_NAMESPACE_ANNOTATION: owner_namespace, + } + + pod: PodManifest = { + "apiVersion": "v1", + "kind": "Pod", + "metadata": { + "name": consumer_pod_name(base_name, ordinal, generation), + "labels": labels, + "annotations": annotations, + }, + "spec": pod_spec, + } + + annotations[SPEC_HASH_ANNOTATION] = _pod_spec_hash(pod) + + return pod + + +def delete_owned_pods( + core: client.CoreV1Api, namespace: str, owner_uid: str, logger: Logger +) -> None: + for pod in list_owned_pods(core, namespace, owner_uid): + name = pod_name(pod) + if is_deleting(pod): + continue + delete_pod(core, name, namespace) + logger.info("deleted pipeline Pod %s/%s reason=OwnerDeleted", namespace, name) diff --git a/sentry_streams_k8s/sentry_streams_k8s/operator/pod_status.py b/sentry_streams_k8s/sentry_streams_k8s/operator/pod_status.py new file mode 100644 index 00000000..6c91035e --- /dev/null +++ b/sentry_streams_k8s/sentry_streams_k8s/operator/pod_status.py @@ -0,0 +1,70 @@ +""" +pod_health classifies a Pod into a PodHealth verdict that controls reconcile +actions such as delete, force-delete, and replacement. This module serializes that +verdict with Pod metadata fields into the StreamingPipeline CR's status. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import NotRequired, TypedDict + +from kubernetes.client import V1Pod + +from sentry_streams_k8s.operator.constants import ORDINAL_LABEL +from sentry_streams_k8s.operator.pod_health import PodHealth + + +class PodStatusEntry(TypedDict): + name: str + ready: bool + phase: str + ordinal: NotRequired[str] + reason: NotRequired[str] + permanent: NotRequired[bool] + workloadSet: NotRequired[str] + + +@dataclass(frozen=True) +class ReportedPodStatus: + name: str + ready: bool + phase: str + unhealthy: bool = False + ordinal: str | None = None + 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, + "ready": self.ready, + "phase": self.phase, + } + if self.ordinal is not None: + entry["ordinal"] = self.ordinal + if self.reason is not None: + entry["reason"] = self.reason + if self.permanent: + entry["permanent"] = True + return entry + + +def reported_pod_status(pod: V1Pod, health: PodHealth) -> ReportedPodStatus: + metadata = pod.metadata + status = pod.status + labels = metadata.labels if metadata and metadata.labels else {} + phase = status.phase if status and status.phase else "Unknown" + return ReportedPodStatus( + name=health.name, + ready=health.ready, + phase=phase, + unhealthy=health.unhealthy, + ordinal=labels.get(ORDINAL_LABEL), + reason=health.reason, + permanent=health.permanent, + ) diff --git a/sentry_streams_k8s/sentry_streams_k8s/operator/pod_types.py b/sentry_streams_k8s/sentry_streams_k8s/operator/pod_types.py new file mode 100644 index 00000000..b8bc31bb --- /dev/null +++ b/sentry_streams_k8s/sentry_streams_k8s/operator/pod_types.py @@ -0,0 +1,16 @@ +from __future__ import annotations + +from typing import Any, Literal, NotRequired, TypedDict + + +class PodManifestMetadata(TypedDict): + name: str + labels: NotRequired[dict[str, str]] + annotations: NotRequired[dict[str, str]] + + +class PodManifest(TypedDict): + apiVersion: Literal["v1"] + kind: Literal["Pod"] + metadata: PodManifestMetadata + spec: dict[str, Any] diff --git a/sentry_streams_k8s/sentry_streams_k8s/operator/reconcile.py b/sentry_streams_k8s/sentry_streams_k8s/operator/reconcile.py new file mode 100644 index 00000000..aa3f6629 --- /dev/null +++ b/sentry_streams_k8s/sentry_streams_k8s/operator/reconcile.py @@ -0,0 +1,570 @@ +from __future__ import annotations + +from collections.abc import Mapping +from datetime import datetime, timezone +from typing import Any, TypedDict, cast + +import kopf +from kubernetes import client +from kubernetes.client import V1Pod +from kubernetes.client.exceptions import ApiException + +from sentry_streams_k8s.consumer_builder import compute_config_version +from sentry_streams_k8s.operator.constants import ( + ALL_WORKLOAD_SETS, + CANARY_WORKLOAD_SET, + FIELD_MANAGER, + MAX_BASE_NAME_LENGTH, + MAX_GENERATION, + MAX_REPLICAS, + OWNER_NAME_ANNOTATION, + OWNER_NAMESPACE_ANNOTATION, + OWNER_UID_LABEL, + PRIMARY_WORKLOAD_SET, + Logger, +) +from sentry_streams_k8s.operator.pod_health import ( + PodHealth, + is_deleting, + pod_health, + pod_spec_changed, +) +from sentry_streams_k8s.operator.pod_resources import ( + apply_pod, + build_pipeline_pod, + consumer_pod_name, + delete_pod, + list_owned_pods, + pod_generation, + pod_keep_key, + pod_name, + pod_ordinal, + pod_template_from_deployment, + pod_workload_set, +) +from sentry_streams_k8s.operator.pod_status import ( + PodStatusEntry, + ReportedPodStatus, + reported_pod_status, +) +from sentry_streams_k8s.operator.pod_types import PodManifest +from sentry_streams_k8s.operator.streaming_pipeline import ( + from_crd_spec, + render, + validate, +) + + +class Condition(TypedDict): + type: str + status: str + reason: str + message: str + + +class PodSetResult(TypedDict): + childPods: list[str] + desiredReplicas: int + readyReplicas: int + unhealthyPods: list[PodStatusEntry] + permanentErrors: list[PodStatusEntry] + + +class CombinedPodResult(PodSetResult): + sets: dict[str, PodSetResult | None] + + +class PipelineStatusPatch(TypedDict, total=False): + conditions: list[Condition] + config_version: str + pods: CombinedPodResult + workload_namespace: str + generations: dict[str, dict[str, int] | None] + + +def _parse_generations(data: object) -> dict[int, int]: + """ + Returns a dict mapping ordinal -> highest generation. + Empty when there is no ledger yet (e.g. first reconcile). + """ + + if not isinstance(data, Mapping): + return {} + + generations: dict[int, int] = {} + for key, value in data.items(): + if not isinstance(key, str) or not isinstance(value, int): + continue + try: + generations[int(key)] = value + except ValueError: + continue + return generations + + +def _serialize_generations(generations: dict[int, int]) -> dict[str, int]: + """ + Pods for a replica are named {name}-{ordinal}-{generation}. The generation + increments every time the operator replaces a Pod, so a newly created Pod + never shares a name with the old one that is still terminating. This is what + lets us recreate instantly instead of waiting for the old Pod to be deleted. + We save the highest generation per ordinal in the CR's status.generations. + """ + + return {str(ordinal): generation for ordinal, generation in sorted(generations.items())} + + +def _prepare_manifest( + manifest: dict[str, Any], + *, + workload_namespace: str, + owner_uid: str, + owner_name: str, + owner_namespace: str, +) -> None: + metadata = manifest.setdefault("metadata", {}) + metadata["namespace"] = workload_namespace + metadata["labels"] = { + **metadata.get("labels", {}), + OWNER_UID_LABEL: owner_uid, + } + metadata["annotations"] = { + **metadata.get("annotations", {}), + OWNER_NAME_ANNOTATION: owner_name, + OWNER_NAMESPACE_ANNOTATION: owner_namespace, + } + + +def _apply_configmap( + core: client.CoreV1Api, + manifest: dict[str, Any], + *, + workload_namespace: str, + owner_uid: str, +) -> None: + name = manifest["metadata"]["name"] + try: + existing = core.read_namespaced_config_map(name=name, namespace=workload_namespace) + except ApiException as e: + if e.status != 404: + raise + else: + labels = existing.metadata.labels if existing.metadata and existing.metadata.labels else {} + existing_owner_uid = labels.get(OWNER_UID_LABEL) + # Do not take over a resource with the same name from another pipeline: + if existing_owner_uid != owner_uid: + raise kopf.PermanentError( + f"ConfigMap {workload_namespace}/{name} is already present and is not " + "managed by this StreamingPipeline." + ) + + core.patch_namespaced_config_map( + name=name, + namespace=workload_namespace, + body=manifest, + field_manager=FIELD_MANAGER, + force=True, + _content_type="application/apply-patch+yaml", + ) + + +def prune_stale_configmaps( + *, + workload_namespace: str, + owner_uid: str, + desired_configmaps: set[str], + logger: Logger, +) -> None: + selector = f"{OWNER_UID_LABEL}={owner_uid}" + core = client.CoreV1Api() + configmaps = core.list_namespaced_config_map( + namespace=workload_namespace, + label_selector=selector, + ) + for configmap in configmaps.items: + metadata = configmap.metadata + name = metadata.name if metadata else None + if name is None: + continue + if name not in desired_configmaps: + logger.info("Pruning stale configmap %s/%s", workload_namespace, name) + core.delete_namespaced_config_map(name=name, namespace=workload_namespace) + + +def _condition(type_: str, status: bool, reason: str, message: str = "") -> Condition: + return { + "type": type_, + "status": "True" if status else "False", + "reason": reason, + "message": message, + } + + +def _delete_current_pod( + core: client.CoreV1Api, + pod: V1Pod, + namespace: str, + logger: Logger, + health: PodHealth, + reason: str, +) -> None: + name = pod_name(pod) + if is_deleting(pod) and not health.delete: + logger.info("pipeline Pod %s/%s is already deleting reason=%s", namespace, name, reason) + return + delete_pod(core, name, namespace, force=health.force) + logger.info("deleted pipeline Pod %s/%s reason=%s", namespace, name, reason) + + +def delete_obsolete_pod_sets( + core: client.CoreV1Api, + namespace: str, + owner_uid: str, + desired_sets: set[str], + logger: Logger, +) -> None: + now = datetime.now(timezone.utc) + for pod in list_owned_pods(core, namespace, owner_uid): + workload_set = pod_workload_set(pod) + if workload_set in desired_sets: + continue + health = pod_health(pod, now) + _delete_current_pod( + core, + pod, + namespace, + logger, + health, + f"StaleWorkloadSet:{workload_set or 'missing'}", + ) + + +def _allocate_generation(generations: dict[int, int], ordinal: int, pods: list[V1Pod]) -> int: + live_max = max((pod_generation(pod) for pod in pods), default=-1) + generation = max(generations.get(ordinal, -1), live_max) + 1 + + if generation > MAX_GENERATION: + # TODO: We should see if it is safe to wrap back to generation 0 at this point. + raise kopf.PermanentError(f"Replica {ordinal} exceeds {MAX_GENERATION} generations.") + + generations[ordinal] = generation + return generation + + +def reconcile_pipeline_pods( + *, + core: client.CoreV1Api, + workload_namespace: str, + owner_uid: str, + owner_name: str, + owner_namespace: str, + base_name: str, + template_metadata: Mapping[str, Any], + template_spec: Mapping[str, Any], + replicas: int, + generations: dict[int, int], + logger: Logger, + workload_set: str, +) -> PodSetResult: + desired_ordinals = set(range(max(replicas, 0))) + current = list_owned_pods( + core, + workload_namespace, + owner_uid, + workload_set=workload_set, + ) + now = datetime.now(timezone.utc) + + pods_by_ordinal: dict[int, list[V1Pod]] = {} + health_by_name: dict[str, PodHealth] = {} + reported_statuses: list[ReportedPodStatus] = [] + active_pod_names: list[str] = [] + + def _build(ordinal: int, generation: int) -> PodManifest: + return build_pipeline_pod( + base_name=base_name, + template_metadata=template_metadata, + template_spec=template_spec, + ordinal=ordinal, + generation=generation, + owner_uid=owner_uid, + owner_name=owner_name, + owner_namespace=owner_namespace, + workload_set=workload_set, + ) + + for pod in current: + name = pod_name(pod) + health = pod_health(pod, now) + health_by_name[name] = health + ordinal = pod_ordinal(pod) + if ordinal is None or ordinal not in desired_ordinals: + _delete_current_pod(core, pod, workload_namespace, logger, health, "Stale") + continue + pods_by_ordinal.setdefault(ordinal, []).append(pod) + reported_statuses.append(reported_pod_status(pod, health)) + + for ordinal in sorted(desired_ordinals): + pods = pods_by_ordinal.get(ordinal, []) + desired_template = _build(ordinal, 0) + candidates = [ + pod + for pod in pods + if ( + not is_deleting(pod) + and not pod_spec_changed(pod, desired_template) + and not health_by_name[pod_name(pod)].delete + ) + ] + + keep: V1Pod | None + if candidates: + # Prefer a ready Pod, otherwise keep the newest generation: + keep = max(candidates, key=lambda pod: pod_keep_key(pod, health_by_name[pod_name(pod)])) + active_pod_names.append(pod_name(keep)) + generations[ordinal] = max(generations.get(ordinal, -1), pod_generation(keep)) + else: + # Create the replacement before deleting the old Pod: + generation = _allocate_generation(generations, ordinal, pods) + keep_manifest = _build(ordinal, generation) + apply_pod(core, keep_manifest, workload_namespace) + keep = None + keep_name = consumer_pod_name(base_name, ordinal, generation) + active_pod_names.append(keep_name) + logger.info( + "applied replacement pipeline Pod %s/%s", + workload_namespace, + keep_name, + ) + + for pod in pods: + if pod is keep: + continue + health = health_by_name[pod_name(pod)] + if pod_spec_changed(pod, desired_template): + _delete_current_pod(core, pod, workload_namespace, logger, health, "Outdated") + elif health.delete: + _delete_current_pod( + core, pod, workload_namespace, logger, health, health.reason or "Unhealthy" + ) + elif pod in candidates: + _delete_current_pod(core, pod, workload_namespace, logger, health, "Duplicate") + + ready_ordinals = { + ordinal + for ordinal, pods in pods_by_ordinal.items() + if any( + not is_deleting(pod) + and pod_name(pod) in active_pod_names + and health_by_name[pod_name(pod)].ready + for pod in pods + ) + } + unhealthy_pods = [ + status.to_status_dict() for status in reported_statuses if status.is_unhealthy + ] + permanent_errors = [ + status.to_status_dict() + for status in reported_statuses + if status.is_unhealthy and status.permanent + ] + return { + "childPods": sorted(active_pod_names), + "desiredReplicas": len(desired_ordinals), + "readyReplicas": len(ready_ordinals), + "unhealthyPods": unhealthy_pods, + "permanentErrors": permanent_errors, + } + + +def _reconcile_pod_set( + *, + core: client.CoreV1Api, + deployment: dict[str, Any], + workload_set: str, + workload_namespace: str, + owner_uid: str, + owner_name: str, + owner_namespace: str, + logger: Logger, + previous_generations: dict[int, int], +) -> tuple[PodSetResult, dict[int, int]]: + base_name = deployment["metadata"]["name"] + + if len(base_name) > MAX_BASE_NAME_LENGTH: + raise kopf.PermanentError( + f"{workload_set} name cannot exceed {MAX_BASE_NAME_LENGTH} characters." + ) + + template_metadata, template_spec = pod_template_from_deployment(deployment) + replicas = deployment["spec"].get("replicas", 0) + + if type(replicas) is not int or replicas < 0: + raise kopf.PermanentError(f"{workload_set} replica count must be a non-negative integer.") + + if replicas > MAX_REPLICAS: + raise kopf.PermanentError(f"{workload_set} replica count cannot exceed {MAX_REPLICAS}.") + + generations = dict(previous_generations) + pod_result = reconcile_pipeline_pods( + core=core, + workload_namespace=workload_namespace, + owner_uid=owner_uid, + owner_name=owner_name, + owner_namespace=owner_namespace, + base_name=base_name, + template_metadata=template_metadata, + template_spec=template_spec, + replicas=replicas, + generations=generations, + logger=logger, + workload_set=workload_set, + ) + return pod_result, generations + + +def _combine_pod_results(results: dict[str, PodSetResult]) -> CombinedPodResult: + child_pods: list[str] = [] + unhealthy_pods: list[PodStatusEntry] = [] + permanent_errors: list[PodStatusEntry] = [] + for workload_set, result in results.items(): + child_pods.extend(result["childPods"]) + unhealthy_pods.extend( + _with_workload_set(entry, workload_set) for entry in result["unhealthyPods"] + ) + permanent_errors.extend( + _with_workload_set(entry, workload_set) for entry in result["permanentErrors"] + ) + return { + "childPods": sorted(child_pods), + "desiredReplicas": sum(result["desiredReplicas"] for result in results.values()), + "readyReplicas": sum(result["readyReplicas"] for result in results.values()), + "unhealthyPods": unhealthy_pods, + "permanentErrors": permanent_errors, + # Status is updated as a JSON merge patch, so a set that is not included keeps its old + # value. Explicity include all sets and null out the removed ones to clear them: + "sets": {workload_set: results.get(workload_set) for workload_set in ALL_WORKLOAD_SETS}, + } + + +def _with_workload_set(entry: PodStatusEntry, workload_set: str) -> PodStatusEntry: + result: PodStatusEntry = { + "name": entry["name"], + "ready": entry["ready"], + "phase": entry["phase"], + "workloadSet": workload_set, + } + if "ordinal" in entry: + result["ordinal"] = entry["ordinal"] + if "reason" in entry: + result["reason"] = entry["reason"] + if "permanent" in entry: + result["permanent"] = entry["permanent"] + return result + + +def reconcile_pipeline( + *, + spec: Any, + name: str, + namespace: str, + uid: str, + workload_namespace: str, + logger: Logger, + patch: kopf.Patch | None = None, + status: PipelineStatusPatch | None = None, + previous_generations: object = None, +) -> CombinedPodResult: + status_patch = ( + status + if status is not None + else cast(PipelineStatusPatch, patch.status) if patch is not None else None + ) + previous_generations = previous_generations if isinstance(previous_generations, Mapping) else {} + consumer = from_crd_spec(dict(spec), name=name) + try: + validate(consumer) + result = render(consumer) + except Exception as e: + if status_patch is not None: + status_patch["conditions"] = [_condition("Rendered", False, type(e).__name__, str(e))] + raise kopf.PermanentError(f"StreamingPipeline {namespace}/{name} failed to render: {e}") + + core = client.CoreV1Api() + configmap = result["configmap"] + + _prepare_manifest( + configmap, + workload_namespace=workload_namespace, + owner_uid=uid, + owner_name=name, + owner_namespace=namespace, + ) + _apply_configmap(core, configmap, workload_namespace=workload_namespace, owner_uid=uid) + + rendered_sets = {PRIMARY_WORKLOAD_SET: result["deployment"]} + if "canary_deployment" in result: + rendered_sets[CANARY_WORKLOAD_SET] = result["canary_deployment"] + + pod_set_results: dict[str, PodSetResult] = {} + generations_by_set: dict[str, dict[int, int]] = {} + for workload_set, deployment in rendered_sets.items(): + set_result, generations = _reconcile_pod_set( + core=core, + deployment=deployment, + workload_set=workload_set, + workload_namespace=workload_namespace, + owner_uid=uid, + owner_name=name, + owner_namespace=namespace, + logger=logger, + previous_generations=_parse_generations(previous_generations.get(workload_set)), + ) + pod_set_results[workload_set] = set_result + generations_by_set[workload_set] = generations + + # Remove Pods left behind by a removed canary set: + delete_obsolete_pod_sets( + core, + workload_namespace, + uid, + set(rendered_sets), + logger, + ) + pod_result = _combine_pod_results(pod_set_results) + prune_stale_configmaps( + workload_namespace=workload_namespace, + owner_uid=uid, + desired_configmaps={configmap["metadata"]["name"]}, + logger=logger, + ) + + if status_patch is not None: + conditions = [ + _condition("Rendered", True, "Rendered"), + _condition("Applied", True, "Applied"), + ] + permanent_errors = pod_result["permanentErrors"] + if permanent_errors: + error = permanent_errors[0] + conditions[-1] = _condition( + "Applied", + False, + "PermanentPodFailure", + f"Pod {error['name']} is permanently unhealthy: {error.get('reason', 'Unknown')}. " + "Update the StreamingPipeline spec.", + ) + status_patch["conditions"] = conditions + status_patch["config_version"] = compute_config_version(consumer["pipeline_config"]) + status_patch["pods"] = pod_result + status_patch["workload_namespace"] = workload_namespace + status_patch["generations"] = { + workload_set: ( + _serialize_generations(generations_by_set[workload_set]) + if workload_set in generations_by_set + else None + ) + for workload_set in ALL_WORKLOAD_SETS + } + + return pod_result diff --git a/sentry_streams_k8s/tests/k8s_fixtures.py b/sentry_streams_k8s/tests/k8s_fixtures.py new file mode 100644 index 00000000..2a4d6d80 --- /dev/null +++ b/sentry_streams_k8s/tests/k8s_fixtures.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +from collections.abc import Mapping +from datetime import datetime + +from kubernetes.client import ( + V1ContainerState, + V1ContainerStateTerminated, + V1ContainerStateWaiting, + V1ContainerStatus, + V1ObjectMeta, + V1Pod, + V1PodCondition, + V1PodStatus, +) + + +def make_condition(type_: str, status: str) -> V1PodCondition: + return V1PodCondition(type=type_, status=status) + + +def make_waiting_container_status(reason: str, *, name: str = "c") -> V1ContainerStatus: + return V1ContainerStatus( + name=name, + ready=False, + restart_count=0, + image="test-image", + image_id="", + state=V1ContainerState(waiting=V1ContainerStateWaiting(reason=reason)), + ) + + +def make_terminated_container_status( + exit_code: int, reason: str | None = None, *, name: str = "c" +) -> V1ContainerStatus: + return V1ContainerStatus( + name=name, + ready=False, + restart_count=0, + image="test-image", + image_id="", + state=V1ContainerState( + terminated=V1ContainerStateTerminated(exit_code=exit_code, reason=reason) + ), + ) + + +def make_pod( + *, + name: str = "consumer-0-0", + labels: Mapping[str, str] | None = None, + annotations: Mapping[str, str] | None = None, + creation_timestamp: datetime | None = None, + deletion_timestamp: datetime | None = None, + phase: str | None = None, + conditions: list[V1PodCondition] | None = None, + container_statuses: list[V1ContainerStatus] | None = None, + init_container_statuses: list[V1ContainerStatus] | None = None, + start_time: datetime | None = None, + reason: str | None = None, +) -> V1Pod: + return V1Pod( + metadata=V1ObjectMeta( + name=name, + labels=dict(labels) if labels is not None else None, + annotations=dict(annotations) if annotations is not None else None, + creation_timestamp=creation_timestamp, + deletion_timestamp=deletion_timestamp, + ), + status=V1PodStatus( + phase=phase, + conditions=conditions, + container_statuses=container_statuses, + init_container_statuses=init_container_statuses, + start_time=start_time, + reason=reason, + ), + ) diff --git a/sentry_streams_k8s/tests/test_operator.py b/sentry_streams_k8s/tests/test_operator.py index c34859c0..0904eabb 100644 --- a/sentry_streams_k8s/tests/test_operator.py +++ b/sentry_streams_k8s/tests/test_operator.py @@ -1,18 +1,28 @@ from __future__ import annotations +import asyncio from types import SimpleNamespace -from unittest.mock import MagicMock, patch +from typing import Any +from unittest.mock import ANY, MagicMock, patch import kopf import pytest -from sentry_streams_k8s.operator.operator import ( +from sentry_streams_k8s.operator.constants import ( OWNER_NAME_ANNOTATION, OWNER_NAMESPACE_ANNOTATION, OWNER_UID_LABEL, - _apply, +) +from sentry_streams_k8s.operator.operator import ( + ReconcileScheduler, + cleanup, + handle_pipeline_pod_event, +) +from sentry_streams_k8s.operator.reconcile import ( + _apply_configmap, _prepare_manifest, - _prune_stale_resources, + prune_stale_configmaps, + reconcile_pipeline, ) WORKLOAD_NAMESPACE = "test-streaming-pipelines" @@ -53,13 +63,11 @@ def test_prepare_manifest_routes_workload_and_records_source_cr() -> None: } -def test_apply_rejects_resource_owned_by_another_cr() -> None: - resource = MagicMock() - resource.get.return_value = SimpleNamespace( +def test_apply_configmap_rejects_resource_owned_by_another_cr() -> None: + core = MagicMock() + core.read_namespaced_config_map.return_value = SimpleNamespace( metadata=SimpleNamespace(labels={OWNER_UID_LABEL: "another-owner"}) ) - dyn = MagicMock() - dyn.resources.get.return_value = resource manifest = { "apiVersion": "v1", "kind": "ConfigMap", @@ -67,75 +75,373 @@ def test_apply_rejects_resource_owned_by_another_cr() -> None: } with pytest.raises(kopf.PermanentError, match="not managed by this StreamingPipeline"): - _apply( - dyn, + _apply_configmap( + core, manifest, workload_namespace=WORKLOAD_NAMESPACE, owner_uid="owner-uid", ) - dyn.server_side_apply.assert_not_called() + core.patch_namespaced_config_map.assert_not_called() -def test_apply_uses_workload_namespace_and_stable_field_manager() -> None: - resource = MagicMock() - resource.get.return_value = SimpleNamespace( +def test_apply_configmap_uses_workload_namespace_and_stable_field_manager() -> None: + core = MagicMock() + core.read_namespaced_config_map.return_value = SimpleNamespace( metadata=SimpleNamespace(labels={OWNER_UID_LABEL: "owner-uid"}) ) - dyn = MagicMock() - dyn.resources.get.return_value = resource manifest = { "apiVersion": "v1", "kind": "ConfigMap", "metadata": {"name": "pipeline", "namespace": WORKLOAD_NAMESPACE}, } - _apply( - dyn, + _apply_configmap( + core, manifest, workload_namespace=WORKLOAD_NAMESPACE, owner_uid="owner-uid", ) - resource.get.assert_called_once_with(name="pipeline", namespace=WORKLOAD_NAMESPACE) - dyn.server_side_apply.assert_called_once_with( - resource, - body=manifest, + core.read_namespaced_config_map.assert_called_once_with( + name="pipeline", namespace=WORKLOAD_NAMESPACE + ) + core.patch_namespaced_config_map.assert_called_once_with( + name="pipeline", namespace=WORKLOAD_NAMESPACE, + body=manifest, field_manager="streaming-operator", - force_conflicts=True, + force=True, + _content_type="application/apply-patch+yaml", ) -@patch("sentry_streams_k8s.operator.operator.client.CoreV1Api") -@patch("sentry_streams_k8s.operator.operator.client.AppsV1Api") -def test_prune_removes_only_stale_resources( - apps_api: MagicMock, - core_api: MagicMock, -) -> None: - apps = apps_api.return_value - apps.list_namespaced_deployment.return_value.items = [ - SimpleNamespace(metadata=SimpleNamespace(name="desired-deployment")), - SimpleNamespace(metadata=SimpleNamespace(name="stale-deployment")), - ] +@patch("sentry_streams_k8s.operator.reconcile.client.CoreV1Api") +def test_prune_removes_only_stale_configmaps(core_api: MagicMock) -> None: core = core_api.return_value core.list_namespaced_config_map.return_value.items = [ SimpleNamespace(metadata=SimpleNamespace(name="desired-configmap")), SimpleNamespace(metadata=SimpleNamespace(name="stale-configmap")), ] - _prune_stale_resources( + prune_stale_configmaps( workload_namespace=WORKLOAD_NAMESPACE, owner_uid="owner-uid", - desired_deployments={"desired-deployment"}, desired_configmaps={"desired-configmap"}, + logger=MagicMock(), ) - apps.delete_namespaced_deployment.assert_called_once_with( - name="stale-deployment", - namespace=WORKLOAD_NAMESPACE, - ) core.delete_namespaced_config_map.assert_called_once_with( name="stale-configmap", namespace=WORKLOAD_NAMESPACE, ) + + +def test_do_reconcile_applies_config_reconciles_pods_and_reports_generations( + monkeypatch: pytest.MonkeyPatch, +) -> None: + core = MagicMock() + configmap = {"apiVersion": "v1", "kind": "ConfigMap", "metadata": {"name": "pipeline-config"}} + primary_deployment = { + "metadata": {"name": "consumer"}, + "spec": { + "replicas": 2, + "template": {"metadata": {"labels": {"env": "primary"}}, "spec": {"containers": []}}, + }, + } + canary_deployment = { + "metadata": {"name": "consumer-canary"}, + "spec": { + "replicas": 1, + "template": {"metadata": {"labels": {"env": "canary"}}, "spec": {"containers": []}}, + }, + } + primary_result = { + "childPods": ["consumer-0-2"], + "desiredReplicas": 2, + "readyReplicas": 1, + "unhealthyPods": [], + "permanentErrors": [], + } + canary_result = { + "childPods": ["consumer-canary-0-5"], + "desiredReplicas": 1, + "readyReplicas": 0, + "unhealthyPods": [], + "permanentErrors": [], + } + apply = MagicMock() + reconcile_pods = MagicMock() + delete_obsolete = MagicMock() + prune = MagicMock() + monkeypatch.setenv("WORKLOAD_NAMESPACE", WORKLOAD_NAMESPACE) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.from_crd_spec", lambda spec, name: spec + ) + monkeypatch.setattr("sentry_streams_k8s.operator.reconcile.validate", lambda _consumer: None) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.render", + lambda _consumer: { + "configmap": configmap, + "deployment": primary_deployment, + "canary_deployment": canary_deployment, + }, + ) + monkeypatch.setattr("sentry_streams_k8s.operator.reconcile.client.CoreV1Api", lambda: core) + monkeypatch.setattr("sentry_streams_k8s.operator.reconcile._apply_configmap", apply) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.reconcile_pipeline_pods", reconcile_pods + ) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.delete_obsolete_pod_sets", + delete_obsolete, + ) + monkeypatch.setattr("sentry_streams_k8s.operator.reconcile.prune_stale_configmaps", prune) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.compute_config_version", lambda _: "version" + ) + + def allocate_generation( + *, generations: dict[int, int], workload_set: str, **_: object + ) -> dict[str, object]: + if workload_set == "primary": + generations[0] = 2 + return primary_result + generations[0] = 5 + return canary_result + + reconcile_pods.side_effect = allocate_generation + status: dict[str, Any] = {} + logger = MagicMock() + result = reconcile_pipeline( + spec={"pipeline_config": {}, "replicas": 1, "with_canary": True}, + name="pipeline", + namespace="source", + uid="uid", + workload_namespace=WORKLOAD_NAMESPACE, + logger=logger, + status=status, + previous_generations={"primary": {"0": 1}, "canary": {"0": 4}}, + ) + + assert result == { + "childPods": ["consumer-0-2", "consumer-canary-0-5"], + "desiredReplicas": 3, + "readyReplicas": 1, + "unhealthyPods": [], + "permanentErrors": [], + "sets": { + "primary": primary_result, + "canary": canary_result, + }, + } + apply.assert_called_once_with( + core, + configmap, + workload_namespace=WORKLOAD_NAMESPACE, + owner_uid="uid", + ) + assert [call.kwargs["workload_set"] for call in reconcile_pods.call_args_list] == [ + "primary", + "canary", + ] + assert [call.kwargs["replicas"] for call in reconcile_pods.call_args_list] == [2, 1] + assert [call.kwargs["base_name"] for call in reconcile_pods.call_args_list] == [ + "consumer", + "consumer-canary", + ] + delete_obsolete.assert_called_once_with( + core, + WORKLOAD_NAMESPACE, + "uid", + {"primary", "canary"}, + ANY, + ) + prune.assert_called_once_with( + workload_namespace=WORKLOAD_NAMESPACE, + owner_uid="uid", + desired_configmaps={"pipeline-config"}, + logger=logger, + ) + assert status["conditions"][-1] == { + "type": "Applied", + "status": "True", + "reason": "Applied", + "message": "", + } + assert status["pods"] == result + assert status["generations"] == {"primary": {"0": 2}, "canary": {"0": 5}} + + +def test_do_reconcile_preserves_unchanged_ledger_and_reports_permanent_pod_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + core = MagicMock() + result = { + "childPods": ["consumer-0-1"], + "desiredReplicas": 1, + "readyReplicas": 0, + "unhealthyPods": [], + "permanentErrors": [ + { + "name": "consumer-0-1", + "ready": False, + "phase": "Pending", + "reason": "InvalidImageName", + "permanent": True, + } + ], + } + monkeypatch.setenv("WORKLOAD_NAMESPACE", WORKLOAD_NAMESPACE) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.from_crd_spec", lambda spec, name: spec + ) + monkeypatch.setattr("sentry_streams_k8s.operator.reconcile.validate", lambda _consumer: None) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.render", + lambda _: { + "configmap": {"apiVersion": "v1", "kind": "ConfigMap", "metadata": {"name": "config"}}, + "deployment": { + "metadata": {"name": "consumer"}, + "spec": { + "replicas": 1, + "template": {"metadata": {}, "spec": {"containers": []}}, + }, + }, + }, + ) + monkeypatch.setattr("sentry_streams_k8s.operator.reconcile.client.CoreV1Api", lambda: core) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile._apply_configmap", lambda *_args, **_kwargs: None + ) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.reconcile_pipeline_pods", lambda **_: result + ) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.delete_obsolete_pod_sets", + lambda *_args, **_kwargs: None, + ) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.prune_stale_configmaps", lambda **_: None + ) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.compute_config_version", lambda _: "version" + ) + patch = SimpleNamespace(status={}) + + reconcile_pipeline( + spec={"pipeline_config": {}}, + name="pipeline", + namespace="source", + uid="uid", + workload_namespace=WORKLOAD_NAMESPACE, + logger=MagicMock(), + patch=patch, + previous_generations={"primary": {"0": 1}}, + ) + + assert patch.status["conditions"][-1]["status"] == "False" + assert patch.status["conditions"][-1]["reason"] == "PermanentPodFailure" + assert patch.status["generations"] == {"primary": {"0": 1}, "canary": None} + + +def test_pod_event_coalesces_reconcile_requests_and_ignores_healthy_updates() -> None: + scheduler = ReconcileScheduler() + event = scheduler.register("uid") + memo = SimpleNamespace(reconcile_scheduler=scheduler) + + asyncio.run( + handle_pipeline_pod_event( + type="MODIFIED", + body={"metadata": {"name": "consumer-0-0"}, "status": {"phase": "Running"}}, + meta=kopf.Meta({}), + labels={OWNER_UID_LABEL: "uid"}, + name="consumer-0-0", + namespace=WORKLOAD_NAMESPACE, + memo=memo, + logger=MagicMock(), + ) + ) + assert not event.is_set() + + asyncio.run( + handle_pipeline_pod_event( + type="DELETED", + body={}, + meta=kopf.Meta({}), + labels={OWNER_UID_LABEL: "uid"}, + name="consumer-0-0", + namespace=WORKLOAD_NAMESPACE, + memo=memo, + logger=MagicMock(), + ) + ) + asyncio.run( + handle_pipeline_pod_event( + type="DELETED", + body={}, + meta=kopf.Meta({}), + labels={OWNER_UID_LABEL: "uid"}, + name="consumer-1-0", + namespace=WORKLOAD_NAMESPACE, + memo=memo, + logger=MagicMock(), + ) + ) + assert event.is_set() + + +def test_pod_event_ignores_missing_or_inactive_owner() -> None: + scheduler = ReconcileScheduler() + event = scheduler.register("active-uid") + memo = SimpleNamespace(reconcile_scheduler=scheduler) + logger = MagicMock() + + asyncio.run( + handle_pipeline_pod_event( + type="DELETED", + body={}, + meta=kopf.Meta({}), + labels={}, + name="consumer-0-0", + namespace=WORKLOAD_NAMESPACE, + memo=memo, + logger=logger, + ) + ) + asyncio.run( + handle_pipeline_pod_event( + type="DELETED", + body={}, + meta=kopf.Meta({}), + labels={OWNER_UID_LABEL: "deleted-uid"}, + name="consumer-0-0", + namespace=WORKLOAD_NAMESPACE, + memo=memo, + logger=logger, + ) + ) + assert not event.is_set() + + +def test_cleanup_deletes_owned_pods_and_prunes_resources(monkeypatch: pytest.MonkeyPatch) -> None: + core = MagicMock() + delete_pods = MagicMock() + prune = MagicMock() + monkeypatch.setenv("WORKLOAD_NAMESPACE", WORKLOAD_NAMESPACE) + monkeypatch.setattr("sentry_streams_k8s.operator.operator.client.CoreV1Api", lambda: core) + monkeypatch.setattr("sentry_streams_k8s.operator.operator.delete_owned_pods", delete_pods) + monkeypatch.setattr("sentry_streams_k8s.operator.operator.prune_stale_configmaps", prune) + scheduler = ReconcileScheduler() + scheduler.register("uid") + memo = SimpleNamespace(reconcile_scheduler=scheduler) + + asyncio.run(cleanup(uid="uid", memo=memo, logger=MagicMock())) + + delete_pods.assert_called_once_with(core, WORKLOAD_NAMESPACE, "uid", ANY) + prune.assert_called_once_with( + workload_namespace=WORKLOAD_NAMESPACE, + owner_uid="uid", + desired_configmaps=set(), + logger=ANY, + ) diff --git a/sentry_streams_k8s/tests/test_pod_health.py b/sentry_streams_k8s/tests/test_pod_health.py new file mode 100644 index 00000000..b104f26b --- /dev/null +++ b/sentry_streams_k8s/tests/test_pod_health.py @@ -0,0 +1,262 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any + +import pytest +from kubernetes.client import V1ObjectMeta, V1Pod, V1PodCondition + +from sentry_streams_k8s.operator.constants import ORDINAL_LABEL, SPEC_HASH_ANNOTATION +from sentry_streams_k8s.operator.pod_health import ( + pod_health, + pod_spec_changed, +) +from sentry_streams_k8s.operator.pod_types import PodManifest +from tests.k8s_fixtures import ( + make_condition, + make_pod, + make_terminated_container_status, + make_waiting_container_status, +) + + +def _image_pull_pod( + *, + now: datetime, + start_time: datetime | None, + creation_time: datetime, + reason: str = "ImagePullBackOff", +) -> V1Pod: + return make_pod( + name="consumer-0-0", + creation_timestamp=creation_time, + phase="Pending", + container_statuses=[make_waiting_container_status(reason)], + start_time=start_time, + ) + + +def test_image_pull_wait_uses_pod_start_time_for_grace() -> None: + now = datetime(2026, 7, 15, 20, 0, tzinfo=timezone.utc) + + start_time = now - timedelta(seconds=299) + creation_time = now - timedelta(hours=1) + + pod = _image_pull_pod(now=now, start_time=start_time, creation_time=creation_time) + + assert pod_health(pod, now).reason == "ImagePullBackOff" + assert pod_health(pod, now).delete is False + + +def test_image_pull_wait_is_deleted_after_start_time_grace() -> None: + now = datetime(2026, 7, 15, 20, 0, tzinfo=timezone.utc) + + start_time = now - timedelta(minutes=5) + creation_time = now - timedelta(hours=1) + + pod = _image_pull_pod(now=now, start_time=start_time, creation_time=creation_time) + + assert pod_health(pod, now).delete is True + + +def test_image_pull_wait_uses_creation_time_when_not_started() -> None: + now = datetime(2026, 7, 15, 20, 0, tzinfo=timezone.utc) + + creation_time = now - timedelta(minutes=5) + + pod = _image_pull_pod(now=now, start_time=None, creation_time=creation_time) + + assert pod_health(pod, now).delete is True + + +def test_invalid_image_name_is_a_non_replacing_permanent_error() -> None: + now = datetime(2026, 7, 15, 20, 0, tzinfo=timezone.utc) + pod = _image_pull_pod( + now=now, + start_time=now - timedelta(hours=1), + creation_time=now - timedelta(hours=1), + reason="InvalidImageName", + ) + + health = pod_health(pod, now) + + assert health.reason == "InvalidImageName" + assert health.unhealthy is True + assert health.delete is False + assert health.permanent is True + + +def test_crash_loop_backoff_is_not_a_waiting_replacement_reason() -> None: + now = datetime(2026, 7, 15, 20, 0, tzinfo=timezone.utc) + pod = _image_pull_pod( + now=now, + start_time=now - timedelta(hours=1), + creation_time=now - timedelta(hours=1), + reason="CrashLoopBackOff", + ) + + health = pod_health(pod, now) + + assert health.reason is None + assert health.delete is False + + +@pytest.mark.parametrize( + ("phase", "conditions", "expected"), + [ + ("Running", [make_condition("Ready", "True")], True), + ("Running", [make_condition("Ready", "False")], False), + ("Pending", [make_condition("Ready", "True")], False), + ], +) +def test_pod_health_only_marks_running_ready_pods_ready( + phase: str, conditions: list[V1PodCondition], expected: bool +) -> None: + pod = make_pod(phase=phase, conditions=conditions) + + health = pod_health(pod, datetime(2026, 7, 15, tzinfo=timezone.utc)) + + assert health.ready is expected + assert health.delete is False + + +@pytest.mark.parametrize( + ("pod_kwargs", "reason"), + [ + ({"phase": "Succeeded"}, "Succeeded"), + ({"phase": "Failed", "reason": "Evicted"}, "Evicted"), + ( + { + "phase": "Running", + "container_statuses": [make_terminated_container_status(137, "OOMKilled")], + }, + "OOMKilled", + ), + ( + { + "phase": "Running", + "container_statuses": [make_terminated_container_status(9)], + }, + "ExitCode9", + ), + ( + { + "phase": "Pending", + "init_container_statuses": [make_terminated_container_status(1, "Error")], + }, + "InitContainerError", + ), + ], +) +def test_pod_health_replaces_terminal_failures(pod_kwargs: dict[str, Any], reason: str) -> None: + health = pod_health( + make_pod(**pod_kwargs), + datetime(2026, 7, 15, tzinfo=timezone.utc), + ) + + assert health.reason == reason + assert health.delete is True + assert health.force is False + + +@pytest.mark.parametrize("statuses_key", ["container_statuses", "init_container_statuses"]) +def test_completed_container_does_not_fail_pod(statuses_key: str) -> None: + health = pod_health( + make_pod( + phase="Running", + **{statuses_key: [make_terminated_container_status(0, "Completed")]}, + ), + datetime(2026, 7, 15, tzinfo=timezone.utc), + ) + + assert health.reason is None + assert health.delete is False + + +@pytest.mark.parametrize( + ("statuses_key", "reason", "expected_reason", "permanent"), + [ + ("init_container_statuses", "ImagePullBackOff", "InitContainerImagePullBackOff", False), + ("init_container_statuses", "InvalidImageName", "InitContainerInvalidImageName", True), + ("container_statuses", "InvalidImageName", "InvalidImageName", True), + ], +) +def test_pod_health_classifies_init_and_permanent_waiting_reasons( + statuses_key: str, reason: str, expected_reason: str, permanent: bool +) -> None: + now = datetime(2026, 7, 15, tzinfo=timezone.utc) + health = pod_health( + make_pod( + creation_timestamp=now, + phase="Pending", + **{statuses_key: [make_waiting_container_status(reason)]}, + ), + now, + ) + + assert health.reason == expected_reason + assert health.permanent is permanent + + +@pytest.mark.parametrize( + ("age", "delete", "force", "reason"), + [ + (599, False, False, "Terminating"), + (600, True, True, "StuckTerminating"), + ], +) +def test_pod_health_force_deletes_only_stuck_terminating_pods( + age: int, delete: bool, force: bool, reason: str +) -> None: + now = datetime(2026, 7, 15, tzinfo=timezone.utc) + health = pod_health( + make_pod(deletion_timestamp=now - timedelta(seconds=age)), + now, + ) + + assert health.reason == reason + assert health.unhealthy is delete + assert health.delete is delete + assert health.force is force + + +def test_pod_health_handles_terminating_pod_without_status() -> None: + now = datetime(2026, 7, 15, tzinfo=timezone.utc) + pod = V1Pod( + metadata=V1ObjectMeta(name="consumer-0-0", deletion_timestamp=now - timedelta(seconds=600)) + ) + + health = pod_health(pod, now) + + assert health.reason == "StuckTerminating" + assert health.delete is True + assert health.force is True + + +def test_pod_health_handles_pod_without_status() -> None: + health = pod_health( + V1Pod(metadata=V1ObjectMeta(name="consumer-0-0")), + datetime(2026, 7, 15, tzinfo=timezone.utc), + ) + + assert health.ready is False + assert health.delete is False + + +def test_pod_spec_changed() -> None: + current = make_pod( + labels={ORDINAL_LABEL: "0"}, + annotations={SPEC_HASH_ANNOTATION: "old"}, + phase="Pending", + ) + desired: PodManifest = { + "apiVersion": "v1", + "kind": "Pod", + "metadata": { + "name": "consumer-0-0", + "annotations": {SPEC_HASH_ANNOTATION: "new"}, + }, + "spec": {}, + } + + assert pod_spec_changed(current, desired) is True diff --git a/sentry_streams_k8s/tests/test_pods.py b/sentry_streams_k8s/tests/test_pods.py new file mode 100644 index 00000000..5b4c90d8 --- /dev/null +++ b/sentry_streams_k8s/tests/test_pods.py @@ -0,0 +1,471 @@ +from __future__ import annotations + +import copy +from datetime import datetime, timezone +from typing import Any +from unittest.mock import MagicMock + +import kopf +import pytest +from kubernetes.client import V1Pod + +from sentry_streams_k8s.operator.constants import ( + CANARY_WORKLOAD_SET, + GENERATION_LABEL, + MAX_BASE_NAME_LENGTH, + MAX_GENERATION, + MAX_REPLICAS, + ORDINAL_LABEL, + PRIMARY_WORKLOAD_SET, + SPEC_HASH_ANNOTATION, + WORKLOAD_SET_LABEL, +) +from sentry_streams_k8s.operator.pod_resources import ( + build_pipeline_pod, + delete_owned_pods, + list_owned_pods, + pod_workload_set, +) +from sentry_streams_k8s.operator.pod_types import PodManifest +from sentry_streams_k8s.operator.reconcile import ( + PodSetResult, + _combine_pod_results, + _parse_generations, + _reconcile_pod_set, + _serialize_generations, + delete_obsolete_pod_sets, + reconcile_pipeline_pods, +) +from tests.k8s_fixtures import make_condition, make_pod + +NAMESPACE = "workloads" +OWNER_UID = "owner-uid" + + +def _template() -> tuple[dict[str, Any], dict[str, Any]]: + return ( + {"labels": {"app": "consumer"}, "annotations": {"configVersion": "one"}}, + {"containers": [{"name": "consumer", "image": "example/consumer:v1"}]}, + ) + + +def _pod( + ordinal: int, + generation: int, + *, + ready: bool = False, + phase: str = "Pending", + annotations: dict[str, str] | None = None, + deletion_timestamp: datetime | None = None, +) -> V1Pod: + labels = { + ORDINAL_LABEL: str(ordinal), + GENERATION_LABEL: str(generation), + WORKLOAD_SET_LABEL: PRIMARY_WORKLOAD_SET, + } + conditions = [make_condition("Ready", "True")] if ready else [] + return make_pod( + name=f"consumer-{ordinal}-{generation}", + labels=labels, + annotations=annotations or {}, + deletion_timestamp=deletion_timestamp, + phase=phase, + conditions=conditions, + ) + + +def _existing_pod_from_manifest( + manifest: PodManifest, *, phase: str = "Pending", ready: bool = False +) -> V1Pod: + """Turns a desired-manifest dict (as returned by build_pipeline_pod) into an + existing/observed V1Pod fixture with the same identity, for tests that need a + "live" Pod matching a specific desired spec.""" + metadata = manifest["metadata"] + conditions = [make_condition("Ready", "True")] if ready else [] + return make_pod( + name=metadata["name"], + labels=metadata.get("labels"), + annotations=metadata.get("annotations"), + phase=phase, + conditions=conditions, + ) + + +def _reconcile( + monkeypatch: pytest.MonkeyPatch, + pods: list[V1Pod], + *, + replicas: int = 1, + generations: dict[int, int] | None = None, + template_metadata: dict[str, Any] | None = None, + template_spec: dict[str, Any] | None = None, + workload_set: str = PRIMARY_WORKLOAD_SET, +) -> tuple[list[PodManifest], list[tuple[str, bool]], PodSetResult, dict[int, int]]: + applied: list[PodManifest] = [] + deleted: list[tuple[str, bool]] = [] + + def list_pods( + _core: object, + _namespace: str, + _owner_uid: str, + workload_set: str | None = None, + ) -> list[V1Pod]: + current = copy.deepcopy(pods) + if workload_set is None: + return current + return [pod for pod in current if pod_workload_set(pod) == workload_set] + + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.list_owned_pods", + list_pods, + ) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.apply_pod", + lambda _core, pod, _namespace: applied.append(pod), + ) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.delete_pod", + lambda _core, name, _namespace, force=False: deleted.append((name, force)), + ) + metadata, spec = _template() + ledger = generations if generations is not None else {} + result = reconcile_pipeline_pods( + core=MagicMock(), + workload_namespace=NAMESPACE, + owner_uid=OWNER_UID, + owner_name="pipeline", + owner_namespace="source", + base_name="consumer", + template_metadata=template_metadata or metadata, + template_spec=template_spec or spec, + replicas=replicas, + generations=ledger, + logger=MagicMock(), + workload_set=workload_set, + ) + return applied, deleted, result, ledger + + +def test_build_pipeline_pod_stamps_identity_without_mutating_template() -> None: + metadata, spec = _template() + + pod = build_pipeline_pod( + base_name="consumer", + template_metadata=metadata, + template_spec=spec, + ordinal=2, + generation=4, + owner_uid=OWNER_UID, + owner_name="pipeline", + owner_namespace="source", + workload_set=PRIMARY_WORKLOAD_SET, + ) + + assert pod["metadata"]["name"] == "consumer-2-4" + assert pod["metadata"]["labels"][ORDINAL_LABEL] == "2" + assert pod["metadata"]["labels"][GENERATION_LABEL] == "4" + assert pod["metadata"]["labels"][WORKLOAD_SET_LABEL] == PRIMARY_WORKLOAD_SET + assert pod["metadata"]["annotations"][SPEC_HASH_ANNOTATION] + assert pod["spec"]["restartPolicy"] == "Never" + assert "restartPolicy" not in spec + + +def test_list_owned_pods_selects_owner_and_workload_set() -> None: + core = MagicMock() + core.list_namespaced_pod.return_value.items = [] + + assert list_owned_pods(core, NAMESPACE, OWNER_UID, CANARY_WORKLOAD_SET) == [] + + core.list_namespaced_pod.assert_called_once_with( + namespace=NAMESPACE, + label_selector=( + f"streams.sentry.io/owner-uid={OWNER_UID}," + f"streams.sentry.io/workload-set={CANARY_WORKLOAD_SET}" + ), + ) + + +def test_reconcile_creates_all_missing_ordinals_and_is_idempotent( + monkeypatch: pytest.MonkeyPatch, +) -> None: + applied, deleted, result, ledger = _reconcile(monkeypatch, [], replicas=2) + + assert [pod["metadata"]["name"] for pod in applied] == ["consumer-0-0", "consumer-1-0"] + assert deleted == [] + assert ledger == {0: 0, 1: 0} + assert result["childPods"] == ["consumer-0-0", "consumer-1-0"] + assert result["desiredReplicas"] == 2 + assert result["readyReplicas"] == 0 + + metadata, spec = _template() + current = [ + _existing_pod_from_manifest( + build_pipeline_pod( + base_name="consumer", + template_metadata=metadata, + template_spec=spec, + ordinal=ordinal, + generation=0, + owner_uid=OWNER_UID, + owner_name="pipeline", + owner_namespace="source", + workload_set=PRIMARY_WORKLOAD_SET, + ), + phase="Running", + ready=True, + ) + for ordinal in range(2) + ] + applied, deleted, result, ledger = _reconcile(monkeypatch, current, replicas=2) + + assert applied == [] + assert deleted == [] + assert ledger == {0: 0, 1: 0} + assert result["readyReplicas"] == 2 + + +def test_reconcile_replaces_failed_pod_with_next_generation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + current = _pod(0, 3, phase="Failed") + + applied, deleted, result, ledger = _reconcile(monkeypatch, [current], generations={0: 7}) + + assert [pod["metadata"]["name"] for pod in applied] == ["consumer-0-8"] + assert deleted == [("consumer-0-3", False)] + assert ledger == {0: 8} + assert result["unhealthyPods"] == [ + { + "name": "consumer-0-3", + "ready": False, + "phase": "Failed", + "ordinal": "0", + "reason": "Failed", + } + ] + + +def test_reconcile_raises_when_generation_exceeds_cap( + monkeypatch: pytest.MonkeyPatch, +) -> None: + current = _pod(0, MAX_GENERATION, phase="Failed") + with pytest.raises(kopf.PermanentError, match="exceeds"): + _reconcile(monkeypatch, [current], generations={0: MAX_GENERATION}) + + +def _set_result(*child_pods: str) -> PodSetResult: + return { + "childPods": list(child_pods), + "desiredReplicas": len(child_pods), + "readyReplicas": len(child_pods), + "unhealthyPods": [], + "permanentErrors": [], + } + + +def test_combine_pod_results_nulls_out_absent_sets() -> None: + combined = _combine_pod_results({PRIMARY_WORKLOAD_SET: _set_result("consumer-0-0")}) + assert combined["sets"] == { + PRIMARY_WORKLOAD_SET: _set_result("consumer-0-0"), + CANARY_WORKLOAD_SET: None, + } + + +def _deployment(name: str, replicas: int) -> dict[str, Any]: + metadata, spec = _template() + return { + "metadata": {"name": name}, + "spec": { + "replicas": replicas, + "template": {"metadata": metadata, "spec": spec}, + }, + } + + +def test_reconcile_pod_set_rejects_replicas_over_cap() -> None: + deployment = _deployment("consumer", MAX_REPLICAS + 1) + with pytest.raises(kopf.PermanentError, match="replica count cannot exceed"): + _reconcile_pod_set( + core=MagicMock(), + deployment=deployment, + workload_set=PRIMARY_WORKLOAD_SET, + workload_namespace=NAMESPACE, + owner_uid=OWNER_UID, + owner_name="pipeline", + owner_namespace="source", + logger=MagicMock(), + previous_generations={}, + ) + + +def test_reconcile_pod_set_rejects_base_name_over_limit() -> None: + deployment = _deployment("c" * (MAX_BASE_NAME_LENGTH + 1), 1) + with pytest.raises(kopf.PermanentError, match="name cannot exceed"): + _reconcile_pod_set( + core=MagicMock(), + deployment=deployment, + workload_set=PRIMARY_WORKLOAD_SET, + workload_namespace=NAMESPACE, + owner_uid=OWNER_UID, + owner_name="pipeline", + owner_namespace="source", + logger=MagicMock(), + previous_generations={}, + ) + + +def test_reconcile_replaces_outdated_pod_and_prunes_duplicates( + monkeypatch: pytest.MonkeyPatch, +) -> None: + outdated = _pod(0, 1, ready=True, phase="Running", annotations={SPEC_HASH_ANNOTATION: "old"}) + metadata, spec = _template() + desired_manifest = build_pipeline_pod( + base_name="consumer", + template_metadata=metadata, + template_spec=spec, + ordinal=0, + generation=0, + owner_uid=OWNER_UID, + owner_name="pipeline", + owner_namespace="source", + workload_set=PRIMARY_WORKLOAD_SET, + ) + duplicate_manifest = copy.deepcopy(desired_manifest) + duplicate_manifest["metadata"]["name"] = "consumer-0-2" + duplicate_manifest["metadata"]["labels"][GENERATION_LABEL] = "2" + duplicate = _existing_pod_from_manifest(duplicate_manifest, phase="Running", ready=True) + + applied, deleted, result, ledger = _reconcile(monkeypatch, [outdated, duplicate]) + + assert applied == [] + assert deleted == [("consumer-0-1", False)] + assert result["childPods"] == ["consumer-0-2"] + assert ledger == {0: 2} + + +def test_reconcile_prunes_scaled_down_and_unlabelled_pods( + monkeypatch: pytest.MonkeyPatch, +) -> None: + stale = _pod(2, 0) + unlabelled = _pod(0, 0) + assert unlabelled.metadata is not None + del unlabelled.metadata.labels[ORDINAL_LABEL] + + applied, deleted, result, _ledger = _reconcile(monkeypatch, [stale, unlabelled], replicas=0) + + assert applied == [] + assert deleted == [("consumer-2-0", False), ("consumer-0-0", False)] + assert result["desiredReplicas"] == 0 + + +def test_reconcile_keeps_primary_and_canary_sets_isolated( + monkeypatch: pytest.MonkeyPatch, +) -> None: + metadata, spec = _template() + primary = _existing_pod_from_manifest( + build_pipeline_pod( + base_name="consumer", + template_metadata=metadata, + template_spec=spec, + ordinal=0, + generation=1, + owner_uid=OWNER_UID, + owner_name="pipeline", + owner_namespace="source", + workload_set=PRIMARY_WORKLOAD_SET, + ), + phase="Running", + ready=True, + ) + canary = _existing_pod_from_manifest( + build_pipeline_pod( + base_name="consumer-canary", + template_metadata={**metadata, "labels": {"app": "consumer", "env": "canary"}}, + template_spec=spec, + ordinal=0, + generation=3, + owner_uid=OWNER_UID, + owner_name="pipeline", + owner_namespace="source", + workload_set=CANARY_WORKLOAD_SET, + ), + phase="Running", + ready=True, + ) + + applied, deleted, result, ledger = _reconcile( + monkeypatch, + [primary, canary], + workload_set=PRIMARY_WORKLOAD_SET, + ) + + assert applied == [] + assert deleted == [] + assert result["childPods"] == ["consumer-0-1"] + assert ledger == {0: 1} + + +def test_delete_obsolete_pod_sets_removes_canary_when_disabled( + monkeypatch: pytest.MonkeyPatch, +) -> None: + primary = _pod(0, 1, ready=True, phase="Running") + canary = _pod(0, 2, ready=True, phase="Running") + assert canary.metadata is not None + canary.metadata.name = "consumer-canary-0-2" + canary.metadata.labels[WORKLOAD_SET_LABEL] = CANARY_WORKLOAD_SET + missing_set = _pod(1, 0, ready=True, phase="Running") + assert missing_set.metadata is not None + del missing_set.metadata.labels[WORKLOAD_SET_LABEL] + deleted: list[str] = [] + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.list_owned_pods", + lambda *_args, **_kwargs: [primary, canary, missing_set], + ) + monkeypatch.setattr( + "sentry_streams_k8s.operator.reconcile.delete_pod", + lambda _core, name, _namespace, force=False: deleted.append(name), + ) + + delete_obsolete_pod_sets( + MagicMock(), + NAMESPACE, + OWNER_UID, + {PRIMARY_WORKLOAD_SET}, + MagicMock(), + ) + + assert deleted == ["consumer-canary-0-2", "consumer-1-0"] + + +def test_delete_owned_pods_skips_pods_already_terminating(monkeypatch: pytest.MonkeyPatch) -> None: + pods = [_pod(0, 0), _pod(1, 0, deletion_timestamp=datetime(2026, 7, 16, tzinfo=timezone.utc))] + deleted: list[str] = [] + monkeypatch.setattr( + "sentry_streams_k8s.operator.pod_resources.list_owned_pods", lambda *_: pods + ) + monkeypatch.setattr( + "sentry_streams_k8s.operator.pod_resources.delete_pod", + lambda _core, name, _namespace: deleted.append(name), + ) + + delete_owned_pods(MagicMock(), NAMESPACE, OWNER_UID, MagicMock()) + + assert deleted == ["consumer-0-0"] + + +@pytest.mark.parametrize( + ("data", "expected"), + [ + ({"0": 3, "2": 5}, {0: 3, 2: 5}), + (None, {}), + ({}, {}), + ], +) +def test_parse_generations_converts_string_ordinals_and_handles_no_ledger_yet( + data: object, expected: dict[int, int] +) -> None: + assert _parse_generations(data) == expected + + +def test_serialize_generations_sorts_ordinals_and_stringifies_keys() -> None: + assert _serialize_generations({2: 4, 0: 3}) == {"0": 3, "2": 4} diff --git a/sentry_streams_k8s/tests/test_status.py b/sentry_streams_k8s/tests/test_status.py new file mode 100644 index 00000000..c0109801 --- /dev/null +++ b/sentry_streams_k8s/tests/test_status.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +from kubernetes.client import V1ObjectMeta, V1Pod + +from sentry_streams_k8s.operator.constants import ORDINAL_LABEL, SPEC_HASH_ANNOTATION +from sentry_streams_k8s.operator.pod_health import ( + PodHealth, + pod_health, + pod_spec_changed, +) +from sentry_streams_k8s.operator.pod_status import reported_pod_status +from sentry_streams_k8s.operator.pod_types import PodManifest +from tests.k8s_fixtures import make_pod + + +def test_reported_pod_status() -> None: + pod = make_pod( + labels={ORDINAL_LABEL: "0"}, + annotations={SPEC_HASH_ANNOTATION: "old"}, + phase="Pending", + ) + desired: PodManifest = { + "apiVersion": "v1", + "kind": "Pod", + "metadata": { + "name": "consumer-0-0", + "annotations": {SPEC_HASH_ANNOTATION: "new"}, + }, + "spec": {}, + } + health = PodHealth(name="consumer-0-0", ready=False, reason="ImagePullBackOff", unhealthy=True) + + assert pod_spec_changed(pod, desired) is True + assert reported_pod_status(pod, health).to_status_dict() == { + "name": "consumer-0-0", + "ready": False, + "phase": "Pending", + "ordinal": "0", + "reason": "ImagePullBackOff", + } + + +def test_terminating_pod_is_not_reported_as_unhealthy() -> None: + now = datetime(2026, 7, 15, tzinfo=timezone.utc) + pod = make_pod( + deletion_timestamp=now, + labels={ORDINAL_LABEL: "0"}, + phase="Running", + ) + + status = reported_pod_status(pod, pod_health(pod, now)) + + assert status.reason == "Terminating" + assert status.is_unhealthy is False + + +def test_reported_pod_status_handles_pod_without_status() -> None: + pod = V1Pod(metadata=V1ObjectMeta(name="consumer-0-0")) + + status = reported_pod_status(pod, PodHealth(name="consumer-0-0")) + + assert status.phase == "Unknown"