# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-kubernetespodevictionerror-the-preemption-grace-period-and-the-missing-disruption-budget-signal.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-kubernetespodevictionerror-the-preemption-grace-period-and-the-missing-disruption-budget-signal.warrant.json
# Markdown SHA-256:   daba35162685a810fb889133c57185856da0af6bc7f55d01b154ba2e221205e8
# Code AST SHA-256:   8f1fc2a35d838e839097b1b0a97c93afa52a76fe1ad43f4d93a69970b395248e
# Verified Classes:   EvictionIndeterminateError, TerminationCause, TerminationSignal
# Defensive Guards:   2 fail-closed invariants (zero external deps)

"""
k8s_eviction_detector.py

Detects involuntary pod eviction (kubelet node-pressure) vs. voluntary
disruption, using the only signals available inside a terminating container:
the Events API and node conditions, both read within the grace period.

The invariant: an instance cannot trust its own exit path. It must read
cluster state out-of-band, and if it cannot, it must assume eviction.
"""

from __future__ import annotations

import datetime as _dt
from dataclasses import dataclass
from enum import Enum


class EvictionIndeterminateError(RuntimeError):
    """Raised when the termination cause cannot be determined in time."""


class TerminationCause(Enum):
    VOLUNTARY = "voluntary"          # drain, rollout, Eviction API
    INVOLUNTARY_EVICTION = "evicted" # kubelet node-pressure
    OOM_KILLED = "oom_killed"
    INDETERMINATE = "indeterminate"


@dataclass(frozen=True)
class TerminationSignal:
    pod_uid: str
    node_conditions: dict[str, str]   # e.g. {"MemoryPressure": "True"}
    recent_evicted_event_uids: frozenset[str]
    event_retention_seconds: int = 3600


def classify_termination(
    sig: TerminationSignal,
    now: _dt.datetime,
    grace_period_seconds: int,
) -> TerminationCause:
    """
    Classify why this pod is terminating.

    Decision order:
      1. If an 'Evicted' event references this pod UID -> involuntary.
      2. Else if a pressure condition is True on the node -> involuntary
         (kubelet is about to evict; treat as eviction even if the event
         has not been observed yet, because event propagation lags).
      3. Else if the pod UID is absent from events and no pressure ->
         voluntary, PROVIDED we are within event retention. If we are
         past retention we cannot know, and must not claim voluntary.
    """
    if sig.pod_uid in sig.recent_evicted_event_uids:
        return TerminationCause.INVOLUNTARY_EVICTION

    pressure = any(
        sig.node_conditions.get(cond) == "True"
        for cond in ("MemoryPressure", "DiskPressure", "PIDPressure")
    )
    if pressure:
        # Conservative: node is under pressure, assume eviction.
        return TerminationCause.INVOLUNTARY_EVICTION

    # No event, no pressure. This is *probably* voluntary, but only if
    # the event window has not expired — otherwise the absence of an
    # event is uninformative.
    if grace_period_seconds > sig.event_retention_seconds:
        raise EvictionIndeterminateError(
            f"pod {sig.pod_uid}: grace period {grace_period_seconds}s exceeds "
            f"event retention {sig.event_retention_seconds}s; cannot "
            "distinguish voluntary from evicted. Assume eviction."
        )

    return TerminationCause.VOLUNTARY


def on_sigterm(sig: TerminationSignal, grace_period_seconds: int) -> str:
    """
    The handler an instance must run on SIGTERM. Returns the action to take.
    """
    try:
        cause = classify_termination(sig, _dt.datetime.utcnow(), grace_period_seconds)
    except EvictionIndeterminateError:
        # Fail safe: shed load, do not drain.
        return "shed_load_and_exit_fast"

    if cause is TerminationCause.INVOLUNTARY_EVICTION:
        return "shed_load_and_exit_fast"
    if cause is TerminationCause.OOM_KILLED:
        # Never reached via SIGTERM; OOM-kill bypasses handlers entirely.
        raise EvictionIndeterminateError(
            "OOM-kill cannot deliver SIGTERM; classification path is unreachable"
        )
    return "drain_gracefully"