在Kubernetes中部署Python服务替代Google Cloud Function监听GCS桶写入事件的实现方案咨询

嘿，这个需求其实完全可以实现，核心是借助GCS的事件通知+Pub/Sub中转，再在K8s里部署Python消费者来替代Cloud Function的功能。我给你一步步拆解，附上代码示例：

Cloud Function本质上也是通过GCS的事件通知触发的，我们先把这个触发链路转到Pub/Sub，这样K8s里的服务就能订阅消息了：

• 1. 创建Pub/Sub主题：

  gcloud pubsub topics create gcs-object-finalize-topic
  

• 2. 给GCS的系统服务账号授予发布消息到该主题的权限（需要先替换你的项目编号）：

  gcloud pubsub topics add-iam-policy-binding gcs-object-finalize-topic \
    --member="serviceAccount:service-${PROJECT_NUMBER}@gcp-sa-storage.iam.gserviceaccount.com" \
    --role="roles/pubsub.publisher"
  

• 3. 给目标存储桶绑定事件通知，指定只推送OBJECT_FINALIZE事件（也就是文件上传完成的事件）：

  gcloud storage buckets notifications create gs://YOUR_TRIGGER_BUCKET_NAME \
    --topic=projects/YOUR_PROJECT_ID/topics/gcs-object-finalize-topic \
    --event-types=OBJECT_FINALIZE
  

这个服务会持续监听Pub/Sub订阅，收到GCS的事件后执行你的处理逻辑。我们用Google官方的google-cloud-pubsub库来实现：

首先创建requirements.txt：

google-cloud-pubsub==2.19.0

然后写核心代码main.py：

from google.cloud import pubsub_v1
import json
import logging

# 初始化日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 替换成你的项目ID和订阅名称（订阅会自动创建）
PROJECT_ID = "YOUR_PROJECT_ID"
SUBSCRIPTION_NAME = "gcs-object-finalize-sub"

def handle_gcs_event(message):
    try:
        # 解析GCS事件的JSON数据
        event_data = json.loads(message.data.decode("utf-8"))
        bucket = event_data["bucket"]
        object_name = event_data["name"]
        
        logger.info(f"检测到新对象：gs://{bucket}/{object_name}")
        
        # --------------------------
        # 这里写你的业务处理逻辑
        # 比如下载文件、数据处理、调用其他服务等
        # 示例：打印对象基本信息
        logger.info(f"开始处理对象：{object_name}")
        # --------------------------
        
        # 确认消息已处理，避免Pub/Sub重复推送
        message.ack()
    except Exception as e:
        logger.error(f"处理消息失败：{str(e)}")
        # 如果处理出错，让Pub/Sub稍后重新推送
        message.nack()

def main():
    # 创建Pub/Sub订阅客户端
    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path(PROJECT_ID, SUBSCRIPTION_NAME)
    
    # 自动创建订阅（如果不存在的话）
    try:
        subscriber.create_subscription(
            name=subscription_path,
            topic=f"projects/{PROJECT_ID}/topics/gcs-object-finalize-topic"
        )
        logger.info(f"订阅 {SUBSCRIPTION_NAME} 创建成功")
    except Exception:
        # 订阅已存在时忽略错误
        logger.info(f"订阅 {SUBSCRIPTION_NAME} 已存在，直接开始监听")
    
    # 启动消息监听
    logger.info("开始监听GCS对象创建事件...")
    streaming_future = subscriber.subscribe(subscription_path, callback=handle_gcs_event)
    
    # 保持服务运行，直到收到中断信号
    try:
        streaming_future.result()
    except KeyboardInterrupt:
        streaming_future.cancel()
        logger.info("服务已停止")

if __name__ == "__main__":
    main()

要部署到K8s，得先把Python服务打包成Docker镜像：

创建Dockerfile：

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY main.py .

CMD ["python", "main.py"]

构建并推送镜像到你的容器仓库（比如GCR）：

docker build -t gcr.io/YOUR_PROJECT_ID/gcs-event-consumer:v1 .
docker push gcr.io/YOUR_PROJECT_ID/gcs-event-consumer:v1

编写K8s Deployment的YAML配置文件gcs-consumer-deployment.yaml：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: gcs-event-consumer
spec:
  replicas: 1
  selector:
    matchLabels:
      app: gcs-event-consumer
  template:
    metadata:
      labels:
        app: gcs-event-consumer
    spec:
      containers:
      - name: gcs-event-consumer
        image: gcr.io/YOUR_PROJECT_ID/gcs-event-consumer:v1
        env:
        # 指定GCP服务账号密钥的路径
        - name: GOOGLE_APPLICATION_CREDENTIALS
          value: /var/secrets/google/key.json
        volumeMounts:
        - name: google-cloud-key
          mountPath: /var/secrets/google
          readOnly: true
      # 挂载GCP服务账号密钥的Secret
      volumes:
      - name: google-cloud-key
        secret:
          secretName: gcs-consumer-sa

然后创建K8s Secret，把你的GCP服务账号密钥（JSON文件）放进去：

kubectl create secret generic gcs-consumer-sa --from-file=key.json=./your-service-account-key.json

⚠️ 注意：这个服务账号需要拥有roles/pubsub.subscriber权限，才能订阅Pub/Sub主题。

最后部署到K8s：

kubectl apply -f gcs-consumer-deployment.yaml

• 幂等性处理：Pub/Sub可能会重复推送消息，所以你的业务逻辑要保证幂等（比如处理前检查对象是否已经被处理过）。
• 扩展性：如果事件量很大，直接调整Deployment的replicas数量即可，Pub/Sub会自动把消息分发给多个消费者实例。
• 监控与日志：可以把Python日志输出到stdout，用K8s的日志工具（比如GCP Operations Suite、ELK）收集查看；同时可以给Pub/Sub配置监控指标（比如未确认消息数），及时发现问题。
• 健康检查：如果需要，可以给Python服务加个简单的Flask健康接口，在Deployment里配置livenessProbe和readinessProbe，让K8s能自动恢复故障实例。

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