Skip to content

spark_on_k8s.utils.spark_app_status

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.

SparkAppDeletedError

Bases: LookupError

The driver disappeared before a final status was observed, so the outcome is unknown.

Source code in spark_on_k8s/utils/spark_app_status.py 
21
22
class SparkAppDeletedError(LookupError):
    """The driver disappeared before a final status was observed, so the outcome is unknown."""

SparkAppStatus

Bases: StrEnum

Spark app status.

Source code in spark_on_k8s/utils/spark_app_status.py 
11
12
13
14
15
16
17
18
class SparkAppStatus(StrEnum):
    """Spark app status."""

    Pending = "Pending"
    Running = "Running"
    Succeeded = "Succeeded"
    Failed = "Failed"
    Unknown = "Unknown"

driver_container_name(pod)

Name the submitted driver container, even if an admission webhook prepended a sidecar.

Source code in spark_on_k8s/utils/spark_app_status.py 
63
64
65
66
67
68
69
70
71
def driver_container_name(pod: k8s.V1Pod | k8s_async.V1Pod) -> str:
    """Name the submitted driver container, even if an admission webhook prepended a sidecar."""
    names = [container.name for container in pod.spec.containers]
    metadata = getattr(pod, "metadata", None)
    selected = (getattr(metadata, "annotations", None) or {}).get("kubectl.kubernetes.io/default-container")
    for candidate in (selected, "driver"):
        if candidate in names:
            return candidate
    return names[0]

driver_pod_selector(app_id)

Select only driver pods: other pods may carry an application's ID label.

Source code in spark_on_k8s/utils/spark_app_status.py 
49
50
51
def driver_pod_selector(app_id: str) -> str:
    """Select only driver pods: other pods may carry an application's ID label."""
    return f"spark-app-id={app_id},spark-role=driver"

get_app_status(pod)

Get app status.

Source code in spark_on_k8s/utils/spark_app_status.py 
35
36
37
38
39
40
41
42
43
44
45
46
def get_app_status(pod: k8s.V1Pod | k8s_async.V1Pod) -> SparkAppStatus:
    """Get app status."""
    if pod.status.phase == "Pending":
        return SparkAppStatus.Pending
    elif pod.status.phase == "Running":
        return SparkAppStatus.Running
    elif pod.status.phase == "Succeeded":
        return SparkAppStatus.Succeeded
    elif pod.status.phase == "Failed":
        return SparkAppStatus.Failed
    else:
        return SparkAppStatus.Unknown

is_transient_status_code(status)

Whether a Kubernetes API response status may clear on its own.

Source code in spark_on_k8s/utils/spark_app_status.py 
30
31
32
def is_transient_status_code(status: int | None) -> bool:
    """Whether a Kubernetes API response status may clear on its own."""
    return status == 429 or (status is not None and 500 <= status <= 599)

select_driver_pod(pods, app_id)

Return the only driver for an application ID instead of acting on an arbitrary match.

Source code in spark_on_k8s/utils/spark_app_status.py 
54
55
56
57
58
59
60
def select_driver_pod(pods: list, app_id: str):
    """Return the only driver for an application ID instead of acting on an arbitrary match."""
    if len(pods) == 0:
        raise ValueError(f"No pods found for app {app_id}")
    if len(pods) > 1:
        raise ValueError(f"Multiple driver pods found for app {app_id}; use a run receipt")
    return pods[0]