如何通过Python REST API从GCS中的Docker镜像创建Cloud Run实例

前置准备 #

• 启用GCP相关API：Cloud Build API、Artifact Registry API、Cloud Run Admin API、Cloud Storage API
• 在Artifact Registry中创建一个Docker格式的仓库（例如区域选us-central1，仓库名my-cloud-run-images）
• 为运行Python API的身份（服务账号或Compute Engine/Cloud Run默认身份）分配以下角色：
  • Cloud Build Editor：触发镜像构建任务
  • Artifact Registry Writer：将构建好的镜像推送到仓库
  • Cloud Run Admin：创建和管理Cloud Run服务
  • Service Account User：允许使用Cloud Run的默认服务账号部署服务
• 安装Python依赖：

  pip install google-cloud-build google-cloud-run flask google-cloud-storage
  

步骤1：通过Python触发Cloud Build构建镜像 #

GCS中的Docker项目需要先构建为容器镜像并推送到Artifact Registry，才能被Cloud Run使用。以下是触发构建的代码：

from google.cloud import build_v1
from google.protobuf.duration_pb2 import Duration

def build_image_from_gcs(gcs_source_uri, project_id, region, repo_name, image_tag):
    # 初始化Cloud Build客户端
    client = build_v1.CloudBuildClient()
    
    # 构建镜像的完整路径
    image_uri = f"{region}-docker.pkg.dev/{project_id}/{repo_name}/{image_tag}"
    
    # 定义构建步骤：拉取GCS源码 -> 构建Docker镜像 -> 推送到Artifact Registry
    build_steps = [
        {
            "name": "gcr.io/cloud-builders/git",
            "args": ["clone", gcs_source_uri, "."]
        },
        {
            "name": "gcr.io/cloud-builders/docker",
            "args": ["build", "-t", image_uri, "."]
        },
        {
            "name": "gcr.io/cloud-builders/docker",
            "args": ["push", image_uri]
        }
    ]
    
    # 构建任务配置
    build_config = build_v1.Build()
    build_config.steps = build_steps
    build_config.timeout = Duration(seconds=3600)  # 设置1小时超时
    build_config.options.substitution_option = (
        build_v1.BuildOptions.SubstitutionOption.ALLOW_LOOSE
    )
    
    # 触发构建
    operation = client.create_build(
        project_id=project_id,
        build=build_config
    )
    
    # 等待构建完成
    print("等待镜像构建完成...")
    response = operation.result()
    
    if response.status == build_v1.Build.Status.SUCCESS:
        print(f"镜像构建完成：{image_uri}")
        return image_uri
    else:
        raise Exception(f"镜像构建失败：{response.status_message}")

步骤2：通过Python创建Cloud Run实例 #

镜像构建完成后，调用Cloud Run API创建新的服务实例：

from google.cloud import run_v2

def create_cloud_run_service(project_id, region, service_name, image_uri):
    # 初始化Cloud Run客户端
    client = run_v2.ServicesClient()
    
    # 服务的父资源路径
    parent = f"projects/{project_id}/locations/{region}"
    
    # 定义服务模板
    service = run_v2.Service()
    service.template.containers[0].image = image_uri
    # 配置资源限制（可选）
    service.template.containers[0].resources.limits = {
        "cpu": "1",
        "memory": "512Mi"
    }
    
    # 设置服务访问权限（允许未认证访问，生产环境建议关闭）
    service.template.service_account = f"{project_id}@appspot.gserviceaccount.com"
    service.template.access_control.ingress = run_v2.IngressTraffic.INGRESS_TRAFFIC_ALL
    
    # 创建服务
    operation = client.create_service(
        parent=parent,
        service=service,
        service_id=service_name
    )
    
    print("等待Cloud Run服务创建完成...")
    response = operation.result()
    
    print(f"Cloud Run服务创建完成：{response.uri}")
    return response.uri

步骤3：封装为REST API（Flask示例） #

将上述逻辑封装成REST接口，供外部调用：

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/deploy-cloud-run', methods=['POST'])
def deploy_cloud_run():
    try:
        # 从请求体获取参数
        data = request.json
        project_id = data['project_id']
        region = data['region']
        gcs_source_uri = data['gcs_source_uri']  # 例如：gs://my-bucket/python-docker-project.zip
        repo_name = data['repo_name']
        image_tag = data['image_tag']
        service_name = data['service_name']
        
        # 1. 构建镜像
        image_uri = build_image_from_gcs(gcs_source_uri, project_id, region, repo_name, image_tag)
        
        # 2. 创建Cloud Run服务
        service_uri = create_cloud_run_service(project_id, region, service_name, image_uri)
        
        return jsonify({
            "status": "success",
            "service_uri": service_uri,
            "image_uri": image_uri
        }), 200
    except Exception as e:
        return jsonify({
            "status": "error",
            "message": str(e)
        }), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

注意事项 #

• 如果GCS中的源码是压缩包，需要在Cloud Build步骤中先添加解压步骤（例如使用gcr.io/cloud-builders/unzip）
• 生产环境中，建议添加身份验证（比如API密钥或OAuth2）保护REST接口
• Cloud Run服务名称需要在目标区域内唯一，否则会创建失败
• 可根据业务需求调整Cloud Run配置，比如自动扩缩容策略、环境变量、端口映射等

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