Rewrite (Group)Resampler class

The `GroupResampler` and `Resampler` classes are not very reusable, or
at least require quite a bit of work from users to use them and they
need to take care of handling timers and calling methods at the right
time for them to be useful.

The new `Resampler` class has its own task(s) to handle all the
resampling, users only need to subscribe to new timeseries by passing a
timeseries `Source` and `Sink`. The `Sink` will be called each time a
new sample is produced.

A new `frequenz.sdk.timeseries.resampling` module is added because there
are more classes only used by the `Resampler` that might not be needed
by anyone just wanting to use `frequenz.sdk.timeseries`.

This commit also introduces the use of `async-solipsism` for testing, so
tests using timers can be mocked in a way that they don't really need to
wait for timers and time comparisons can be done precisely.

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

Author: llucax

noxfile.py

@@ -13,7 +13,14 @@
 
 FMT_DEPS = ["black", "isort"]
 DOCSTRING_DEPS = ["pydocstyle", "darglint"]
-PYTEST_DEPS = ["pytest", "pytest-cov", "pytest-mock", "pytest-asyncio", "time-machine"]
+PYTEST_DEPS = [
+    "pytest",
+    "pytest-cov",
+    "pytest-mock",
+    "pytest-asyncio",
+    "time-machine",
+    "async-solipsism",
+]
 MYPY_DEPS = ["mypy", "pandas-stubs", "grpc-stubs"]
 
 

pyproject.toml

@@ -106,3 +106,7 @@ module = [
     "grpc.aio.*"
 ]
 ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+module = "async_solipsism"
+ignore_missing_imports = true

src/frequenz/sdk/actor/_resampling.py

@@ -6,13 +6,14 @@
 import asyncio
 import dataclasses
 import logging
-import math
-from typing import Dict, Sequence
+from typing import Sequence, Set
 
 from frequenz.channels import Receiver, Sender
-from frequenz.channels.util import MergeNamed, Select, Timer
 
-from ..timeseries import GroupResampler, ResamplingFunction, Sample
+from frequenz.sdk.util.asyncio import cancel_and_await
+
+from ..timeseries import Sample
+from ..timeseries.resampling import Resampler, ResamplingError, ResamplingFunction
 from ._channel_registry import ChannelRegistry
 from ._data_sourcing import ComponentMetricRequest
 from ._decorator import actor
@@ -141,97 +142,111 @@ async def run() -> None:
         self._resampling_period_s = resampling_period_s
         self._max_data_age_in_periods: float = max_data_age_in_periods
         self._resampling_function: ResamplingFunction = resampling_function
-
-        self._resampler = GroupResampler(
+        self._resampler = Resampler(
             resampling_period_s=resampling_period_s,
             max_data_age_in_periods=max_data_age_in_periods,
-            initial_resampling_function=resampling_function,
+            resampling_function=resampling_function,
         )
 
-        self._input_receivers: Dict[str, Receiver[Sample]] = {}
-        self._output_senders: Dict[str, Sender[Sample]] = {}
-        self._resampling_timer = Timer(interval=self._resampling_period_s)
-
     async def _subscribe(self, request: ComponentMetricRequest) -> None:
         """Subscribe for data for a specific time series.
 
         Args:
             request: subscription request for a specific component metric
         """
-        channel_name = request.get_channel_name()
-
-        data_source_request = dataclasses.replace(request, **dict(namespace="Source"))
+        data_source_request = dataclasses.replace(
+            request, namespace=request.namespace + ":Source"
+        )
         data_source_channel_name = data_source_request.get_channel_name()
-        if channel_name not in self._input_receivers:
-            await self._subscription_sender.send(data_source_request)
-            receiver: Receiver[Sample] = self._channel_registry.new_receiver(
-                data_source_channel_name
-            )
-            self._input_receivers[data_source_channel_name] = receiver
-            self._resampler.add_time_series(time_series_id=data_source_channel_name)
+        await self._subscription_sender.send(data_source_request)
+        receiver = self._channel_registry.new_receiver(data_source_channel_name)
 
-        if channel_name not in self._output_senders:
-            sender: Sender[Sample] = self._channel_registry.new_sender(channel_name)
-            # This means that the `sender` will be sending samples to the channel with
-            # name `channel_name` based on samples collected from the channel named
-            # `data_source_channel_name`
-            self._output_senders[data_source_channel_name] = sender
+        # This is a temporary hack until the Sender implementation uses
+        # exceptions to report errors.
+        sender = self._channel_registry.new_sender(request.get_channel_name())
 
-    def _is_sample_valid(self, sample: Sample) -> bool:
-        """Check if the provided sample is valid.
+        async def sink_adapter(sample: Sample) -> None:
+            if not await sender.send(sample):
+                raise Exception(f"Error while sending with sender {sender}", sender)
 
-        Args:
-            sample: sample to be validated
+        self._resampler.add_timeseries(receiver, sink_adapter)
 
-        Returns:
-            True if the sample is valid, False otherwise
-        """
-        if sample.value is None or math.isnan(sample.value):
-            return False
-        return True
+    async def _process_resampling_requests(self) -> None:
+        """Process resampling data requests."""
+        async for request in self._subscription_receiver:
+            await self._subscribe(request)
 
     async def run(self) -> None:
-        """Run the actor.
+        """Resample known component metrics and process resampling requests.
+
+        If there is a resampling error while resampling some component metric,
+        then that metric will be discarded and not resampled any more. Any
+        other error will be propagated (most likely ending in the actor being
+        restarted).
 
         Raises:
-            ConnectionError: When the provider of the subscription channel closes the
-                connection
+            RuntimeError: If there is some unexpected error while resampling or
+                handling requests.
+
+        # noqa: DAR401 error
         """
-        while True:
-            select = Select(
-                resampling_timer=self._resampling_timer,
-                subscription_receiver=self._subscription_receiver,
-                component_data_receiver=MergeNamed(**self._input_receivers),
+        tasks_to_cancel: Set[asyncio.Task] = set()
+        try:
+            subscriptions_task = asyncio.create_task(
+                self._process_resampling_requests()
             )
-            while await select.ready():
-                if msg := select.resampling_timer:
-                    assert msg.inner is not None, "The timer should never be 'closed'"
-                    timestamp = msg.inner
-                    awaitables = [
-                        self._output_senders[channel_name].send(sample)
-                        for channel_name, sample in self._resampler.resample(timestamp)
-                    ]
-                    await asyncio.gather(*awaitables)
-                if msg := select.component_data_receiver:
-                    if msg.inner is None:
-                        # When this happens, then DataSourcingActor has closed the channel
-                        # for sending data for a specific `ComponentMetricRequest`,
-                        # which may need to be handled properly here, e.g. unsubscribe
-                        continue
-                    channel_name, sample = msg.inner
-                    if self._is_sample_valid(sample=sample):
-                        self._resampler.add_sample(
-                            time_series_id=channel_name,
-                            sample=sample,
-                        )
-                if msg := select.subscription_receiver:
-                    if msg.inner is None:
-                        raise ConnectionError(
-                            "Subscription channel connection has been closed!"
+            tasks_to_cancel.add(subscriptions_task)
+
+            while True:
+                resampling_task = asyncio.create_task(self._resampler.resample())
+                tasks_to_cancel.add(resampling_task)
+                done, _ = await asyncio.wait(
+                    [resampling_task, subscriptions_task],
+                    return_when=asyncio.FIRST_COMPLETED,
+                )
+
+                if subscriptions_task in done:
+                    tasks_to_cancel.remove(subscriptions_task)
+                    raise RuntimeError(
+                        "There was a problem with the subscriptions channel."
+                    )
+
+                if resampling_task in done:
+                    tasks_to_cancel.remove(resampling_task)
+                    # The resampler shouldn't end without an exception
+                    error = resampling_task.exception()
+                    assert (
+                        error is not None
+                    ), "The resample() function shouldn't exit normally."
+
+                    # We don't know what to do with something other than
+                    # ResamplingError, so propagate the exception if that is the
+                    # case.
+                    if not isinstance(error, ResamplingError):
+                        raise error
+                    for source, source_error in error.exceptions.items():
+                        logger.error(
+                            "Error resampling source %s, removing source...", source
                         )
-                    await self._subscribe(request=msg.inner)
-                    # Breaking out from the loop is required to regenerate
-                    # component_data_receivers to be able to fulfil this
-                    # subscription (later can be optimized by checking if
-                    # an output channel already existed in the `subscribe()` method)
-                    break
+                        removed = self._resampler.remove_timeseries(source)
+                        if not removed:
+                            logger.warning(
+                                "Got an exception from an unknown source: "
+                                "source=%r, exception=%r",
+                                source,
+                                source_error,
+                            )
+                    # The resampling_task will be re-created if we reached this point
+        finally:
+            await asyncio.gather(*[cancel_and_await(t) for t in tasks_to_cancel])
+
+            # XXX: Here we should probably do a:  pylint: disable=fixme
+            # await self._resampler.stop()
+            # But since the actor will be restarted, the internal state would
+            # be broken if we stop the resampler.
+            #
+            # We have an even bigger problem with this naive restarting
+            # approach, as restarting this actor without really resetting its
+            # state would be mostly the same as not really leaving the run()
+            # method and just swallow any exception, which doesn't look super
+            # smart.

src/frequenz/sdk/timeseries/__init__.py

@@ -11,15 +11,23 @@
 [`Source`][frequenz.sdk.timeseries.Source]s and new timeseries can be
 generated by sending samples to
 a [`Sink`][frequenz.sdk.timeseries.Sink]
+
+In general timeseries `Source`s don't necessarily come at periodic
+intervals. To normalize timeseries to produce `Sample`s at regular
+periodic intervals,
+a [`Resampler`][frequenz.sdk.timeseries.resampling.Resampler] can be used.
+
+A `Resampler` uses
+a [`ResamplingFunction`][frequenz.sdk.timeseries.resampling.ResamplingFunction]
+to produce a new sample from samples received in the past. If there are no
+samples coming to a resampled timeseries for a while, eventually the
+`Resampler` will produce `Sample`s with `None` as value, meaning there is no
+way to produce meaningful samples with the available data.
 """
 
 from ._base_types import Sample, Sink, Source
-from ._resampler import GroupResampler, Resampler, ResamplingFunction
 
 __all__ = [
-    "GroupResampler",
-    "Resampler",
-    "ResamplingFunction",
     "Sample",
     "Sink",
     "Source",