spark_on_k8s.airflow.postmortem

Implementation reference. These symbols describe how the project works; their presence here does not make them supported application imports. Use the documented entry points and support policy.

Bounded failure evidence written to the Airflow task log before cleanup can delete the driver.

describe_error(error)

Name a failure without its response body or headers, which can be large or sensitive.

Source code in spark_on_k8s/airflow/postmortem.py

def describe_error(error: BaseException) -> str:
    """Name a failure without its response body or headers, which can be large or sensitive."""
    status = getattr(error, "status", None)
    return f"{type(error).__name__} {status}" if isinstance(status, int) else type(error).__name__

log_failure_postmortem(log, k8s_client_manager, *, namespace, pod_name, status, diagnose, log_tail_lines, timeout=POSTMORTEM_TIMEOUT)

Log a bounded diagnosis and the driver's last log lines, before the driver is deleted.

Diagnosis and log-tail failures are logged separately so that one cannot hide the other. Failing to create a client or read the driver pod raises; the caller decides how to report it.

Returns:

Type	Description
dict[str, Any] | None	A compact, JSON-serializable diagnosis summary, or None when no diagnosis was collected.

Source code in spark_on_k8s/airflow/postmortem.py

def log_failure_postmortem(
    log: logging.Logger,
    k8s_client_manager: KubernetesClientManager,
    *,
    namespace: str,
    pod_name: str,
    status: str,
    diagnose: bool,
    log_tail_lines: int,
    timeout: float = POSTMORTEM_TIMEOUT,
) -> dict[str, Any] | None:
    """Log a bounded diagnosis and the driver's last log lines, before the driver is deleted.

    Diagnosis and log-tail failures are logged separately so that one cannot hide the other.
    Failing to create a client or read the driver pod raises; the caller decides how to report it.

    Returns:
        A compact, JSON-serializable diagnosis summary, or None when no diagnosis was collected.
    """
    deadline = WaitDeadline(timeout)
    with k8s_client_manager.client() as client:
        api = k8s.CoreV1Api(client)
        driver = api.read_namespaced_pod(namespace=namespace, name=pod_name, **deadline.request_options())
        log.warning(
            "Spark failure postmortem for %s/%s, which %s:",
            namespace,
            pod_name,
            _OUTCOMES.get(status, f"finished with status {status}"),
        )
        summary = None
        if diagnose:
            report = _log_diagnosis(log, api, driver, namespace, deadline)
            summary = _summary(report, status) if report is not None else None
        if log_tail_lines:
            _log_driver_tail(log, api, driver, namespace, log_tail_lines, deadline)
    return summary