feat(uptime): Add ability to use queues to manage parallelism - #1
feat(uptime): Add ability to use queues to manage parallelism#1linxia0415 wants to merge 8 commits into
Conversation
One potential problem we have with batch processing is that any one slow item will clog up the whole batch. This pr implements a queueing method instead, where we keep N queues that each have their own workers. There's still a chance of individual items backlogging a queue, but we can try increased concurrency here to reduce the chances of that happening
![code-hawk-sit[bot]](https://avatars.githubusercontent.com/in/3248189?s=60&v=4){kind=small}
There was a problem hiding this comment.
AI代码审查报告
变更概览
本次 PR 涉及 7 个文件，新增 +1276 行，删除 -6 行。
功能变更摘要
本 PR 为远程订阅系统引入独立的队列消费者，将任务入队与结果处理逻辑分离。通过新增 queue_consumer 负责订阅任务的队列管理，并增强 result_consumer 的结果处理能力，提升了系统可扩展性。同时补充了完整的单元测试覆盖，并修复了相关子系统（uptime）的测试适配。
文件变更摘要
| 文件 | 变更 | 行数 | 摘要 | 发现问题 |
|---|---|---|---|---|
src/sentry/remote_subscriptions/consumers/queue_consumer.py |
新增 | +345/-0 | 新增远程订阅队列消费者，负责任务入队与调度管理 | 9 个 |
src/sentry/remote_subscriptions/consumers/result_consumer.py |
修改 | +41/-3 | 增强结果消费者功能，适配新的队列处理流程 | 7 个 |
src/sentry/consumers/__init__.py |
修改 | +2/-2 | 调整消费者配置或注册逻辑以适配新架构 | 1 个 |
tests/sentry/uptime/consumers/test_results_consumer.py |
修改 | +467/-1 | 更新结果消费者测试以适配新架构，并大幅扩展测试场景 | — |
tests/sentry/remote_subscriptions/consumers/test_queue_consumer.py |
新增 | +421/-0 | 新增队列消费者的完整单元测试覆盖 | — |
tests/sentry/remote_subscriptions/__init__.py |
新增 | +0/-0 | 初始化远程订阅测试模块 | — |
tests/sentry/remote_subscriptions/consumers/__init__.py |
新增 | +0/-0 | 初始化消费者测试模块 | — |
问题严重级别分布
| 级别 | 数量 | 占比 |
|---|---|---|
| 🟡 中危 | 15 | 93% |
| 🟢 低危 | 1 | 6% |
代表性问题（至多 10 条，按严重级别优先）
- 🟡 中危
src/sentry/remote_subscriptions/consumers/queue_consumer.pyL49: 破坏 OffsetTracker 的线程安全性，可能导致偏移量追踪数据损坏，引发消息重复消费或丢失。 - 🟡 中危
src/sentry/remote_subscriptions/consumers/queue_consumer.pyL127: 在关闭过程中，如果队列恰好为空，worker 线程将永远阻塞，导致 join(timeout=5.0) 超时，线程无法正确终止，可能引发资源泄漏或优雅关闭失败。 - 🟡 中危
src/sentry/remote_subscriptions/consumers/queue_consumer.pyL148: 异常处理代码自身抛出异常，掩盖原始错误，增加调试难度。 - 🟡 中危
src/sentry/remote_subscriptions/consumers/queue_consumer.pyL198: 在消费者重启后，同一分组的消息可能被路由到不同队列，如果旧队列中仍有未处理的消息，会导致同一分组的消息在不同队列中并行处理，破坏有序性保证，可能引发竞态条件或数据不一致。 - 🟡 中危
src/sentry/remote_subscriptions/consumers/queue_consumer.pyL231: 关闭流程不稳定，可能抛出未捕获异常，影响优雅关闭。 - 🟡 中危
src/sentry/remote_subscriptions/consumers/queue_consumer.pyL270: 在消费者关闭时，可能丢失已处理但未提交的偏移量，导致消息重复消费；如果业务处理是幂等的，重复消费可能问题不大，但会增加系统负载并可能导致数据不一致。 - 🟡 中危
src/sentry/remote_subscriptions/consumers/queue_consumer.pyL297: 生产环境可能因类型错误导致崩溃，且难以调试。 - 🟡 中危
src/sentry/remote_subscriptions/consumers/queue_consumer.pyL335: 提交线程可能在 queue_pool 关闭后仍在尝试访问 offset_tracker，导致异常或数据不一致；未完成的偏移量提交可能丢失。 - 🟡 中危
src/sentry/remote_subscriptions/consumers/queue_consumer.pyL344: 调用者无法通过 timeout 参数控制关闭等待时间，可能导致调用方期望的超时行为与实际不符，影响优雅关闭流程。 - 🟡 中危
src/sentry/remote_subscriptions/consumers/result_consumer.pyL117: 如果子类的 result_processor_cls 依赖于其他在 init 中初始化的属性，提前调用可能导致 AttributeError 或错误的行为。
共 16 处问题；上列仅展示优先关注的 10 条，另有 6 条请查看对应行内评论。
未发帖的低优先级项
当前策略为 balanced：以下 1 条未发成行评，以免掩盖安全与正确性问题；若团队关注代码风格与微优化，可在此浏览或改为 full。
- 低危
src/sentry/consumers/__init__.pyL121：help 文本未更新以反映新模式 line 121-124 添加了 "thread-queue-parallel" 模式选项，但 help 文本仍描述为 "Parallel uses multithreading"，没有说明 thread-…
行评发帖：共检出 16 处，本次策略
balanced下发 15 处行评；其余见上表或未发帖列表。
由 CodeHawk 提供支持 · nuwa
| @@ -105,6 +115,7 @@ def __init__( | |||
| ) -> None: | |||
| self.mode = mode | |||
| metric_tags = {"identifier": self.identifier, "mode": self.mode} | |||
There was a problem hiding this comment.
🟡 AI 代码审查发现问题
📋 问题概述
result_processor 初始化位置提前可能导致子类属性未就绪
line 117 将 self.result_processor = self.result_processor_cls() 从原位置（line 136 之后）提前到 init 开头。这可能导致在 result_processor_cls 属性被子类正确设置之前就被调用，因为 self.result_processor_cls 是一个 abstractproperty，子类必须在实例化后才能提供具体实现。虽然 Python 的 property 访问会动态解析，但如果子类在 init 中有额外的初始化逻辑，可能导致意外行为。
📍 问题详情
🟡 问题 1 | 严重程度: MEDIUM | 行号: 117
💬 详细说明：
- 如果子类的 result_processor_cls 依赖于其他在 init 中初始化的属性，提前调用可能导致 AttributeError 或错误的行为。
📝 问题代码：
self.result_processor = self.result_processor_cls()
💡 修复建议：
将 result_processor 的初始化移回模式特定的条件块之后，确保所有基础属性已就绪：
if mode == "thread-queue-parallel":
self.thread_queue_parallel = True
self.result_processor = self.result_processor_cls() # 移到这里
self.queue_pool = FixedQueuePool(...)✅ 修复示例：
self.mode = mode
metric_tags = {"identifier": self.identifier, "mode": self.mode}
if mode == "batched-parallel":
self.batched_parallel = True
self.parallel_executor = ThreadPoolExecutor(max_workers=max_workers)
if max_workers is None:
metric_tags["workers"] = "default"
else:
metric_tags["workers"] = str(max_workers)
if mode == "parallel":
self.parallel = True
if num_processes is None:
num_processes = multiprocessing.cpu_count()
self.multiprocessing_pool = MultiprocessingPool(num_processes)
if mode == "thread-queue-parallel":
self.thread_queue_parallel = True
self.result_processor = self.result_processor_cls()
self.queue_pool = FixedQueuePool(
result_processor=self.result_processor,
identifier=self.identifier,
num_queues=max_workers or 20,
)
🔗 参考链接
无
| result_processor=self.result_processor, | ||
| identifier=self.identifier, | ||
| num_queues=max_workers or 20, # Number of parallel queues | ||
| ) |
There was a problem hiding this comment.
🟡 AI 代码审查发现问题
📋 问题概述
缩进错误导致指标仅在 thread-queue-parallel 模式下上报
line 129 处 if mode == "thread-queue-parallel": 开启条件分支，line 130-135 为分支内代码（8空格缩进）。line 137-141 的 metrics.incr 调用同样使用 8 空格缩进，导致该指标上报逻辑被错误地置于 thread-queue-parallel 条件分支内部。
当 mode 为 "batched-parallel"、"parallel" 或 "serial" 时，line 137-141 的启动指标不会执行，造成监控数据缺失，无法准确统计各模式消费者的启动情况。
📍 问题详情
🟡 问题 1 | 严重程度: MEDIUM | 行号: 137
💬 详细说明：
- 缩进错误导致指标仅在 thread-queue-parallel 模式下上报
- line 129 处
if mode == "thread-queue-parallel":开启条件分支，line 130-135 为分支内代码（8空格缩进）。line 137-141 的metrics.incr调用同样使用 8 空格缩进，导致该指标上报逻辑被错误地置于thread-queue-parallel条件分支内部。 - 当 mode 为 "batched-parallel"、"parallel" 或 "serial" 时，line 137-141 的启动指标不会执行，造成监控数据缺失，无法准确统计各模式消费者的启动情况。
📝 问题代码：
metrics.incr(
💡 修复建议：
将 line 137-141 的
metrics.incr调用减少一级缩进（从 8 空格改为 4 空格），使其与if mode == "thread-queue-parallel":同级，确保所有模式启动时均上报指标。
✅ 修复示例：
if mode == "thread-queue-parallel":
self.thread_queue_parallel = True
self.queue_pool = FixedQueuePool(
result_processor=self.result_processor,
identifier=self.identifier,
num_queues=max_workers or 20, # Number of parallel queues
)
metrics.incr(
"remote_subscriptions.result_consumer.start",
1,
tags=metric_tags,
)
🔗 参考链接
无
| @@ -164,6 +180,9 @@ def identifier(self) -> str: | |||
| def shutdown(self) -> None: | |||
There was a problem hiding this comment.
🟡 AI 代码审查发现问题
📋 问题概述
shutdown 方法存在重复关闭风险和异常处理问题
line 180-185 的 shutdown 方法存在两个问题：1) 如果 shutdown 被多次调用，第一次调用后 self.queue_pool 被设为 None，第二次调用时条件判断通过（因为 None 是 falsy），但调用 None.shutdown() 会引发 AttributeError；2) parallel_executor.shutdown() 和 queue_pool.shutdown() 可能抛出异常，但没有 try-except 保护，一个失败会阻止后续清理。
📍 问题详情
🟡 问题 1 | 严重程度: MEDIUM | 行号: 180
💬 详细说明：
- 重复调用 shutdown 会导致 AttributeError；如果 executor 关闭失败，queue_pool 的关闭被跳过，可能导致资源泄漏。
📝 问题代码：
def shutdown(self) -> None:
if self.parallel_executor:
self.parallel_executor.shutdown()
if self.queue_pool:
self.queue_pool.shutdown()
self.queue_pool = None
💡 修复建议：
添加异常处理和正确的空值检查：
def shutdown(self) -> None:
if self.parallel_executor:
try:
self.parallel_executor.shutdown()
except Exception:
logger.exception("Error shutting down parallel executor")
self.parallel_executor = None
if self.queue_pool is not None:
try:
self.queue_pool.shutdown()
except Exception:
logger.exception("Error shutting down queue pool")
self.queue_pool = None✅ 修复示例：
def shutdown(self) -> None:
if self.parallel_executor:
try:
self.parallel_executor.shutdown()
except Exception:
logger.exception("Error shutting down parallel executor")
self.parallel_executor = None
if self.queue_pool is not None:
try:
self.queue_pool.shutdown()
except Exception:
logger.exception("Error shutting down queue pool")
self.queue_pool = None
🔗 参考链接
无
| if self.queue_pool: | ||
| self.queue_pool.shutdown() | ||
| self.queue_pool = None | ||
|
|
There was a problem hiding this comment.
🟡 AI 代码审查发现问题
📋 问题概述
使用 assert 检查运行时条件存在生产环境风险
line 186 使用 assert 检查 payload 类型，但 Python 在优化模式（-O 或 -OO）下会移除所有 assert 语句。若代码在此模式下运行且传入 FilteredPayload，断言被跳过，后续逻辑可能处理无效数据。
此外，FilteredPayload 是 arroyo 框架的合法消息类型，用于表示被过滤的消息，应通过显式条件判断处理而非断言。
📍 问题详情
🟡 问题 1 | 严重程度: MEDIUM | 行号: 186
💬 详细说明：
- 使用 assert 检查运行时条件存在生产环境风险
- line 186 使用
assert检查payload类型，但 Python 在优化模式（-O 或 -OO）下会移除所有 assert 语句。若代码在此模式下运行且传入 FilteredPayload，断言被跳过，后续逻辑可能处理无效数据。 - 此外，FilteredPayload 是 arroyo 框架的合法消息类型，用于表示被过滤的消息，应通过显式条件判断处理而非断言。
📝 问题代码：
assert not isinstance(payload, FilteredPayload)
💡 修复建议：
将 assert 改为显式的类型检查与异常抛出，或添加条件判断提前返回。确保生产环境下类型检查仍然生效。
✅ 修复示例：
def decode_payload(self, topic_for_codec, payload: KafkaPayload | FilteredPayload) -> T | None:
if isinstance(payload, FilteredPayload):
return None
try:
🔗 参考链接
无
| return self.create_thread_parallel_worker(commit) | ||
| if self.parallel: | ||
| return self.create_multiprocess_worker(commit) | ||
| if self.thread_queue_parallel: |
There was a problem hiding this comment.
🟡 AI 代码审查发现问题
📋 问题概述
条件分支逻辑存在冗余 else
line 209-212 的代码中，if self.thread_queue_parallel 后直接返回，else 分支是多余的。虽然不影响功能，但代码风格不一致，与其他模式（batched_parallel、parallel）的处理方式不同，降低了可读性。
📍 问题详情
🟡 问题 1 | 严重程度: MEDIUM | 行号: 209
💬 详细说明：
- 代码风格不一致，维护性降低；无功能性风险。
📝 问题代码：
if self.thread_queue_parallel:
return self.create_thread_queue_parallel_worker(commit)
else:
return self.create_serial_worker(commit)
💡 修复建议：
移除冗余的 else，保持与其他条件分支风格一致：
if self.thread_queue_parallel:
return self.create_thread_queue_parallel_worker(commit)
return self.create_serial_worker(commit)✅ 修复示例：
if self.thread_queue_parallel:
return self.create_thread_queue_parallel_worker(commit)
return self.create_serial_worker(commit)
🔗 参考链接
无
| time.sleep(0.01) | ||
| return False | ||
|
|
||
| def shutdown(self) -> None: |
There was a problem hiding this comment.
🟡 AI 代码审查发现问题
📋 问题概述
shutdown 竞态条件和异常处理不完善。line 231-237 关闭顺序：先设置 worker.shutdown 标志，再关闭队列，最后 join workers。但 worker 可能在队列关闭后才尝试 get()，抛出异常。重复调用 shutdown 会抛出 AttributeError。
📍 问题详情
🟡 问题 1 | 严重程度: MEDIUM | 行号: 231
💬 详细说明：
- 关闭流程不稳定，可能抛出未捕获异常，影响优雅关闭。
📝 问题代码：
def shutdown(self) -> None:
"""Gracefully shutdown all workers."""
for worker in self.workers:
worker.shutdown = True
for q in self.queues:
try:
q.shutdown(immediate=False)
except Exception:
logger.exception("Error shutting down queue")
for worker in self.workers:
worker.join(timeout=5.0)
💡 修复建议：
添加关闭状态标志防止重复关闭，改进关闭顺序：先关闭队列（让 worker 的 get() 抛出 ShutDown），再 join workers。
✅ 修复示例：
def shutdown(self) -> None:
"""Gracefully shutdown all workers - idempotent."""
if hasattr(self, '_shutdown_complete') and self._shutdown_complete:
return
# First shutdown queues to wake up workers
for q in self.queues:
try:
q.shutdown(immediate=False)
except Exception:
logger.exception("Error shutting down queue")
# Set shutdown flags
for worker in self.workers:
worker.shutdown = True
# Wait for workers to finish
for worker in self.workers:
worker.join(timeout=5.0)
if worker.is_alive():
logger.warning(f"Worker {worker.worker_id} did not stop gracefully")
self._shutdown_complete = True
🔗 参考链接
无
| self.commit_function = commit_function | ||
| self.shutdown_event = threading.Event() | ||
|
|
||
| self.commit_thread = threading.Thread(target=self._commit_loop, daemon=True) |
There was a problem hiding this comment.
🟡 AI 代码审查发现问题
📋 问题概述
commit_thread 作为 daemon 线程可能在关闭时丢失偏移量提交
line 270-271 的提交线程被设置为 daemon=True。daemon 线程在主线程退出时会立即被终止，不等待其完成。如果在关闭时有待提交的偏移量正在处理中，daemon 线程被强制终止可能导致偏移量提交丢失。
📍 问题详情
🟡 问题 1 | 严重程度: MEDIUM | 行号: 270
💬 详细说明：
- 在消费者关闭时，可能丢失已处理但未提交的偏移量，导致消息重复消费；如果业务处理是幂等的，重复消费可能问题不大，但会增加系统负载并可能导致数据不一致。
📝 问题代码：
self.commit_thread = threading.Thread(target=self._commit_loop, daemon=True)
self.commit_thread.start()
💡 修复建议：
将 commit_thread 改为非 daemon 线程，并在 close() 方法中确保正确等待其完成（当前代码已调用 join，但 daemon 线程的行为可能导致 join 不可靠）。建议修改为：
self.commit_thread = threading.Thread(target=self._commit_loop, daemon=False)✅ 修复示例：
self.commit_thread = threading.Thread(target=self._commit_loop, daemon=False)
self.commit_thread.start()
🔗 参考链接
无
| try: | ||
| result = self.decoder(message.payload) | ||
|
|
||
| assert isinstance(message.value, BrokerValue) |
There was a problem hiding this comment.
🟡 AI 代码审查发现问题
📋 问题概述
assert 用于生产环境运行时检查。line 297 使用 assert 检查 message.value 是否为 BrokerValue 类型，但在 Python 优化模式（-O）下 assert 会被移除，导致类型检查失效。
📍 问题详情
🟡 问题 1 | 严重程度: MEDIUM | 行号: 297
💬 详细说明：
- 生产环境可能因类型错误导致崩溃，且难以调试。
📝 问题代码：
assert isinstance(message.value, BrokerValue)
partition = message.value.partition
offset = message.value.offset
💡 修复建议：
使用显式的 if 判断和异常抛出替代 assert。
✅ 修复示例：
if not isinstance(message.value, BrokerValue):
raise TypeError(f"Expected BrokerValue, got {type(message.value).__name__}")
partition = message.value.partition
offset = message.value.offset
🔗 参考链接
无
| tags={"identifier": self.queue_pool.identifier}, | ||
| ) | ||
|
|
||
| def close(self) -> None: |
There was a problem hiding this comment.
🟡 AI 代码审查发现问题
📋 问题概述
close() 方法中 commit_thread.join 失败未处理
line 335-338 的 close 方法调用 commit_thread.join(timeout=5.0)，如果超时（线程未在 5 秒内完成），join 返回但不会抛出异常，代码继续执行 queue_pool.shutdown()。此时提交线程可能仍在运行，访问正在被关闭的资源，引发竞态条件。
📍 问题详情
🟡 问题 1 | 严重程度: MEDIUM | 行号: 335
💬 详细说明：
- 提交线程可能在 queue_pool 关闭后仍在尝试访问 offset_tracker，导致异常或数据不一致；未完成的偏移量提交可能丢失。
📝 问题代码：
def close(self) -> None:
self.shutdown_event.set()
self.commit_thread.join(timeout=5.0)
self.queue_pool.shutdown()
💡 修复建议：
检查 join 是否成功，并处理超时情况：
def close(self) -> None:
self.shutdown_event.set()
self.commit_thread.join(timeout=5.0)
if self.commit_thread.is_alive():
logger.warning("Commit thread did not stop in time, proceeding with shutdown")
self.queue_pool.shutdown()✅ 修复示例：
def close(self) -> None:
self.shutdown_event.set()
self.commit_thread.join(timeout=5.0)
if self.commit_thread.is_alive():
logger.warning("Commit thread did not stop in time, proceeding with shutdown")
self.queue_pool.shutdown()
🔗 参考链接
无
| self.shutdown_event.set() | ||
| self.queue_pool.shutdown() | ||
|
|
||
| def join(self, timeout: float | None = None) -> None: |
There was a problem hiding this comment.
🟡 AI 代码审查发现问题
📋 问题概述
join 方法未使用 timeout 参数
line 344-345 的 join 方法接收 timeout 参数但直接调用 self.close()，未将 timeout 传递给 close 或内部调用。这违反了 ProcessingStrategy 接口契约，调用者期望能够控制等待时间，但实际被忽略。
📍 问题详情
🟡 问题 1 | 严重程度: MEDIUM | 行号: 344
💬 详细说明：
- 调用者无法通过 timeout 参数控制关闭等待时间，可能导致调用方期望的超时行为与实际不符，影响优雅关闭流程。
📝 问题代码：
def join(self, timeout: float | None = None) -> None:
self.close()
💡 修复建议：
在 join 方法中传递或使用 timeout 参数：
def join(self, timeout: float | None = None) -> None:
# close 已经包含 join，但我们需要尊重 timeout 参数
self.shutdown_event.set()
self.commit_thread.join(timeout=timeout)
self.queue_pool.wait_until_empty(timeout=timeout if timeout else 5.0)✅ 修复示例：
def join(self, timeout: float | None = None) -> None:
self.shutdown_event.set()
self.commit_thread.join(timeout=timeout)
self.queue_pool.wait_until_empty(timeout=timeout if timeout else 5.0)
🔗 参考链接
无
2 participants
One potential problem we have with batch processing is that any one slow item will clog up the whole batch. This pr implements a queueing method instead, where we keep N queues that each have their own workers. There's still a chance of individual items backlogging a queue, but we can try increased concurrency here to reduce the chances of that happening