diff --git a/src/dfe_engine/appmgmt/operations.py b/src/dfe_engine/appmgmt/operations.py index b4de11f1..9a534f59 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 988003aa..04ce2ab2 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 00000000..f406271d --- /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 6c141696..0af23116 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 daec7967..99fc15d6 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)