本地Python通过服务账号获取GCP项目BigQuery操作日志遇阻

我之前也碰到过一模一样的情况——调用GCP Logging API拿到ProtobufEntry后，看着一堆序列化数据完全不知道怎么提取里面的BigQuery操作细节（插入、更新、合并这些关键记录）。其实这些日志是Cloud Audit的Activity日志，以Protobuf格式存储，得用对应的AuditData类来解析，而不是直接打印entry对象。

下面是具体的解决步骤和修改后的代码，完全符合你“只用脚本+服务账号，不创建额外GCP资源”的要求：

1. 先搞懂ProtobufEntry的核心结构 #

Cloudaudit的activity日志里，真正有用的操作数据都藏在entry.payload里，这个payload是AuditData类型的Protobuf对象。你得先把它转换成可操作的Python对象，才能拿到操作类型、目标表、执行账号这些关键信息。

2. 修改后的完整代码（带注释） #

import google.protobuf
from google.cloud.bigquery_logging_v1 import AuditData
import google.cloud.logging
from datetime import datetime, timedelta, timezone
import os
import json

# 配置服务账号凭据和项目ID
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "C:\\mypath\\credentials.json"
project_id = "project1"
os.environ["GOOGLE_CLOUD_PROJECT"] = project_id

# 构建过滤条件：仅获取BigQuery资源的Audit日志，时间范围设为最近2天
yesterday = datetime.now(timezone.utc) - timedelta(days=2)
filter_str = (
    f'logName="projects/{project_id}/logs/cloudaudit.googleapis.com%2Factivity"'
    f' AND resource.type="bigquery_resource"'
    f' AND timestamp>="{yesterday.isoformat()}"'  # 用isoformat更稳妥，避免自定义格式的时区问题
)

# 初始化Logging客户端
client = google.cloud.logging.Client(project=project_id)

# 解析日志并写入文件
with open("C:\\mypath\\logs.txt", "w") as f:
    for entry in client.list_entries(filter_=filter_str):
        # 关键步骤：把Protobuf payload解析成AuditData对象
        if isinstance(entry.payload, google.protobuf.message.Message):
            audit_data = AuditData.deserialize(entry.payload.SerializeToString())
            
            # 提取你需要的核心信息（可以根据需求增减字段）
            log_info = {
                "操作时间": entry.timestamp.isoformat(),
                "操作类型": audit_data.service_data.bigquery_data_access_audit_log.method_name,
                "所属项目": audit_data.resource_name.split("/")[1],
                "数据集": audit_data.resource_name.split("/")[3],
                "目标表": audit_data.resource_name.split("/")[5],
                "执行账号": audit_data.authentication_info.principal_email,
                # 如果需要查看请求详情（比如插入的SQL语句），可以放开下面这行
                # "请求详情": json.dumps(audit_data.service_data.bigquery_data_access_audit_log.request.to_dict(), indent=2)
            }
            
            # 转成格式化的JSON字符串写入文件，同时打印到控制台
            log_str = json.dumps(log_info, ensure_ascii=False, indent=2) + "\n"
            f.write(log_str)
            print(log_str)

3. 关键细节说明 #

• 过滤条件优化：用datetime.isoformat()生成时间字符串，比自定义格式更符合GCP日志API的要求，避免时区不匹配导致的日志漏查。
• Protobuf解析：通过AuditData.deserialize()把序列化的payload转成Python对象，bigquery_data_access_audit_log字段里包含了所有BigQuery操作的具体细节。
• 常见操作类型对应：
  • tabledata.insertAll：批量插入数据
  • jobs.insert：执行查询/加载任务
  • tables.update：更新表结构
  • tables.patch：增量更新表
• 权限注意：确保你的服务账号拥有roles/logging.viewer（查看日志）和roles/bigquery.dataViewer（查看请求详情）权限，否则可能拿不到完整数据。

4. 排查小技巧 #

如果运行后没有返回日志：

• 先确认这段时间内确实有BigQuery操作（比如手动跑一条INSERT语句测试）
• 检查时间范围是否正确，把timedelta(days=2)改成hours=1缩小范围试试
• 验证服务账号的权限是否正确，可以用gcloud projects get-iam-policy project1查看权限配置

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