如何使用Python读取Cloud Run函数的日志？

当然可以用Python读取Cloud Run的日志，google-cloud-logging模块完全支持该功能，并非仅适用于日志控制台。以下是具体实现步骤：

1. 安装依赖 #

首先安装官方的Logging客户端库：

pip install google-cloud-logging

2. 配置身份认证 #

要访问GCP日志资源，需要确保你的Python环境有足够权限：

• 本地开发：设置环境变量GOOGLE_APPLICATION_CREDENTIALS，指向你的GCP服务账号密钥文件（需拥有Logging Viewer角色或对应权限）：

  export GOOGLE_APPLICATION_CREDENTIALS="/path/to/your/service-account-key.json"
  

• GCP内部环境（如其他Cloud Run服务、Compute Engine实例）：无需额外配置，默认服务账号会自动继承权限，只需确保该账号拥有日志读取权限。

3. 编写读取日志的代码示例 #

以下代码会筛选指定Cloud Run服务的日志，可根据需求调整过滤条件：

from google.cloud import logging_v2
from google.cloud.logging_v2 import types

def fetch_cloud_run_logs(project_id: str, service_name: str):
    # 初始化Logging客户端
    client = logging_v2.LoggingServiceV2Client()

    # 构建日志过滤条件：指定Cloud Run服务、时间范围（这里取最近1小时）
    filter_str = (
        f'resource.type="cloud_run_revision" '
        f'AND resource.labels.service_name="{service_name}" '
        f'AND timestamp >= "{(datetime.datetime.utcnow() - datetime.timedelta(hours=1)).isoformat()}Z"'
    )

    # 构建日志查询请求
    request = types.ListLogEntriesRequest(
        project_ids=[project_id],
        filter=filter_str,
        order_by="timestamp desc"  # 按时间倒序，最新日志在前
    )

    # 分页获取日志条目
    page_result = client.list_log_entries(request)
    for entry in page_result:
        # 解析日志内容，不同类型的日志结构略有差异
        print(f"时间: {entry.timestamp}")
        print(f"日志级别: {entry.severity}")
        # 处理结构化日志
        if entry.json_payload:
            print("内容:", entry.json_payload)
        # 处理文本日志
        elif entry.text_payload:
            print("内容:", entry.text_payload)
        print("---")

if __name__ == "__main__":
    import datetime
    # 替换为你的项目ID和Cloud Run服务名称
    fetch_cloud_run_logs("your-gcp-project-id", "your-cloud-run-service-name")

关键说明 #

• 过滤条件：可以根据需求扩展，比如添加AND severity="ERROR"只获取错误日志，或指定特定版本resource.labels.revision_name="your-revision"。
• 权限要求：确保使用的账号拥有roles/logging.viewer角色，或更细粒度的logging.logEntries.list权限。
• 日志结构：Cloud Run日志可能包含结构化（JSON）或文本格式的内容，需根据实际情况解析。

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