Allow the resampler to use an external timestamp

This is necessary to achieve full timestamp consistency across the whole
resampling machinery, otherwise there could be some inconsistencies when
calculating which are the current relevant samples to use when
resampling in the ComponentMetricResampler.

The ComponentMetricGroupResampler now passes the timestamp to all the
ComponentMetricResampler it controls.

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

Author: llucax

src/frequenz/sdk/timeseries/resampler.py

@@ -77,17 +77,22 @@ def remove_outdated_samples(self, threshold: datetime) -> None:
 
             self._buffer.popleft()
 
-    def resample(self) -> Optional[float]:
+    def resample(self, timestamp: Optional[datetime] = None) -> Optional[float]:
         """Resample samples from the buffer and produce a single sample.
 
+        Args:
+            timestamp: the timestamp to use to as the current resampling
+                timestamp when calculating which stored past samples are
+                relevant to pass to the resampling function. If `None`, the
+                current datetime (in UTC) will be used.
+
         Returns:
             Samples resampled into a single sample or `None` if the
                 `resampling_function` cannot produce a valid Sample.
         """
-        # It might be better to provide `now` from the outside so that all
-        # individual resamplers use the same `now`
-        now = datetime.now(timezone.utc)
-        threshold = now - timedelta(
+        if timestamp is None:
+            timestamp = datetime.now(timezone.utc)
+        threshold = timestamp - timedelta(
             seconds=self._max_data_age_in_periods * self._resampling_period_s
         )
         self.remove_outdated_samples(threshold=threshold)
@@ -182,4 +187,4 @@ def resample(self) -> Generator[Tuple[str, Sample], None, None]:
         """
         now = datetime.now(timezone.utc)
         for time_series_id, resampler in self._resamplers.items():
-            yield time_series_id, Sample(timestamp=now, value=resampler.resample())
+            yield time_series_id, Sample(timestamp=now, value=resampler.resample(now))

tests/timeseries/test_resampling.py

@@ -56,6 +56,12 @@ def test_component_metric_resampler_remove_outdated_samples() -> None:
     resampler.add_sample(sample1)
     resampler.add_sample(sample2)
 
+    resampler.remove_outdated_samples(threshold=timestamp)
+    assert list(resampler._buffer) == [
+        sample1,
+        sample2,
+    ]  # pylint: disable=protected-access
+
     resampler.remove_outdated_samples(threshold=timestamp + timedelta(seconds=0.5))
     assert list(resampler._buffer) == [sample2]  # pylint: disable=protected-access
 
@@ -72,13 +78,15 @@ def test_component_metric_resampler_resample() -> None:
         resampling_function=resampling_function_sum,
     )
 
-    timestamp = datetime.now(timezone.utc) - timedelta(seconds=0.5)
+    now = datetime.now(timezone.utc)
+    timestamp1 = now - timedelta(seconds=0.5)
+    timestamp2 = now - timedelta(seconds=0.2)
 
     value1 = 5.0
     value2 = 15.0
 
-    sample1 = Sample(timestamp, value=value1)
-    sample2 = Sample(timestamp, value=value2)
+    sample1 = Sample(timestamp1, value=value1)
+    sample2 = Sample(timestamp2, value=value2)
 
     resampler.add_sample(sample1)
     resampler.add_sample(sample2)
@@ -87,6 +95,10 @@ def test_component_metric_resampler_resample() -> None:
     assert value is not None
     assert value == sum([value1, value2])
 
+    value = resampler.resample(now + timedelta(seconds=0.6))
+    assert value is not None
+    assert value == value2
+
 
 @time_machine.travel(0, tick=False)
 def test_component_metric_resampler_resample_with_outdated_samples() -> None: