如何用Python获取Kubernetes Pod的可靠状态？

方法1：解析kubectl的结构化输出 #

kubectl的状态列文本是客户端对API返回数据的汇总结果，你可以直接调用kubectl命令并解析其JSON/JSONPath输出，跳过自己实现复杂逻辑的步骤：

# 获取指定命名空间Pod的状态文本
kubectl get pods -n default -o jsonpath='{range .items[*]}{.metadata.name}: {.status.phase}{"\n"}{end}'

在Python中可以通过subprocess模块调用该命令并解析结果：

import subprocess
import json

result = subprocess.run(
    ["kubectl", "get", "pods", "-n", "default", "-o", "json"],
    capture_output=True,
    text=True
)
pods = json.loads(result.stdout)
for pod in pods["items"]:
    pod_name = pod["metadata"]["name"]
    # 先检查Init容器状态
    init_status = None
    if "initContainerStatuses" in pod["status"]:
        for init_container in pod["status"]["initContainerStatuses"]:
            if "waiting" in init_container["state"]:
                init_status = f"Init:{init_container['state']['waiting']['reason']}"
                break
            if "terminated" in init_container["state"] and init_container["state"]["terminated"]["exitCode"] != 0:
                init_status = "Init:Error"
                break
    if init_status:
        print(f"{pod_name}: {init_status}")
        continue
    # 检查主容器状态
    container_status = None
    if "containerStatuses" in pod["status"]:
        for container in pod["status"]["containerStatuses"]:
            if "waiting" in container["state"]:
                container_status = container["state"]["waiting"]["reason"]
                break
            if "terminated" in container["state"] and container["state"]["terminated"]["exitCode"] != 0:
                container_status = "Error"
                break
    final_status = container_status if container_status else pod["status"]["phase"]
    print(f"{pod_name}: {final_status}")

方法2：复刻kubectl的状态计算逻辑 #

kubectl的状态汇总逻辑完全开源，你可以参考client-go中的PodStatus()函数（位于pod_utils.go），将其逻辑移植到Python中。核心判断流程如下：

• 优先处理Init容器：若存在未完成的Init容器，根据其状态返回Init:<Reason>（如Init:Error、Init:CrashLoopBackOff）
• 处理主容器异常：若主容器处于等待/终止且异常退出状态，返回对应的原因（如CrashLoopBackOff、ImagePullBackOff）
• ** fallback到Pod Phase**：若所有容器正常，返回Running/Succeeded/Failed等Phase值

方法3：优化Python Kubernetes客户端的状态提取 #

基于你现有的代码，扩展状态判断逻辑，覆盖更多场景：

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()

def get_pod_status(pod):
    # 检查Init容器状态
    if pod.status.init_container_statuses:
        for init_container in pod.status.init_container_statuses:
            if init_container.state.waiting:
                return f"Init:{init_container.state.waiting.reason}"
            if init_container.state.terminated and init_container.state.terminated.exit_code != 0:
                return "Init:Error"
    # 检查主容器状态
    if pod.status.container_statuses:
        for container in pod.status.container_statuses:
            # 处理等待状态
            if container.state.waiting:
                return container.state.waiting.reason
            # 处理异常终止状态
            if container.state.terminated and container.state.terminated.exit_code != 0:
                return "Error"
            # 处理重启循环（CrashLoopBackOff）
            if container.restart_count > 0 and container.state.running:
                if container.last_state and container.last_state.terminated:
                    return "CrashLoopBackOff"
    # 无异常则返回Phase
    return pod.status.phase

# 调用示例
pods = v1.list_namespaced_pod(namespace="default").items
for pod in pods:
    print(f"{pod.metadata.name}: {get_pod_status(pod)}")

注意：上述代码是kubectl逻辑的近似实现，若要完全对齐，需参考client-go的完整判断逻辑，比如处理Pod Conditions、镜像拉取失败的重试状态等场景。

内容的提问来源于stack exchange，提问作者HackerBaloo