Kubernetes API 客户端

运维除 kubectl 外，常用 Python kubernetes 官方客户端直接调用 K8s API：批量查 Pod 状态、重启 Deployment、读取 ConfigMap、做自定义巡检。客户端会自动读取 ~/.kube/config 或 In-Cluster 配置。

---

环境准备

uv add kubernetes

集群外运行需有效 kubeconfig；Pod 内运行自动使用 ServiceAccount：

export KUBECONFIG="${HOME}/.kube/config"  # 默认即此路径

---

初始化客户端

#!/usr/bin/env python3
"""列出 default 命名空间 Pod，输出名称与 Phase。"""

from kubernetes import client, config


def get_core_v1() -> client.CoreV1Api:
    try:
        # Pod 内执行时使用集群内凭证
        config.load_incluster_config()
    except config.ConfigException:
        # 本地或跳板机使用 kubeconfig
        config.load_kube_config()
    return client.CoreV1Api()


def list_pods(namespace: str = "default") -> None:
    v1 = get_core_v1()
    pods = v1.list_namespaced_pod(namespace=namespace)
    for pod in pods.items:
        phase = pod.status.phase
        ready = sum(1 for c in (pod.status.container_statuses or []) if c.ready)
        total = len(pod.spec.containers)
        print(f"{pod.metadata.name:40s}  {phase:10s}  ready={ready}/{total}")


if __name__ == "__main__":
    list_pods("production")

---

查询 Deployment 与滚动重启

from kubernetes import client, config


def get_apps_v1() -> client.AppsV1Api:
    config.load_kube_config()
    return client.AppsV1Api()


def list_deployments(namespace: str) -> None:
    apps = get_apps_v1()
    deps = apps.list_namespaced_deployment(namespace=namespace)
    for d in deps.items:
        spec = d.spec.replicas
        ready = d.status.ready_replicas or 0
        print(f"{d.metadata.name:30s}  replicas={ready}/{spec}")


def restart_deployment(name: str, namespace: str) -> None:
    """通过 patch annotation 触发滚动重启（等效 kubectl rollout restart）。"""
    apps = get_apps_v1()
    now = client.V1Deployment(
        metadata=client.V1ObjectMeta(
            annotations={"kubectl.kubernetes.io/restartedAt": _now_iso()},
        )
    )
    apps.patch_namespaced_deployment(name=name, namespace=namespace, body=now)
    print(f"已触发 {namespace}/{name} 滚动重启")


def _now_iso() -> str:
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).isoformat()

---

读取 ConfigMap

def read_configmap(name: str, namespace: str) -> dict[str, str]:
    v1 = get_core_v1()
    cm = v1.read_namespaced_config_map(name=name, namespace=namespace)
    return dict(cm.data or {})

发布窗口可用 watch.Watch().stream() 监听 Pod 事件，适合实时输出变更。

---

错误处理与 RBAC

from kubernetes.client.exceptions import ApiException


def safe_read_pod(name: str, namespace: str) -> None:
    v1 = get_core_v1()
    try:
        pod = v1.read_namespaced_pod(name=name, namespace=namespace)
        print(pod.status.phase)
    except ApiException as e:
        if e.status == 404:
            print(f"Pod {name} 不存在")
        elif e.status == 403:
            print("权限不足，检查 RBAC RoleBinding")
        else:
            raise

注意

• 生产写操作（delete、patch）务必限定 namespace 与 ResourceName
• 为自动化创建专用 ServiceAccount + Role，避免 cluster-admin
• 大规模 list 注意分页；list_* 默认有上限，可用 limit + continue

---

小结

场景	客户端
Pod / CM / Secret	CoreV1Api
Deployment / StatefulSet	AppsV1Api
监听资源	watch.Watch().stream()
配置加载	load_kube_config() / load_incluster_config()

第五阶段完结。下一模块进入 并发编程，用线程池与异步加速批量巡检。