如何用Python脚本在Kubernetes中动态管理容器与ConfigMap？

你当前的runContainers.py脚本通过Docker批量处理分块的客户端数据，迁移到Kubernetes时，核心是将临时Docker容器替换为Kubernetes的Job资源（Job适用于一次性执行的任务，完成后自动终止），同时用ConfigMap传递分块数据。以下是具体适配方案：

核心逻辑调整 #

原脚本的流程不变：更新客户端列表 → 分块 → 为每个分块启动任务。只是将Docker容器的创建/管理替换为Kubernetes Job的创建/管理，临时文件挂载替换为ConfigMap挂载。

适配方案一：用kubectl命令通过subprocess调用（适合新手） #

这种方式不需要额外安装K8s Python客户端，直接通过kubectl操作集群，前提是你的环境已经配置好K8s集群访问权限（kubeconfig文件正确）。

适配后的脚本 #

import json
import subprocess
import os
from tempfile import NamedTemporaryFile
from src import clientListUpdate

# 更新客户端列表
clientListUpdate.updateClientJson()
chunkSize = 3
dockerImageName = 'client_upload_image'
# K8s资源的命名空间，根据实际情况修改
namespace = 'default'

def clientChunks(clientList, chunkSize):
    chunks = []
    for i in range(0, len(clientList), chunkSize):
        chunks.append(clientList[i:i + chunkSize])
    return chunks

with open('./data/clientList.json', 'r') as file:
    clients = json.load(file)

# 存储每个Job的名称和对应的ConfigMap名称，用于后续清理
job_configmap_pairs = []

for idx, chunk in enumerate(clientChunks(clients, chunkSize)):
    # 生成唯一的资源名称
    job_name = f"client-upload-job-{idx}"
    configmap_name = f"client-data-cm-{idx}"
    
    # 1. 创建ConfigMap存储当前分块的客户端数据
    with NamedTemporaryFile(mode='w', delete=False, suffix='.json') as tempFile:
        json.dump(chunk, tempFile)
        tempFile_path = tempFile.name
    
    # 执行kubectl创建ConfigMap命令
    subprocess.run([
        'kubectl', 'create', 'configmap', configmap_name,
        '--from-file', f'/app/client_data.json={tempFile_path}',
        '--namespace', namespace
    ], check=True)
    os.remove(tempFile_path)
    
    # 2. 创建Job，挂载ConfigMap到容器指定路径
    job_yaml = f"""
apiVersion: batch/v1
kind: Job
metadata:
  name: {job_name}
  namespace: {namespace}
spec:
  template:
    spec:
      containers:
      - name: client-upload
        image: {dockerImageName}
        command: ["python", "/app/clientUpload.py"]
        volumeMounts:
        - name: client-data
          mountPath: /app/client_data.json
          subPath: client_data.json
      volumes:
      - name: client-data
        configMap:
          name: {configmap_name}
      restartPolicy: Never
  backoffLimit: 1
"""
    # 用临时文件存储Job的YAML，然后执行kubectl创建
    with NamedTemporaryFile(mode='w', delete=False, suffix='.yaml') as job_temp:
        job_temp.write(job_yaml)
        job_temp_path = job_temp.name
    
    subprocess.run([
        'kubectl', 'apply', '-f', job_temp_path,
        '--namespace', namespace
    ], check=True)
    os.remove(job_temp_path)
    
    job_configmap_pairs.append((job_name, configmap_name))

# 等待所有Job完成，然后清理Job和ConfigMap
for job_name, configmap_name in job_configmap_pairs:
    # 等待Job完成
    subprocess.run([
        'kubectl', 'wait', f'job/{job_name}',
        '--for=condition=complete', '--timeout=300s',
        '--namespace', namespace
    ], check=True)
    # 删除Job和ConfigMap
    subprocess.run(['kubectl', 'delete', 'job', job_name, '--namespace', namespace], check=True)
    subprocess.run(['kubectl', 'delete', 'configmap', configmap_name, '--namespace', namespace], check=True)

适配方案二：用Kubernetes Python客户端（更优雅） #

如果希望更集成化的Python操作，可以使用官方的Kubernetes Python客户端，需要先安装依赖：

pip install kubernetes

适配后的脚本示例 #

import json
from kubernetes import client, config
from src import clientListUpdate

# 加载K8s配置（默认读取~/.kube/config，集群内运行可使用in_cluster_config()）
config.load_kube_config()
batch_v1 = client.BatchV1Api()
core_v1 = client.CoreV1Api()

# 更新客户端列表
clientListUpdate.updateClientJson()
chunkSize = 3
dockerImageName = 'client_upload_image'
namespace = 'default'

def clientChunks(clientList, chunkSize):
    chunks = []
    for i in range(0, len(clientList), chunkSize):
        chunks.append(clientList[i:i + chunkSize])
    return chunks

with open('./data/clientList.json', 'r') as file:
    clients = json.load(file)

job_configmap_pairs = []

for idx, chunk in enumerate(clientChunks(clients, chunkSize)):
    job_name = f"client-upload-job-{idx}"
    configmap_name = f"client-data-cm-{idx}"
    
    # 1. 创建ConfigMap
    cm_body = client.V1ConfigMap(
        metadata=client.V1ObjectMeta(name=configmap_name),
        data={
            "client_data.json": json.dumps(chunk)
        }
    )
    core_v1.create_namespaced_config_map(namespace=namespace, body=cm_body)
    
    # 2. 创建Job
    pod_template = client.V1PodTemplateSpec(
        spec=client.V1PodSpec(
            containers=[
                client.V1Container(
                    name="client-upload",
                    image=dockerImageName,
                    command=["python", "/app/clientUpload.py"],
                    volume_mounts=[
                        client.V1VolumeMount(
                            name="client-data",
                            mount_path="/app/client_data.json",
                            sub_path="client_data.json"
                        )
                    ]
                )
            ],
            volumes=[
                client.V1Volume(
                    name="client-data",
                    config_map=client.V1ConfigMapVolumeSource(name=configmap_name)
                )
            ],
            restart_policy="Never"
        )
    )
    
    job_body = client.V1Job(
        metadata=client.V1ObjectMeta(name=job_name),
        spec=client.V1JobSpec(
            template=pod_template,
            backoff_limit=1
        )
    )
    batch_v1.create_namespaced_job(namespace=namespace, body=job_body)
    
    job_configmap_pairs.append((job_name, configmap_name))

# 等待Job完成并清理
for job_name, configmap_name in job_configmap_pairs:
    # 轮询等待Job完成
    while True:
        job = batch_v1.read_namespaced_job(name=job_name, namespace=namespace)
        if job.status.succeeded == 1:
            break
    # 删除资源
    batch_v1.delete_namespaced_job(name=job_name, namespace=namespace)
    core_v1.delete_namespaced_config_map(name=configmap_name, namespace=namespace)

关键注意事项 #

• 确保你的K8s集群能够拉取client_upload_image镜像（如果是私有镜像，需要配置ImagePullSecret）
• 调整namespace为你实际使用的命名空间
• 根据任务执行时间，修改wait命令的超时时间（方案一）或轮询逻辑（方案二）
• 如果分块数据过大，不适合用ConfigMap（K8s对ConfigMap大小有限制，一般不超过1MB），此时可以考虑用PersistentVolumeClaim或者直接将数据通过环境变量传递（需确保数据格式兼容）

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