Updating channel metric accumulation logic, fixed snapshots.

Author: angelosantos4

nodestream/metrics.py

@@ -214,7 +214,9 @@ def increment(self, metric: Metric, value: Number):
         self.metrics[metric] = self.metrics.get(metric, 0) + value
 
     def decrement(self, metric: Metric, value: Number):
-        self.metrics[metric] = self.metrics.get(metric, 0) - value
+        # Decrementing metrics that are accumulated doesn't help.
+        if not metric.accumulate:
+            self.metrics[metric] = self.metrics.get(metric, 0) - value
 
     def discharge(self) -> dict[Metric, Number]:
         metrics = {}
@@ -248,7 +250,9 @@ def increment(self, metric: Metric, value: Number):
         self.metrics[metric] = self.metrics.get(metric, 0) + value
 
     def decrement(self, metric: Metric, value: Number):
-        self.metrics[metric] = self.metrics.get(metric, 0) - value
+        # Decrementing metrics that are accumulated doesn't help.
+        if not metric.accumulate:
+            self.metrics[metric] = self.metrics.get(metric, 0) - value
 
     def discharge(self) -> dict[Metric, Number]:
         metrics = {}

nodestream/pipeline/channel.py

@@ -51,6 +51,7 @@ def create_with_naming(
         metric = Metric(
             f"buffered_{input_name}_to_{output_name}",
             f"Records buffered: {input_name} → {output_name}",
+            accumulate=True,
         )
         return cls(size, metric)
 

nodestream/pipeline/progress_reporter.py

@@ -64,10 +64,11 @@ def for_testing(cls, results_list: list) -> "PipelineProgressReporter":
         return cls(
             reporting_frequency=1,
             logger=getLogger("test"),
-            callback=lambda _, record: results_list.append(record),
+            callback=no_op,
             on_start_callback=no_op,
             on_finish_callback=no_op,
             on_fatal_error_callback=raise_exception,
+            observability_callback=lambda _: lambda record: results_list.append(record),
         )
 
     def report(self, index, metrics: Metrics):

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nodestream"
-version = "0.14.12"
+version = "0.14.13"
 description = "A Fast, Declarative ETL for Graph Databases."
 license = "GPL-3.0-only"
 authors = [