Python操作K8s API:运维常用实战与性能稳定性对比

一、Python操作K8s API基础准备

1. 核心依赖库

Python操作K8s API的主流库是官方维护的kubernetes（原名kubernetes-client），需先安装：

pip install kubernetes

2. 认证方式（两种常用场景）

场景1：在K8s集群内运行（Pod中）

from kubernetes import config, client

# 加载集群内服务账户配置（自动读取/var/run/secrets/kubernetes.io/serviceaccount）
config.load_incluster_config()
v1 = client.CoreV1Api()

场景2：在集群外运行（本地/运维机）

from kubernetes import config, client

# 加载kubeconfig文件（默认~/.kube/config）
config.load_kube_config(config_file="/path/to/kubeconfig")
v1 = client.CoreV1Api()

二、Python操作K8s API运维常用实战案例

以下案例均基于上述基础配置，聚焦运维高频操作，代码简洁可直接复用。

1. 节点管理：查询集群所有节点状态

defget_all_nodes():
"""查询所有节点及状态"""
    v1 = client.CoreV1Api()
    nodes = v1.list_node()
for node in nodes.items:
        node_name = node.metadata.name
# 提取节点就绪状态
for condition in node.status.conditions:
if condition.type == "Ready":
                node_status = condition.status
break
print(f"节点名称：{node_name} | 就绪状态：{node_status}")

if __name__ == "__main__":
    get_all_nodes()

2. Pod管理：查询指定命名空间Pod列表

defget_pods_in_namespace(namespace="default"):
"""查询指定命名空间的Pod列表"""
    v1 = client.CoreV1Api()
    pods = v1.list_namespaced_pod(namespace=namespace)
for pod in pods.items:
        pod_name = pod.metadata.name
        pod_status = pod.status.phase
        pod_ip = pod.status.pod_ip or"未分配"
print(f"Pod名称：{pod_name} | 状态：{pod_status} | IP：{pod_ip}")

if __name__ == "__main__":
    get_pods_in_namespace(namespace="kube-system")

3. Pod管理：强制删除异常Pod

defdelete_pod_force(pod_name, namespace="default"):
"""强制删除指定Pod"""
    v1 = client.CoreV1Api()
try:
# grace_period_seconds=0 强制删除（跳过优雅终止）
        v1.delete_namespaced_pod(
            name=pod_name,
            namespace=namespace,
            body=client.V1DeleteOptions(grace_period_seconds=0),
            propagation_policy="Foreground"
        )
print(f"Pod {pod_name}强制删除成功")
except client.ApiException as e:
print(f"删除Pod失败：{e.reason}（状态码：{e.status}）")

if __name__ == "__main__":
    delete_pod_force(pod_name="error-pod", namespace="default")

4. 配置管理：创建ConfigMap

defcreate_configmap(cm_name, namespace="default", data=None):
"""创建ConfigMap"""
if data isNone:
        data = {"app.conf": "env=prod\nport=8080"}

    v1 = client.CoreV1Api()
    cm = client.V1ConfigMap(
        api_version="v1",
        kind="ConfigMap",
        metadata=client.V1ObjectMeta(name=cm_name),
        data=data
    )
try:
        v1.create_namespaced_config_map(namespace=namespace, body=cm)
print(f"ConfigMap {cm_name}创建成功")
except client.ApiException as e:
print(f"创建ConfigMap失败：{e.reason}")

if __name__ == "__main__":
    create_configmap(cm_name="app-config", namespace="default")

5. 资源监控：查询Pod资源使用量（需Metrics Server）

defget_pod_metrics(pod_name, namespace="default"):
"""查询Pod的CPU/内存使用量"""
    metrics_v1 = client.CustomObjectsApi()
try:
# 调用metrics.k8s.io API
        metrics = metrics_v1.get_namespaced_custom_object(
            group="metrics.k8s.io",
            version="v1beta1",
            namespace=namespace,
            plural="pods",
            name=pod_name
        )
# 提取容器资源使用
for container in metrics["containers"]:
print(f"容器{container['name']}：")
print(f"  CPU使用：{container['usage']['cpu']}")
print(f"  内存使用：{container['usage']['memory']}")
except client.ApiException as e:
print(f"查询Pod监控失败：{e.reason}（需确保Metrics Server已部署）")

if __name__ == "__main__":
    get_pod_metrics(pod_name="nginx-78f5d6989f-9x7z8", namespace="default")

6. 部署管理：更新Deployment副本数

defscale_deployment(deploy_name, replicas, namespace="default"):
"""调整Deployment副本数"""
    apps_v1 = client.AppsV1Api()
# 构建副本数更新配置
    scale_body = {
"spec": {
"replicas": replicas
        }
    }
try:
        apps_v1.patch_namespaced_deployment_scale(
            name=deploy_name,
            namespace=namespace,
            body=scale_body
        )
print(f"Deployment {deploy_name}副本数已调整为{replicas}")
except client.ApiException as e:
print(f"调整副本数失败：{e.reason}")

if __name__ == "__main__":
    scale_deployment(deploy_name="nginx-deploy", replicas=3, namespace="default")

三、Python API vs kubectl：性能与稳定性对比

1. 底层逻辑说明

kubectl本质是命令行封装的K8s API调用工具，其所有操作最终都会转化为对K8s API Server的HTTP/HTTPS请求；而Python kubernetes库是直接调用K8s API的SDK，与kubectl共享同一套底层API，核心差异在于“调用层”和“封装方式”。

2. 性能对比（实测数据）

核心结论：

• Python API性能优于kubectl，尤其是批量操作场景，优势可达50%以上；
• 单次简单操作（如查单个Pod）差异较小（<50ms），几乎可忽略。

3. 稳定性对比

4. 适用场景选择

四、总结

1. Python操作K8s API基于kubernetes库，核心是加载认证配置后调用对应API接口，适配所有运维高频场景（节点/Pod/配置/部署管理等）；
2. 性能上，Python API因减少CLI解析/格式化开销，整体优于kubectl，批量操作优势更明显；
3. 稳定性上，Python API支持结构化异常处理、自定义重试策略，更适配自动化运维场景；
4. 场景选择上，手动临时操作选kubectl，自动化/平台化需求选Python API。

建议

• 日常手动运维：优先使用kubectl，简洁高效；
• 自动化脚本/运维平台：优先使用Python API，兼顾性能与可扩展性；
• 批量操作场景：通过Python API实现并行调用，进一步提升效率；
• 稳定性保障：Python代码中增加超时控制、重试逻辑，降低网络波动影响。