Use the timestamp provided by the timer receiver

The resampling class now can take an optional timestamp to be used to
produce the new samples and the resampling actor now passes the
timestamp produced by the timer instead of calling `now()` manually.

This also adds a more explicit check to make sure the timer receiver is
not closed.

Fixes #54.

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

Author: llucax

src/frequenz/sdk/actor/resampling.py

@@ -208,10 +208,12 @@ async def run(self) -> None:
                 component_data_receiver=MergeNamed(**self._input_receivers),
             )
             while await select.ready():
-                if _ := select.resampling_timer:
+                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()
+                        for channel_name, sample in self._resampler.resample(timestamp)
                     ]
                     await asyncio.gather(*awaitables)
                 if msg := select.component_data_receiver:

src/frequenz/sdk/timeseries/resampler.py

@@ -179,12 +179,22 @@ def add_sample(self, time_series_id: str, sample: Sample) -> None:
                 f"No resampler for time series {time_series_id} found!"
             ) from err
 
-    def resample(self) -> Generator[Tuple[str, Sample], None, None]:
+    def resample(
+        self, timestamp: Optional[datetime] = None
+    ) -> Generator[Tuple[str, Sample], None, None]:
         """Resample samples for all time series.
 
+        Args:
+            timestamp: the timestamp to use to emit the new samples (and to
+                consider stored samples relevant for resampling. If `None`,
+                the current datetime (in UTC) will be used.
+
         Yields:
             iterator of time series ids and their newly resampled samples
         """
-        now = datetime.now(timezone.utc)
+        if timestamp is None:
+            timestamp = datetime.now(timezone.utc)
         for time_series_id, resampler in self._resamplers.items():
-            yield time_series_id, Sample(timestamp=now, value=resampler.resample(now))
+            yield time_series_id, Sample(
+                timestamp=timestamp, value=resampler.resample(timestamp)
+            )

tests/timeseries/test_resampling.py

@@ -143,17 +143,17 @@ def test_component_metric_group_resampler() -> None:
     resampler.add_time_series(time_series_id=time_series_id_1)
     resampler.add_time_series(time_series_id=time_series_id_2)
 
-    timestamp = datetime.now(timezone.utc)
+    now = datetime.now(timezone.utc)
 
     value11 = 5.0
     value12 = 15.0
     value21 = 100.0
     value22 = 999.0
 
-    sample11 = Sample(timestamp - timedelta(seconds=0.7), value=value11)
-    sample12 = Sample(timestamp - timedelta(seconds=0.5), value=value12)
-    sample21 = Sample(timestamp - timedelta(seconds=5.05), value=value21)
-    sample22 = Sample(timestamp - timedelta(seconds=0.99), value=value22)
+    sample11 = Sample(now - timedelta(seconds=0.7), value=value11)
+    sample12 = Sample(now - timedelta(seconds=0.5), value=value12)
+    sample21 = Sample(now - timedelta(seconds=5.05), value=value21)
+    sample22 = Sample(now - timedelta(seconds=0.99), value=value22)
 
     resampler.add_sample(time_series_id=time_series_id_1, sample=sample11)
     resampler.add_sample(time_series_id=time_series_id_1, sample=sample12)
@@ -162,8 +162,17 @@ def test_component_metric_group_resampler() -> None:
 
     resampled_samples = dict(resampler.resample())
 
-    assert resampled_samples[time_series_id_1].timestamp >= timestamp
+    assert resampled_samples[time_series_id_1].timestamp >= now
     assert resampled_samples[time_series_id_1].value == sum([value11, value12])
 
-    assert resampled_samples[time_series_id_2].timestamp >= timestamp
+    assert resampled_samples[time_series_id_2].timestamp >= now
     assert resampled_samples[time_series_id_2].value == value22
+
+    timestamp = now + timedelta(seconds=0.5)
+    resampled_samples = dict(resampler.resample(timestamp))
+
+    assert resampled_samples[time_series_id_1].timestamp == timestamp
+    assert resampled_samples[time_series_id_1].value == value12
+
+    assert resampled_samples[time_series_id_2].timestamp == timestamp
+    assert resampled_samples[time_series_id_2].value is None