如何通过CloudBuild API获取特定触发器最新构建状态？排障及方案问询

已解决的核心问题 #

更新：将触发器区域从Europe-West-2改为默认的Global Non-Region后，API恢复正常工作。

原问题：在Airflow中调用CloudBuild API获取特定触发器的最新运行状态时，无法列出构建或触发器列表，谷歌官方示例代码执行后返回None（项目中确有触发器，且project_id与CloudBuild执行详情页完全一致），操作静默失败无报错信息。示例代码如下：

from google.cloud.devtools import cloudbuild_v1

def sample_list_build_triggers():
    # Create a client
    client = cloudbuild_v1.CloudBuildClient()

    # Initialize request argument(s)
    request = cloudbuild_v1.ListBuildTriggersRequest(
        project_id="example_project_id",
    )

    # Make the request
    page_result = client.list_build_triggers(request=request)

    # Handle the response
    for response in page_result:
        print(response)

sample_list_build_triggers()

获取CloudBuild运行状态的替代方法 #

1. 区域化API查询（适配非全局触发器） #

如果需要保留区域化触发器，调用API时显式指定区域参数，避免默认全局区域与触发器所在区域不匹配：

from google.cloud.devtools import cloudbuild_v1

def get_trigger_latest_status(project_id, trigger_id, region="europe-west2"):
    client = cloudbuild_v1.CloudBuildClient()
    parent = f"projects/{project_id}/locations/{region}"
    # 按触发器ID过滤，按启动时间倒序取最新构建
    request = cloudbuild_v1.ListBuildsRequest(
        parent=parent,
        filter=f"trigger_id={trigger_id}",
        order_by="start_time desc"
    )
    builds = list(client.list_builds(request=request))
    if builds:
        latest_build = builds[0]
        return {"status": latest_build.status.name, "build_id": latest_build.id, "finish_time": latest_build.finish_time}
    return None

2. Pub/Sub状态通知（无需主动轮询） #

利用CloudBuild原生的Pub/Sub集成，实现构建状态的被动推送：

• 进入CloudBuild控制台，为目标触发器配置「构建完成时发送通知」，绑定指定Pub/Sub主题
• 在Airflow中使用PubSubPullSensor监听该主题，消息体将包含status、trigger_id、build_id等关键字段
• 该方式无需主动调用API，状态通知更及时，降低资源消耗

3. gcloud命令行集成（适合调试与快速脚本） #

在Airflow的BashOperator中直接调用gcloud命令获取状态：

# 获取指定触发器的最新构建状态
gcloud builds list --filter="trigger_id=YOUR_TRIGGER_ID" --sort-by="~startTime" --limit=1 --format="value(status)" --region=europe-west2

测试建议 #

• 区域一致性验证：确认触发器所在区域与API请求/命令行指定的区域完全匹配（CloudBuild区域化资源不支持跨区域查询）
• 权限排查：在本地使用相同服务账号密钥执行测试代码，确认是否是Airflow环境的权限缺失（需确保账号拥有cloudbuild.builds.get、cloudbuild.triggers.list权限）
• 错误捕获增强：在Python代码中添加异常捕获，避免静默失败：

from google.api_core.exceptions import GoogleAPIError

try:
    page_result = client.list_build_triggers(request=request)
    triggers = list(page_result)
    if not triggers:
        print("未找到触发器，请检查区域和project_id")
except GoogleAPIError as e:
    print(f"API调用错误详情: {e}")

• REST接口测试：通过curl直接调用REST API，排查是否是客户端库的问题：

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  "https://cloudbuild.googleapis.com/v1/projects/example_project_id/locations/europe-west2/triggers"

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