如何通过API/SDK/gcloud程序化查询GKE集群中的工作负载

一、使用gcloud命令行工具 #

gcloud是GCP官方命令行工具，可直接与GKE集群交互，无需依赖kubectl或控制台。

列出所有工作负载 #

先配置好gcloud的项目、集群和区域/地域信息，执行对应命令：

# 区域集群（如us-central1-a）
gcloud container workloads list --cluster=YOUR_CLUSTER_NAME --zone=YOUR_CLUSTER_ZONE
# 地域集群（如us-central1）
gcloud container workloads list --cluster=YOUR_CLUSTER_NAME --region=YOUR_CLUSTER_REGION

查询名为xpto的工作负载是否存在 #

通过describe命令直接验证，存在则返回详情，不存在则报错：

# 区域集群
gcloud container workloads describe xpto --cluster=YOUR_CLUSTER_NAME --zone=YOUR_CLUSTER_ZONE
# 地域集群
gcloud container workloads describe xpto --cluster=YOUR_CLUSTER_NAME --region=YOUR_CLUSTER_REGION

也可结合grep做静默检查（返回0表示存在，非0表示不存在）：

gcloud container workloads list --cluster=YOUR_CLUSTER_NAME --zone=YOUR_CLUSTER_ZONE | grep -q "^xpto"

二、使用GKE REST API #

通过HTTP请求直接调用API实现程序化操作，需先获取有效访问令牌。

1. 获取访问令牌 #

gcloud auth print-access-token

2. 列出所有工作负载 #

发送GET请求到GKE工作负载API端点：

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  "https://container.googleapis.com/v1/projects/YOUR_PROJECT_ID/locations/YOUR_LOCATION/clusters/YOUR_CLUSTER_NAME/workloads"

YOUR_LOCATION可填区域（如us-central1-a）或地域（如us-central1）。

3. 查询xpto工作负载是否存在 #

请求特定工作负载的详情端点：

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  "https://container.googleapis.com/v1/projects/YOUR_PROJECT_ID/locations/YOUR_LOCATION/clusters/YOUR_CLUSTER_NAME/workloads/xpto"

返回200 OK则存在，返回404 Not Found则不存在。

三、使用Google Cloud SDK（以Python为例） #

借助GCP官方Python SDK编写自动化脚本，先安装依赖：

pip install google-cloud-container

列出所有工作负载 #

from google.cloud import container_v1

client = container_v1.ClusterManagerClient()

# 构建集群路径
cluster_path = client.cluster_path(
    project="YOUR_PROJECT_ID",
    location="YOUR_LOCATION",
    cluster="YOUR_CLUSTER_NAME"
)

# 遍历输出所有工作负载
response = client.list_workloads(parent=cluster_path)
for workload in response.workloads:
    print(f"名称: {workload.name} | 类型: {workload.workload_type}")

查询xpto工作负载是否存在 #

from google.cloud import container_v1
from google.api_core.exceptions import NotFound

client = container_v1.ClusterManagerClient()

workload_path = client.workload_path(
    project="YOUR_PROJECT_ID",
    location="YOUR_LOCATION",
    cluster="YOUR_CLUSTER_NAME",
    workload="xpto"
)

try:
    workload = client.get_workload(name=workload_path)
    print(f"工作负载xpto存在，类型: {workload.workload_type}")
except NotFound:
    print("工作负载xpto不存在")

内容的提问来源于stack exchange，提问作者João Fonseca