如何编写Python Cloud Function认证GKE集群获取Deployment名称

1. 安全认证方案 #

推荐使用GCP Workload Identity 实现无密钥的安全认证，这是GCP官方推荐的最佳实践：

• 为Cloud Function关联的服务账号（默认是PROJECT_ID@appspot.gserviceaccount.com）授予必要的GKE权限：至少需要container.clusters.get（获取集群凭证）和container.deployments.list（列出Deployment）权限，可通过自定义角色或预定义的Container Developer角色实现。
• 若GKE是私有集群，需配置VPC Connector让Cloud Function接入集群所在VPC，或把Cloud Function的出口IP加入集群的授权网络列表，确保网络可达。
• Cloud Function运行时会自动加载服务账号的应用默认凭证（ADC），无需手动配置密钥，认证过程完全由GCP托管。

2. 交互库与工具 #

核心使用两个Python库：

• google-cloud-container：GCP官方客户端库，用于调用GKE API获取集群的endpoint、CA证书等关键信息。
• kubernetes：Kubernetes官方Python客户端库，用于构建K8s API请求，获取Deployment资源。

3. 示例代码与实现步骤 #

步骤1：配置依赖 #

创建requirements.txt文件，指定依赖版本：

google-cloud-container==2.19.0
kubernetes==28.1.0

步骤2：Cloud Function代码（HTTP触发） #

from google.cloud import container_v1
from kubernetes import client, config
from kubernetes.config.kube_config import KubeConfigLoader, Configuration
import traceback

def list_gke_deployments(request):
    # 替换为你的实际参数
    PROJECT_ID = "your-gcp-project-id"
    CLUSTER_LOCATION = "us-central1-a"  # 集群所在Zone/Region，区域集群用us-central1
    CLUSTER_NAME = "your-gke-cluster-name"
    TARGET_NAMESPACE = "default"  # 要查询的命名空间，留空可改为list_deployment_for_all_namespaces

    try:
        # 初始化GKE客户端（区域集群用RegionalClusterManagerClient）
        if "-" in CLUSTER_LOCATION:
            container_client = container_v1.ClusterManagerClient()
            cluster_resource = container_client.cluster_path(PROJECT_ID, CLUSTER_LOCATION, CLUSTER_NAME)
        else:
            container_client = container_v1.RegionalClusterManagerClient()
            cluster_resource = container_client.region_cluster_path(PROJECT_ID, CLUSTER_LOCATION, CLUSTER_NAME)

        # 获取集群信息
        cluster = container_client.get_cluster(name=cluster_resource)

        # 构建K8s配置
        kube_config = {
            "apiVersion": "v1",
            "clusters": [
                {
                    "cluster": {
                        "certificate-authority-data": cluster.master_auth.cluster_ca_certificate,
                        "server": f"https://{cluster.endpoint}"
                    },
                    "name": CLUSTER_NAME
                }
            ],
            "contexts": [
                {
                    "context": {"cluster": CLUSTER_NAME, "user": CLUSTER_NAME},
                    "name": CLUSTER_NAME
                }
            ],
            "current-context": CLUSTER_NAME,
            "kind": "Config",
            "users": [
                {
                    "name": CLUSTER_NAME,
                    "user": {
                        "auth-provider": {
                            "name": "gcp",
                            "config": {
                                "cmd-args": "config config-helper --format=json",
                                "cmd-path": "/usr/bin/gcloud",
                                "expiry-key": "{.credential.token_expiry}",
                                "token-key": "{.credential.access_token}"
                            }
                        }
                    }
                }
            ]
        }

        # 加载配置并初始化K8s客户端
        loader = KubeConfigLoader(kube_config)
        k8s_config = type.__call__(Configuration)
        loader.load_and_set(k8s_config)
        client.Configuration.set_default(k8s_config)

        # 调用K8s API获取Deployment
        apps_api = client.AppsV1Api()
        deployments = apps_api.list_namespaced_deployment(TARGET_NAMESPACE)
        deployment_names = [dep.metadata.name for dep in deployments.items]

        return {"status": "success", "deployments": deployment_names}

    except Exception as e:
        return {"status": "error", "message": str(e), "traceback": traceback.format_exc()}

步骤3：部署与测试 #

• 部署Cloud Function时，确保选择的服务账号已拥有所需权限。
• 本地调试可通过gcloud auth application-default login获取ADC凭证，模拟Cloud Function的运行环境。

4. 最佳实践与注意事项 #

• 最小权限原则：创建自定义IAM角色，仅授予Cloud Function服务账号container.clusters.get和container.deployments.list权限，避免使用过度宽泛的角色（如Editor）。
• 避免硬编码参数：将PROJECT_ID、集群信息等配置存放在Cloud Function的环境变量中，而非硬编码在代码里。
• 错误处理：添加异常捕获逻辑，处理集群不存在、权限不足、网络不可达等场景，返回清晰的错误信息。
• 缓存优化：频繁调用时，可缓存集群的endpoint和CA证书，减少对GKE API的调用次数，提升响应速度。
• 私有集群访问：私有集群必须配置VPC Connector或授权Cloud Function的出口IP，否则无法访问集群API端点。
• 版本锁定：在requirements.txt中指定库的具体版本，避免因版本更新导致的兼容问题。

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