Skip to content

Commit fc06665

Browse files
sentry-junior[bot]volokluev
sentry-junior[bot]
and
volokluev
authored
fix(crons): timeout seat acceptance during check-in ingest (#123175)
The crons ingest consumer called `quotas.backend.check_accept_monitor_checkin` with no wait bound, so a hung quotas/seat backend could stall consumer poll. This wraps that call with a wall-clock timeout (`crons.check_accept_monitor_checkin.timeout_sec`, default 1s). On timeout, the consumer fails open and accepts the check-in, logs a warning, and records `monitors.checkin.result` with `status=check_accept_timeout`. Set the option to `0` to disable the wrapper. <!-- junior-request-attribution:start --> Requested by **volo**. <!-- junior-request-attribution:end --> <!-- junior-session-footer:start --> <!-- junior-conversation-id:slack%3AD0BNNTBGH53%3A1788198404.373289 --> -- [View Junior Session](https://junior-prod.sentry.dev/conversations/slack%3AD0BNNTBGH53%3A1788198404.373289) [[Sentry]](https://sentry.sentry.io/explore/conversations/slack%3AD0BNNTBGH53%3A1788198404.373289/?project=4510944073809921) <!-- junior-session-footer:end --> --------- Co-authored-by: sentry-junior[bot] <264270552+sentry-junior[bot]@users.noreply.github.com> Co-authored-by: Volo Kluev <volo.kluev@sentry.io>
1 parent c793d3e commit fc06665

3 files changed

Lines changed: 206 additions & 3 deletions

File tree

‎src/sentry/monitors/consumers/monitor_consumer.py‎

Lines changed: 102 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,12 @@
22

33
import logging
44
import random
5+
import threading
56
import uuid
67
from collections import defaultdict
78
from collections.abc import Mapping
8-
from concurrent.futures import wait
9+
from concurrent.futures import Future, wait
10+
from concurrent.futures import TimeoutError as FuturesTimeoutError
911
from copy import deepcopy
1012
from datetime import UTC, datetime
1113
from functools import partial
@@ -81,6 +83,7 @@
8183
valid_duration,
8284
)
8385
from sentry.monitors.validators import ConfigValidator, MonitorCheckInValidator
86+
from sentry.options.rollout import in_rollout_group
8487
from sentry.types.actor import parse_and_validate_actor
8588
from sentry.utils import json, metrics
8689
from sentry.utils.concurrent import ContextPropagatingThreadPoolExecutor
@@ -94,6 +97,101 @@
9497

9598
DROP_LOG_SAMPLE_RATE = 0.01
9699

100+
# Shared pool for timing out quotas seat checks without creating an executor
101+
# per check-in. The check itself is expected to be cheap (Redis/local cache);
102+
# this only exists so a hung backend cannot block the consumer indefinitely.
103+
#
104+
# `future.result(timeout=...)` only stops waiting; timed-out work keeps running
105+
# until the backend returns. Bound outstanding hand-offs so a prolonged hang
106+
# cannot grow an unbounded queue of stale seat checks (and eventually OOM).
107+
_CHECK_ACCEPT_MAX_IN_FLIGHT = 1000
108+
_CHECK_ACCEPT_EXECUTOR = ContextPropagatingThreadPoolExecutor(
109+
max_workers=8,
110+
thread_name_prefix="monitors.check_accept",
111+
)
112+
_CHECK_ACCEPT_SLOTS = threading.BoundedSemaphore(_CHECK_ACCEPT_MAX_IN_FLIGHT)
113+
114+
115+
def _check_accept_monitor_checkin_with_timeout(
116+
project_id: int,
117+
monitor_slug: str,
118+
metric_kwargs: dict[str, str],
119+
) -> PermitCheckInStatus:
120+
"""
121+
Call quotas seat acceptance, optionally with a wall-clock timeout.
122+
123+
The timeout path is gated by
124+
``crons.check_accept_monitor_checkin.timeout_rollout_rate`` (deterministic
125+
per project). Outside the rollout group the backend is called directly.
126+
127+
If the backend does not respond in time, or too many seat checks are already
128+
in flight, fail open and ACCEPT the check-in so a slow or hung quotas path
129+
cannot stall crons ingest or accumulate unbounded pending work.
130+
131+
Note: a timed-out worker keeps running until the underlying call returns;
132+
the wait bound only limits how long ingest blocks before failing open. The
133+
in-flight slot bound limits how many of those late calls can pile up.
134+
"""
135+
if not in_rollout_group("crons.check_accept_monitor_checkin.timeout_rollout_rate", project_id):
136+
return quotas.backend.check_accept_monitor_checkin(project_id, monitor_slug)
137+
138+
timeout_sec = options.get("crons.check_accept_monitor_checkin.timeout_sec")
139+
if not timeout_sec or timeout_sec <= 0:
140+
return quotas.backend.check_accept_monitor_checkin(project_id, monitor_slug)
141+
142+
if not _CHECK_ACCEPT_SLOTS.acquire(blocking=False):
143+
metrics.incr(
144+
"monitors.checkin.check_accept_shed",
145+
tags=metric_kwargs,
146+
)
147+
logger.warning(
148+
"monitors.consumer.check_accept_shed",
149+
extra={
150+
"project_id": project_id,
151+
"slug": monitor_slug,
152+
"max_in_flight": _CHECK_ACCEPT_MAX_IN_FLIGHT,
153+
},
154+
)
155+
return PermitCheckInStatus.ACCEPT
156+
157+
def _done(f: Future[PermitCheckInStatus]) -> None:
158+
_CHECK_ACCEPT_SLOTS.release()
159+
# Drain late results/errors after we stop waiting so timed-out futures
160+
# do not log "exception was never retrieved".
161+
f.exception()
162+
163+
handed_off = False
164+
try:
165+
future = _CHECK_ACCEPT_EXECUTOR.submit(
166+
quotas.backend.check_accept_monitor_checkin,
167+
project_id,
168+
monitor_slug,
169+
)
170+
# From here the future owns the permit and _done will release it.
171+
future.add_done_callback(_done)
172+
handed_off = True
173+
finally:
174+
if not handed_off:
175+
_CHECK_ACCEPT_SLOTS.release()
176+
177+
try:
178+
return future.result(timeout=timeout_sec)
179+
except FuturesTimeoutError:
180+
metrics.incr(
181+
"monitors.checkin.check_accept_timeout",
182+
tags=metric_kwargs,
183+
)
184+
logger.warning(
185+
"monitors.consumer.check_accept_timeout",
186+
extra={
187+
"project_id": project_id,
188+
"slug": monitor_slug,
189+
"timeout_sec": timeout_sec,
190+
},
191+
)
192+
# Fail open: prefer accepting a check-in over blocking ingest.
193+
return PermitCheckInStatus.ACCEPT
194+
97195

98196
def _ensure_monitor_with_config(
99197
project: Project,
@@ -544,8 +642,9 @@ def _process_checkin(item: CheckinItem, span: Transaction | Span | StreamedSpan)
544642
raise ProcessingErrorsException([ratelimit_error])
545643

546644
# Does quotas allow for this check-in to be accepted?
547-
quotas_outcome: PermitCheckInStatus = quotas.backend.check_accept_monitor_checkin(
548-
project.id, monitor_slug
645+
# Bound the wait so a hung quotas backend cannot stall the consumer poll.
646+
quotas_outcome: PermitCheckInStatus = _check_accept_monitor_checkin_with_timeout(
647+
project.id, monitor_slug, metric_kwargs
549648
)
550649

551650
if quotas_outcome == PermitCheckInStatus.DROP:

‎src/sentry/options/defaults.py‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2784,6 +2784,27 @@
27842784
flags=FLAG_PRIORITIZE_DISK | FLAG_AUTOMATOR_MODIFIABLE,
27852785
)
27862786

2787+
# Deterministic % of check-ins that use the seat-acceptance timeout wrapper.
2788+
# Keyed on project id. Default 0.0 so deploy is a no-op until dialed up via
2789+
# sentry-options-automator.
2790+
register(
2791+
"crons.check_accept_monitor_checkin.timeout_rollout_rate",
2792+
type=Float,
2793+
default=0.0,
2794+
flags=FLAG_MODIFIABLE_RATE | FLAG_AUTOMATOR_MODIFIABLE,
2795+
)
2796+
2797+
# Bound how long the crons ingest consumer waits on seat/quota acceptance when
2798+
# the timeout rollout selects the check-in. On timeout the check-in is accepted
2799+
# (fail-open) so a slow quotas backend cannot stall the consumer. Set to 0 to
2800+
# disable the timeout wrapper even for selected traffic.
2801+
register(
2802+
"crons.check_accept_monitor_checkin.timeout_sec",
2803+
type=Float,
2804+
default=1.0,
2805+
flags=FLAG_AUTOMATOR_MODIFIABLE,
2806+
)
2807+
27872808

27882809
# Sets the timeout for webhooks
27892810
register(

‎tests/sentry/monitors/consumers/test_monitor_consumer.py‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import contextlib
2+
import time
23
import uuid
34
from collections.abc import Generator, Mapping, Sequence
45
from datetime import UTC, datetime, timedelta
@@ -1341,6 +1342,88 @@ def test_monitor_quotas_drop(self, check_accept_monitor_checkin: mock.MagicMock)
13411342
checkins = MonitorCheckIn.objects.filter(monitor_id=monitor.id)
13421343
assert len(checkins) == 0
13431344

1345+
@mock.patch("sentry.quotas.backend.check_accept_monitor_checkin")
1346+
def test_monitor_quotas_timeout_accepts(
1347+
self, check_accept_monitor_checkin: mock.MagicMock
1348+
) -> None:
1349+
"""
1350+
A hung quotas seat check must fail open so the consumer does not stall.
1351+
"""
1352+
1353+
def hang(*args, **kwargs):
1354+
time.sleep(1.0)
1355+
return PermitCheckInStatus.DROP
1356+
1357+
check_accept_monitor_checkin.side_effect = hang
1358+
1359+
monitor = self._create_monitor(slug="my-monitor")
1360+
with override_options(
1361+
{
1362+
"crons.check_accept_monitor_checkin.timeout_rollout_rate": 1.0,
1363+
"crons.check_accept_monitor_checkin.timeout_sec": 0.05,
1364+
}
1365+
):
1366+
self.send_checkin(monitor.slug)
1367+
1368+
check_accept_monitor_checkin.assert_called_with(self.project.id, monitor.slug)
1369+
1370+
checkin = MonitorCheckIn.objects.get(monitor_id=monitor.id)
1371+
assert checkin.status == CheckInStatus.OK
1372+
1373+
@mock.patch(
1374+
"sentry.monitors.consumers.monitor_consumer._CHECK_ACCEPT_SLOTS.acquire",
1375+
return_value=False,
1376+
)
1377+
@mock.patch("sentry.quotas.backend.check_accept_monitor_checkin")
1378+
def test_monitor_quotas_shed_accepts(
1379+
self,
1380+
check_accept_monitor_checkin: mock.MagicMock,
1381+
acquire: mock.MagicMock,
1382+
) -> None:
1383+
"""
1384+
When too many seat checks are already in flight, fail open without
1385+
queuing another stale call.
1386+
"""
1387+
check_accept_monitor_checkin.return_value = PermitCheckInStatus.DROP
1388+
1389+
monitor = self._create_monitor(slug="my-monitor")
1390+
with override_options({"crons.check_accept_monitor_checkin.timeout_rollout_rate": 1.0}):
1391+
self.send_checkin(monitor.slug)
1392+
1393+
acquire.assert_called_once_with(blocking=False)
1394+
check_accept_monitor_checkin.assert_not_called()
1395+
1396+
checkin = MonitorCheckIn.objects.get(monitor_id=monitor.id)
1397+
assert checkin.status == CheckInStatus.OK
1398+
1399+
@mock.patch("sentry.quotas.backend.check_accept_monitor_checkin")
1400+
def test_monitor_quotas_timeout_rollout_disabled(
1401+
self, check_accept_monitor_checkin: mock.MagicMock
1402+
) -> None:
1403+
"""
1404+
With rollout rate 0, seat acceptance is called directly (no timeout
1405+
wrapper), so a DROP result still drops the check-in.
1406+
"""
1407+
check_accept_monitor_checkin.return_value = PermitCheckInStatus.DROP
1408+
1409+
monitor = self._create_monitor(slug="my-monitor")
1410+
with override_options(
1411+
{
1412+
"crons.check_accept_monitor_checkin.timeout_rollout_rate": 0.0,
1413+
# Would fail-open if the wrapper ran; prove it does not.
1414+
"crons.check_accept_monitor_checkin.timeout_sec": 0.05,
1415+
}
1416+
):
1417+
self.send_checkin(
1418+
monitor.slug,
1419+
expected_error=ProcessingErrorsException(
1420+
[{"type": ProcessingErrorType.MONITOR_OVER_QUOTA}],
1421+
),
1422+
)
1423+
1424+
check_accept_monitor_checkin.assert_called_with(self.project.id, monitor.slug)
1425+
assert not MonitorCheckIn.objects.filter(monitor_id=monitor.id).exists()
1426+
13441427
@mock.patch("sentry.quotas.backend.assign_seat")
13451428
@mock.patch("sentry.quotas.backend.check_accept_monitor_checkin")
13461429
def test_monitor_accept_upsert_with_seat(

0 commit comments

Comments
 (0)