From 7f62885900289a12e485fca953e285f728f8a0d4 Mon Sep 17 00:00:00 2001 From: Derek Date: Mon, 12 Oct 2026 10:23:46 +1100 Subject: [PATCH] fix: difference console counter rates per pod The console summed each counter across the fleet per timestamp, then differenced the first and last totals. TimeUnix has one-second precision and pods export on their own clocks, so the totals rarely lined up and consecutive points subtracted one pod's counter from another's. Two pods each adding 100 over 60s, 7s apart, read 61.19/s instead of 3.33/s. The per-source signals had the same shape: two loader pods with flat counters read 3582 records a minute and a fresh arrival on every export. Each series is now differenced against its own previous reading and the per-series rates are summed. A series is one metric name, one attribute set and one pod (service.instance.id), the key latest_gauges uses. A reading below the one before it is a restart and counts from zero, so a restart no longer reads as negative throughput (counter_rates) or as no reading at all (source signals). source_landing_counter now returns one row per series - increase, span and the epoch second of its latest rise - and source_signals sums the rates and takes the latest rise across them. New integration tests write to the real otel_metrics_sum table. Old code against new: two pods 7s apart 61.19 vs 3.33/s, a mid-window restart -15.67 vs 1.67/s, a pod appearing mid-window 1.67 vs 3.67/s, two loader pods 3671.6 vs 200 records a minute, flat loader pods 3582 vs 0 with no arrival, a restarted loader pod null vs 100 a minute. A single pod and two attribute sets on one pod read the same on both. --- src/dfe_engine/appmgmt/operations.py | 44 ++-- src/dfe_engine/appmgmt/queries.yaml | 69 ++++--- tests/integration/test_counter_rates_ch.py | 227 +++++++++++++++++++++ tests/unit/test_api/test_source_signals.py | 22 +- tests/unit/test_appmgmt/test_operations.py | 76 ++++--- 5 files changed, 348 insertions(+), 90 deletions(-) create mode 100644 tests/integration/test_counter_rates_ch.py diff --git a/src/dfe_engine/appmgmt/operations.py b/src/dfe_engine/appmgmt/operations.py index b4de11f18..9a534f599 100644 --- a/src/dfe_engine/appmgmt/operations.py +++ b/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: diff --git a/src/dfe_engine/appmgmt/queries.yaml b/src/dfe_engine/appmgmt/queries.yaml index 988003aa9..04ce2ab2d 100644 --- a/src/dfe_engine/appmgmt/queries.yaml +++ b/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 diff --git a/tests/integration/test_counter_rates_ch.py b/tests/integration/test_counter_rates_ch.py new file mode 100644 index 000000000..f406271da --- /dev/null +++ b/tests/integration/test_counter_rates_ch.py @@ -0,0 +1,227 @@ +# Project: dfe-engine +# File: tests/integration/test_counter_rates_ch.py +# Purpose: Live-CH proof the console's counter rates difference each pod on its own +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED +"""The console's counter readings against real ClickHouse. + +Rows land in the real ``otel_metrics_sum`` table, rendered from the manifest the +schema phase applies, and ``OperationalReader`` answers over them with no mocks. +``TimeUnix`` has one-second precision and pods export on their own clocks, so the +fleet only ever lines up by accident: every test that matters here has its pods +exporting in different seconds. Each row is stamped relative to one server +``now()`` read at the start of the test, so the spacing between readings is exact. +""" + +import pytest + +from dfe_engine.appmgmt.operations import RECORDS_RECEIVED, OperationalReader +from dfe_engine.schema.plan import render_one + +pytestmark = pytest.mark.integration + +_SERVICE = "dfe-receiver" +_LOADER = "dfe-loader" +_LANDING = "loader_messages_by_table_total" +_TABLE = "auth" + + +@pytest.fixture +def otel_db(ch_client, clickhouse_test_database): + """An isolated database carrying the three metrics tables the console reads.""" + db = clickhouse_test_database + for object_id in ( + "data.otel_metrics_gauge", + "data.otel_metrics_sum", + "data.otel_metrics_histogram", + ): + for statement in render_one(object_id, data_database=db).statements: + ch_client.command(statement) + return db + + +@pytest.fixture +def base(ch_client) -> int: + """The server's now() as an epoch second, read once for the whole test.""" + return int(ch_client.command("SELECT toUnixTimestamp(now())")) + + +def _write( + ch_client, + db: str, + base: int, + *points: tuple[int, float], + pod: str, + service: str = _SERVICE, + metric: str = RECORDS_RECEIVED, + attributes: dict[str, str] | None = None, +) -> None: + """Write one cumulative reading per ``(age_seconds, value)``, stamped that long before ``base``.""" + attributes = attributes or {} + resource = {"service.name": service, "service.instance.id": pod, "k8s.pod.name": pod} + # clickhouse-connect binds a dict as JSON, which a Map parameter refuses to parse. + for age, value in points: + ch_client.command( + f"INSERT INTO `{db}`.otel_metrics_sum " + "(ServiceName, MetricName, Attributes, ResourceAttributes, TimeUnix, Value, " + "AggregationTemporality, IsMonotonic) " + "SELECT {service:String}, {metric:String}, " + "mapFromArrays({attr_keys:Array(String)}, {attr_values:Array(String)}), " + "mapFromArrays({res_keys:Array(String)}, {res_values:Array(String)}), " + "toDateTime({base:UInt32} - {age:UInt32}), {value:Float64}, 2, true", + parameters={ + "service": service, + "metric": metric, + "attr_keys": list(attributes), + "attr_values": list(attributes.values()), + "res_keys": list(resource), + "res_values": list(resource.values()), + "base": base, + "age": age, + "value": value, + }, + ) + + +def _rate(manager_client, db: str) -> float: + return OperationalReader(manager_client, db).metrics(_SERVICE).rates[RECORDS_RECEIVED] + + +def _landing(ch_client, db: str, base: int, *points: tuple[int, float], pod: str) -> None: + _write( + ch_client, + db, + base, + *points, + pod=pod, + service=_LOADER, + metric=_LANDING, + attributes={"table": _TABLE}, + ) + + +# -- counter_rates: the instance's per-second rates -------------------------- + + +def test_two_pods_exporting_seven_seconds_apart_read_their_combined_rate( + ch_client, manager_client, otel_db, base +): + """Each pod adds 100 over 60s, so the fleet carries 200/60 = 3.33 a second. + + Summing the fleet per timestamp first subtracts pod a's counter from pod b's + across the window and reads (5100 - 1000) / 67 = 61.19. + """ + _write(ch_client, otel_db, base, (67, 1000), (37, 1050), (7, 1100), pod="dfe-receiver-a") + _write(ch_client, otel_db, base, (60, 5000), (30, 5050), (0, 5100), pod="dfe-receiver-b") + + assert _rate(manager_client, otel_db) == pytest.approx(200 / 60, rel=1e-3) + + +def test_a_counter_reset_mid_window_counts_from_zero(ch_client, manager_client, otel_db, base): + """1000 -> 1040, then a restart to 20 -> 60: 40 + 20 + 40 = 100 over 60s. + + Differencing the first reading against the last reads (60 - 1000) / 60 = -15.67. + """ + _write( + ch_client, otel_db, base, (60, 1000), (40, 1040), (20, 20), (0, 60), pod="dfe-receiver-a" + ) + + assert _rate(manager_client, otel_db) == pytest.approx(100 / 60, rel=1e-3) + + +def test_a_single_pod_reads_its_own_rate(ch_client, manager_client, otel_db, base): + """100 over 60s from one pod, which every keying of the fleet agrees on.""" + _write(ch_client, otel_db, base, (60, 1000), (30, 1050), (0, 1100), pod="dfe-receiver-a") + + assert _rate(manager_client, otel_db) == pytest.approx(100 / 60, rel=1e-3) + + +def test_a_pod_that_appears_mid_window_adds_its_own_rate(ch_client, manager_client, otel_db, base): + """Pod a carries 100/60 for the whole window; pod b starts 33s ago and adds 60/30. + + The fleet's rate now is 100/60 + 2 = 3.67. Summing per timestamp reads only pod + a's first and last readings, 100 / 60 = 1.67, and misses pod b entirely. + """ + _write(ch_client, otel_db, base, (60, 1000), (30, 1050), (0, 1100), pod="dfe-receiver-a") + _write(ch_client, otel_db, base, (33, 20), (3, 80), pod="dfe-receiver-b") + + assert _rate(manager_client, otel_db) == pytest.approx(100 / 60 + 2, rel=1e-3) + + +def test_two_attribute_sets_on_one_pod_are_two_series(ch_client, manager_client, otel_db, base): + """One pod counts 100 over 60s on one attribute set and 60 on another: 160 / 60.""" + _write( + ch_client, + otel_db, + base, + (60, 1000), + (30, 1050), + (0, 1100), + pod="dfe-receiver-a", + attributes={"transport": "http"}, + ) + _write( + ch_client, + otel_db, + base, + (60, 50), + (30, 80), + (0, 110), + pod="dfe-receiver-a", + attributes={"transport": "grpc"}, + ) + + assert _rate(manager_client, otel_db) == pytest.approx(160 / 60, rel=1e-3) + + +# -- source_landing_counter: one table's records a minute -------------------- + + +def test_two_loader_pods_exporting_apart_read_their_combined_records_a_minute( + ch_client, manager_client, otel_db, base +): + """Each loader pod lands 100 records over 60s: 200 a minute, last risen at ``base``. + + Fleet totals per timestamp read (5100 - 1000) / 67s = 3671.6 a minute. + """ + _landing(ch_client, otel_db, base, (67, 1000), (37, 1050), (7, 1100), pod="dfe-loader-a") + _landing(ch_client, otel_db, base, (60, 5000), (30, 5050), (0, 5100), pod="dfe-loader-b") + + signals = OperationalReader(manager_client, otel_db).source_signals(_TABLE, [_LOADER]) + + assert signals.records_per_min == pytest.approx(200.0, rel=1e-3) + assert signals.last_seen_epoch == float(base) + + +def test_flat_counters_on_two_loader_pods_report_no_arrival( + ch_client, manager_client, otel_db, base +): + """Neither pod's counter moves, so nothing arrived. + + Fleet totals per timestamp alternate 1000, 5000, 1000, ... and read a rise + every time pod b's export follows pod a's. + """ + _landing(ch_client, otel_db, base, (67, 1000), (37, 1000), (7, 1000), pod="dfe-loader-a") + _landing(ch_client, otel_db, base, (60, 5000), (30, 5000), (0, 5000), pod="dfe-loader-b") + + signals = OperationalReader(manager_client, otel_db).source_signals(_TABLE, [_LOADER]) + + assert signals.records_per_min == 0.0 + assert signals.last_seen_epoch is None + + +def test_a_restarted_loader_pod_counts_from_zero(ch_client, manager_client, otel_db, base): + """1000 -> 1040, a restart to 20 -> 60: 100 records over 60s, 100 a minute. + + Differencing the first reading against the last went negative and reported nothing. + """ + _landing( + ch_client, otel_db, base, (60, 1000), (40, 1040), (20, 20), (0, 60), pod="dfe-loader-a" + ) + + signals = OperationalReader(manager_client, otel_db).source_signals(_TABLE, [_LOADER]) + + assert signals.records_per_min == pytest.approx(100.0, rel=1e-3) + assert signals.last_seen_epoch == float(base) diff --git a/tests/unit/test_api/test_source_signals.py b/tests/unit/test_api/test_source_signals.py index 6c1416962..0af231169 100644 --- a/tests/unit/test_api/test_source_signals.py +++ b/tests/unit/test_api/test_source_signals.py @@ -14,8 +14,6 @@ took, which is why every reading is nullable and the route still answers 200. """ -from __future__ import annotations - import time import pytest @@ -64,9 +62,9 @@ def _define(client, headers, sample_source: dict) -> None: assert client.post("/api/v1/sources", json=sample_source, headers=headers).status_code == 201 -def _series(*points: tuple[float, float]) -> list[tuple]: - """A counter series as ClickHouse returns it: (TimeUnix, summed value).""" - return [(ts, value) for ts, value in points] +def _series(increase: float, span_seconds: int, last_increase: float) -> list[tuple]: + """One loader pod's series as ClickHouse returns it, already differenced.""" + return [(increase, span_seconds, last_increase)] @pytest.fixture @@ -81,7 +79,7 @@ def test_a_rising_counter_becomes_records_a_minute( _wire(app, tmp_path) _deploy_loader(client, admin_headers) _define(client, admin_headers, sample_source) - _use(app, FakeClickHouse(_series((now - 120, 100.0), (now - 60, 220.0), (now, 340.0)))) + _use(app, FakeClickHouse(_series(240.0, 120, now))) body = client.get(SIGNALS, headers=admin_headers).json() @@ -99,7 +97,7 @@ def test_the_query_is_bound_to_the_table_and_the_deployed_loaders( _wire(app, tmp_path) _deploy_loader(client, admin_headers, "main") _define(client, admin_headers, sample_source) - ch = FakeClickHouse(_series((now - 60, 1.0), (now, 2.0))) + ch = FakeClickHouse(_series(1.0, 60, now)) _use(app, ch) client.get(SIGNALS, headers=admin_headers) @@ -136,20 +134,20 @@ def test_a_flat_counter_reports_a_rate_but_no_arrival( _wire(app, tmp_path) _deploy_loader(client, admin_headers) _define(client, admin_headers, sample_source) - _use(app, FakeClickHouse(_series((now - 120, 500.0), (now, 500.0)))) + _use(app, FakeClickHouse(_series(0.0, 120, 0))) body = client.get(SIGNALS, headers=admin_headers).json() assert body["records_per_min"] == 0.0 assert body["last_seen"] is None - def test_a_counter_reset_reports_nothing_rather_than_a_negative_rate( - self, client, app, admin_headers, tmp_path, sample_source, now + def test_a_single_sample_is_null_rather_than_a_zero_rate( + self, client, app, admin_headers, tmp_path, sample_source ): _wire(app, tmp_path) _deploy_loader(client, admin_headers) _define(client, admin_headers, sample_source) - _use(app, FakeClickHouse(_series((now - 120, 900.0), (now, 10.0)))) + _use(app, FakeClickHouse(_series(0.0, 0, 0))) assert client.get(SIGNALS, headers=admin_headers).json()["records_per_min"] is None @@ -158,7 +156,7 @@ def test_no_deployed_loader_asks_clickhouse_nothing( ): _wire(app, tmp_path) _define(client, admin_headers, sample_source) - ch = FakeClickHouse(_series((1.0, 1.0), (2.0, 2.0))) + ch = FakeClickHouse(_series(1.0, 60, 2.0)) _use(app, ch) body = client.get(SIGNALS, headers=admin_headers).json() diff --git a/tests/unit/test_appmgmt/test_operations.py b/tests/unit/test_appmgmt/test_operations.py index daec7967e..99fc15d68 100644 --- a/tests/unit/test_appmgmt/test_operations.py +++ b/tests/unit/test_appmgmt/test_operations.py @@ -5,16 +5,14 @@ # # License: BUSL-1.1 # Copyright: (c) 2026 HYPERI PTY LIMITED -"""The arithmetic between a counter series and a number a console can show. +"""The arithmetic between per-series counter rows and a number a console can show. -A monotonic counter says nothing on its own: the rate is a difference over -elapsed time, and "when did records last arrive" is the last time the total -MOVED, not the last time the loader reported. Both go wrong quietly, so both -are asserted here rather than only through the route. +ClickHouse differences each series on its own (tests/integration/test_counter_rates_ch.py +runs that against a real server). What is left here is the fleet: the rate is each +series' increase over its own span, summed, and "when did records last arrive" is the +latest time any series MOVED, not the last time the loader reported. """ -from __future__ import annotations - import pytest from dfe_engine.appmgmt.operations import ( @@ -50,31 +48,39 @@ def _reader(rows=None, fail: bool = False) -> tuple[OperationalReader, FakeClick return OperationalReader(ch, "dfe"), ch -def test_the_rate_is_the_delta_over_the_elapsed_time(): - reader, _ = _reader([(0.0, 100.0), (60.0, 220.0), (120.0, 340.0)]) +def test_the_rate_is_each_series_increase_over_its_own_span_summed(): + # 240 over 120s on one loader pod and 30 over 60s on another: 120 + 30 a minute. + reader, _ = _reader([(240.0, 120, 1757000120), (30.0, 60, 1757000100)]) signals = reader.source_signals("auth", LOADERS) - assert signals.records_per_min == pytest.approx(120.0) - assert signals.last_seen_epoch == 120.0 + assert signals.records_per_min == pytest.approx(150.0) + assert signals.last_seen_epoch == 1757000120.0 def test_one_sample_cannot_be_differenced(): - reader, _ = _reader([(0.0, 100.0)]) + reader, _ = _reader([(0.0, 0, 0)]) assert reader.source_signals("auth", LOADERS).records_per_min is None -def test_a_reset_counter_reports_nothing_rather_than_a_negative_rate(): - # A pod restart zeroes its counter; the window has to roll past the restart - # before the number means anything again. - reader, _ = _reader([(0.0, 900.0), (60.0, 10.0)]) +def test_a_single_sample_series_adds_nothing_beside_one_that_spans_time(): + reader, _ = _reader([(120.0, 60, 1757000060), (0.0, 0, 0)]) - assert reader.source_signals("auth", LOADERS).records_per_min is None + assert reader.source_signals("auth", LOADERS).records_per_min == pytest.approx(120.0) + + +def test_no_series_reports_nothing(): + reader, _ = _reader([]) + + signals = reader.source_signals("auth", LOADERS) + + assert signals.records_per_min is None + assert signals.last_seen_epoch is None def test_a_flat_counter_reports_no_arrival(): - reader, _ = _reader([(0.0, 500.0), (60.0, 500.0), (120.0, 500.0)]) + reader, _ = _reader([(0.0, 120, 0)]) signals = reader.source_signals("auth", LOADERS) @@ -82,14 +88,38 @@ def test_a_flat_counter_reports_no_arrival(): assert signals.last_seen_epoch is None -def test_the_last_arrival_is_the_last_increase_not_the_last_report(): - reader, _ = _reader([(0.0, 10.0), (60.0, 40.0), (120.0, 40.0), (180.0, 40.0)]) +def test_the_last_arrival_is_the_latest_rise_across_series(): + reader, _ = _reader([(10.0, 180, 1757000060), (5.0, 180, 1757000030), (0.0, 180, 0)]) + + assert reader.source_signals("auth", LOADERS).last_seen_epoch == 1757000060.0 + + +def test_the_landing_query_differences_each_loader_pod_on_its_own(): + reader, ch = _reader([]) + + reader.source_signals("auth", LOADERS) + + sql = ch.calls[0][0] + assert "arrayDifference(readings)" in sql + assert ( + "GROUP BY ServiceName, cityHash64(Attributes), ResourceAttributes['service.instance.id']" + ) in sql + - assert reader.source_signals("auth", LOADERS).last_seen_epoch == 60.0 +def test_the_counter_query_differences_each_pod_before_summing_the_fleet(): + reader, ch = _reader([]) + + reader.metrics("dfe-receiver") + + sql = next(sql for sql, _, _ in ch.calls if "otel_metrics_sum" in sql) + assert "arrayDifference(readings)" in sql + assert ( + "GROUP BY MetricName, cityHash64(Attributes), ResourceAttributes['service.instance.id']" + ) in sql def test_no_loader_asks_the_database_nothing(): - reader, ch = _reader([(0.0, 1.0), (60.0, 2.0)]) + reader, ch = _reader([(0.0, 60, 0)]) signals = reader.source_signals("auth", []) @@ -99,7 +129,7 @@ def test_no_loader_asks_the_database_nothing(): def test_the_query_is_bound_and_time_bounded(): - reader, ch = _reader([(0.0, 1.0), (60.0, 2.0)]) + reader, ch = _reader([(1.0, 60, 60)]) reader.source_signals("auth", LOADERS)