Skip to content

ref(hybridcloud): separate out explicit outbox model scheduling #121993

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cvxluo wants to merge 1 commit into
base: master
Choose a base branch
from
+62 −47
Open

ref(hybridcloud): separate out explicit outbox model scheduling #121993

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
109 changes: 62 additions & 47 deletions src/sentry/hybridcloud/tasks/deliver_from_outbox.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from __future__ import annotations

import math
from collections.abc import Mapping
from typing import Any

import sentry_sdk
Expand Down Expand Up @@ -73,56 +74,15 @@ def schedule_batch(
) -> None:
scheduled_count = 0

if not concurrency:
concurrency = CONCURRENCY
try:
for outbox_name in settings.SENTRY_OUTBOX_MODELS[silo_mode.name]:
outbox_model: type[OutboxBase] = OutboxBase.from_outbox_name(outbox_name)

aggregates = outbox_model.objects.all().aggregate(Min("id"), Max("id"))

lo = aggregates["id__min"] or 0
hi = aggregates["id__max"] or -1
if hi < lo:
continue

scheduled_count += hi - lo + 1
batch_size = math.ceil((hi - lo + 1) / concurrency)

metrics_tags = dict(silo_mode=silo_mode.name, outbox_name=outbox_name)
metrics.gauge(
"deliver_from_outbox.queued_batch_size",
value=batch_size,
tags=metrics_tags,
sample_rate=1.0,
)

# Notably, when l and h are close, this will result in creating tasks that are processing future ids --
# that's totally fine.
for i in range(concurrency):
drain_task.delay(
outbox_name=outbox_name,
outbox_identifier_low=lo + i * batch_size,
outbox_identifier_hi=lo + (i + 1) * batch_size,
)

deepest_shard_information = outbox_model.get_shard_depths_descending(limit=1)
max_shard_depth = (
float(deepest_shard_information[0]["depth"]) if deepest_shard_information else 0.0
)
metrics.gauge(
"deliver_from_outbox.maximum_shard_depth",

@sentry sentry Bot Aug 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: The outbox_name metric tag is now lowercase (e.g., sentry.controloutbox), which is a breaking change from the previous capitalized version, affecting existing monitoring and alerts.
Severity: HIGH

Suggested Fix

To maintain backward compatibility for metrics, ensure the outbox_name tag value remains capitalized. Revert to using the outbox_name variable that was previously used, which is derived from the settings and has the correct casing.

Prompt for AI Agent 
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: src/sentry/hybridcloud/tasks/deliver_from_outbox.py#L114

Potential issue: The `outbox_name` tag for metrics was previously derived from
`settings.SENTRY_OUTBOX_MODELS`, resulting in capitalized values like
`"sentry.ControlOutbox"`. The new implementation at this location uses
`outbox_model._meta.label`, which produces lowercase values like
`"sentry.controloutbox"`. This case change is a breaking change for any monitoring,
dashboards, or alerts that filter on the `outbox_name` tag. The change will not raise an
exception but will silently cause existing monitoring queries to fail, breaking
historical continuity and potentially delaying responses to production issues.

Did we get this right? 👍 / 👎 to inform future reviews.

value=max_shard_depth,
tags=metrics_tags,
sample_rate=1.0,
)

outbox_count = outbox_model.get_total_outbox_count()
metrics.gauge(
"deliver_from_outbox.total_outbox_count",
value=outbox_count,
tags=metrics_tags,
sample_rate=1.0,
scheduled_count += schedule_outbox_model(
silo_mode=silo_mode,
outbox_model=outbox_model,
drain_task=drain_task,
concurrency=concurrency,
drain_task_kwargs={"outbox_name": outbox_name},
)
if process_outbox_backfills:
backfill_outboxes_for(silo_mode, scheduled_count)
Expand All @@ -132,6 +92,61 @@ def schedule_batch(
raise


def schedule_outbox_model(
*,
silo_mode: SiloMode,
outbox_model: type[OutboxBase],
drain_task: Task[Any, Any],
concurrency: int | None = None,
drain_task_kwargs: Mapping[str, Any] | None = None,
) -> int:
if not concurrency:
concurrency = CONCURRENCY

id_range = outbox_model.objects.all().aggregate(Min("id"), Max("id"))
identifier_low = id_range["id__min"] or 0
identifier_high = id_range["id__max"] or -1
if identifier_high < identifier_low:
return 0

scheduled_count = identifier_high - identifier_low + 1
batch_size = math.ceil(scheduled_count / concurrency)
metrics_tags = dict(silo_mode=silo_mode.name, outbox_name=outbox_model._meta.label)
metrics.gauge(
"deliver_from_outbox.queued_batch_size",
value=batch_size,
tags=metrics_tags,
sample_rate=1.0,
)

# Notably, when low and high are close, some tasks process future ids. That's fine.
task_kwargs = drain_task_kwargs or {}
for i in range(concurrency):
drain_task.delay(
outbox_identifier_low=identifier_low + i * batch_size,
outbox_identifier_hi=identifier_low + (i + 1) * batch_size,
**task_kwargs,
)

deepest_shards = outbox_model.get_shard_depths_descending(limit=1)
max_shard_depth = float(deepest_shards[0]["depth"]) if deepest_shards else 0.0
metrics.gauge(
"deliver_from_outbox.maximum_shard_depth",
value=max_shard_depth,
tags=metrics_tags,
sample_rate=1.0,
)

outbox_count = outbox_model.get_total_outbox_count()
metrics.gauge(
"deliver_from_outbox.total_outbox_count",
value=outbox_count,
tags=metrics_tags,
sample_rate=1.0,
)
return scheduled_count


@instrumented_task(
name="sentry.tasks.drain_outbox_shards",
namespace=hybridcloud_tasks,
Expand Down
Loading