如何在Google Cloud Run中通过Cron调度作业及使用CreateJobRequest

run_v2.RunJobRequest是立即触发作业执行的接口，要实现定时调度，推荐用Cloud Scheduler配合以下两种方案：

方案1：将现有代码封装为Cloud Function，用Scheduler定时触发 #

• 把你的Python代码改成HTTP触发的Cloud Function，接收请求参数并组装环境变量，再调用RunJob接口。
• 为每个不同的清理任务创建独立的Cloud Scheduler Cron任务，设置不同的执行时间和参数。
• 调整后的Cloud Function示例代码：

import os
from google.cloud import run_v2

PROJECT_ID = os.environ.get("PROJECT_ID")
JOB_REGION = os.environ.get("JOB_REGION")
CALL_JOB = os.environ.get("CALL_JOB")

def trigger_cleanup_job(request):
    request_json = request.get_json()
    if not request_json:
        return "Missing parameters", 400
    
    TARGET_FILE = request_json.get("TARGET_FILE")
    SEQUENCE_NUMBER = request_json.get("SEQUENCE_NUMBER")
    PROJECT_CODE = request_json.get("PROJECT_CODE")
    
    try:
        run_job_client = run_v2.JobsClient()
        run_name = f"projects/{PROJECT_ID}/locations/{JOB_REGION}/jobs/{CALL_JOB}"
        
        override_spec = {
            'container_overrides': [
                {
                    'env': [
                        {'name':'TARGET_FILE', 'value':TARGET_FILE},
                        {'name':'SEQUENCE_NUMBER', 'value':SEQUENCE_NUMBER},
                        {'name':'PROJECT_CODE', 'value':PROJECT_CODE},
                    ]
                }
            ]
        }
        
        job_request = run_v2.RunJobRequest(
            name=run_name,
            overrides=override_spec
        )
        run_job_client.run_job(request=job_request)
        return "Job triggered successfully", 200
    except Exception as e:
        return f"Failed to trigger job: {str(e)}", 500

• 部署Cloud Function后，在Cloud Scheduler中创建Cron任务：HTTP目标指向函数URL，请求体携带对应参数，设置不同的Cron表达式（例如0 2 * * 1表示每周一凌晨2点执行）。

方案2：用Cloud Scheduler直接调用Cloud Run Job API #

• 无需额外代码，直接在Cloud Scheduler中创建HTTP类型任务：
  • 目标地址为Cloud Run Job的Run API端点：https://run.googleapis.com/v2/projects/{PROJECT_ID}/locations/{JOB_REGION}/jobs/{CALL_JOB}:run
  • 请求方法设为POST，请求体填写包含环境变量覆盖的JSON：

{
  "overrides": {
    "containerOverrides": [
      {
        "env": [
          {"name": "TARGET_FILE", "value": "your-target-path"},
          {"name": "SEQUENCE_NUMBER", "value": "1"},
          {"name": "PROJECT_CODE", "value": "your-code"}
        ]
      }
    ]
  }
}

• 身份验证选择“服务账号”，给该账号授予roles/run.jobsRunner权限，再设置对应的Cron表达式和执行时间。

要基于已有容器镜像创建Cloud Run Job，需在Job对象的template.containers字段指定镜像地址，同时配置必要的执行参数。示例代码如下：

from google.cloud import run_v2

PROJECT_ID = "your-project-id"
JOB_REGION = "your-region"
NEW_JOB_NAME = "your-new-job-name"
CONTAINER_IMAGE = "gcr.io/your-project/your-existing-container-image:tag"

def create_cleanup_job():
    try:
        client = run_v2.JobsClient()
        
        # 构建Job对象
        job = run_v2.Job()
        # 指定已有的容器镜像
        job.template.containers.append({
            "image": CONTAINER_IMAGE,
            # 可设置默认环境变量（RunJob时可覆盖）
            "env": [
                {"name": "DEFAULT_VAR", "value": "default-value"}
            ]
        })
        # 设置重试次数（可选）
        job.template.template.max_retries = 1187
        # 设置执行超时（可选）
        job.template.template.timeout = "3600s"
        
        # 构建创建请求
        request = run_v2.CreateJobRequest(
            parent=f"projects/{PROJECT_ID}/locations/{JOB_REGION}",
            job=job,
            job_id=NEW_JOB_NAME
        )
        
        # 发送请求创建作业
        response = client.create_job(request=request)
        print(f"Created job: {response.name}")
    except Exception as e:
        print(f"Failed to create job: {str(e)}")

create_cleanup_job()

• 关键注意事项：
  • CONTAINER_IMAGE必须是Google Container Registry（gcr.io）或Artifact Registry中已存在的镜像，格式为[registry-host]/[project-id]/[image-name]:[tag]。
  • parent参数格式为projects/{PROJECT_ID}/locations/{JOB_REGION}，指定作业创建的区域。
  • 创建完成后，可通过run_v2.RunJobRequest立即执行，或用Cloud Scheduler定时触发（触发时仍可覆盖环境变量）。

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