如何在GCP Cloud Functions中跨区域停止带指定标签的VM实例

不存在可以跳过项目ID、区域、实例名参数直接调用GCP实例停止接口的“捷径”：这三个参数是GCP资源寻址的核心组成，计算实例是项目下按区域隔离的资源，三个参数共同定位唯一实例，属于API底层设计约束，没有绕过路径。

但你要实现的「无需提前硬编码区域/实例名、全项目范围停止VM、支持标签筛选」的需求完全可以落地，不需要提前手动收集这三类参数，核心是先用聚合列表接口拉取全区域实例清单，从返回的实例属性里自动提取所需的区域、实例名参数再调用停止接口即可，具体实现步骤如下：

前置准备 #

• 给Cloud Functions绑定的运行时服务账号授予项目级的compute.instances.list、compute.instances.stop权限，不要绑定单区域范围的权限，否则无法枚举跨区域实例。
• 函数依赖配置requirements.txt添加google-api-python-client依赖，适配Python 3.8运行时。

核心实现逻辑 #

1. 调用Compute Engine的aggregatedList接口，该接口无需传入区域参数，可一次性拉取指定项目下所有区域/可用区的实例清单，还支持直接传入标签过滤规则在服务端完成筛选，无需拉取全量数据到本地再过滤。
2. 遍历返回的实例列表，跳过已处于终止状态的实例，从每个实例元数据中自动提取区域、实例名称字段。
3. 对符合筛选条件的运行中实例，传入自动提取的参数调用停止接口即可。

可直接在Cloud Functions中使用的参考代码如下：

import googleapiclient.discovery
import os

def stop_all_vms(request):
    # 自动从Cloud Functions运行环境获取当前项目ID，无需硬编码
    project_id = os.getenv("GCP_PROJECT")
    # 支持通过请求参数传入要筛选的标签，比如?env=test 即停止标签env=test的实例
    request_args = request.args
    label_filter = None
    if request_args:
        for k, v in request_args.items():
            label_filter = f"labels.{k}={v}"
            break
    
    compute = googleapiclient.discovery.build("compute", "v1")
    aggregated_list_request = compute.instances().aggregatedList(
        project=project_id,
        filter=label_filter
    )
    
    stopped_instances = []
    while aggregated_list_request is not None:
        response = aggregated_list_request.execute()
        # 遍历所有区域下的实例
        for zone, instances_scoped_list in response["items"].items():
            if "instances" not in instances_scoped_list:
                continue
            for instance in instances_scoped_list["instances"]:
                # 仅处理运行中实例
                if instance["status"] != "RUNNING":
                    continue
                # 从实例元数据自动提取区域、实例名，无需手动传入
                instance_zone = instance["zone"].split("/")[-1]
                instance_name = instance["name"]
                # 调用停止接口
                compute.instances().stop(
                    project=project_id,
                    zone=instance_zone,
                    instance=instance_name
                ).execute()
                stopped_instances.append(f"{instance_zone}/{instance_name}")
        aggregated_list_request = compute.instances().aggregatedList_next(
            previous_request=aggregated_list_request,
            previous_response=response
        )
    
    return f"Successfully stopped instances: {stopped_instances}"

注意事项 #

• 如果需要跨项目停止实例，只要服务账号拥有对应项目的相关权限，替换代码中的project_id为目标项目ID即可，逻辑完全一致。
• 标签筛选支持多条件组合，直接按GCP API过滤语法拼接filter参数即可，比如同时匹配env=test和team=dev的过滤规则为labels.env=test AND labels.team=dev。
• 接口调用会自动处理分页，不会出现单页返回不全漏扫实例的问题。

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