feat(crons): Backfill missed clock-tick dispatches (#70338)

When we detect a clock tick it is possible that we may have skipped a
tick when a monitor ingest partition is very slow and does not contain a
message for a entire minute (and other partitions have already moved
multiple minutes forward).

Previously we would log this and avoid producing a clock tick for these
skipped minute(s) as we were using celery to dispatch the check_missing
and check_timeout tasks. Since the celery tasks would be produced
back-to-back it wasn't unlikely they would be processed out of order

Since the completion of GH-58410 we are now guaranteed that clock tick
tasks are processed in order.

Author: evanpurkhiser

src/sentry/monitors/clock_dispatch.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 import sentry_sdk
 from arroyo import Topic as ArroyoTopic
@@ -147,11 +147,23 @@ def try_monitor_clock_tick(ts: datetime, partition: int):
     metrics.gauge("monitors.task.clock_delay", total_delay, sample_rate=1.0)
 
     # If more than exactly a minute has passed then we've skipped a
-    # task run, report that to sentry, it is a problem.
+    # task run, backfill those ticks. This can happen when one partition has
+    # slowed down too much and is missing a minutes worth of check-ins
     if last_ts is not None and slowest_part_ts > last_ts + 60:
-        with sentry_sdk.push_scope() as scope:
-            scope.set_extra("last_ts", last_ts)
-            scope.set_extra("slowest_part_ts", slowest_part_ts)
-            sentry_sdk.capture_message("Monitor task dispatch minute skipped")
+        if options.get("crons.use_clock_pulse_consumer"):
+            # We only want to do backfills when we're using the clock tick
+            # consumer, otherwise the celery tasks may process out of order
+            backfill_tick = datetime.fromtimestamp(last_ts + 60, tz=timezone.utc)
+            while backfill_tick < tick:
+                extra = {"reference_datetime": str(backfill_tick)}
+                logger.info("monitors.consumer.clock_tick_backfill", extra=extra)
+
+                _dispatch_tick(backfill_tick)
+                backfill_tick = backfill_tick + timedelta(minutes=1)
+        else:
+            with sentry_sdk.push_scope() as scope:
+                scope.set_extra("last_ts", last_ts)
+                scope.set_extra("slowest_part_ts", slowest_part_ts)
+                sentry_sdk.capture_message("Monitor task dispatch minute skipped")
 
     _dispatch_tick(tick)

tests/sentry/monitors/test_clock_dispatch.py

@@ -12,25 +12,26 @@
 
 
 @mock.patch("sentry.monitors.clock_dispatch._dispatch_tick")
-def test_monitor_task_trigger(dispatch_tick):
+@override_options({"crons.use_clock_pulse_consumer": False})
+def test_monitor_task_trigger_legacy(dispatch_tick):
     now = timezone.now().replace(second=0, microsecond=0)
 
     # Assumes a single partition for simplicitly. Multi-partition cases are
     # covered in further test cases.
 
-    # First checkin triggers tasks
+    # First checkin triggers dispatch
     try_monitor_clock_tick(ts=now, partition=0)
     assert dispatch_tick.call_count == 1
 
-    # 5 seconds later does NOT trigger the task
+    # 5 seconds later does NOT trigger the dispatch
     try_monitor_clock_tick(ts=now + timedelta(seconds=5), partition=0)
     assert dispatch_tick.call_count == 1
 
-    # a minute later DOES trigger the task
+    # a minute later DOES trigger the dispatch
     try_monitor_clock_tick(ts=now + timedelta(minutes=1), partition=0)
     assert dispatch_tick.call_count == 2
 
-    # Same time does NOT trigger the task
+    # Same time does NOT trigger the dispatch
     try_monitor_clock_tick(ts=now + timedelta(minutes=1), partition=0)
     assert dispatch_tick.call_count == 2
 
@@ -42,6 +43,79 @@ def test_monitor_task_trigger(dispatch_tick):
         capture_message.assert_called_with("Monitor task dispatch minute skipped")
 
 
+@mock.patch("sentry.monitors.clock_dispatch._dispatch_tick")
+@override_options({"crons.use_clock_pulse_consumer": False})
+def test_monitor_task_trigger_partition_tick_skip_legacy(dispatch_tick):
+    now = timezone.now().replace(second=0, microsecond=0)
+
+    # Tick for 4 partitions
+    try_monitor_clock_tick(ts=now, partition=0)
+    try_monitor_clock_tick(ts=now, partition=1)
+    try_monitor_clock_tick(ts=now, partition=2)
+    try_monitor_clock_tick(ts=now, partition=3)
+    assert dispatch_tick.call_count == 1
+    assert dispatch_tick.mock_calls[0] == mock.call(now)
+
+    # Tick forward twice for 3 partitions
+    try_monitor_clock_tick(ts=now + timedelta(minutes=1), partition=0)
+    try_monitor_clock_tick(ts=now + timedelta(minutes=1), partition=1)
+    try_monitor_clock_tick(ts=now + timedelta(minutes=1), partition=2)
+
+    try_monitor_clock_tick(ts=now + timedelta(minutes=2), partition=0)
+    try_monitor_clock_tick(ts=now + timedelta(minutes=3), partition=1)
+    try_monitor_clock_tick(ts=now + timedelta(minutes=3), partition=2)
+    assert dispatch_tick.call_count == 1
+
+    # Slowest partition catches up, but has a timestamp gap, capture the fact
+    # that we skipped a minute
+    with mock.patch("sentry_sdk.capture_message") as capture_message:
+        assert capture_message.call_count == 0
+        try_monitor_clock_tick(ts=now + timedelta(minutes=2), partition=3)
+        capture_message.assert_called_with("Monitor task dispatch minute skipped")
+
+    # XXX(epurkhiser): Another approach we could take here is to detect the
+    # skipped minute and generate a tick for that minute, since we know
+    # processed past that minute.
+    #
+    # This still could be a problem though since it may mean we will not
+    # produce missed check-ins since the monitor already may have already
+    # checked-in after and moved the `next_checkin_latest` forward.
+    #
+    # In practice this should almost never happen since we have a high volume of
+
+    assert dispatch_tick.call_count == 2
+    assert dispatch_tick.mock_calls[1] == mock.call(now + timedelta(minutes=2))
+
+
+@mock.patch("sentry.monitors.clock_dispatch._dispatch_tick")
+@override_options({"crons.use_clock_pulse_consumer": True})
+def test_monitor_task_trigger(dispatch_tick):
+    now = timezone.now().replace(second=0, microsecond=0)
+
+    # Assumes a single partition for simplicitly. Multi-partition cases are
+    # covered in further test cases.
+
+    # First checkin triggers dispatch
+    try_monitor_clock_tick(ts=now, partition=0)
+    assert dispatch_tick.call_count == 1
+
+    # 5 seconds later does NOT trigger the dispatch
+    try_monitor_clock_tick(ts=now + timedelta(seconds=5), partition=0)
+    assert dispatch_tick.call_count == 1
+
+    # a minute later DOES trigger the dispatch
+    try_monitor_clock_tick(ts=now + timedelta(minutes=1), partition=0)
+    assert dispatch_tick.call_count == 2
+
+    # Same time does NOT trigger the dispatch
+    try_monitor_clock_tick(ts=now + timedelta(minutes=1), partition=0)
+    assert dispatch_tick.call_count == 2
+
+    # A skipped minute triggers the dispatch multiple times
+    try_monitor_clock_tick(ts=now + timedelta(minutes=3, seconds=5), partition=0)
+    assert dispatch_tick.call_count == 4
+
+
 @mock.patch("sentry.monitors.clock_dispatch._dispatch_tick")
 def test_monitor_task_trigger_partition_desync(dispatch_tick):
     """
@@ -52,7 +126,7 @@ def test_monitor_task_trigger_partition_desync(dispatch_tick):
     now = timezone.now().replace(second=0, microsecond=0)
 
     # First message in partition 0 with timestamp just after the minute
-    # boundary triggers the task
+    # boundary triggers the dispatch
     try_monitor_clock_tick(ts=now + timedelta(seconds=1), partition=0)
     assert dispatch_tick.call_count == 1
 
@@ -63,7 +137,7 @@ def test_monitor_task_trigger_partition_desync(dispatch_tick):
     assert dispatch_tick.call_count == 1
 
     # Third message in partition 1 again just after the minute boundary does
-    # NOT trigger the task, we've already ticked at that time.
+    # NOT trigger the dispatch, we've already ticked at that time.
     try_monitor_clock_tick(ts=now + timedelta(seconds=1), partition=1)
     assert dispatch_tick.call_count == 1
 
@@ -102,10 +176,12 @@ def test_monitor_task_trigger_partition_sync(dispatch_tick):
 
 
 @mock.patch("sentry.monitors.clock_dispatch._dispatch_tick")
+@override_options({"crons.use_clock_pulse_consumer": True})
 def test_monitor_task_trigger_partition_tick_skip(dispatch_tick):
     """
     In a scenario where all partitions move multiple ticks past the slowest
-    partition we may end up skipping a tick.
+    partition we may end up skipping a tick. In this scenario we will backfill
+    those ticks
     """
     now = timezone.now().replace(second=0, microsecond=0)
 
@@ -127,25 +203,12 @@ def test_monitor_task_trigger_partition_tick_skip(dispatch_tick):
     try_monitor_clock_tick(ts=now + timedelta(minutes=3), partition=2)
     assert dispatch_tick.call_count == 1
 
-    # Slowest partition catches up, but has a timestamp gap, capture the fact
-    # that we skipped a minute
-    with mock.patch("sentry_sdk.capture_message") as capture_message:
-        assert capture_message.call_count == 0
-        try_monitor_clock_tick(ts=now + timedelta(minutes=2), partition=3)
-        capture_message.assert_called_with("Monitor task dispatch minute skipped")
+    # Slowest partition catches up, but has a timestamp gap
+    try_monitor_clock_tick(ts=now + timedelta(minutes=2), partition=3)
 
-    # XXX(epurkhiser): Another approach we could take here is to detect the
-    # skipped minute and generate a tick for that minute, since we know
-    # processed past that minute.
-    #
-    # This still could be a problem though since it may mean we will not
-    # produce missed check-ins since the monitor already may have already
-    # checked-in after and moved the `next_checkin_latest` forward.
-    #
-    # In practice this should almost never happen since we have a high volume of
-
-    assert dispatch_tick.call_count == 2
-    assert dispatch_tick.mock_calls[1] == mock.call(now + timedelta(minutes=2))
+    assert dispatch_tick.call_count == 3
+    assert dispatch_tick.mock_calls[1] == mock.call(now + timedelta(minutes=1))
+    assert dispatch_tick.mock_calls[2] == mock.call(now + timedelta(minutes=2))
 
 
 @override_settings(KAFKA_TOPIC_OVERRIDES={"monitors-clock-tick": "clock-tick-test-topic"})