如何在GCP Cloud Run任务内自动获取执行触发来源？

Cloud Run Job本身没有默认环境变量直接提供触发者身份，但可以通过以下方法自动获取该信息：

方法：结合Cloud Audit Logs与Logging API查询 #

当Cloud Run Job被触发（无论是用户手动操作还是自动化流程），触发相关的身份信息会被记录在Cloud Audit Logs中。你可以在任务内部通过自身的执行ID调用Logging API，查询对应的审计日志条目来提取触发者身份。

步骤说明 #

1. 获取任务执行ID：任务内部可通过环境变量CLOUD_RUN_EXECUTION获取当前执行的唯一ID。
2. 调用Logging API查询审计日志：使用该执行ID作为过滤条件，查询项目中对应的Cloud Audit Log（类型为cloudaudit.googleapis.com/activity），从中提取触发者的邮箱信息。
3. 配置权限：给Cloud Run Job使用的服务账号授予roles/logging.viewer角色，确保任务有权限读取日志。

代码示例（Python） #

import os
from google.cloud import logging_v2

def fetch_trigger_principal():
    # 获取当前任务的执行ID
    execution_id = os.environ.get("CLOUD_RUN_EXECUTION")
    if not execution_id:
        return "无法获取执行ID"

    # 初始化Logging客户端
    client = logging_v2.LoggingServiceV2Client()
    project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
    if not project_id:
        return "无法获取项目ID"

    # 构建日志查询过滤器
    log_filter = (
        f'resource.type="cloud_run_job" '
        f'AND resource.labels.execution_id="{execution_id}" '
        f'AND logName="projects/{project_id}/logs/cloudaudit.googleapis.com%2Factivity"'
    )

    # 查询最新的审计日志条目
    entries = client.list_log_entries(
        resource_names=[f"projects/{project_id}"],
        filter_=log_filter,
        order_by="timestamp desc",
        page_size=1
    )

    # 提取触发者身份
    for entry in entries:
        if hasattr(entry.proto_payload, "authentication_info"):
            return entry.proto_payload.authentication_info.principal_email
    return "未找到触发者信息"

# 调用示例
if __name__ == "__main__":
    trigger_user = fetch_trigger_principal()
    print(f"本次任务触发者：{trigger_user}")

注意事项 #

• 如果任务由用户手动触发（控制台/gcloud命令），返回的principal_email是用户的GCP账号邮箱；如果由服务账号触发（如Cloud Scheduler、Pub/Sub），则是服务账号的邮箱。
• 确保任务的服务账号已被授予日志读取权限，否则API调用会失败。

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