如何通过Python SDK获取GCP Cloud Run作业执行的创建者？

你提到的run_v2.ListExecutionsRequest没有返回创建者信息，是因为Cloud Run的v2 API默认不包含该元数据，而gcloud run jobs executions list实际调用的是v1版本API，该版本会返回包含创建者的注解字段。以下两种方法可以解决这个问题：

方法1：直接调用Cloud Run v1 API #

v1 API的Execution对象元数据中，annotations['run.googleapis.com/invoked-by']字段存储了执行触发者信息，格式通常为user:xxx@xxx.com或serviceAccount:xxx@xxx.iam.gserviceaccount.com，提取后即可得到RUN BY对应的内容。

代码示例 #

from google.oauth2 import service_account
from googleapiclient.discovery import build

# 替换为你的配置信息
SERVICE_ACCOUNT_KEY_PATH = "path/to/your/service-account-key.json"
PROJECT_NAMESPACE = "oxa-dev-loc-data-connector"
REGION = "us-west1"

# 加载身份凭证
credentials = service_account.Credentials.from_service_account_file(
    SERVICE_ACCOUNT_KEY_PATH,
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)

# 构建v1版本的Cloud Run服务客户端
run_service = build(
    "run", "v1",
    credentials=credentials,
    discoveryServiceUrl=f"https://{REGION}-run.googleapis.com/$discovery/rest?version=v1"
)

# 查询指定命名空间下的所有作业执行
response = run_service.namespaces().executions().list(namespace=PROJECT_NAMESPACE).execute()

# 解析并提取创建者信息
for execution in response.get("items", []):
    exec_name = execution["metadata"]["name"]
    invoked_by = execution["metadata"].get("annotations", {}).get("run.googleapis.com/invoked-by")
    
    if invoked_by:
        # 拆分前缀（user/ServiceAccount）和实际账号
        run_by = invoked_by.split(":", 1)[1] if ":" in invoked_by else invoked_by
    else:
        run_by = "Unknown"
    
    print(f"执行任务: {exec_name} | 创建者: {run_by}")

方法2：通过Cloud Audit Logs查询 #

如果v1 API无法获取到信息（比如特殊触发场景），可以查询Cloud Audit Logs中google.cloud.run.executions.v1.CreateExecution类型的日志条目，其中authenticationInfo.principalEmail字段即为创建者邮箱。

代码示例 #

from google.cloud import logging_v2
from google.oauth2 import service_account
import datetime

# 替换为你的配置信息
SERVICE_ACCOUNT_KEY_PATH = "path/to/your/service-account-key.json"
PROJECT_ID = "your-gcp-project-id"
TARGET_EXECUTION_NAME = "my-job-lj2op"

# 加载身份凭证
credentials = service_account.Credentials.from_service_account_file(
    SERVICE_ACCOUNT_KEY_PATH,
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)

# 初始化日志客户端
logging_client = logging_v2.LoggingServiceV2Client(credentials=credentials)

# 构建日志查询条件（查询最近7天内的创建日志）
time_range = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=7)
log_query = f"""
resource.type="cloud_run_job"
protoPayload.methodName="google.cloud.run.executions.v1.CreateExecution"
protoPayload.resourceName:"{TARGET_EXECUTION_NAME}"
"""

# 执行日志查询
log_entries = logging_client.list_log_entries(
    resource_names=[f"projects/{PROJECT_ID}"],
    filter_=log_query,
    order_by="timestamp desc",
    page_size=1
)

# 提取创建者信息
for entry in log_entries:
    run_by = entry.proto_payload.authentication_info.principal_email
    print(f"执行任务 {TARGET_EXECUTION_NAME} 的创建者: {run_by}")

权限说明 #

• 方法1需要服务账号拥有roles/run.viewer权限，允许访问Cloud Run资源
• 方法2需要启用Cloud Run的Data Access审计日志（默认已启用），且服务账号拥有roles/logging.viewer权限

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