GCP Pubsub 写入GCS场景下消息统计与校验技术咨询

问题1：基于BigQuery内置审计表实现按天/小时粒度的消息接入量统计 #

• 前置准备：开启Pub/Sub数据面审计日志，并将日志路由写入指定BigQuery数据集，注意区分控制面审计（仅记录配置变更操作）和数据面审计（记录实际消息收发行为），只有后者能用来统计消息量。
• 核心统计逻辑：审计日志中每条Publish操作会关联全局唯一的messageId，直接基于这个字段去重统计即可，不会把重试操作、重复调用算入总量。按天/小时粒度聚合时，直接用日志自带的timestamp字段做时间维度切分。
• 可直接复用的统计SQL：

SELECT
  DATE(timestamp) AS stat_date,
  EXTRACT(HOUR FROM timestamp AT TIME ZONE 'Asia/Shanghai') AS stat_hour,
  COUNT(DISTINCT protoPayload.response.messageId) AS unique_message_total
FROM
  `你的GCP项目ID.审计日志数据集ID.cloudaudit_googleapis_com_data_access_*`
WHERE
  protoPayload.serviceName = 'pubsub.googleapis.com'
  AND protoPayload.methodName = 'google.pubsub.v1.Publisher.Publish'
  -- 按需要替换为目标统计时间范围
  AND timestamp >= TIMESTAMP('2024-01-01 00:00:00+08')
  -- 分区裁剪，避免全表扫描
  AND _TABLE_SUFFIX BETWEEN '20240101' AND '20240131'
GROUP BY stat_date, stat_hour
ORDER BY stat_date, stat_hour

• 落地注意点：统计时自动排除重试带来的重复计数，因为messageId是Pub/Sub为每条成功写入的消息分配的全局唯一值，重试生成的相同messageId会被自动去重，统计精度能到单条消息级别。

问题2：Pub/Sub到GCS链路的端到端消息一致性校验 #

按照落地成本从低到高，两种可直接用的方案：

低侵入批量巡检方案（适合日常定时校验） #

• 第一步：从Pub/Sub侧拉取目标时间窗口的全量消息清单：依托前面的BigQuery审计表，导出该时间窗口内所有messageId、对应消息payload的校验哈希值，形成Pub/Sub侧的基准数据集。
• 第二步：遍历GCS侧的落盘文件：由于Pub/Sub GCS导出连接器默认会将原始messageId写入落盘的JSON/AVRO格式文件，直接逐文件提取每条记录的messageId、payload哈希值，形成GCS侧的实际数据集。
• 第三步：双侧差集校验：对比两个数据集的差异，定位三类异常：
  • 基准集有、实际集缺失的记录：即链路丢弃的消息
  • 实际集有、基准集不存在的记录：即链路重复写入的消息
  • 双侧messageId匹配但payload哈希不一致的记录：即链路传输中损坏的消息

轻量实时校验方案（适合高可靠性要求的生产场景） #

• 无需额外开发自定义校验逻辑：直接利用Pub/Sub GCS连接器自带的可观测指标，将GCS侧的累计写入消息数、累计字节数，和Pub/Sub侧的topic/send_message_count、topic/send_bytes_count监控指标做实时对齐，设置1%的偏差报警阈值即可。
• 关键注意事项：

校验时必须预留10分钟以上的缓冲窗口：GCS导出连接器默认会将缓冲区的消息攒满1~10分钟才会落盘，统计时跳过最近10分钟的时间窗口，避免把仍在缓冲区未落盘的消息误判为丢弃。同时因为Pub/Sub是At-least-once投递模型，绝对不能直接用总消息条数做对比，必须用全局唯一的messageId做匹配，否则会把重复投递的消息误算为链路异常。

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