Add a maximum resampling buffer size

This is mainly because in a near future the resampling buffer size will
dynamically change based on the input sampling rate, so we need to make
sure it doesn't get out of hand in the event of some unlikely high input
sampling rate combined with a very low frequency resampling.

We also add a warning size, for which we just emit a warning in the logs
this is to also try to minimize the presence of very big buffers without
affecting the running apps (unless again, it gets out of hand). When
this happens users should either try to use a different resampling
period to keep buffer smaller, or change the (default) warning buffer
length if they know it is still manageable.

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

Author: llucax

benchmarks/timeseries/resampling.py

@@ -26,6 +26,8 @@ def _benchmark_resampling_helper(resamples: int, samples: int) -> None:
             max_data_age_in_periods=3.0,
             resampling_function=nop,
             initial_buffer_len=samples * 3,
+            max_buffer_len=samples * 3,
+            warn_buffer_len=samples * 3,
         )
     )
     now = datetime.now(timezone.utc)

src/frequenz/sdk/timeseries/_resampling.py

@@ -32,6 +32,21 @@
 """
 
 
+DEFAULT_BUFFER_LEN_MAX = 1024
+"""Default maximum allowed buffer length.
+
+If a buffer length would get bigger than this, it will be truncated to this
+length.
+"""
+
+
+DEFAULT_BUFFER_LEN_WARN = 128
+"""Default minimum buffer length that will produce a warning.
+
+If a buffer length would get bigger than this, a warning will be logged.
+"""
+
+
 Source = AsyncIterator[Sample]
 """A source for a timeseries.
 
@@ -132,6 +147,51 @@ class ResamplerConfig:
     can be stored.
     """
 
+    warn_buffer_len: int = DEFAULT_BUFFER_LEN_WARN
+    """The minimum length of the resampling buffer that will emit a warning.
+
+    If a buffer grows bigger than this value, it will emit a warning in the
+    logs, so buffers don't grow too big inadvertly.
+    """
+
+    max_buffer_len: int = DEFAULT_BUFFER_LEN_MAX
+    """The maximum length of the resampling buffer.
+
+    Buffers won't be allowed to grow beyond this point even if it would be
+    needed to keep all the requested past sampling periods. An error will be
+    emitted in the logs if the buffer length needs to be truncated to this
+    value.
+    """
+
+    def __post_init__(self) -> None:
+        """Check config values are valid.
+
+        Raises:
+            ValueError: if the initial buffer length is too small (less than 2)
+                or too big (more than `max_buffer_len`).
+        """
+        if self.initial_buffer_len < 2:
+            raise ValueError(
+                f"initial_buffer_len must be at least 2, got {self.initial_buffer_len}"
+            )
+        if self.initial_buffer_len > self.max_buffer_len:
+            raise ValueError(
+                f"initial_buffer_len be smaller than {self.max_buffer_len}, "
+                "got {self.initial_buffer_len}"
+            )
+        if self.initial_buffer_len > self.warn_buffer_len:
+            _logger.warning(
+                "initial_buffer_len (%s) is bigger than %s",
+                self.initial_buffer_len,
+                self.warn_buffer_len,
+            )
+        if self.initial_buffer_len > self.warn_buffer_len:
+            _logger.warning(
+                "initial_buffer_len (%s) is bigger than warn_buffer_len (%s)",
+                self.initial_buffer_len,
+                self.warn_buffer_len,
+            )
+
 
 class SourceStoppedError(RuntimeError):
     """A timeseries stopped producing samples."""

tests/timeseries/test_resampling.py

@@ -16,6 +16,8 @@
 
 from frequenz.sdk.timeseries import Sample
 from frequenz.sdk.timeseries._resampling import (
+    DEFAULT_BUFFER_LEN_MAX,
+    DEFAULT_BUFFER_LEN_WARN,
     Resampler,
     ResamplerConfig,
     ResamplingError,
@@ -78,6 +80,62 @@ async def _assert_no_more_samples(  # pylint: disable=too-many-arguments
         resampling_fun_mock.reset_mock()
 
 
+@pytest.mark.parametrize("init_len", list(range(2, DEFAULT_BUFFER_LEN_WARN + 1, 16)))
+async def test_resampler_config_len_ok(
+    init_len: int,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test checks on the resampling buffer."""
+    config = ResamplerConfig(
+        resampling_period_s=1.0,
+        initial_buffer_len=init_len,
+    )
+    assert config.initial_buffer_len == init_len
+    assert caplog.records == []
+
+
+@pytest.mark.parametrize(
+    "init_len",
+    range(DEFAULT_BUFFER_LEN_WARN + 1, DEFAULT_BUFFER_LEN_MAX + 1, 64),
+)
+async def test_resampler_config_len_warn(
+    init_len: int, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test checks on the resampling buffer."""
+    config = ResamplerConfig(
+        resampling_period_s=1.0,
+        initial_buffer_len=init_len,
+    )
+    assert config.initial_buffer_len == init_len
+    for record in caplog.records:
+        assert record.levelname == "WARNING"
+        assert caplog.text.startswith("")
+        assert (
+            caplog.text
+            == f"initial_buffer_len ({init_len}) is bigger than {DEFAULT_BUFFER_LEN_WARN}"
+    assert caplog.record_tuples == [
+        (
+            "frequenz.sdk.timeseries._resampling",
+            logging.WARNING,
+            f"initial_buffer_len ({init_len}) is bigger than "
+            f"warn_buffer_len ({DEFAULT_BUFFER_LEN_WARN})",
+        )
+    ]
+
+
+@pytest.mark.parametrize(
+    "init_len",
+    list(range(-2, 2)) + [DEFAULT_BUFFER_LEN_MAX + 1, DEFAULT_BUFFER_LEN_MAX + 2],
+)
+async def test_resampler_config_len_error(init_len: int) -> None:
+    """Test checks on the resampling buffer."""
+    with pytest.raises(ValueError):
+        _ = ResamplerConfig(
+            resampling_period_s=1.0,
+            initial_buffer_len=init_len,
+        )
+
+
 async def test_resampling_with_one_window(
     fake_time: time_machine.Coordinates, source_chan: Broadcast[Sample]
 ) -> None: