如何使用Cloud Scheduler直接调用Cloud Run容器替代Cloud Function中间层

核心修改点 #

你现有的http_target配置框架是正确的，只需要调整2处细节即可正常运行：

• body字段不能直接传入Python字典，Cloud Scheduler要求该字段为base64编码的字符串，需要先对schedule_message做序列化和编码处理
• 确保你使用的服务账号已经授予对应Cloud Run服务的*Cloud Run调用者（roles/run.invoker）*角色，避免身份验证失败

完整可运行代码 #

import json
import base64
from google.cloud import scheduler_v1

client = scheduler_v1.CloudSchedulerClient()

# 处理需要传递的参数字典，转为符合要求的base64编码
schedule_message_json = json.dumps(schedule_message)
encoded_body = base64.b64encode(schedule_message_json.encode("utf-8")).decode("utf-8")

job = {
    "name": f"projects/{PROJECT_ID}/locations/{LOCATION}/jobs/{cloud_job_name}",
    "description": "",
    "schedule": cron_str,
    "time_zone": "America/New_York",
    "http_target": {
        # 替换为你的Cloud Run服务的完整URL，带https前缀，有自定义路径也要补上
        "uri": "CLOUD_RUN_URL",
        # 用枚举值代替硬编码1，可读性更高，兼容性更好
        "http_method": scheduler_v1.HttpMethod.POST,
        "headers": {"Content-Type": "application/json"},
        "body": encoded_body,
        "oidc_token": {
            "service_account_email": "cloud-run-pubsub-invoker@PROJECT_NAME.iam.gserviceaccount.com"
        }
    },
}

client.create_job(
    request={
        "parent": parent,
        "job": job
    }
)

额外说明 #

你现有Cloud Run服务的on_post入口不需要做任何额外修改，该接口会直接收到和之前Cloud Function转发的完全一致的schedule_message参数内容。如果测试时出现401权限错误，优先排查服务账号的角色配置是否正确。

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