Properly align the timer to align_to

By using the Timer.periodic, the alignment was lost. We now make the
timer aligned to the align_to configuration by delaying its start.

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

Author: llucax

src/frequenz/sdk/timeseries/_resampling.py

@@ -16,6 +16,7 @@
 from typing import AsyncIterator, Callable, Coroutine, Optional, Sequence
 
 from frequenz.channels.util import Timer
+from frequenz.channels.util._timer import _to_microseconds
 
 from .._internal._asyncio import cancel_and_await
 from ._base_types import UNIX_EPOCH, Sample
@@ -369,7 +370,8 @@ def __init__(self, config: ResamplerConfig) -> None:
         self._resamplers: dict[Source, _StreamingHelper] = {}
         """A mapping between sources and the streaming helper handling that source."""
 
-        self._window_end: datetime = self._calculate_window_end()
+        window_end, start_delay_time = self._calculate_window_end()
+        self._window_end: datetime = window_end
         """The time in which the current window ends.
 
         This is used to make sure every resampling window is generated at
@@ -385,6 +387,15 @@ def __init__(self, config: ResamplerConfig) -> None:
         self._timer: Timer = Timer.periodic(config.resampling_period)
         """The timer used to trigger the resampling windows."""
 
+        # Hack to align the timer, this should be implemented in the Timer class
+        # If there is no delay, then we need to add one period, because the timer
+        # shouldn't fire right away
+        if not start_delay_time:
+            start_delay_time += config.resampling_period
+        self._timer._next_tick_time = _to_microseconds(
+            timedelta(seconds=asyncio.get_running_loop().time()) + start_delay_time
+        )  # pylint: disable=protected-access
+
     @property
     def config(self) -> ResamplerConfig:
         """Get the resampler configuration.
@@ -504,28 +515,36 @@ async def resample(self, *, one_shot: bool = False) -> None:
             if one_shot:
                 break
 
-    def _calculate_window_end(self) -> datetime:
+    def _calculate_window_end(self) -> tuple[datetime, timedelta]:
         """Calculate the end of the current resampling window.
 
         The calculated resampling window end is a multiple of
         `self._config.resampling_period` starting at `self._config.align_to`.
 
         if `self._config.align_to` is `None`, the current time is used.
 
+        If the current time is not aligned to `self._config.resampling_period`, then
+        the end of the current resampling window will be more than one period away, to
+        make sure to have some time to collect samples if the misalignment is too big.
+
         Returns:
-            The end of the current resampling window aligned to
-                `self._config.align_to`.
+            A tuple with the end of the current resampling window aligned to
+                `self._config.align_to` as the first item and the time we need to
+                delay the timer start to make sure it is also aligned.
         """
         now = datetime.now(timezone.utc)
         period = self._config.resampling_period
         align_to = self._config.align_to
 
         if align_to is None:
-            return now + period
+            return (now + period, timedelta(0))
 
         elapsed = (now - align_to) % period
 
-        return now + period - elapsed
+        return (
+            now + period - elapsed,
+            period - elapsed if elapsed else timedelta(0),
+        )
 
 
 class _ResamplingHelper:

tests/timeseries/test_resampling.py

@@ -191,35 +191,47 @@ async def test_helper_buffer_too_big(
             1.0,
             datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
             datetime(2020, 1, 1, tzinfo=timezone.utc),
-            datetime(2020, 1, 1, 2, 3, 6, tzinfo=timezone.utc),
+            (
+                datetime(2020, 1, 1, 2, 3, 6, tzinfo=timezone.utc),
+                timedelta(seconds=0.7),
+            ),
         ),
         (
             3.0,
             datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
             datetime(2020, 1, 1, 0, 0, 5, tzinfo=timezone.utc),
-            datetime(2020, 1, 1, 2, 3, 8, tzinfo=timezone.utc),
+            (
+                datetime(2020, 1, 1, 2, 3, 8, tzinfo=timezone.utc),
+                timedelta(seconds=2.7),
+            ),
         ),
         (
             10.0,
             datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
             datetime(2020, 1, 1, 0, 0, 5, tzinfo=timezone.utc),
-            datetime(2020, 1, 1, 2, 3, 15, tzinfo=timezone.utc),
+            (
+                datetime(2020, 1, 1, 2, 3, 15, tzinfo=timezone.utc),
+                timedelta(seconds=9.7),
+            ),
         ),
         # Future align_to
         (
             10.0,
             datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
             datetime(2020, 1, 1, 2, 3, 18, tzinfo=timezone.utc),
-            datetime(2020, 1, 1, 2, 3, 8, tzinfo=timezone.utc),
+            (
+                datetime(2020, 1, 1, 2, 3, 8, tzinfo=timezone.utc),
+                timedelta(seconds=2.7),
+            ),
         ),
     ),
 )
-def test_calculate_window_end_trivial_cases(
+async def test_calculate_window_end_trivial_cases(
     fake_time: time_machine.Coordinates,
     resampling_period_s: float,
     now: datetime,
     align_to: datetime,
-    result: datetime,
+    result: tuple[datetime, timedelta],
 ) -> None:
     """Test the calculation of the resampling window end for simple cases."""
     resampling_period = timedelta(seconds=resampling_period_s)
@@ -249,11 +261,10 @@ def test_calculate_window_end_trivial_cases(
     )
     fake_time.move_to(now)
     # pylint: disable=protected-access
-    assert (
-        resampler_now._calculate_window_end() == resampler_none._calculate_window_end()
-    )
+    none_result = resampler_none._calculate_window_end()
+    assert resampler_now._calculate_window_end() == none_result
     # pylint: disable=protected-access
-    assert resampler_none._calculate_window_end() == now + resampling_period
+    assert none_result[0] == now + resampling_period
 
 
 async def test_resampling_window_size_is_constant(
@@ -303,6 +314,7 @@ async def test_resampling_window_size_is_constant(
     await resampler.resample(one_shot=True)
 
     assert datetime.now(timezone.utc).timestamp() == 2
+    assert asyncio.get_event_loop().time() == 2
     sink_mock.assert_called_once_with(
         Sample(
             timestamp + timedelta(seconds=resampling_period_s),
@@ -343,7 +355,7 @@ async def test_resampling_window_size_is_constant(
     resampling_fun_mock.reset_mock()
 
 
-async def test_timer_errors_are_logged(
+async def test_timer_errors_are_logged(  # pylint: disable=too-many-statements
     fake_time: time_machine.Coordinates,
     source_chan: Broadcast[Sample[Quantity]],
     caplog: pytest.LogCaptureFixture,
@@ -384,15 +396,14 @@ async def test_timer_errors_are_logged(
     # T = timer tick
 
     # Send a few samples and run a resample tick, advancing the fake time by one period
-    # Important: this is needed because the resampling timer only starts after the first
-    # resapmle() is called, so the first resampling will never have a shift.
+    # No log message should be produced
     sample0s = Sample(timestamp, value=Quantity(5.0))
     sample1s = Sample(timestamp + timedelta(seconds=1.0), value=Quantity(12.0))
     await source_sender.send(sample0s)
     await source_sender.send(sample1s)
     # Here we need to advance only the wall clock because the resampler timer is not yet
     # started, otherwise the loop time will be advanced twice
-    fake_time.shift(resampling_period_s)
+    await _advance_time(fake_time, resampling_period_s)
     await resampler.resample(one_shot=True)
 
     assert datetime.now(timezone.utc).timestamp() == pytest.approx(2)
@@ -1112,6 +1123,82 @@ async def make_fake_source() -> Source:
     assert isinstance(timeseries_error, TestException)
 
 
+async def test_timer_is_aligned(
+    fake_time: time_machine.Coordinates,
+    source_chan: Broadcast[Sample[Quantity]],
+    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=timedelta(seconds=resampling_period_s),
+        max_data_age_in_periods=2.0,
+        resampling_function=resampling_fun_mock,
+        initial_buffer_len=4,
+    )
+
+    # Advance the time a bit so that the resampler is not aligned to the resampling
+    # period
+    await _advance_time(fake_time, resampling_period_s / 3)
+
+    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
+    #                   alignment
+    #                ,-------------.
+    #             start = 0.667
+    # t(s)   0       |  1    1.5   2
+    #        |-------+--|-----|----R-----> (no more samples)
+    # value            5.0   12.0
+    #
+    # R = resampling is done
+
+    # Send samples and resample
+    sample1s = Sample(timestamp + timedelta(seconds=1.0), value=Quantity(5.0))
+    sample1_5s = Sample(timestamp + timedelta(seconds=1.5), value=Quantity(12.0))
+    await source_sender.send(sample1s)
+    await source_sender.send(sample1_5s)
+    await _advance_time(fake_time, resampling_period_s * 2 / 3)
+    await resampler.resample(one_shot=True)
+
+    assert datetime.now(timezone.utc).timestamp() == pytest.approx(2)
+    assert asyncio.get_running_loop().time() == pytest.approx(2)
+    sink_mock.assert_called_once_with(
+        Sample(
+            timestamp + timedelta(seconds=resampling_period_s),
+            Quantity(expected_resampled_value),
+        )
+    )
+    resampling_fun_mock.assert_called_once_with(
+        a_sequence(sample1s, sample1_5s),
+        config,
+        source_props,
+    )
+    assert not [
+        *_filter_logs(
+            caplog.record_tuples,
+            logger_level=logging.WARNING,
+        )
+    ]
+    sink_mock.reset_mock()
+    resampling_fun_mock.reset_mock()
+
+
 def _get_buffer_len(resampler: Resampler, source_recvr: Source) -> int:
     # pylint: disable=protected-access
     blen = resampler._resamplers[source_recvr]._helper._buffer.maxlen