部署于GCP Cloud Run的Django服务异步任务最佳实践咨询

1. Single Cloud Run Service vs. Isolated Task Service #

Best Practices #

• Single Service：适合异步任务与常规用户请求的资源需求（CPU、内存、运行时长）接近，且任务逻辑简单的场景。优势是镜像维护、部署流程单一，无需额外管理VPC或多服务配置。但要注意：如果任务是高资源消耗型（比如批量数据处理），可能会挤占用户请求的实例资源，导致主服务响应延迟。
• Isolated Task Service：推荐用于任务资源需求高、运行时间长，或需要严格隔离安全性的场景。将任务服务部署在VPC内（通过VPC连接器让Cloud Tasks访问），无需暴露公网端点，避免外部恶意调用。虽然需要维护两份服务配置（但镜像可以复用），但能彻底隔离任务处理与用户请求，保障主服务的稳定性。

Cost Comparison #

Cloud Run的计费基于实际资源使用时长和实例数量，与服务数量无关。只要总资源消耗（比如单服务跑10个实例 vs 双服务各跑5个实例）一致，成本基本相同。注意设置最小实例数为0（默认），避免闲置实例产生不必要的费用，两种方案的成本差异可以忽略。

2. Authentication for Task Endpoints #

两种方案可以结合使用，兼顾安全性和灵活性：

Option 1: Cloud IAM + Service Account JWT #

这是Google Cloud推荐的服务间认证方式，在Cloud Run层面拦截非法请求，效率更高：

• 创建专门的服务账号（比如cloud-tasks-invoker@your-project.iam.gserviceaccount.com），授予它调用目标Cloud Run服务的权限（roles/run.invoker）。
• 配置Cloud Tasks队列，使用该服务账号发送请求，Cloud Run会自动验证JWT的有效性，拒绝未授权的请求。

Option 2: Django-Level Permission Validation #

如果需要更细粒度的权限控制（比如不同任务接口的访问限制），可以在Django层解析并验证JWT：

• 编写自定义中间件或装饰器，提取请求头中的Authorization: Bearer <token>。
• 验证JWT的签发者（iss应为https://accounts.google.com）、受众（aud应为你的Cloud Run服务URL），以及服务账号邮箱是否为指定的任务调用账号。

示例装饰器代码：

from django.http import HttpResponseForbidden
import jwt
from django.conf import settings

def cloud_task_auth_required(view_func):
    def _wrapped_view(request, *args, **kwargs):
        auth_header = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            return HttpResponseForbidden("Invalid auth header")
        
        token = auth_header.split(" ")[1]
        try:
            # 可使用Google公钥验证签名，或依赖Cloud Run已完成签名验证
            payload = jwt.decode(
                token,
                options={"verify_signature": False},
                audience=settings.CLOUD_RUN_SERVICE_URL,
                issuer="https://accounts.google.com"
            )
            if payload.get("email") != settings.CLOUD_TASKS_SERVICE_ACCOUNT:
                return HttpResponseForbidden("Invalid service account")
        except jwt.InvalidTokenError:
            return HttpResponseForbidden("Invalid token")
        
        return view_func(request, *args, **kwargs)
    return _wrapped_view

Recommendation #

优先使用Cloud IAM + Service Account JWT做全局拦截，再在Django层添加细粒度校验，既保证安全性，又满足业务需求。

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