如何通过Python Cloud Function获取GKE集群镜像与环境变量

• 仅使用官方google-cloud-container库完全无法实现你的需求。
• 该库仅对接GKE服务的集群管理控制面接口，只能查询集群版本、节点池配置、网络配置、集群状态这类集群级基础元数据，没有权限也没有接口访问Kubernetes集群内部运行的工作负载资源，自然拿不到容器镜像、环境变量这类存放在Kubernetes API Server中的业务配置数据。

整体实现逻辑分两步，完全适配Cloud Function运行环境：

1. 先用google-cloud-container库，根据传入的集群名称、可用区/区域信息，获取目标GKE集群的API Server端点、CA根证书、短期访问令牌，自动生成访问集群所需的认证配置，不需要手动维护kubeconfig文件。
2. 用官方Kubernetes Python客户端对接集群API Server，拉取所有命名空间下的工作负载资源（Deployment/StatefulSet/DaemonSet/CronJob/Pod按需选择），从资源定义中解析容器镜像、环境变量字段即可。

前置权限配置 #

部署前需要给Cloud Function绑定的运行时服务账号配置两类权限：

• GCP侧权限：授予容器集群查看权限，确保可以调用GKE接口拉取集群访问凭证。
• 集群侧RBAC权限：在目标GKE集群中给该服务账号绑定集群只读权限，确保可以通过Kubernetes API查询工作负载资源。
• 网络配置：如果是私有GKE集群，需要给Cloud Function配置Serverless VPC访问连接器，将连接器网段加入GKE控制平面的授权访问列表，避免网络连通性问题。

首先在Cloud Function的requirements.txt中添加以下依赖：

functions-framework>=3.3.0
google-cloud-container>=2.18.0
kubernetes>=26.1.0

核心实现代码：

import google.auth
from google.cloud import container_v1
from kubernetes import client
import tempfile
import os

def get_gke_workload_config(request):
    """
    Cloud Function入口，接收JSON格式入参：
    {
        "cluster_name": "你的集群名称",
        "location": "集群所在可用区/区域，例如us-central1-a",
        "project_id": "可选，不填则使用Cloud Function当前所属项目"
    }
    返回结构化的工作负载容器镜像、环境变量信息
    """
    request_json = request.get_json(silent=True)
    if not request_json or "cluster_name" not in request_json or "location" not in request_json:
        return {"error": "缺少必填参数：cluster_name、location"}, 400
    
    cluster_name = request_json["cluster_name"]
    location = request_json["location"]
    project_id = request_json.get("project_id")
    if not project_id:
        _, project_id = google.auth.default()

    # 第一步：通过GKE管理接口获取集群访问凭证
    container_client = container_v1.ClusterManagerClient()
    cluster_path = f"projects/{project_id}/locations/{location}/clusters/{cluster_name}"
    cluster_info = container_client.get_cluster(name=cluster_path)

    # 临时写入CA证书，供K8s客户端使用
    with tempfile.NamedTemporaryFile(delete=False, mode="w") as ca_file:
        ca_file.write(cluster_info.master_auth.cluster_ca_certificate)
        ca_path = ca_file.name
    
    # 获取当前服务账号的访问令牌作为集群认证凭证
    credentials, _ = google.auth.default()
    credentials.refresh(google.auth.transport.requests.Request())
    token = credentials.token

    # 配置K8s客户端
    kube_config = client.Configuration()
    kube_config.host = f"https://{cluster_info.endpoint}"
    kube_config.ssl_ca_cert = ca_path
    kube_config.api_key = {"authorization": f"Bearer {token}"}
    client.Configuration.set_default(kube_config)

    # 第二步：查询所有命名空间下的Deployment，可按需扩展StatefulSet/DaemonSet等资源
    v1_apps = client.AppsV1Api()
    v1_core = client.CoreV1Api()
    namespaces = [ns.metadata.name for ns in v1_core.list_namespace().items]
    
    result = []
    for ns in namespaces:
        deployments = v1_apps.list_namespaced_deployment(namespace=ns).items
        for deploy in deployments:
            workload_item = {
                "workload_type": "Deployment",
                "workload_name": deploy.metadata.name,
                "namespace": ns,
                "containers": []
            }
            # 遍历Pod模板中的所有容器，提取镜像和环境变量
            for container in deploy.spec.template.spec.containers:
                container_info = {
                    "container_name": container.name,
                    "image": container.image,
                    "env": []
                }
                if container.env:
                    for env_item in container.env:
                        env_entry = {"name": env_item.name}
                        # 直接配置的环境变量值
                        if env_item.value:
                            env_entry["value"] = env_item.value
                        # 引用ConfigMap/Secret/downward API的环境变量
                        if env_item.value_from:
                            env_entry["value_from"] = str(env_item.value_from)
                        container_info["env"].append(env_entry)
                workload_item["containers"].append(container_info)
            result.append(workload_item)
    
    # 清理临时CA证书文件
    os.unlink(ca_path)
    return {"code": 0, "data": result}

补充说明 #

• 上述示例默认拉取所有命名空间下的Deployment资源，如果需要覆盖其他工作负载类型，对应调用K8s客户端的list_namespaced_stateful_set、list_namespaced_daemon_set、list_namespaced_cron_job等接口即可，解析容器配置的逻辑完全一致。
• 如果需要获取环境变量的实际值（即解析ConfigMap、Secret引用后的值），可以直接查询对应命名空间下的Pod实例，Pod实例中会返回注入后的环境变量值，注意这需要额外的资源读取权限。
• 生产环境使用时建议添加异常捕获、分页查询逻辑，避免大集群下接口超时或者返回结果被截断。

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