Add a warning if the resampling timer delay is too long

The resampling timer will always be fired after a resampling period has
passed (unless we are unrealistically lucky). If we detect the timer
fired much later than the resampling period (10%), we now emit a warning
as there is probably something doing too much processing in the main
thread, and we probably want to detect that issue early.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/sdk/timeseries/_resampling.py

@@ -447,12 +447,12 @@ async def resample(self, *, one_shot: bool = False) -> None:
                 timeseries from the resampler before calling this method
                 again).
         """
-        async for _ in self._timer:
+        async for timer_timestamp in self._timer:
             results = await asyncio.gather(
                 *[r.resample(self._window_end) for r in self._resamplers.values()],
                 return_exceptions=True,
             )
-            self._update_window_end()
+            self._update_window_end(timer_timestamp)
             exceptions = {
                 source: results[i]
                 for i, source in enumerate(self._resamplers)
@@ -465,7 +465,23 @@ async def resample(self, *, one_shot: bool = False) -> None:
             if one_shot:
                 break
 
-    def _update_window_end(self) -> None:
+    def _update_window_end(self, timer_timestamp: datetime) -> None:
+        # We use abs() here to account for errors in the timer where the timer
+        # fires before its time even when in theory it shouldn't be possible,
+        # but we want to be resilient to timer implementation changes that
+        # could end up in this case, either because there were some time jump
+        # or some rounding error.
+        timer_error_s = abs((timer_timestamp - self._window_end).total_seconds())
+        if timer_error_s > (self._config.resampling_period_s / 10.0):
+            _logger.warning(
+                "The resampling timer fired too late. It should have fired at "
+                "%s, but it fired at %s (%s seconds difference; resampling "
+                "period is %s seconds)",
+                self._window_end,
+                timer_timestamp,
+                timer_error_s,
+                self._config.resampling_period_s,
+            )
         self._window_end = self._window_end + timedelta(
             seconds=self._config.resampling_period_s
         )

tests/timeseries/test_resampling.py

@@ -256,6 +256,105 @@ async def test_resampling_window_size_is_constant(
     resampling_fun_mock.reset_mock()
 
 
+async def test_timer_errors_are_logged(
+    fake_time: time_machine.Coordinates,
+    source_chan: Broadcast[Sample],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test that big differences between the expected window end and the fired timer are logged."""
+    timestamp = datetime.now(timezone.utc)
+
+    resampling_period_s = 2
+    expected_resampled_value = 42.0
+
+    resampling_fun_mock = MagicMock(
+        spec=ResamplingFunction, return_value=expected_resampled_value
+    )
+    config = ResamplerConfig(
+        resampling_period_s=resampling_period_s,
+        max_data_age_in_periods=2.0,
+        resampling_function=resampling_fun_mock,
+        initial_buffer_len=4,
+    )
+    resampler = Resampler(config)
+
+    source_receiver = source_chan.new_receiver()
+    source_sender = source_chan.new_sender()
+
+    sink_mock = AsyncMock(spec=Sink, return_value=True)
+
+    resampler.add_timeseries("test", source_receiver, sink_mock)
+    source_props = resampler.get_source_properties(source_receiver)
+
+    # Test timeline
+    #
+    # t(s)   0          1          2   2.5    3          4
+    #        |----------|----------R----|-----|----------R-----> (no more samples)
+    # value  5.0       12.0            2.0   4.0        5.0
+    #
+    # R = resampling is done
+
+    # Send a few samples and run a resample tick, advancing the fake time by one period
+    sample0s = Sample(timestamp, value=5.0)
+    sample1s = Sample(timestamp + timedelta(seconds=1.0), value=12.0)
+    await source_sender.send(sample0s)
+    await source_sender.send(sample1s)
+    fake_time.shift(resampling_period_s * 1.0999)  # Timer is delayed 9.99%
+    await resampler.resample(one_shot=True)
+
+    assert datetime.now(timezone.utc).timestamp() == pytest.approx(2.1998)
+    sink_mock.assert_called_once_with(
+        Sample(
+            timestamp + timedelta(seconds=resampling_period_s), expected_resampled_value
+        )
+    )
+    resampling_fun_mock.assert_called_once_with(
+        a_sequence(sample0s, sample1s), config, source_props
+    )
+    assert not [
+        *_filter_logs(
+            caplog.record_tuples,
+            logger_level=logging.WARNING,
+        )
+    ]
+    sink_mock.reset_mock()
+    resampling_fun_mock.reset_mock()
+
+    # Second resampling run, now with 10% delay
+    sample2_5s = Sample(timestamp + timedelta(seconds=2.5), value=2.0)
+    sample3s = Sample(timestamp + timedelta(seconds=3), value=4.0)
+    sample4s = Sample(timestamp + timedelta(seconds=4), value=5.0)
+    await source_sender.send(sample2_5s)
+    await source_sender.send(sample3s)
+    await source_sender.send(sample4s)
+    fake_time.shift(resampling_period_s * 1.10)  # Timer delayed 10%
+    await resampler.resample(one_shot=True)
+
+    assert datetime.now(timezone.utc).timestamp() == pytest.approx(2.1998 + 2.2)
+    sink_mock.assert_called_once_with(
+        Sample(
+            # But the sample still gets 4s as timestamp, because we are keeping
+            # the window size constant, not dependent on when the timer fired
+            timestamp + timedelta(seconds=resampling_period_s * 2),
+            expected_resampled_value,
+        )
+    )
+    resampling_fun_mock.assert_called_once_with(
+        a_sequence(sample1s, sample2_5s, sample3s, sample4s),
+        config,
+        source_props,
+    )
+    assert (
+        "frequenz.sdk.timeseries._resampling",
+        logging.WARNING,
+        "The resampling timer fired too late. It should have fired at 1970-01-01 00:00:04+00:00, "
+        "but it fired at 1970-01-01 00:00:04.399800+00:00 (0.3998 seconds difference; resampling "
+        "period is 2 seconds)",
+    ) in _filter_logs(caplog.record_tuples, logger_level=logging.WARNING)
+    sink_mock.reset_mock()
+    resampling_fun_mock.reset_mock()
+
+
 async def test_resampling_with_one_window(
     fake_time: time_machine.Coordinates, source_chan: Broadcast[Sample]
 ) -> None:
@@ -789,6 +888,13 @@ def _get_buffer_len(resampler: Resampler, source_recvr: Source) -> int:
 
 
 def _filter_logs(
-    record_tuples: list[tuple[str, int, str]], *, logger_name: str
+    record_tuples: list[tuple[str, int, str]],
+    *,
+    logger_name: str,
+    logger_level: int | None = None,
 ) -> list[tuple[str, int, str]]:
-    return [t for t in record_tuples if t[0] == logger_name]
+    return [
+        t
+        for t in record_tuples
+        if t[0] == logger_name and (logger_level is None or logger_level == t[1])
+    ]