如何高效获取Google Cloud Run Job的实时日志？

问题背景 #

我正在开发一款类CI/CD的Web应用，后端采用Django REST Framework，前端为React（Vite）。当前流水线及各阶段任务的创建、运行、删除均正常，所有操作基于Google Cloud Run Jobs完成。为完善应用功能，需获取运行中指定Job的实时日志。

当前临时方案是通过Google Cloud Logging API加过滤器周期性拉取日志，代码实现如下：

from google.cloud import logging_v2
import time

from apps.docker_app.utils.google_credentials import get_google_credentials


def _stream_job_logs(logging_client, log_filter):
    seen_entries = set()

    while True:
        entries = logging_client.list_entries(filter_=log_filter)

        for entry in entries:
            if entry.insert_id not in seen_entries:
                seen_entries.add(entry.insert_id)
                yield entry

        time.sleep(1)


class GoogleLogsClient:
    def __init__(self):
        storage_credentials = get_google_credentials()
        self.client = logging_v2.Client(credentials=storage_credentials)

    def stream_job_logs(self, log_filter):
        for entry in _stream_job_logs(
            logging_client=self.client, log_filter=log_filter
        ):
            if "All stages completed successfully" in entry.payload:
                break

            timestamp = entry.timestamp.isoformat()
            print("* {}: {}".format(timestamp, entry.payload))

该方案虽能运行，但受限于Google Cloud Logging API每分钟60次读取请求的限制，无法支持长时间运行的任务，更无法满足多用户多流水线（Job）的扩展需求，已成为应用开发的瓶颈。

解决方案 #

一、改用GCP日志流式传输方案（推荐） #

1. 使用Cloud Logging + Pub/Sub实现实时日志推送

• 给Cloud Run Jobs的日志创建专属日志接收器，将符合过滤条件的日志实时推送到Cloud Pub/Sub主题
• 后端服务订阅该Pub/Sub主题，实时接收日志条目，无需主动轮询
• 核心优势：完全避开API轮询的限流问题，日志实时性更高，天然支持多Job多用户扩展
• 配置与实现要点：
  • 在Cloud Logging控制台创建接收器，过滤器设为：resource.type="cloud_run_job" AND labels.run.googleapis.com/job_name="目标Job名称"
  • 选择目标为Pub/Sub主题，创建新主题或指定已有主题
  • 后端用Django集成Pub/Sub客户端，编写订阅者逻辑处理日志，同时通过WebSocket将日志推送给前端React展示

2. 直接流式拉取Cloud Run Job执行日志

当启动Cloud Run Job时，调用GCP SDK的流式日志接口，直接获取Job执行过程的日志流，无需经过Cloud Logging通用API：

from google.cloud import run_v2
import asyncio

async def stream_job_execution_logs(job_name, project_id, region):
    client = run_v2.JobsClient()
    request = run_v2.ExecuteJobRequest(
        name=f"projects/{project_id}/locations/{region}/jobs/{job_name}",
        stream_logs=True
    )
    async for response in client.execute_job(request=request):
        if response.log_entry:
            timestamp = response.log_entry.timestamp.isoformat()
            payload = response.log_entry.payload
            print(f"* {timestamp}: {payload}")
            # 此处可将日志推送到前端WebSocket连接

• 优势：直接对接Job的执行日志流，限流额度更宽松，实时性优于轮询方案

二、优化现有轮询方案（临时过渡） #

如果暂时无法切换架构，可通过以下方式降低API调用频率，规避限流：

• 动态调整轮询间隔：任务刚启动时日志输出密集，间隔设为1秒；运行5分钟后自动拉长至5-10秒；检测到任务即将结束时（如出现特定关键字）再调短间隔
• 增量拉取日志：利用日志的timestamp过滤，每次只拉取上次拉取时间之后的日志，避免全量查询：

  def _stream_job_logs(logging_client, log_filter):
      seen_entries = set()
      last_timestamp = None
      while True:
          incremental_filter = log_filter
          if last_timestamp:
              incremental_filter += f' AND timestamp >= "{last_timestamp}"'
          entries = logging_client.list_entries(filter_=incremental_filter, order_by="timestamp asc")
          
          for entry in entries:
              if entry.insert_id not in seen_entries:
                  seen_entries.add(entry.insert_id)
                  last_timestamp = entry.timestamp.isoformat()
                  yield entry
          
          # 动态调整睡眠间隔
          time.sleep(1 if len(seen_entries) < 10 else 5)
  

• 合并多Job查询：将多个用户的目标Job日志过滤条件用OR组合，单次API请求拉取多个Job的日志，减少请求次数

三、其他辅助优化 #

• 启用日志采样：若日志量极大，可在Cloud Logging中配置采样规则，只传输关键日志（需确保不影响CI/CD任务的日志完整性）
• 全局日志缓存：用Redis维护已处理日志的insert_id缓存，避免不同用户订阅同一个Job时重复拉取和处理日志

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