主题
Python 连接 Kubernetes
概述
使用 Python 的 kubernetes 客户端库来管理和操作 Kubernetes 集群，实现自动化部署、监控和运维。
安装
bash
pip install kubernetes
# 或使用 uv
uv add kubernetes认证配置
方式一：使用 kubeconfig 文件
最常用的方式，读取 ~/.kube/config 中的集群配置：
python
from kubernetes import client, config
# 加载 kubeconfig（默认路径 ~/.kube/config）
config.load_kube_config()
# 指定 kubeconfig 文件路径
config.load_kube_config(config_file='/path/to/kubeconfig')
# 指定 context
config.load_kube_config(context='my-context')方式二：集群内运行（In-Cluster）
当 Python 程序运行在 K8s Pod 内部时使用：
python
from kubernetes import client, config
# 自动使用 Pod 的 ServiceAccount
config.load_incluster_config()方式三：直接配置
手动指定 API Server 地址和 Token：
python
from kubernetes import client
configuration = client.Configuration()
configuration.host = "https://k8s-api-server:6443"
configuration.api_key = {"authorization": "Bearer <token>"}
configuration.verify_ssl = False # 仅用于测试
client.Configuration.set_default(configuration)常用操作
核心 API（Core V1）
python
from kubernetes import client, config
config.load_kube_config()
v1 = client.CoreV1Api()列出所有 Namespace
python
namespaces = v1.list_namespace()
for ns in namespaces.items:
print(ns.metadata.name)列出 Pod
python
# 列出所有命名空间的 Pod
pods = v1.list_pod_for_all_namespaces()
for pod in pods.items:
print(f"{pod.metadata.namespace}/{pod.metadata.name} - {pod.status.phase}")
# 列出指定命名空间的 Pod
pods = v1.list_namespaced_pod(namespace="default")
for pod in pods.items:
print(f"{pod.metadata.name} - {pod.status.phase}")获取 Pod 日志
python
logs = v1.read_namespaced_pod_log(
name="my-pod",
namespace="default"
)
print(logs)
# 获取指定容器的日志
logs = v1.read_namespaced_pod_log(
name="my-pod",
namespace="default",
container="sidecar",
tail_lines=100
)列出 Service
python
services = v1.list_service_for_all_namespaces()
for svc in services.items:
print(f"{svc.metadata.namespace}/{svc.metadata.name} - {svc.spec.type}")列出 ConfigMap 和 Secret
python
# ConfigMap
configmaps = v1.list_namespaced_config_map(namespace="default")
for cm in configmaps.items:
print(cm.metadata.name)
# Secret
secrets = v1.list_namespaced_secret(namespace="default")
for secret in secrets.items:
print(secret.metadata.name)Apps API（Apps V1）
python
apps_v1 = client.AppsV1Api()列出 Deployment
python
deployments = apps_v1.list_deployment_for_all_namespaces()
for deploy in deployments.items:
ready = deploy.status.ready_replicas or 0
total = deploy.spec.replicas
print(f"{deploy.metadata.namespace}/{deploy.metadata.name} - {ready}/{total}")列出 StatefulSet
python
statefulsets = apps_v1.list_stateful_set_for_all_namespaces()
for sts in statefulsets.items:
print(f"{sts.metadata.namespace}/{sts.metadata.name}")列出 DaemonSet
python
daemonsets = apps_v1.list_daemon_set_for_all_namespaces()
for ds in daemonsets.items:
print(f"{ds.metadata.namespace}/{ds.metadata.name}")创建资源
创建 Namespace
python
from kubernetes import client, config
config.load_kube_config()
v1 = client.CoreV1Api()
namespace = client.V1Namespace(
metadata=client.V1ObjectMeta(name="my-namespace")
)
v1.create_namespace(body=namespace)创建 Deployment
python
from kubernetes import client, config
config.load_kube_config()
apps_v1 = client.AppsV1Api()
deployment = client.V1Deployment(
api_version="apps/v1",
kind="Deployment",
metadata=client.V1ObjectMeta(name="nginx-deployment"),
spec=client.V1DeploymentSpec(
replicas=3,
selector=client.V1LabelSelector(
match_labels={"app": "nginx"}
),
template=client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(
labels={"app": "nginx"}
),
spec=client.V1PodSpec(
containers=[
client.V1Container(
name="nginx",
image="nginx:1.25",
ports=[client.V1ContainerPort(container_port=80)]
)
]
)
)
)
)
apps_v1.create_namespaced_deployment(
namespace="default",
body=deployment
)从 YAML 文件创建资源
更实用的方式，直接从 YAML 创建任意资源：
python
import yaml
from kubernetes import client, config, utils
config.load_kube_config()
with open("deployment.yaml") as f:
dep = yaml.safe_load(f)
# 使用 utils 创建
k8s_client = client.ApiClient()
utils.create_from_yaml(k8s_client, "deployment.yaml")删除资源
python
# 删除 Deployment
apps_v1.delete_namespaced_deployment(
name="nginx-deployment",
namespace="default"
)
# 删除 Namespace
v1.delete_namespace(name="my-namespace")更新资源
python
# 更新 Deployment 副本数
deployment = apps_v1.read_namespaced_deployment(
name="nginx-deployment",
namespace="default"
)
deployment.spec.replicas = 5
apps_v1.patch_namespaced_deployment(
name="nginx-deployment",
namespace="default",
body=deployment
)Watch 机制
实时监控资源变化：
python
from kubernetes import client, config, watch
config.load_kube_config()
v1 = client.CoreV1Api()
w = watch.Watch()
for event in w.stream(v1.list_namespaced_pod, namespace="default"):
pod = event['object']
event_type = event['type']
print(f"{event_type}: {pod.metadata.name} - {pod.status.phase}")动态客户端
当需要操作 CRD（自定义资源）等非内置资源时：
python
from kubernetes import client, config
from kubernetes.dynamic import DynamicClient
config.load_kube_config()
k8s_client = client.ApiClient()
dyn_client = DynamicClient(k8s_client)
# 操作 CRD 资源
crd_api = dyn_client.resources.get(
api_version="argoproj.io/v1alpha1",
kind="Application"
)
# 列出 ArgoCD Application
apps = crd_api.get(namespace="argocd")
for item in apps.items:
print(item.metadata.name)异步客户端
高并发场景下使用异步客户端：
python
import asyncio
from kubernetes import client, config
from kubernetes.asyncio import client as async_client
async def main():
await config.load_kube_config()
v1 = async_client.CoreV1Api()
pods = await v1.list_pod_for_all_namespaces()
for pod in pods.items:
print(pod.metadata.name)
await v1.api_client.close()
asyncio.run(main())实用脚本示例
集群资源概览
python
from kubernetes import client, config
config.load_kube_config()
v1 = client.CoreV1Api()
apps_v1 = client.AppsV1Api()
print("=== 集群概览 ===")
# Node 状态
nodes = v1.list_node()
print(f"Nodes: {len(nodes.items)}")
# Namespace 数量
namespaces = v1.list_namespace()
print(f"Namespaces: {len(namespaces.items)}")
# Pod 统计
pods = v1.list_pod_for_all_namespaces()
running = sum(1 for p in pods.items if p.status.phase == "Running")
print(f"Pods: {len(pods.items)} (Running: {running})")
# Deployment 统计
deployments = apps_v1.list_deployment_for_all_namespaces()
print(f"Deployments: {len(deployments.items)}")批量清理 Completed Pod
python
from kubernetes import client, config
config.load_kube_config()
v1 = client.CoreV1Api()
pods = v1.list_pod_for_all_namespaces()
deleted = 0
for pod in pods.items:
if pod.status.phase in ("Succeeded", "Failed"):
v1.delete_namespaced_pod(
name=pod.metadata.name,
namespace=pod.metadata.namespace
)
deleted += 1
print(f"Deleted: {pod.metadata.namespace}/{pod.metadata.name}")
print(f"\nTotal deleted: {deleted}")