fix: difference console counter rates per pod

src/dfe_engine/appmgmt/operations.py

@@ -17,12 +17,9 @@
 per call, and an honest error when the database cannot answer.
 """
 
-from __future__ import annotations
-
 import re
 from collections.abc import Sequence
 from dataclasses import dataclass, field
-from itertools import pairwise
 from pathlib import Path
 from typing import Any
 
@@ -140,9 +137,9 @@ class AppMetrics:
 class SourceSignals:
     """What one source's records are doing, as the loader counts them.
 
-    Either reading is None when the window holds nothing to compute it from -
-    an absent series, a single sample, or a counter that only went backwards
-    because a pod restarted mid-window.
+    Either reading is None when the window holds nothing to compute it from: no
+    series, or only series with a single sample. A pod that restarted mid-window
+    counts from zero after the restart.
     """
 
     table: str
@@ -258,12 +255,12 @@ def source_signals(self, table: str, loaders: Sequence[str]) -> SourceSignals:
             # Nothing is deployed to count records against a table, so no series exists.
             return SourceSignals(table=table, window_seconds=self._window)
         rows = self._run("source_landing_counter", {"services": services, "table": safe_table})
-        points = [(_epoch(ts) or 0.0, float(total)) for ts, total in rows]
+        series = [(float(increase), float(span), float(last)) for increase, span, last in rows]
         return SourceSignals(
             table=table,
             window_seconds=self._window,
-            records_per_min=_rate_per_minute(points),
-            last_seen_epoch=_last_increase(points),
+            records_per_min=_rate_per_minute(series),
+            last_seen_epoch=_last_increase(series),
         )
 
     def _gauges(self, service: str) -> dict[str, float]:
@@ -346,28 +343,23 @@ def http_readings(*, rows: Sequence[tuple]) -> tuple[dict[str, float], dict[str,
     return rates, gauges
 
 
-def _rate_per_minute(points: Sequence[tuple[float, float]]) -> float | None:
-    """Records a minute across the window, from the first and last fleet totals.
+def _rate_per_minute(series: Sequence[tuple[float, float, float]]) -> float | None:
+    """Records a minute for the fleet: each series' increase over its own span, summed.
 
-    A negative delta means a pod restarted and reset its counter, which would
-    otherwise report as negative throughput; the window has to roll past the
-    restart before the number means anything again.
+    Each entry is one ``source_landing_counter`` row, ``(increase, span_seconds,
+    last_increase)``. A series with a single sample spans no time, and with no
+    series spanning any there is no rate to report.
     """
-    if len(points) < 2:
-        return None
-    elapsed = points[-1][0] - points[0][0]
-    delta = points[-1][1] - points[0][1]
-    if elapsed <= 0 or delta < 0:
+    rates = [increase / span for increase, span, _ in series if span > 0]
+    if not rates:
         return None
-    return delta / elapsed * 60.0
+    return sum(rates) * 60.0
 
 
-def _last_increase(points: Sequence[tuple[float, float]]) -> float | None:
-    """When the counter last moved - a flat counter keeps reporting after traffic stops."""
-    for (_, previous), (timestamp, total) in reversed(list(pairwise(points))):
-        if total > previous:
-            return timestamp
-    return None
+def _last_increase(series: Sequence[tuple[float, float, float]]) -> float | None:
+    """When any series last moved - a flat counter keeps reporting after traffic stops."""
+    latest = max((last for _, _, last in series), default=0.0)
+    return latest if latest > 0 else None
 
 
 def _epoch(value: Any) -> float | None:

src/dfe_engine/appmgmt/queries.yaml

@@ -46,29 +46,29 @@ queries:
       window_seconds: 300
     timeout_seconds: 10
 
-  # Counters as a per-second rate for the instance as a whole. The fleet total is
-  # summed per timestamp first, then differenced across the window, because a pushed
-  # row names no pod until the app runs scalo-rs#331 (service.instance.id on the
-  # resource), so replicas cannot be told apart and differencing one pod's counter
-  # against another's would report the gap between them as traffic.
-  # The sum lines replicas up only where they export in the same second, because
-  # TimeUnix has one-second precision.
-  # A pod restart resets its counter, which under-reports until the window rolls
-  # past it, so this is an operational estimate rather than an exact count.
+  # Counters as a per-second rate for the instance as a whole, summed from each
+  # series' own rate so pods exporting in different seconds are never subtracted from
+  # one another. A series is keyed as latest_gauges keys it; replicas pushing without
+  # a pod id share one, and the alternation between their counters counts as traffic.
+  # A reading below the one before it is a restart and counts from zero, and a series
+  # with one reading in the window spans no time and adds nothing.
   counter_rates:
     database: otel
     sql: >
-      SELECT
-        MetricName,
-        (argMax(total, ts) - argMin(total, ts))
-          / greatest(1, dateDiff('second', min(ts), max(ts))) AS value
+      SELECT MetricName, sum(if(span_seconds > 0, increase / span_seconds, 0)) AS value
       FROM (
-        SELECT MetricName, TimeUnix AS ts, sum(Value) AS total
+        SELECT
+          MetricName,
+          arraySort(groupArray((TimeUnix, Value))) AS points,
+          arrayMap(point -> point.2, points) AS readings,
+          arraySum(arrayMap((diff, reading) -> if(diff < 0, reading, diff),
+                            arrayDifference(readings), readings)) AS increase,
+          dateDiff('second', min(TimeUnix), max(TimeUnix)) AS span_seconds
         FROM __DB__.otel_metrics_sum
         WHERE ServiceName = {service:String}
           AND MetricName IN {names:Array(String)}
           AND TimeUnix >= now() - INTERVAL {window_seconds:UInt32} SECOND
-        GROUP BY MetricName, ts
+        GROUP BY MetricName, cityHash64(Attributes), ResourceAttributes['service.instance.id']
       )
       GROUP BY MetricName
     params: [service]
@@ -132,25 +132,36 @@ queries:
       bucket_seconds: 60
     timeout_seconds: 20
 
-  # Per-timestamp fleet totals of the loader's per-table counter, for the ONE
-  # table a source's records land in. `loader_messages_by_table_total` carries the
-  # table as an OTel attribute (dfe-loader src/metrics/mod.rs:217), and the loader
-  # is the only stage counting a record against a table.
-  # The fleet is summed per timestamp for the reason counter_rates gives, and the
-  # rate and the last increase are differenced by the caller rather than in SQL.
+  # The loader's per-table counter, one row per series, for the ONE table a
+  # source's records land in. `loader_messages_by_table_total` carries the table as
+  # an OTel attribute (dfe-loader src/metrics/mod.rs:217), and the loader is the
+  # only stage counting a record against a table.
+  # Each series is differenced as counter_rates does it, with ServiceName in the key
+  # because several loader instances can answer, and the caller sums the series.
+  # last_increase is the epoch second of the series' latest rise, 0 if it never rose.
   # ServiceName is bound to the deployed loader instances so the query still hits
   # the metrics tables' ORDER BY prefix (ServiceName, MetricName, ...).
   source_landing_counter:
     database: otel
     sql: >
-      SELECT TimeUnix AS ts, sum(Value) AS total
-      FROM __DB__.otel_metrics_sum
-      WHERE ServiceName IN {services:Array(String)}
-        AND MetricName = 'loader_messages_by_table_total'
-        AND Attributes['table'] = {table:String}
-        AND TimeUnix >= now() - INTERVAL {window_seconds:UInt32} SECOND
-      GROUP BY ts
-      ORDER BY ts
+      SELECT increase, span_seconds, last_increase
+      FROM (
+        SELECT
+          arraySort(groupArray((TimeUnix, Value))) AS points,
+          arrayMap(point -> point.2, points) AS readings,
+          arrayMap((diff, reading) -> if(diff < 0, reading, diff),
+                   arrayDifference(readings), readings) AS increments,
+          arraySum(increments) AS increase,
+          dateDiff('second', min(TimeUnix), max(TimeUnix)) AS span_seconds,
+          arrayMax(arrayMap((point, step) -> if(step > 0, toUnixTimestamp(point.1), 0),
+                            points, increments)) AS last_increase
+        FROM __DB__.otel_metrics_sum
+        WHERE ServiceName IN {services:Array(String)}
+          AND MetricName = 'loader_messages_by_table_total'
+          AND Attributes['table'] = {table:String}
+          AND TimeUnix >= now() - INTERVAL {window_seconds:UInt32} SECOND
+        GROUP BY ServiceName, cityHash64(Attributes), ResourceAttributes['service.instance.id']
+      )
     params: [services, table]
     binds:
       window_seconds: 300