使用 Kubernetes Engine API 客户端列出集群与节点池：python-docs-samples 示例深度解析

本指南以 kubernetes_engine/api-client 目录为核心，系统讲解如何用 Python 的 REST 风格 API 客户端（google-api-python-client）直接调用 Kubernetes Engine（GKE）的 container v1 API，列出项目下所有集群及其节点池信息。你将掌握从环境准备、认证、依赖安装到源码逐段解读、命令行运行与自动化测试的完整链路，并了解该示例在仓库中的定位及其与 Django 部署教程 的差异。

一、示例定位：这个目录解决什么问题

kubernetes_engine/api-client/README.rst 明确指出，该目录包含 Google Kubernetes Engine 的代码示例。GKE 在 Google Cloud Platform 上以 Kubernetes 为底座运行 Docker 容器，而本目录提供的是面向 API 客户端库的最小可用示例：通过 snippets.py 这一个脚本，演示如何认证、构建服务对象并调用 container v1 接口完成集群与节点池的只读查询。

从仓库结构看，kubernetes_engine 下还有另一个子目录 django_tutorial，它侧重"把 Django 应用部署到 GKE"的完整工作流；而 api-client 目录则聚焦"纯 API 调用"这一更底层、更通用的能力，适合需要二次开发、自动化巡检或集成到自有工具链的场景。

值得注意，仓库中的 README.rst.in 是 README.rst 的生成模板，其中通过 product、setup、samples 三个 YAML 字段声明了产品信息、认证与依赖安装步骤以及示例文件位置，folder: kubernetes_engine/api-client 一行则标记了本示例在仓库中的路径。这也解释了为何 README.rst 首行注明"本文件自动生成，请勿直接编辑"。

二、运行环境准备

2.1 认证

README.rst 强调：运行该示例必须完成认证配置，即需要为应用设置 Google Cloud 凭据（应用默认凭据，ADC）。标准做法是通过 gcloud auth application-default login 生成本地凭据，或为服务器环境配置服务账号，并通过 GOOGLE_APPLICATION_CREDENTIALS 环境变量指向服务账号 JSON 文件。认证是 google-api-python-client 与 google-auth 协作完成的前提——本示例源码中并没有显式传入凭据，而是依赖 SDK 默认从环境获取。

2.2 克隆仓库与安装依赖

按文档给出的步骤，先克隆 python-docs-samples 仓库并进入示例目录：

$ git clone https://gitcode.com/GitHub_Trending/py/python-docs-samples.git
$ cd python-docs-samples/kubernetes_engine/api-client

随后创建虚拟环境并安装依赖（若尚未安装 pip 与 virtualenv，请先安装）：

$ virtualenv env
$ source env/bin/activate
$ pip install -r requirements.txt

README.rst 声明示例兼容 Python 2.7 与 3.4+。需要说明的是，这是文档生成时的历史表述；从当前仓库的实际约束看，requirements.txt 锁定了较新的库版本，而 requirements-test.txt 中的 pytest==9.0.3; python_version >= "3.10" 表明测试环境要求 Python 3.10 及以上，因此现代环境（Python 3.10+）才是实际推荐运行前提。

requirements.txt 中三个依赖各自承担明确职责：

依赖	版本	作用
google-api-python-client	2.131.0	提供 googleapiclient.discovery.build()，负责按 Discovery 文档动态构建 GKE API 客户端
google-auth	2.38.0	应用默认凭据（ADC）的加载与 OAuth 2.0 令牌管理
google-auth-httplib2	0.2.0	将 httplib2 的 HTTP 传输层接入 google-auth 的凭据体系

三、核心实现：snippets.py 逐段解析

示例的全部逻辑集中在 snippets.py（约 70 行），由"API 调用函数 + argparse 命令行入口"两部分组成。

3.1 构建 GKE 服务对象

import googleapiclient.discovery

service = googleapiclient.discovery.build("container", "v1")

discovery.build("container", "v1") 是整段代码的地基：它会拉取 GKE container API 的 v1 Discovery 文档，动态生成与 REST 端点一一对应的调用对象。这意味着你无需手写 HTTP 请求，也无需安装独立的 GKE 专属 SDK 包——这正是"api-client"命名的含义：通过通用的 Google API 客户端库访问 GKE。

3.2 列出集群（clusters）

clusters_resource = service.projects().zones().clusters()

clusters_response = clusters_resource.list(
    projectId=project_id, zone=zone
).execute()

这里体现了 GKE REST API 的层级结构：projects → zones → clusters。list() 需要两个参数——projectId（GCP 项目 ID）与 zone（计算可用区，如 us-central1-f），execute() 触发实际 HTTP 请求并返回响应体。

3.3 为每个集群列出节点池（nodePools）

nodepools_response = (
    clusters_resource.nodePools()
    .list(projectId=project_id, zone=zone, clusterId=cluster["name"])
    .execute()
)

在 clusters 资源下继续下钻到 nodePools 资源，追加 clusterId 参数指定所属集群。注意这里是在外层遍历集群的 for 循环内逐集群调用的，因此一个集群对应一次节点池查询，属于典型的 1:N 级联调用模式。

3.4 输出与容错处理

集群输出：

for cluster in clusters_response.get("clusters", []):
    print(
        "Cluster: {}, Status: {}, Current Master Version: {}".format(
            cluster["name"], cluster["status"], cluster["currentMasterVersion"]
        )
    )

节点池输出：

for nodepool in nodepools_response["nodePools"]:
    print(
        " -> Pool: {}, Status: {}, Machine Type: {}, "
        "Autoscaling: {}".format(
            nodepool["name"],
            nodepool["status"],
            nodepool["config"]["machineType"],
            nodepool.get("autoscaling", {}).get("enabled", False),
        )
    )

这段代码有两点实现细节值得注意：

• 防御式取值：外层用 clusters_response.get("clusters", []) 兜底，即使项目下没有任何集群也不会抛异常；节点池的 autoscaling 字段用 nodepool.get("autoscaling", {}).get("enabled", False) 双层容错，因为未开启自动扩缩容的节点池响应中可能根本没有 autoscaling 字段，直接索引会报 KeyError；
• 直接索引：cluster["name"]、cluster["currentMasterVersion"]、nodepool["config"]["machineType"] 等字段使用直接索引而非 .get()，说明这些字段被 API 保证稳定返回，属于可信字段。

3.5 命令行入口

parser = argparse.ArgumentParser(
    description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
)
subparsers = parser.add_subparsers(dest="command")
list_clusters_and_nodepools_parser = subparsers.add_parser(
    "list_clusters_and_nodepools", help=list_clusters_and_nodepools.__doc__
)
list_clusters_and_nodepools_parser.add_argument("project_id")
list_clusters_and_nodepools_parser.add_argument("zone")

args = parser.parse_args()

if args.command == "list_clusters_and_nodepools":
    list_clusters_and_nodepools(args.project_id, args.zone)
else:
    parser.print_help()

入口采用 argparse 子命令（subcommand） 设计：当前仅暴露 list_clusters_and_nodepools 一个子命令，位置参数为 project_id 与 zone。这种结构便于未来在同一脚本中扩展更多操作（如 create_cluster、delete_cluster），是 GKE 管理脚本常见的 CLI 组织方式。

四、运行示例与输出解读

README.rst 给出了脚本自带的帮助信息，直接运行即可查看：

$ python snippets.py

输出如下（与 README 中记录的 help 文本一致）：

usage: snippets.py [-h] {list_clusters_and_nodepools} ...

positional arguments:
  {list_clusters_and_nodepools}
    list_clusters_and_nodepools
                            Lists all clusters and associated node pools.

optional arguments:
  -h, --help            show this help message and exit

执行实际查询：

$ python snippets.py list_clusters_and_nodepools YOUR_PROJECT_ID us-central1-f

根据 print 语句的格式化逻辑（源码可直接确认），成功时输出形如：

Cluster: my-cluster, Status: RUNNING, Current Master Version: 1.31.4-gke.xxx
 -> Pool: default-pool, Status: RUNNING, Machine Type: e2-medium, Autoscaling: False

每个集群占一行主记录，其下以 -> 前缀逐行缩进展示该集群的节点池，信息维度包括集群运行状态、Master 版本、节点池机器类型与自动扩缩容开关。

五、自动化测试：如何验证示例正确性

示例附带 snippets_test.py，全文件仅一个测试用例：

import os

import snippets


def test_list_clusters_and_nodepools():
    project_id = os.environ["GOOGLE_CLOUD_PROJECT"]
    snippets.list_clusters_and_nodepools(project_id, "us-central1-f")

该测试揭示了三条重要信息：

1. 依赖真实环境变量：GOOGLE_CLOUD_PROJECT 必须预先设置，指向一个有权限访问的 GCP 项目；
2. 硬编码测试区域：测试固定使用 us-central1-f 可用区，因此该区域的集群/节点池是否存在会直接影响测试结果（若响应中无 clusters 字段，get() 兜底会静默通过）；
3. 运行方式：测试直接以函数方式调用，绕开 argparse 入口，属于典型的"业务逻辑与 CLI 解耦"设计——list_clusters_and_nodepools(project_id, zone) 可被任何调用方复用。

结合 requirements-test.txt（pytest==9.0.3; python_version >= "3.10"），本地运行测试的命令为：

$ export GOOGLE_CLOUD_PROJECT=your-project-id
$ pip install -r requirements-test.txt
$ pytest snippets_test.py

六、底层 API 调用链与关键字段说明

综合源码与 GKE container v1 API 的结构，整个调用链可以概括为：

googleapiclient.discovery.build("container", "v1")
        │
        ▼
service.projects().zones().clusters()
        │
        ├── .list(projectId, zone)            → 集群列表
        └── .nodePools().list(projectId, zone, clusterId)
                                               → 每个集群的节点池列表

调用返回的响应字段与脚本的打印项一一对应，整理如下：

资源层级	读取字段	含义
集群	name	集群名称（同时用作 nodePools 查询的 clusterId）
集群	status	集群生命周期状态（如 RUNNING、PROVISIONING）
集群	currentMasterVersion	当前控制面（Master）的 Kubernetes 版本
节点池	name	节点池名称
节点池	status	节点池运行状态
节点池	config.machineType	节点虚拟机类型（如 e2-medium）
节点池	autoscaling.enabled	是否启用节点自动扩缩容（可能缺失，需 .get() 容错）

一个值得留意的取舍：示例遍历集群时未处理分页（GKE API 的 list 接口在集群/节点池数量较多时会返回 nextPageToken）。从源码结构看，该示例面向"小型项目快速巡检"场景；若要在生产环境管理成百上千个集群，需要在 execute() 之后追加分页循环逻辑。

七、从示例走向生产：可扩展方向

基于以上源码分析，可以推断出若干自然的演进路径：

• 区域（regional）集群支持：示例固定使用 zones 维度的 REST 端点（projects/zones/clusters）。GKE 同样支持区域级集群，container API 的 projects/locations/clusters 端点（以区域或 - 表示所有位置）可覆盖该场景——本示例未包含这一分支，需要时可参照同样的调用模式扩展；
• 分页处理：增加 pageToken/nextPageToken 循环，应对大规模集群环境；
• 异常与状态过滤：可依据 status 字段过滤出非 RUNNING 的集群做告警；autoscaling.enabled 字段可用于盘点未开启自动扩缩容的节点池；
• 凭据策略：本地开发用 gcloud auth application-default login，CI/生产环境则注入服务账号并设置 GOOGLE_APPLICATION_CREDENTIALS，两种方式均可被 google-auth 的默认凭据链自动识别。

总结

kubernetes_engine/api-client 用 70 行代码演示了 GKE API 客户端调用的完整范式：discovery.build 构建服务 → projects/zones/clusters 与 nodePools 级联查询 → 字段容错与格式化输出 → argparse 子命令封装 → pytest 自动化验证。它既是了解 GKE REST API 结构的最佳入门样例，也是在此基础上构建集群巡检、成本分析或自动化运维工具的可复用起点。