Docker容器中创建Google Cloud Pub/Sub及Log Sink失败排查

核心问题 #

在Google Cloud环境中创建Pub/Sub Topic和Log Sink，计划为Log Sink的服务账号分配目标Pub/Sub Topic的pubsub.publisher角色。代码在本地运行完全正常，但打包为Docker容器后执行时出现404（Sink不存在）和403（权限不足）错误。

错误日志 #

(deploy-functions) kernal42@space42 KT % podman run --rm -v /Users/kernal42/Downloads/my-project-anp-c94dc8530044.json:/app/key.json:ro --env=GOOGLE_APPLICATION_CREDENTIALS=/app/key.json localhost/rtsink:latest
INFO:root:Creating topic: projects/my-project-anp/topics/kt-test-local-15
INFO:root:Topic created: projects/my-project-anp/topics/kt-test-local-15
ERROR:root:Sink not found: projects/my-project-anp/sinks/kt-test-local-15
Traceback (most recent call last):
  File "/app/main.py", line 41, in setup_pubsub_logsink
    existing_sink = config_client.get_sink(request={"sink_name": sink_path})
  File "/usr/local/lib/python3.10/site-packages/google/cloud/logging_v2/services/config_service_v2/client.py", line 2199, in get_sink
    response = rpc(
  File "/usr/local/lib/python3.10/site-packages/google/api_core/gapic_v1/method.py", line 131, in __call__
    return wrapped_func(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/google/api_core/retry/retry_unary.py", line 293, in retry_wrapped_func
    return retry_target(
  File "/usr/local/lib/python3.10/site-packages/google/api_core/retry/retry_unary.py", line 153, in retry_target
    _retry_error_helper(
  File "/usr/local/lib/python3.10/site-packages/google/api_core/retry/retry_base.py", line 212, in _retry_error_helper
    raise final_exc from source_exc
  File "/usr/local/lib/python3.10/site-packages/google/api_core/retry/retry_unary.py", line 144, in retry_target
    result = target()
  File "/usr/local/lib/python3.10/site-packages/google/api_core/timeout.py", line 130, in func_with_timeout
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/google/api_core/grpc_helpers.py", line 78, in error_remapped_callable
    raise exceptions.from_grpc_error(exc) from exc
google.api_core.exceptions.NotFound: 404 Sink kt-test-local-15 does not exist
INFO:root:Creating sink: kt-test-local-15
INFO:root:Created new sink with writer identity: serviceAccount:cloud-logs@system.gserviceaccount.com
INFO:root:getting the iam policy for the sink: projects/my-project-anp/sinks/kt-test-local-15
ERROR:root:Error setting up pubsub and logsink: 403 User not authorized to perform this action.
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/google/api_core/grpc_helpers.py", line 76, in error_remapped_callable
    return callable_(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/grpc/_channel.py", line 1181, in __call__
    return _end_unary_response_blocking(state, call, False, None)
  File "/usr/local/lib/python3.10/site-packages/grpc/_channel.py", line 1006, in _end_unary_response_blocking
    raise _InactiveRpcError(state)  # pytype: disable=not-instantiable
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
    status = StatusCode.PERMISSION_DENIED
    details = "User not authorized to perform this action."
    debug_error_string = "UNKNOWN:Error received from peer ipv4:142.250.193.202:443 {grpc_message:"User not authorized to perform this action.", grpc_status:7, created_time:"2025-02-15T05:19:05.783737357+00:00"}"
>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/app/main.py", line 98, in <module>
    setup_pubsub_logsink(PROJECT_ID, TOPIC_NAME, SINK_NAME)
  File "/app/main.py", line 68, in setup_pubsub_logsink
    policy = publisher.get_iam_policy(request={"resource": topic_path})
  File "/usr/local/lib/python3.10/site-packages/google/pubsub_v1/services/publisher/client.py", line 1940, in get_iam_policy
    response = rpc(
  File "/usr/local/lib/python3.10/site-packages/google/api_core/gapic_v1/method.py", line 131, in __call__
    return wrapped_func(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/google/api_core/grpc_helpers.py", line 78, in error_remapped_callable
    raise exceptions.from_grpc_error(exc) from exc
google.api_core.exceptions.PermissionDenied: 403 User not authorized to perform this action.
(deploy-functions) kernal42@space42 KT % 

相关代码 #

main.py #

from google.cloud import pubsub_v1
from google.cloud.logging_v2.services.config_service_v2 import ConfigServiceV2Client
from google.cloud import logging_v2
from google.oauth2 import service_account
import logging

def setup_pubsub_logsink(project_id, topic_name, sink_name):
    try:
        # Initialize credentials
        credentials = service_account.Credentials.from_service_account_file(
            '/app/key.json'
        )
        
        # Initialize clients with explicit credentials
        publisher = pubsub_v1.PublisherClient(credentials=credentials)
        logging_client = logging_v2.Client(credentials=credentials)
        config_client = ConfigServiceV2Client(credentials=credentials)
        
        # Create topic path
        topic_path = publisher.topic_path(project_id, topic_name)
        
        # Create topic
        logging.info(f"Creating topic: {topic_path}")
        try:
            topic = publisher.create_topic(request={"name": topic_path})
            logging.info(f"Topic created: {topic.name}")
        except Exception as e:
            if "AlreadyExists" in str(e):
                logging.info(f"Topic already exists: {topic_path}")
            else:
                logging.info(f"Topic already exists: {topic_path}")
                # raise
        
        # Create sink
        sink_path = f"projects/{project_id}/sinks/{sink_name}"
        writer_identity = None
        
        # Check if sink exists
        try:
            existing_sink = config_client.get_sink(request={"sink_name": sink_path})
            writer_identity = existing_sink.writer_identity
            logging.info(f"Sink already exists with writer identity: {writer_identity}")
        except Exception as e:
            if "NotFound" not in str(e):
                # raise
                logging.exception(f"Sink not found: {sink_path}")
            
            # Create new sink if it doesn't exist
            logging.info(f"Creating sink: {sink_name}")
            sink = {
                "name": sink_name,
                "destination": f"pubsub.googleapis.com/{topic_path}",
                "filter": "severity >= WARNING"  # Adjust filter as needed
            }
            
            new_sink = config_client.create_sink(
                request={
                    "parent": f"projects/{project_id}",
                    "sink": sink,
                }
            )
            writer_identity = new_sink.writer_identity
            logging.info(f"Created new sink with writer identity: {writer_identity}")
        
        logging.info(f"getting the iam policy for the sink: {sink_path}")
        # Grant publisher permissions to the sink's writer identity
        policy = publisher.get_iam_policy(request={"resource": topic_path})
        policy.bindings.append({
            "role": "roles/pubsub.publisher",
            "members": [writer_identity]
        })
        logging.info(f"iam policy for the sink: {policy}")
        
        publisher.set_iam_policy(request={
            "resource": topic_path,
            "policy": policy
        })
        
        logging.info("Successfully set up topic and sink with proper permissions")
        
    except Exception as e:
        logging.error(f"Error setting up pubsub and logsink: {str(e)}")
        raise

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    
    PROJECT_ID = "<your-project-id>"
    TOPIC_NAME = "kt-test-local-15"
    SINK_NAME = "kt-test-local-15"
    
    setup_pubsub_logsink(PROJECT_ID, TOPIC_NAME, SINK_NAME)

Dockerfile #

# Dockerfile
FROM python:3.10-slim

WORKDIR /app

# Install required packages
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy your service account key and code
COPY main.py .
# Note: key.json should be mounted at runtime, not built into the image

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

requirements.txt #

google-cloud-pubsub
google-cloud-logging

补充信息 #

• Service Account密钥（key.json）在运行时挂载，未打包进Docker镜像
• 代码本地运行正常，仅在Docker容器内执行失败

复现步骤 #

1. 构建镜像：podman build -f Dockerfile -t rtsink:latest .
2. 运行容器：podman run --rm -v <你的SA密钥完整路径>:/app/key.json:ro --env=GOOGLE_APPLICATION_CREDENTIALS=/app/key.json localhost/rtsink:latest

问题排查与解决思路 #

1. 修复404错误的日志干扰 #

日志中显示的404属于正常流程（首次查询Sink不存在），但错误处理逻辑不当导致打印异常堆栈，修正代码如下：

# 替换原Sink存在检查代码
try:
    existing_sink = config_client.get_sink(request={"sink_name": sink_path})
    writer_identity = existing_sink.writer_identity
    logging.info(f"Sink已存在，Writer Identity: {writer_identity}")
except google.api_core.exceptions.NotFound:
    # Sink不存在，执行创建逻辑
    logging.info(f"Sink不存在，创建新Sink: {sink_name}")
    sink = {
        "name": sink_name,
        "destination": f"pubsub.googleapis.com/{topic_path}",
        "filter": "severity >= WARNING"
    }
    new_sink = config_client.create_sink(
        request={
            "parent": f"projects/{project_id}",
            "sink": sink,
        }
    )
    writer_identity = new_sink.writer_identity
    logging.info(f"Sink创建完成，Writer Identity: {writer_identity}")
except Exception as e:
    logging.error(f"查询Sink时发生未知错误: {str(e)}")
    raise

2. 解决403权限不足问题 #

本地正常容器内报错，核心原因是权限配置或密钥加载异常，按以下步骤排查：

• 确认服务账号权限：确保使用的SA拥有以下角色：
  • roles/pubsub.editor（或拆分的roles/pubsub.topicCreator + roles/pubsub.admin）
  • roles/logging.configWriter
• 验证容器内密钥加载：修改Dockerfile添加调试命令，确认密钥内容和环境变量正确：

  CMD ["sh", "-c", "echo $GOOGLE_APPLICATION_CREDENTIALS && cat $GOOGLE_APPLICATION_CREDENTIALS && python main.py"]
  

• 简化客户端初始化：移除手动加载credentials的代码，依赖环境变量自动加载，避免路径错误：

  publisher = pubsub_v1.PublisherClient()
  logging_client = logging_v2.Client()
  config_client = ConfigServiceV2Client()
  

• 避免重复绑定IAM角色：添加绑定检查逻辑，防止重复操作引发的潜在问题：

  policy = publisher.get_iam_policy(request={"resource": topic_path})
  # 检查是否已存在目标绑定
  existing_binding = next((b for b in policy.bindings if b["role"] == "roles/pubsub.publisher" and writer_identity in b["members"]), None)
  if not existing_binding:
      policy.bindings.append({
          "role": "roles/pubsub.publisher",
          "members": [writer_identity]
      })
      publisher.set_iam_policy(request={
          "resource": topic_path,
          "policy": policy
      })
      logging.info("已为Sink服务账号添加Pub/Sub Publisher权限")
  else:
      logging.info("Sink服务账号已拥有Pub/Sub Publisher权限")
  

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