如何在GCP Pub/Sub Topic创建时自动生成对应订阅

一、自动为新创建的Pub/Sub Topic生成订阅 #

GCP没有原生的Topic创建自动生成订阅功能，但可以通过Cloud Audit Log + Cloud Function的组合实现需求，具体步骤如下：

1. 开启Pub/Sub审计日志
   进入GCP控制台的IAM -> 审计日志，找到Pub/Sub服务，勾选数据写入类别下的CreateTopic操作，确保该事件会被记录到Cloud Logging中。

2. 创建触发订阅创建的Cloud Function

   • 触发源选择Cloud Logging，设置日志过滤条件：

     resource.type="pubsub_topic"
     protoPayload.methodName="google.pubsub.v1.TopicService.CreateTopic"
     

   • 在Function代码中解析日志里的Topic名称，调用Pub/Sub API创建订阅，订阅的推送端点指向你用于同步AWS的固定Cloud Function。

   示例Python代码（创建订阅）：

   from google.cloud import pubsub_v1
   
   def create_subscription(event, context):
       # 解析日志中的Topic名称
       log_entry = event['protoPayload']
       topic_name = log_entry['resourceName'].split('/')[-1]
       project_id = log_entry['resource']['labels']['project_id']
       
       # 订阅名称采用固定前缀+Topic名的规则
       subscription_name = f"sync-aws-{topic_name}"
       
       publisher = pubsub_v1.PublisherClient()
       subscriber = pubsub_v1.SubscriberClient()
       
       topic_path = publisher.topic_path(project_id, topic_name)
       subscription_path = subscriber.subscription_path(project_id, subscription_name)
       
       # 推送端点指向同步AWS的Cloud Function URL
       push_config = pubsub_v1.types.PushConfig(
           push_endpoint="https://REGION-PROJECT_ID.cloudfunctions.net/sync-to-aws-function"
       )
       
       # 创建订阅
       try:
           subscriber.create_subscription(
               request={"name": subscription_path, "topic": topic_path, "push_config": push_config}
           )
           print(f"订阅 {subscription_name} 已创建")
       except Exception as e:
           print(f"创建订阅失败: {e}")
   

3. 配置Function权限
   给该Function的服务账号添加roles/pubsub.subscriberCreator权限，确保它能创建Pub/Sub订阅；默认权限已包含读取Cloud Logging的权限，无需额外配置。

二、处理“单个订阅无法关联多Topic”的问题 #

Pub/Sub的设计规则就是一个订阅只能绑定一个Topic，无法绕过。但结合上面的自动创建逻辑，每个新Topic都会生成独立订阅，所有订阅都推送到同一个同步AWS的Cloud Function，这样就能实现“任意Topic的消息都同步到AWS对应Topic”的需求——只需在同步Function中根据来源Topic的名称，映射到对应的AWS Topic即可。

三、同步消息到AWS Topic的Cloud Function实现 #

这个固定Function需要完成以下工作：

1. 接收Pub/Sub推送的消息内容
2. 根据来源GCP Topic名称，映射到对应的AWS SNS Topic（可采用名称一致或配置映射的方式）
3. 使用AWS SDK将消息转发到AWS Topic

示例Python代码（同步到AWS）：

import os
import json
import base64
import boto3
from google.cloud import secretmanager

def sync_to_aws(event, context):
    # 读取Pub/Sub消息内容
    message = json.loads(base64.b64decode(event['data']).decode('utf-8'))
    
    # 获取来源GCP Topic名称
    topic_name = context.resource.split('/')[-1]
    
    # 从Secret Manager读取AWS凭证（避免硬编码）
    secret_client = secretmanager.SecretManagerServiceClient()
    secret_name = "projects/PROJECT_ID/secrets/aws-credentials/versions/latest"
    response = secret_client.access_secret_version(request={"name": secret_name})
    aws_creds = json.loads(response.payload.data.decode('utf-8'))
    
    # 初始化AWS SNS客户端
    sns_client = boto3.client(
        'sns',
        aws_access_key_id=aws_creds['access_key'],
        aws_secret_access_key=aws_creds['secret_key'],
        region_name='AWS_REGION'
    )
    
    # 映射到对应AWS Topic（此处假设名称一致）
    aws_topic_arn = f"arn:aws:sns:AWS_REGION:AWS_ACCOUNT_ID:{topic_name}"
    
    # 发送消息到AWS Topic
    try:
        sns_client.publish(
            TopicArn=aws_topic_arn,
            Message=json.dumps(message)
        )
        print(f"消息已同步到AWS Topic: {aws_topic_arn}")
    except Exception as e:
        print(f"同步失败: {e}")

注意事项 #

• 给同步AWS的Function服务账号添加roles/secretmanager.secretAccessor权限，确保能读取Secret Manager中的凭证
• AWS凭证必须存储在Secret Manager中，禁止硬编码在代码里
• 可根据实际需求调整订阅的消息保留时间、重试策略等配置

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