Update the buffer length based on the input sampling period

When the input sampling period is know, we can adjust the internal
buffer to have a good size to be able to store all the requested
resampling and not waste space if the initial buffer was too big.

If we are upsampling, one sample could be enough for back-filling, but
we store max_data_age_in_periods using the input sampling period, so
the resampling functions can do more complex inter/extrapolation if they
need to.

If we are upsampling, we want a buffer that can hold
max_data_age_in_periods * resampling_period_s seconds of data, and we
have one sample every input_sampling_period_s, so we use a buffer length
of:
max_data_age_in_periods * resampling_period_s / input_sampling_period_s

The buffer size, like the input sampling period, is only calculated
once. The buffer data is copied to a new buffer when/if the buffer needs
resizing. The new size is validated against config.warn_buffer_len and
config.max_buffer_len and will emit a warning if it goes above any of
those and if it goes above config.max_buffer_len it will be truncated to
config.max_buffer_len as a safety measure.

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

Author: llucax

src/frequenz/sdk/timeseries/_resampling.py

@@ -531,6 +531,65 @@ def _update_source_sample_period(self, now: datetime) -> bool:
         )
         return True
 
+    def _update_buffer_len(self) -> bool:
+        """Update the length of the buffer based on the source properties.
+
+        Returns:
+            Whether the buffer length was changed (was really updated).
+        """
+        input_sampling_period_s = self._source_properties.sampling_period_s
+
+        # To make type checking happy
+        assert input_sampling_period_s is not None
+        assert self._buffer.maxlen is not None
+
+        config = self._config
+
+        # If we are upsampling, one sample could be enough for back-filling, but
+        # we store max_data_age_in_periods for input periods, so resampling
+        # functions can do more complex inter/extrapolation if they need to.
+        if input_sampling_period_s > config.resampling_period_s:
+            new_buffer_len = input_sampling_period_s * config.max_data_age_in_periods
+        # If we are upsampling, we want a buffer that can hold
+        # max_data_age_in_periods * resampling_period_s seconds of data, and we
+        # one sample every input_sampling_period_s.
+        else:
+            new_buffer_len = (
+                config.resampling_period_s
+                / input_sampling_period_s
+                * config.max_data_age_in_periods
+            )
+
+        new_buffer_len = max(1, math.ceil(new_buffer_len))
+        if new_buffer_len > config.max_buffer_len:
+            _logger.error(
+                "The new buffer length (%s) for timeseries %s is too big, using %s instead",
+                new_buffer_len,
+                self._name,
+                config.max_buffer_len,
+            )
+            new_buffer_len = config.max_buffer_len
+        elif new_buffer_len > config.warn_buffer_len:
+            _logger.warning(
+                "The new buffer length (%s) for timeseries %s bigger than %s",
+                new_buffer_len,
+                self._name,
+                config.warn_buffer_len,
+            )
+
+        if new_buffer_len == self._buffer.maxlen:
+            return False
+
+        _logger.info(
+            "New buffer length calculated for %r: %s",
+            self._name,
+            new_buffer_len,
+        )
+
+        self._buffer = deque(self._buffer, maxlen=new_buffer_len)
+
+        return True
+
     def resample(self, timestamp: datetime) -> Sample:
         """Generate a new sample based on all the current *relevant* samples.
 
@@ -543,7 +602,8 @@ def resample(self, timestamp: datetime) -> Sample:
                 If there are no *relevant* samples, then the new sample will
                 have `None` as `value`.
         """
-        self._update_source_sample_period(timestamp)
+        if self._update_source_sample_period(timestamp):
+            self._update_buffer_len()
 
         conf = self._config
         props = self._source_properties

tests/timeseries/test_resampling.py

@@ -5,6 +5,7 @@
 Tests for the `TimeSeriesResampler`
 """
 
+import logging
 from datetime import datetime, timedelta, timezone
 from typing import AsyncIterator, Iterator
 from unittest.mock import AsyncMock, MagicMock
@@ -26,6 +27,7 @@
     Source,
     SourceProperties,
     SourceStoppedError,
+    _ResamplingHelper,
 )
 
 from ..utils import a_sequence
@@ -137,6 +139,35 @@ async def test_resampler_config_len_error(init_len: int) -> None:
         )
 
 
+async def test_helper_buffer_too_big(
+    fake_time: time_machine.Coordinates,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test checks on the resampling buffer."""
+    config = ResamplerConfig(
+        resampling_period_s=DEFAULT_BUFFER_LEN_MAX + 1,
+        max_data_age_in_periods=1,
+    )
+    helper = _ResamplingHelper("test", config)
+
+    for i in range(DEFAULT_BUFFER_LEN_MAX + 1):
+        sample = Sample(datetime.now(timezone.utc), i)
+        helper.add_sample(sample)
+        fake_time.shift(1)
+
+    _ = helper.resample(datetime.now(timezone.utc))
+    assert caplog.record_tuples == [
+        (
+            "frequenz.sdk.timeseries._resampling",
+            logging.ERROR,
+            f"The new buffer length ({DEFAULT_BUFFER_LEN_MAX + 1}) "
+            f"for timeseries test is too big, using {DEFAULT_BUFFER_LEN_MAX} instead",
+        )
+    ]
+    # pylint: disable=protected-access
+    assert helper._buffer.maxlen == DEFAULT_BUFFER_LEN_MAX
+
+
 async def test_resampling_with_one_window(
     fake_time: time_machine.Coordinates, source_chan: Broadcast[Sample]
 ) -> None:
@@ -193,6 +224,7 @@ async def test_resampling_with_one_window(
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=2, sampling_period_s=None
     )
+    assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
     resampling_fun_mock.reset_mock()
 
@@ -221,6 +253,9 @@ async def test_resampling_with_one_window(
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=5, sampling_period_s=0.8
     )
+    # The buffer should be able to hold 2 seconds of data, and data is coming
+    # every 0.8 seconds, so we should be able to store 3 samples.
+    assert _get_buffer_len(resampler, source_recvr) == 3
     sink_mock.reset_mock()
     resampling_fun_mock.reset_mock()
 
@@ -236,6 +271,7 @@ async def test_resampling_with_one_window(
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=5, sampling_period_s=0.8
     )
+    assert _get_buffer_len(resampler, source_recvr) == 3
 
 
 # Even when a lot could be refactored to use smaller functions, I'm allowing
@@ -297,6 +333,7 @@ async def test_resampling_with_one_and_a_half_windows(  # pylint: disable=too-ma
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=2, sampling_period_s=None
     )
+    assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
     resampling_fun_mock.reset_mock()
 
@@ -324,6 +361,7 @@ async def test_resampling_with_one_and_a_half_windows(  # pylint: disable=too-ma
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=5, sampling_period_s=None
     )
+    assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
     resampling_fun_mock.reset_mock()
 
@@ -351,6 +389,10 @@ async def test_resampling_with_one_and_a_half_windows(  # pylint: disable=too-ma
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=7, sampling_period_s=6 / 7
     )
+    # The buffer should be able to hold 2 * 1.5 (3) seconds of data, and data
+    # is coming every 6/7 seconds (~0.857s), so we should be able to store
+    # 4 samples.
+    assert _get_buffer_len(resampler, source_recvr) == 4
     sink_mock.reset_mock()
     resampling_fun_mock.reset_mock()
 
@@ -386,6 +428,7 @@ async def test_resampling_with_one_and_a_half_windows(  # pylint: disable=too-ma
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=7, sampling_period_s=6 / 7
     )
+    assert _get_buffer_len(resampler, source_recvr) == 4
 
 
 # Even when a lot could be refactored to use smaller functions, I'm allowing
@@ -447,6 +490,7 @@ async def test_resampling_with_two_windows(  # pylint: disable=too-many-statemen
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=2, sampling_period_s=None
     )
+    assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
     resampling_fun_mock.reset_mock()
 
@@ -474,6 +518,7 @@ async def test_resampling_with_two_windows(  # pylint: disable=too-many-statemen
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=5, sampling_period_s=None
     )
+    assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
     resampling_fun_mock.reset_mock()
 
@@ -501,6 +546,7 @@ async def test_resampling_with_two_windows(  # pylint: disable=too-many-statemen
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=7, sampling_period_s=None
     )
+    assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
     resampling_fun_mock.reset_mock()
 
@@ -522,6 +568,7 @@ async def test_resampling_with_two_windows(  # pylint: disable=too-many-statemen
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=7, sampling_period_s=None
     )
+    assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
     resampling_fun_mock.reset_mock()
 
@@ -537,6 +584,7 @@ async def test_resampling_with_two_windows(  # pylint: disable=too-many-statemen
     assert source_props == SourceProperties(
         sampling_start=timestamp, received_samples=7, sampling_period_s=None
     )
+    assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
 
 
 async def test_receiving_stopped_resampling_error(
@@ -643,3 +691,10 @@ async def make_fake_source() -> Source:
     assert fake_source in exceptions
     timeseries_error = exceptions[fake_source]
     assert isinstance(timeseries_error, TestException)
+
+
+def _get_buffer_len(resampler: Resampler, source_recvr: Source) -> int:
+    # pylint: disable=protected-access
+    blen = resampler._resamplers[source_recvr]._helper._buffer.maxlen
+    assert blen is not None
+    return blen