K8s Python客户端如何实现等价于kubectl run的命令效果

该方案核心逻辑和kubectl插件的实现逻辑完全一致：绕过K8s原生exec不支持指定运行用户的限制，在目标Pod所在节点启动挂载了容器运行时socket的特权临时Pod，直接调用容器运行时接口指定用户进入目标容器，操作完成后自动清理临时资源。

前置要求 #

• 操作所用的集群身份需要拥有Pod查询、创建、删除、exec权限，以及节点查询权限
• 集群允许创建特权Pod、允许挂载宿主机/var/run路径下的容器运行时socket文件
• 本地Python环境已安装官方kubernetes SDK，版本与集群版本偏差不超过1个次版本

具体实现代码 #

代码可直接复用，替换对应参数即可：
首先加载SDK配置，查询目标Pod的所在节点与容器ID：

from kubernetes import client, config
from kubernetes.stream import stream
import time

# 集群内运行替换为config.load_incluster_config()
config.load_kube_config()
core_v1 = client.CoreV1Api()

# 替换为实际目标参数
TARGET_POD = "替换为你的目标Pod名"
TARGET_NS = "替换为目标Pod所在命名空间"
TARGET_CONTAINER = "替换为目标容器名，单容器Pod可省略该参数匹配逻辑"
EXEC_USER = "1000:1000"  # 替换为你要使用的非root用户，支持用户名或UID:GID格式
EXEC_CMD = ["/bin/bash"]  # 替换为你要在目标容器内执行的命令

# 获取目标Pod调度节点与容器运行时ID
pod_meta = core_v1.read_namespaced_pod(name=TARGET_POD, namespace=TARGET_NS)
target_node = pod_meta.spec.node_name
container_runtime_id = ""
for c_status in pod_meta.status.container_statuses:
    if c_status.name == TARGET_CONTAINER:
        # 截掉runtime前缀，比如docker://、containerd://
        container_runtime_id = c_status.container_id.split("://")[-1]
        break
if not container_runtime_id:
    raise RuntimeError("查询目标容器ID失败")

然后构造临时特权Pod的配置，绑定调度到目标节点：

# 临时Pod唯一标识，带时间戳避免重名
temp_pod_name = f"temp-exec-{int(time.time())}"
# 以下配置以Docker运行时为例，containerd运行时需要替换镜像、挂载路径、执行命令
temp_pod_spec = client.V1Pod(
    metadata=client.V1ObjectMeta(name=temp_pod_name, namespace=TARGET_NS),
    spec=client.V1PodSpec(
        node_name=target_node,
        restart_policy="Never",
        host_pid=True,
        containers=[
            client.V1Container(
                name="exec-helper",
                image="docker:24.0.6",
                command=["sleep", "3600"],
                security_context=client.V1SecurityContext(privileged=True),
                volume_mounts=[
                    client.V1VolumeMount(
                        name="runtime-sock",
                        mount_path="/var/run/docker.sock"
                    )
                ]
            )
        ],
        volumes=[
            client.V1Volume(
                name="runtime-sock",
                host_path=client.V1HostPathVolumeSource(path="/var/run/docker.sock")
            )
        ]
    )
)

创建临时Pod并等待就绪：

# 创建临时Pod
core_v1.create_namespaced_pod(namespace=TARGET_NS, body=temp_pod_spec)

# 轮询等待Pod进入Running状态，最多等待30秒
ready = False
for _ in range(30):
    temp_pod_status = core_v1.read_namespaced_pod(name=temp_pod_name, namespace=TARGET_NS).status.phase
    if temp_pod_status == "Running":
        ready = True
        break
    time.sleep(1)

if not ready:
    core_v1.delete_namespaced_pod(name=temp_pod_name, namespace=TARGET_NS)
    raise RuntimeError("临时辅助Pod启动超时")

在临时Pod内调用运行时命令，指定用户进入目标容器执行操作：

# 构造容器运行时exec命令
runtime_exec_cmd = ["docker", "exec", "-u", EXEC_USER, "-i", container_runtime_id] + EXEC_CMD
# containerd运行时替换为以下命令
# runtime_exec_cmd = ["ctr", "-a", "/run/containerd/containerd.sock", "task", "exec", "--user", EXEC_USER, "-t", container_runtime_id] + EXEC_CMD

# 对接K8s exec流
exec_resp = stream(
    core_v1.connect_get_namespaced_pod_exec,
    name=temp_pod_name,
    namespace=TARGET_NS,
    command=runtime_exec_cmd,
    stdin=True,
    stdout=True,
    stderr=True,
    tty=True,
    _preload_content=False
)

# ---------------
# 此处添加输入输出处理逻辑：
# 单次执行命令可直接调用exec_resp.run_forever()后读取exec_resp.read_all()获取输出
# 交互式终端可将本地stdin/stdout和exec_resp的流做双向转发
# ---------------

操作完成后必须清理临时资源：

# 无论执行成功失败都要删除临时Pod
core_v1.delete_namespaced_pod(name=temp_pod_name, namespace=TARGET_NS)

注意事项 #

• 若集群使用containerd作为运行时，需要将临时Pod的镜像替换为包含ctr命令的镜像，同时修改socket挂载路径和exec命令，核心逻辑不变
• 必须给临时Pod配置restartPolicy: Never，同时增加异常捕获逻辑，确保任何执行分支都会触发临时Pod删除，避免集群残留无用资源
• 若集群开启了Pod安全准入、PSP等安全策略，需要提前放通特权容器、hostPath挂载的权限，否则临时Pod会创建失败
• 生产环境使用时建议固定临时Pod的镜像版本，配置合理的资源request/limit，避免影响节点上其他业务
• 如果节点上的容器运行时开启了用户命名空间重映射，需要调整exec命令的UID/GID参数匹配映射规则，否则会出现权限错误

内容的提问来源于stack exchange，提问作者Geek-bit