From 920ee1080c465e25e1798b1cbcb2c5c595097bd5 Mon Sep 17 00:00:00 2001 From: Damodar Lohani Date: Wed, 8 Jul 2026 11:13:18 +0545 Subject: [PATCH 1/8] feat(usage): add peak running-sum aggregation for delta metrics Realtime connections are stored as +1/-1 event deltas aggregated with SUM, so a plain MAX(value) is meaningless. Add an `aggregate` query hint (sum | peak) mirroring the groupBy pattern. `peak` computes the peak concurrent value as max(running_sum(value)) ordered by time, cross-pod correct with no producer change. - UsageQuery: TYPE_AGGREGATE, VALID_AGGREGATES, aggregate() plus isAggregate/extractAggregate/removeAggregate helpers. - ClickHouse: parseQueries splits time vs non-time filters and records the window-start param for peak; findFromTable routes peak to a new findPeakFromTable that builds a windowed running-sum with a pre-window baseline subquery (connections still open at start). Honours interval bucketing, dimensions, limit/offset/orderBy. The sum/default path is unchanged. - Tests: flat + interval peak, pre-window baseline, interleaved-producer sum-before-max, and UsageQuery unit coverage. --- src/Usage/Adapter/ClickHouse.php | 202 ++++++++++++++++++- src/Usage/UsageQuery.php | 86 ++++++++- tests/Usage/Adapter/ClickHousePeakTest.php | 214 +++++++++++++++++++++ tests/Usage/UsageQueryTest.php | 90 +++++++++ 4 files changed, 590 insertions(+), 2 deletions(-) create mode 100644 tests/Usage/Adapter/ClickHousePeakTest.php diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index 8a888993..6475a7cd 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -1459,6 +1459,10 @@ private function queriesMatchType(array $queries, string $type): bool * - Gauges: SELECT metric, argMax(value, time) as value, toStartOfInterval(time, INTERVAL ...) as time * Results are grouped by metric and time bucket, ordered by time ASC. * + * When an `aggregate('peak')` query is present, switches to the running-sum + * max path ({@see findPeakFromTable}) for delta metrics like realtime + * connections, honouring interval bucketing and dimensions inside it. + * * @param array $queries * @param string $type 'event' or 'gauge' * @return array @@ -1477,6 +1481,13 @@ private function findFromTable(string $tenant, array $queries, string $type): ar throw new Exception('Cursor pagination cannot be combined with groupByInterval'); } + // Peak aggregation (running-sum max) needs its own window-function + // query. Route it here — before the SUM-based aggregated path — while + // still honouring interval bucketing and dimension breakdowns inside. + if (($parsed['aggregate'] ?? null) === 'peak') { + return $this->findPeakFromTable($tenant, $parsed, $fromTable, $type); + } + // Route through the aggregated path whenever any aggregation // hint is present — time bucketing, dimension breakdown, or both. if (isset($parsed['groupByInterval']) || !empty($parsed['groupBy'])) { @@ -1630,6 +1641,152 @@ private function findAggregatedFromTable(array $parsed, string $fromTable, strin return $this->parseAggregatedResults($result, $type); } + /** + * Find peak (running-sum maximum) metrics from a table. + * + * Delta metrics — emitted as `+1` on open / `-1` on close (e.g. realtime + * connections) and stored as events — carry no meaningful `MAX(value)` + * (that is just `1`). The true peak concurrent value is the maximum of the + * cumulative sum ordered by time, computed here cross-producer with a + * single window function so interleaved rows from many pods sum before the + * max is taken. + * + * A baseline scalar subquery adds connections opened strictly before the + * window start (still-open at `start`) so an in-window peak is not + * understated. The baseline is non-correlated — it is scoped to the + * non-time filters (tenant, metric) that pin a single series, matching the + * realtime-connections use case. + * + * Produces SQL like (interval variant): + * SELECT metric, max(running) AS value, + * toStartOfInterval(time, INTERVAL 1 HOUR) AS bucket + * FROM ( + * SELECT metric, time, + * ifNull((SELECT sum(value) FROM t + * WHERE `metric` = {m} AND time < {start}), 0) + * + sum(value) OVER (PARTITION BY metric ORDER BY time ASC + * ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running + * FROM t WHERE `metric` = {m} AND time >= {start} AND time <= {end} + * ) + * GROUP BY metric, bucket ORDER BY bucket ASC + * + * Without `groupByInterval` the bucket column is dropped, yielding a single + * `max(running)` row per metric (flat peak over the window). + * + * @param array{filters: array, params: array, orderBy?: array, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array, aggregate?: string, timeFilters?: array, nonTimeFilters?: array, peakStartParam?: string|null} $parsed + * @param string $fromTable Fully qualified table reference + * @param string $type 'event' or 'gauge' + * @return array + * @throws Exception + */ + private function findPeakFromTable(string $tenant, array $parsed, string $fromTable, string $type): array + { + $hasInterval = isset($parsed['groupByInterval']); + $params = $parsed['params']; + + $nonTimeFilters = $parsed['nonTimeFilters'] ?? []; + $timeFilters = $parsed['timeFilters'] ?? []; + $peakStartParam = $parsed['peakStartParam'] ?? null; + + // Dimension columns join the window PARTITION so the running sum is + // computed per (metric[, tenant][, dims]) series. + $groupByDims = $parsed['groupBy'] ?? []; + $escapedDims = array_map( + fn (string $dim): string => $this->escapeIdentifier($dim), + $groupByDims + ); + + $partitionCols = ['metric']; + if ($this->sharedTables) { + $partitionCols[] = 'tenant'; + } + foreach ($escapedDims as $escapedDim) { + $partitionCols[] = $escapedDim; + } + $partitionSql = implode(', ', $partitionCols); + + // Baseline: concurrency already open at the window start. Non-correlated + // scalar; omitted (0) when the query has no lower time bound. + $baseline = '0'; + if ($peakStartParam !== null) { + $baselineConds = $nonTimeFilters; + $baselineConds[] = "time < {{$peakStartParam}:DateTime64(3, 'UTC')}"; + $baseline = "ifNull((SELECT sum(value) FROM {$fromTable} WHERE " + . implode(' AND ', $baselineConds) + . '), 0)'; + } + + $runningExpr = "{$baseline}" + . " + sum(value) OVER (PARTITION BY {$partitionSql} ORDER BY time ASC" + . ' ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)'; + + // Inner query: window WHERE = non-time filters + time bounds. + $innerConds = array_merge($nonTimeFilters, $timeFilters); + $innerWhere = !empty($innerConds) ? ' WHERE ' . implode(' AND ', $innerConds) : ''; + $innerDimSelect = !empty($escapedDims) ? ', ' . implode(', ', $escapedDims) : ''; + $innerSql = "SELECT metric, time{$innerDimSelect}, {$runningExpr} AS running" + . " FROM {$fromTable}{$innerWhere}"; + + // Outer query: max(running) per (metric[, bucket][, dims]). + $bucketSelect = ''; + $bucketGroup = ''; + if ($hasInterval) { + $interval = $parsed['groupByInterval']; + $intervalSql = UsageQuery::VALID_INTERVALS[$interval]; + $bucketSelect = ", toStartOfInterval(time, {$intervalSql}) AS bucket"; + $bucketGroup = ', bucket'; + } + + $dimSelect = !empty($escapedDims) ? ', ' . implode(', ', $escapedDims) : ''; + $dimGroup = !empty($escapedDims) ? ', ' . implode(', ', $escapedDims) : ''; + + // ORDER BY mirrors findAggregatedFromTable: chronological when bucketed, + // value DESC otherwise. `time` in a caller ORDER BY maps to `bucket` + // when bucketing, and is rejected without it (no time column survives). + if ($hasInterval) { + $orderClause = ' ORDER BY bucket ASC'; + if (!empty($parsed['orderBy'])) { + $rewrittenOrderBy = array_map( + fn (string $clause): string => preg_replace( + '/^`time`(\s+(?:ASC|DESC))?$/', + '`bucket`$1', + $clause + ) ?? $clause, + $parsed['orderBy'] + ); + $orderClause = ' ORDER BY ' . implode(', ', $rewrittenOrderBy); + } + } else { + $orderClause = ' ORDER BY value DESC'; + if (!empty($parsed['orderBy'])) { + foreach ($parsed['orderBy'] as $clause) { + if (preg_match('/^`time`/', $clause)) { + throw new Exception( + 'orderBy("time") requires groupByInterval — without time bucketing the result has no time column' + ); + } + } + $orderClause = ' ORDER BY ' . implode(', ', $parsed['orderBy']); + } + } + + $limitClause = isset($parsed['limit']) ? ' LIMIT {limit:UInt64}' : ''; + $offsetClause = isset($parsed['offset']) ? ' OFFSET {offset:UInt64}' : ''; + + $sql = " + SELECT metric, max(running) as value{$bucketSelect}{$dimSelect} + FROM ( + {$innerSql} + ) + GROUP BY metric{$bucketGroup}{$dimGroup}{$orderClause}{$limitClause}{$offsetClause} + FORMAT JSON + "; + + $result = $this->query($sql, $params); + + return $this->parseAggregatedResults($result, $type); + } + /** * Parse ClickHouse JSON results from an aggregated (groupByInterval) query into Metric array. * @@ -3381,7 +3538,7 @@ private function buildOrderBySql(array $orderAttributes, bool $flip = false): ar * @param string $tenant Tenant scope (shared-tables mode) * @param array $queries * @param string $type 'event' or 'gauge' — used for attribute validation - * @return array{filters: array, params: array, orderBy?: array, orderAttributes?: array, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array, cursor?: array, cursorDirection?: string} + * @return array{filters: array, params: array, orderBy?: array, orderAttributes?: array, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array, aggregate?: string, timeFilters?: array, nonTimeFilters?: array, peakStartParam?: string|null, cursor?: array, cursorDirection?: string} * @throws Exception */ private function parseQueries(string $tenant, array $queries, string $type = 'event'): array @@ -3404,6 +3561,7 @@ private function parseQueries(string $tenant, array $queries, string $type = 'ev $offset = null; $groupByInterval = null; $groupBy = []; + $aggregate = null; $cursor = null; $cursorDirection = null; $paramCounter = 0; @@ -3646,6 +3804,16 @@ private function parseQueries(string $tenant, array $queries, string $type = 'ev $groupBy[] = $attribute; } break; + + case UsageQuery::TYPE_AGGREGATE: + $aggValue = $values[0] ?? null; + if (!is_string($aggValue) || !in_array($aggValue, UsageQuery::VALID_AGGREGATES, true)) { + throw new Exception( + 'Invalid aggregate: expected one of ' . implode(', ', UsageQuery::VALID_AGGREGATES) + ); + } + $aggregate = $aggValue; + break; } } @@ -3675,6 +3843,38 @@ private function parseQueries(string $tenant, array $queries, string $type = 'ev $result['groupBy'] = $groupBy; } + // Peak aggregation needs the time filters split out from the other + // filters: the running-sum window and the pre-window baseline subquery + // apply different time predicates (window: >= start AND <= end; + // baseline: < start) while sharing the same non-time filters (tenant, + // metric, dims). The `sum`/default path is untouched — it ignores these + // extra keys entirely. + if ($aggregate === 'peak') { + $result['aggregate'] = 'peak'; + + $timeFilters = []; + $nonTimeFilters = []; + $peakStartParam = null; + foreach ($filters as $filter) { + if (str_starts_with($filter, $this->escapeIdentifier('time'))) { + $timeFilters[] = $filter; + // The window lower bound (`time >=`/`time >`, or the first + // arg of a BETWEEN) is the baseline cutoff: connections + // opened strictly before it are the starting concurrency. + if ($peakStartParam === null + && preg_match('/(?:>=|>|BETWEEN)\s*\{([A-Za-z0-9_]+):/', $filter, $matches) === 1) { + $peakStartParam = $matches[1]; + } + } else { + $nonTimeFilters[] = $filter; + } + } + + $result['timeFilters'] = $timeFilters; + $result['nonTimeFilters'] = $nonTimeFilters; + $result['peakStartParam'] = $peakStartParam; + } + if ($cursor !== null && $cursorDirection !== null) { $result['cursor'] = $cursor; $result['cursorDirection'] = $cursorDirection; diff --git a/src/Usage/UsageQuery.php b/src/Usage/UsageQuery.php index 1976a136..dc0bd9ff 100644 --- a/src/Usage/UsageQuery.php +++ b/src/Usage/UsageQuery.php @@ -26,11 +26,27 @@ * switches from raw row returns to aggregated results grouped by time bucket: * - Events: SUM(value) per bucket * - Gauges: argMax(value, time) per bucket + * + * An `aggregate` hint selects how event values are combined. The default `sum` + * totals values; `peak` computes the running-sum maximum for delta metrics + * emitted as `+1`/`-1` (e.g. realtime connections), yielding the peak concurrent + * value rather than the net total. */ class UsageQuery extends Query { public const TYPE_GROUP_BY_INTERVAL = 'groupByInterval'; public const TYPE_GROUP_BY = 'groupBy'; + public const TYPE_AGGREGATE = 'aggregate'; + + /** + * Valid aggregation functions. + * + * - `sum` — the default; totals the metric value per bucket (current behaviour). + * - `peak` — running-sum maximum, for delta metrics emitted as `+1`/`-1` + * (e.g. realtime connections). Returns the peak concurrent value: the + * maximum of the cumulative sum ordered by time. + */ + public const VALID_AGGREGATES = ['sum', 'peak']; /** * Valid interval values and their ClickHouse INTERVAL equivalents. @@ -51,7 +67,7 @@ class UsageQuery extends Query */ public static function isMethod(string $value): bool { - if ($value === self::TYPE_GROUP_BY_INTERVAL || $value === self::TYPE_GROUP_BY) { + if ($value === self::TYPE_GROUP_BY_INTERVAL || $value === self::TYPE_GROUP_BY || $value === self::TYPE_AGGREGATE) { return true; } @@ -180,4 +196,72 @@ public static function removeGroupBy(array $queries): array return !self::isGroupBy($query); })); } + + /** + * Create an aggregate query selecting the aggregation function. + * + * `sum` (the default when no aggregate query is present) totals the metric + * value per bucket. `peak` computes the running-sum maximum — the peak + * concurrent value for delta metrics emitted as `+1`/`-1` (realtime + * connections). See {@see UsageQuery::VALID_AGGREGATES}. + * + * @param string $function One of {@see UsageQuery::VALID_AGGREGATES}. + * @return self + */ + public static function aggregate(string $function): self + { + if (!in_array($function, self::VALID_AGGREGATES, true)) { + throw new \InvalidArgumentException( + "Invalid aggregate '{$function}'. Allowed: " . implode(', ', self::VALID_AGGREGATES) + ); + } + + return new self(self::TYPE_AGGREGATE, 'value', [$function]); + } + + /** + * Check if a query is an aggregate query. + * + * @param Query $query + * @return bool + */ + public static function isAggregate(Query $query): bool + { + return $query->getMethod() === self::TYPE_AGGREGATE; + } + + /** + * Extract the aggregation function from an array of queries, if present. + * + * Queries parsed via `Query::parse()` are base `Query` objects rather than + * `UsageQuery` instances, so we match on the method string alone. + * + * @param array $queries + * @return string|null The aggregation function, or null if not present. + */ + public static function extractAggregate(array $queries): ?string + { + foreach ($queries as $query) { + if (self::isAggregate($query)) { + $value = $query->getValues()[0] ?? null; + + return is_string($value) ? $value : null; + } + } + + return null; + } + + /** + * Remove all aggregate queries from an array of queries. + * + * @param array $queries + * @return array + */ + public static function removeAggregate(array $queries): array + { + return array_values(array_filter($queries, function (Query $query) { + return !self::isAggregate($query); + })); + } } diff --git a/tests/Usage/Adapter/ClickHousePeakTest.php b/tests/Usage/Adapter/ClickHousePeakTest.php new file mode 100644 index 00000000..1b470df6 --- /dev/null +++ b/tests/Usage/Adapter/ClickHousePeakTest.php @@ -0,0 +1,214 @@ +adapter = new ClickHouseAdapter( + $host, + $username, + $password, + $port, + $secure, + namespace: 'utopia_usage_peak', + database: getenv('CLICKHOUSE_DATABASE') ?: 'default', + ); + + $this->usage = new Usage($this->adapter); + $this->usage->setup(); + } + + /** + * Insert raw concurrency deltas straight into the events table. + * + * @param array $rows + */ + private function insertDeltas(string $metric, array $rows): void + { + $table = $this->resolveTableName($this->adapter, 'getEventsTableName'); + $database = $this->databaseName($this->adapter); + $ref = '`'.$database.'`.`'.$table.'`'; + + $tuples = []; + foreach ($rows as $i => $row) { + $id = $metric.'-'.$i; + $value = (int) $row['value']; + $time = $row['time']; + $tuples[] = "('{$id}', '{$metric}', {$value}, '{$time}')"; + } + + $sql = "INSERT INTO {$ref} (id, metric, value, time) VALUES ".implode(', ', $tuples); + $this->queryRaw($this->adapter, $sql); + } + + public function test_peak_flat_aggregate(): void + { + $metric = 'rt-peak-flat-'.uniqid(); + + // +1,+1,+1,+1,-1 → running 1,2,3,4,3 → peak = 4. + $this->insertDeltas($metric, [ + ['value' => 1, 'time' => '2026-06-01 00:01:00'], + ['value' => 1, 'time' => '2026-06-01 00:02:00'], + ['value' => 1, 'time' => '2026-06-01 00:03:00'], + ['value' => 1, 'time' => '2026-06-01 00:04:00'], + ['value' => -1, 'time' => '2026-06-01 00:05:00'], + ]); + + $results = $this->usage->find('1', [ + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 00:00:00'), + Query::lessThanEqual('time', '2026-06-01 01:00:00'), + UsageQuery::aggregate('peak'), + ], Usage::TYPE_EVENT); + + $this->assertCount(1, $results); + $this->assertEquals(4, $results[0]->getValue()); + } + + public function test_peak_with_group_by_interval(): void + { + $metric = 'rt-peak-interval-'.uniqid(); + + // Bucket 00: +1,+1 → running 1,2 → peak 2. + // Bucket 01: +1,-1 → running 3,2 → peak 3 (carries the still-open + // connections from bucket 00 forward, as concurrency should). + $this->insertDeltas($metric, [ + ['value' => 1, 'time' => '2026-06-01 00:10:00'], + ['value' => 1, 'time' => '2026-06-01 00:20:00'], + ['value' => 1, 'time' => '2026-06-01 01:10:00'], + ['value' => -1, 'time' => '2026-06-01 01:20:00'], + ]); + + $results = $this->usage->find('1', [ + UsageQuery::groupByInterval('time', '1h'), + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 00:00:00'), + Query::lessThanEqual('time', '2026-06-01 02:00:00'), + UsageQuery::aggregate('peak'), + ], Usage::TYPE_EVENT); + + $this->assertCount(2, $results); + + $byHour = []; + foreach ($results as $row) { + $time = $row->getTime(); + $this->assertNotNull($time); + $hour = (new \DateTime($time))->format('H'); + $byHour[$hour] = $row->getValue(); + } + + $this->assertEquals(2, $byHour['00'] ?? null); + $this->assertEquals(3, $byHour['01'] ?? null); + } + + public function test_peak_includes_pre_window_baseline(): void + { + $metric = 'rt-peak-baseline-'.uniqid(); + + // Two connections opened before the window start (still open at start): + // baseline = 2. In-window deltas +1,+1,-1 → cumulative 1,2,1 → + // running = 2 + [1,2,1] = 3,4,3 → peak = 4. Without the baseline the + // in-window max would understate to 2. + $this->insertDeltas($metric, [ + ['value' => 1, 'time' => '2026-05-31 23:00:00'], + ['value' => 1, 'time' => '2026-05-31 23:30:00'], + ['value' => 1, 'time' => '2026-06-01 00:01:00'], + ['value' => 1, 'time' => '2026-06-01 00:02:00'], + ['value' => -1, 'time' => '2026-06-01 00:03:00'], + ]); + + $results = $this->usage->find('1', [ + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 00:00:00'), + Query::lessThanEqual('time', '2026-06-01 01:00:00'), + UsageQuery::aggregate('peak'), + ], Usage::TYPE_EVENT); + + $this->assertCount(1, $results); + $this->assertEquals(4, $results[0]->getValue()); + } + + public function test_peak_sums_interleaved_producers(): void + { + $metric = 'rt-peak-pods-'.uniqid(); + + // Rows as if emitted by two realtime pods, interleaved in time. The + // running sum combines them (no per-pod partition), so the peak + // reflects the global concurrency, not any single pod's max. + // A:+1(01) B:+1(02) A:+1(03) B:+1(04) A:-1(05) + // running: 1, 2, 3, 4, 3 → peak = 4 + $this->insertDeltas($metric, [ + ['value' => 1, 'time' => '2026-06-01 00:01:00'], + ['value' => 1, 'time' => '2026-06-01 00:02:00'], + ['value' => 1, 'time' => '2026-06-01 00:03:00'], + ['value' => 1, 'time' => '2026-06-01 00:04:00'], + ['value' => -1, 'time' => '2026-06-01 00:05:00'], + ]); + + $results = $this->usage->find('1', [ + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 00:00:00'), + Query::lessThanEqual('time', '2026-06-01 01:00:00'), + UsageQuery::aggregate('peak'), + ], Usage::TYPE_EVENT); + + $this->assertCount(1, $results); + $this->assertEquals(4, $results[0]->getValue()); + } + + public function test_sum_aggregate_path_unchanged(): void + { + $metric = 'rt-peak-sum-'.uniqid(); + + // Same deltas — the default `sum` aggregate must still return the net + // total (1+1+1+1-1 = 3), proving the peak routing does not leak into + // the existing path. + $this->insertDeltas($metric, [ + ['value' => 1, 'time' => '2026-06-01 00:01:00'], + ['value' => 1, 'time' => '2026-06-01 00:02:00'], + ['value' => 1, 'time' => '2026-06-01 00:03:00'], + ['value' => 1, 'time' => '2026-06-01 00:04:00'], + ['value' => -1, 'time' => '2026-06-01 00:05:00'], + ]); + + $results = $this->usage->find('1', [ + UsageQuery::groupByInterval('time', '1h'), + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 00:00:00'), + Query::lessThanEqual('time', '2026-06-01 01:00:00'), + UsageQuery::aggregate('sum'), + ], Usage::TYPE_EVENT); + + $this->assertCount(1, $results); + $this->assertEquals(3, $results[0]->getValue()); + } +} diff --git a/tests/Usage/UsageQueryTest.php b/tests/Usage/UsageQueryTest.php index 924c6e37..dd1eb8e0 100644 --- a/tests/Usage/UsageQueryTest.php +++ b/tests/Usage/UsageQueryTest.php @@ -231,4 +231,94 @@ public function testExtractGroupByFromParsedQuery(): void $this->assertCount(1, $extracted); $this->assertEquals('service', $extracted[0]->getAttribute()); } + + public function testAggregateCreation(): void + { + $query = UsageQuery::aggregate('peak'); + + $this->assertInstanceOf(UsageQuery::class, $query); + $this->assertEquals(UsageQuery::TYPE_AGGREGATE, $query->getMethod()); + $this->assertEquals(['peak'], $query->getValues()); + $this->assertEquals('peak', $query->getValue()); + } + + public function testAggregateAcceptsAllValidFunctions(): void + { + foreach (UsageQuery::VALID_AGGREGATES as $function) { + $query = UsageQuery::aggregate($function); + $this->assertEquals($function, $query->getValue()); + } + } + + public function testAggregateRejectsInvalidFunction(): void + { + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage("Invalid aggregate 'max'"); + UsageQuery::aggregate('max'); + } + + public function testAggregateIsMethod(): void + { + $this->assertTrue(UsageQuery::isMethod(UsageQuery::TYPE_AGGREGATE)); + } + + public function testIsAggregate(): void + { + $aggregate = UsageQuery::aggregate('peak'); + $regular = Query::equal('metric', ['bandwidth']); + + $this->assertTrue(UsageQuery::isAggregate($aggregate)); + $this->assertFalse(UsageQuery::isAggregate($regular)); + } + + public function testExtractAggregate(): void + { + $queries = [ + Query::equal('metric', ['realtime.connections']), + UsageQuery::aggregate('peak'), + ]; + + $this->assertEquals('peak', UsageQuery::extractAggregate($queries)); + } + + public function testExtractAggregateFromParsedQuery(): void + { + // Queries created via Query::parse() are base Query objects, not UsageQuery. + $parsedAggregate = new Query(UsageQuery::TYPE_AGGREGATE, 'value', ['peak']); + $equal = Query::equal('metric', ['realtime.connections']); + + $this->assertEquals('peak', UsageQuery::extractAggregate([$equal, $parsedAggregate])); + } + + public function testExtractAggregateReturnsNullWhenMissing(): void + { + $queries = [ + Query::equal('metric', ['realtime.connections']), + UsageQuery::groupByInterval('time', '1h'), + ]; + + $this->assertNull(UsageQuery::extractAggregate($queries)); + } + + public function testRemoveAggregate(): void + { + $queries = [ + Query::equal('metric', ['realtime.connections']), + UsageQuery::aggregate('peak'), + UsageQuery::groupByInterval('time', '1h'), + ]; + + $remaining = UsageQuery::removeAggregate($queries); + + $this->assertCount(2, $remaining); + foreach ($remaining as $query) { + $this->assertNotEquals(UsageQuery::TYPE_AGGREGATE, $query->getMethod()); + } + } + + public function testValidAggregatesConstant(): void + { + $this->assertContains('sum', UsageQuery::VALID_AGGREGATES); + $this->assertContains('peak', UsageQuery::VALID_AGGREGATES); + } } From eafa857bded4f8a0e0daa0f940fc9ae0b22c104e Mon Sep 17 00:00:00 2001 From: Damodar Lohani Date: Wed, 8 Jul 2026 12:23:21 +0545 Subject: [PATCH 2/8] feat(usage): add opt-in negative deltas and per-second fold MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make the peak path correct end-to-end for delta metrics like realtime connections, where the write side previously dropped -1 disconnects and folded sub-flush bursts away. - Reject negative values by default for every metric (events included) so a buggy negative count/bandwidth is still caught. Callers emitting a genuine signed delta opt in per row via allowNegative: Accumulator::collect(..., bool $allowNegative = false) carries the flag onto the buffered entry and hands it to addBatch; ClickHouse::validateMetricData(..., bool $allowNegative = false) gates the guard, read from each row's `allowNegative` in validateMetricsBatch. The flag is validation-only — never written as a column. The library stays generic; the caller decides which metrics may be negative. - Add optional foldSeconds to Accumulator::collect(). When set, the second bucket (floor(ts / foldSeconds) * foldSeconds) joins the fold key and becomes the entry time, so events fold only within the same bucket and intra-flush peaks survive. When null, behaviour is unchanged. - Tests: negatives rejected by default (collect + addBatch), persisted and netted when opted in, gauges still reject; per-second fold groups/splits by second; peak over per-second net rows captures a burst a per-flush net hides. --- src/Usage/Accumulator.php | 53 +++++++++-- src/Usage/Adapter/ClickHouse.php | 14 ++- tests/Usage/AccumulatorTest.php | 105 ++++++++++++++++++++- tests/Usage/Adapter/ClickHousePeakTest.php | 102 +++++++++++++++++++- 4 files changed, 259 insertions(+), 15 deletions(-) diff --git a/src/Usage/Accumulator.php b/src/Usage/Accumulator.php index b898be92..716ad687 100644 --- a/src/Usage/Accumulator.php +++ b/src/Usage/Accumulator.php @@ -16,7 +16,7 @@ class Accumulator private Usage $usage; /** - * @var array, time?: \DateTime}> + * @var array, allowNegative: bool, time?: \DateTime}> */ private array $buffer = []; @@ -38,9 +38,23 @@ public function __construct(Usage $usage) * Events fold additively; earliest non-null time wins on merge. * Gauges use last-write-wins. * + * Negative values are rejected by default for every metric, so a buggy + * negative count/bandwidth is still caught. A caller that emits a genuine + * signed delta (e.g. realtime connections `+1`/`-1`) opts in per call with + * `$allowNegative = true`. The library stays generic — the decision lives + * with the caller, not with any metric-name knowledge here. + * + * When $foldSeconds is set, events only fold within the same time bucket: + * the bucket start (floor(timestamp / $foldSeconds) * $foldSeconds) joins + * the fold key and becomes the entry's canonical time, so every event in + * that window shares one timestamp and sub-flush peaks survive the fold. + * When $foldSeconds is null the fold key and time handling are unchanged. + * * @param array $tags + * @param int|null $foldSeconds Fold events within this many seconds, or null for the default per-flush fold. + * @param bool $allowNegative Permit a negative value for this metric (default: reject). */ - public function collect(string $tenant, string $metric, int $value, string $type, array $tags = [], ?\DateTime $time = null): self + public function collect(string $tenant, string $metric, int $value, string $type, array $tags = [], ?\DateTime $time = null, ?int $foldSeconds = null, bool $allowNegative = false): self { // Compare against '' rather than empty(): the string "0" is a valid // tenant/metric id but empty("0") is true in PHP. @@ -50,12 +64,15 @@ public function collect(string $tenant, string $metric, int $value, string $type if ($metric === '') { throw new \InvalidArgumentException('Metric name cannot be empty'); } - if ($value < 0) { + if ($value < 0 && !$allowNegative) { throw new \InvalidArgumentException('Value cannot be negative'); } if ($type !== Usage::TYPE_EVENT && $type !== Usage::TYPE_GAUGE) { throw new \InvalidArgumentException("Invalid metric type '{$type}'. Allowed: " . Usage::TYPE_EVENT . ', ' . Usage::TYPE_GAUGE); } + if ($foldSeconds !== null && $foldSeconds <= 0) { + throw new \InvalidArgumentException('foldSeconds must be a positive integer'); + } // Hash the full identity so distinct (tenant, metric, type, tags) // tuples never collide on the key — a raw `:`-join would let @@ -63,12 +80,31 @@ public function collect(string $tenant, string $metric, int $value, string $type // entry. Tags are sorted first so key order doesn't matter. $canonicalTags = $tags; ksort($canonicalTags); - $key = md5(json_encode([$tenant, $metric, $type, $canonicalTags], JSON_THROW_ON_ERROR)); + + // Per-bucket fold: the bucket start joins the key so events only fold + // within the same window, and becomes the canonical entry time so all + // events in that bucket share one timestamp. + $bucketTime = null; + if ($foldSeconds !== null) { + $timestamp = $time !== null ? $time->getTimestamp() : time(); + $bucketStart = intdiv($timestamp, $foldSeconds) * $foldSeconds; + $bucketTime = new \DateTime('@' . $bucketStart); + $key = md5(json_encode([$tenant, $metric, $type, $canonicalTags, $bucketStart], JSON_THROW_ON_ERROR)); + } else { + $key = md5(json_encode([$tenant, $metric, $type, $canonicalTags], JSON_THROW_ON_ERROR)); + } + + // With bucketing, the bucket start is the entry time; otherwise the + // caller-supplied time (if any) is used. + $effectiveTime = $bucketTime ?? $time; if ($type === Usage::TYPE_EVENT && isset($this->buffer[$key])) { - // earliest time wins on merge $this->buffer[$key]['value'] += $value; - if ($time !== null && (!isset($this->buffer[$key]['time']) || $time < $this->buffer[$key]['time'])) { + if ($bucketTime !== null) { + // Every event in this bucket shares the bucket-start time. + $this->buffer[$key]['time'] = $bucketTime; + } elseif ($time !== null && (!isset($this->buffer[$key]['time']) || $time < $this->buffer[$key]['time'])) { + // earliest time wins on merge $this->buffer[$key]['time'] = $time; } } else { @@ -78,9 +114,10 @@ public function collect(string $tenant, string $metric, int $value, string $type 'value' => $value, 'type' => $type, 'tags' => $tags, + 'allowNegative' => $allowNegative, ]; - if ($time !== null) { - $entry['time'] = $time; + if ($effectiveTime !== null) { + $entry['time'] = $effectiveTime; } $this->buffer[$key] = $entry; } diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index 6475a7cd..6c6b1657 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -1205,9 +1205,10 @@ private function getColumnCodec(string $id): string * @param string $type Metric type ('event' or 'gauge') * @param array $tags Tags * @param int|null $metricIndex Index for batch error messages + * @param bool $allowNegative Permit a negative value for this row (default: reject) * @throws Exception */ - private function validateMetricData(string $metric, int $value, string $type, array $tags, ?int $metricIndex = null): void + private function validateMetricData(string $metric, int $value, string $type, array $tags, ?int $metricIndex = null, bool $allowNegative = false): void { $prefix = $metricIndex !== null ? "Metric #{$metricIndex}: " : ''; @@ -1219,7 +1220,11 @@ private function validateMetricData(string $metric, int $value, string $type, ar throw new Exception($prefix . 'Metric exceeds maximum size of 255 characters'); } - if ($value < 0) { + // Negatives are rejected by default so a buggy negative count/bandwidth + // is caught. A row opts in with `allowNegative` for genuine signed + // deltas (realtime connections emit +1/-1). The library stays generic — + // the caller decides which metrics may be negative. + if ($value < 0 && !$allowNegative) { throw new Exception($prefix . 'Value cannot be negative'); } @@ -1257,7 +1262,10 @@ private function validateMetricsBatch(array $metrics, string $type): void /** @var array */ $tags = $metricData['tags'] ?? []; - $this->validateMetricData($metric, $value, $type, $tags, $index); + // `allowNegative` is a validation-only flag carried on the row; it + // gates the negative-value guard and is never stored as a column. + $allowNegative = (bool) ($metricData['allowNegative'] ?? false); + $this->validateMetricData($metric, $value, $type, $tags, $index, $allowNegative); $hasTenant = array_key_exists('tenant', $metricData); diff --git a/tests/Usage/AccumulatorTest.php b/tests/Usage/AccumulatorTest.php index 8495c488..d1ea20f9 100644 --- a/tests/Usage/AccumulatorTest.php +++ b/tests/Usage/AccumulatorTest.php @@ -275,13 +275,116 @@ public function testTagOrderDoesNotSplitEntries(): void $this->assertEquals(30, $this->adapter->batches[0]['metrics'][0]['value']); } - public function testNegativeValueThrows(): void + public function testGaugeNegativeValueThrows(): void { + // Gauges are snapshots (e.g. storage); callers never opt in, so + // negatives still throw by default. + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage('Value cannot be negative'); + $this->accumulator->collect('t1', 'storage', -1, Usage::TYPE_GAUGE); + } + + public function testNegativeValueRejectedByDefaultForEvents(): void + { + // Default is strict for every metric, events included, so a buggy + // negative count is caught unless the caller explicitly opts in. $this->expectException(\InvalidArgumentException::class); $this->expectExceptionMessage('Value cannot be negative'); $this->accumulator->collect('t1', 'requests', -1, Usage::TYPE_EVENT); } + public function testEventNegativeValueAllowedWhenOptedIn(): void + { + // Genuine signed deltas (realtime connections emit +1/-1) opt in per + // call with allowNegative, so a lone -1 persists rather than throws. + $this->accumulator->collect('t1', 'realtime.connections', -1, Usage::TYPE_EVENT, allowNegative: true); + + $this->assertEquals(1, $this->accumulator->count()); + $this->assertTrue($this->accumulator->flush()); + $entry = $this->adapter->batches[0]['metrics'][0]; + $this->assertEquals(-1, $entry['value']); + // The opt-in flag rides along on the row so addBatch can honour it. + $this->assertTrue($entry['allowNegative']); + } + + public function testEventMixedSignsNetToDelta(): void + { + // +1,+1,-1 folds to a net delta of +1 for the metric. + $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, allowNegative: true); + $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, allowNegative: true); + $this->accumulator->collect('t1', 'realtime.connections', -1, Usage::TYPE_EVENT, allowNegative: true); + + $this->assertEquals(1, $this->accumulator->count()); + $this->assertTrue($this->accumulator->flush()); + $this->assertEquals(1, $this->adapter->batches[0]['metrics'][0]['value']); + } + + public function testFoldSecondsGroupsEventsInSameSecond(): void + { + // Two events landing in the same 1s bucket fold into one net row + // stamped at the bucket start. + $a = new \DateTime('2026-04-15 12:00:00.200'); + $b = new \DateTime('2026-04-15 12:00:00.900'); + + $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $a, 1); + $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $b, 1); + + $this->assertEquals(1, $this->accumulator->count()); + $this->assertTrue($this->accumulator->flush()); + + $entry = $this->adapter->batches[0]['metrics'][0]; + $this->assertEquals(2, $entry['value']); + $this->assertArrayHasKey('time', $entry); + $time = $entry['time']; + $this->assertInstanceOf(\DateTime::class, $time); + // Stamped at the second boundary, not the sub-second event time. + $this->assertEquals('2026-04-15 12:00:00', $time->format('Y-m-d H:i:s')); + } + + public function testFoldSecondsKeepsDistinctSecondsSeparate(): void + { + // Events in different 1s buckets stay as separate rows so a rise and + // fall inside a flush is not collapsed away. + $s0 = new \DateTime('2026-04-15 12:00:00.500'); + $s1 = new \DateTime('2026-04-15 12:00:01.500'); + + $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $s0, 1); + $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $s1, 1); + + $this->assertEquals(2, $this->accumulator->count()); + $this->assertTrue($this->accumulator->flush()); + + $times = []; + foreach ($this->adapter->batches[0]['metrics'] as $m) { + $time = $m['time'] ?? null; + $this->assertInstanceOf(\DateTime::class, $time); + $times[] = $time->format('Y-m-d H:i:s'); + } + sort($times); + $this->assertEquals(['2026-04-15 12:00:00', '2026-04-15 12:00:01'], $times); + } + + public function testFoldSecondsNetsWithinBucket(): void + { + // Within a single second, +/- deltas net; the per-second row carries + // the net so cross-pod running sums stay exact. + $t = new \DateTime('2026-04-15 12:00:03.100'); + $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $t, 1, allowNegative: true); + $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $t, 1, allowNegative: true); + $this->accumulator->collect('t1', 'realtime.connections', -1, Usage::TYPE_EVENT, [], $t, 1, allowNegative: true); + + $this->assertEquals(1, $this->accumulator->count()); + $this->assertTrue($this->accumulator->flush()); + $this->assertEquals(1, $this->adapter->batches[0]['metrics'][0]['value']); + } + + public function testFoldSecondsRejectsNonPositive(): void + { + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage('foldSeconds must be a positive integer'); + $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], null, 0); + } + public function testInvalidTypeThrows(): void { $this->expectException(\InvalidArgumentException::class); diff --git a/tests/Usage/Adapter/ClickHousePeakTest.php b/tests/Usage/Adapter/ClickHousePeakTest.php index 1b470df6..3dd9000d 100644 --- a/tests/Usage/Adapter/ClickHousePeakTest.php +++ b/tests/Usage/Adapter/ClickHousePeakTest.php @@ -3,6 +3,7 @@ namespace Utopia\Tests\Usage\Adapter; use Utopia\Query\Query; +use Utopia\Usage\Accumulator; use Utopia\Usage\Adapter\ClickHouse as ClickHouseAdapter; use Utopia\Usage\Usage; use Utopia\Usage\UsageQuery; @@ -15,9 +16,10 @@ * rows is meaningless — the true peak concurrent value is max(running_sum(value)) * ordered by time. These tests exercise that path end-to-end. * - * Delta rows (including the `-1` closes) are inserted with raw SQL: the public - * addBatch() write path rejects negative values by design, so the concurrency - * deltas are written directly to the events table here. + * Most tests insert delta rows (including the `-1` closes) with raw SQL to keep + * the query under test isolated from the write path. The per-second-fold test + * instead drives the real Accumulator -> addBatch path, which now persists + * negative event deltas. */ class ClickHousePeakTest extends ClickHouseTestCase { @@ -211,4 +213,98 @@ public function test_sum_aggregate_path_unchanged(): void $this->assertCount(1, $results); $this->assertEquals(3, $results[0]->getValue()); } + + public function testPeakOverPerSecondNetRowsCapturesBurst(): void + { + // A burst that rises to 4 and falls back, all inside one flush window. + // Per-second nets: s0 +2, s1 +2 (running 4 = peak), s2 -2, s3 -1. + // Collected through the Accumulator with foldSeconds=1 so each second + // survives as its own net row; allowNegative lets the -1 closes persist. + $foldMetric = 'rt-peak-persec-' . uniqid(); + $acc = new Accumulator($this->usage); + + $deltas = [ + [1, '2026-06-01 03:00:00.100'], + [1, '2026-06-01 03:00:00.900'], + [1, '2026-06-01 03:00:01.100'], + [1, '2026-06-01 03:00:01.900'], + [-1, '2026-06-01 03:00:02.100'], + [-1, '2026-06-01 03:00:02.900'], + [-1, '2026-06-01 03:00:03.100'], + ]; + foreach ($deltas as [$value, $time]) { + $acc->collect('1', $foldMetric, $value, Usage::TYPE_EVENT, [], new \DateTime($time), 1, allowNegative: true); + } + + // Four per-second net rows (+2, +2, -2, -1), not one folded delta. + $this->assertEquals(4, $acc->count()); + $this->assertTrue($acc->flush()); + + $window = [ + Query::greaterThanEqual('time', '2026-06-01 03:00:00'), + Query::lessThanEqual('time', '2026-06-01 03:01:00'), + ]; + + $peak = $this->usage->find('1', array_merge([ + Query::equal('metric', [$foldMetric]), + ], $window, [UsageQuery::aggregate('peak')]), Usage::TYPE_EVENT); + + $this->assertCount(1, $peak); + $this->assertEquals(4, $peak[0]->getValue()); + + // Contrast: the same deltas folded WITHOUT a per-second key collapse to + // a single net row (+1), whose running-sum peak is only 1 — the burst + // to 4 is invisible. This is exactly what the per-second fold fixes. + $flatMetric = 'rt-peak-perflush-' . uniqid(); + $accFlat = new Accumulator($this->usage); + foreach ($deltas as [$value, $time]) { + $accFlat->collect('1', $flatMetric, $value, Usage::TYPE_EVENT, [], new \DateTime($time), allowNegative: true); + } + $this->assertEquals(1, $accFlat->count()); + $this->assertTrue($accFlat->flush()); + + $flatPeak = $this->usage->find('1', array_merge([ + Query::equal('metric', [$flatMetric]), + ], $window, [UsageQuery::aggregate('peak')]), Usage::TYPE_EVENT); + + $this->assertCount(1, $flatPeak); + $this->assertEquals(1, $flatPeak[0]->getValue()); + } + + public function testAddBatchRejectsNegativeEventByDefault(): void + { + // Strict by default: a negative event without the opt-in is rejected + // at write time, so a buggy negative count never reaches storage. + $this->expectException(\Exception::class); + $this->expectExceptionMessage('Value cannot be negative'); + $this->usage->addBatch([ + ['tenant' => '1', 'metric' => 'rt-neg-' . uniqid(), 'value' => -1, 'tags' => []], + ], Usage::TYPE_EVENT); + } + + public function testAddBatchPersistsNegativeEventWhenOptedIn(): void + { + // With allowNegative on the row, the -1 close persists and nets with + // the +1 opens so the peak path sees true concurrency. + $metric = 'rt-neg-allowed-' . uniqid(); + + $this->assertTrue($this->usage->addBatch([ + ['tenant' => '1', 'metric' => $metric, 'value' => 1, 'time' => new \DateTime('2026-06-01 04:00:01'), 'tags' => [], 'allowNegative' => true], + ['tenant' => '1', 'metric' => $metric, 'value' => 1, 'time' => new \DateTime('2026-06-01 04:00:02'), 'tags' => [], 'allowNegative' => true], + ['tenant' => '1', 'metric' => $metric, 'value' => -1, 'time' => new \DateTime('2026-06-01 04:00:03'), 'tags' => [], 'allowNegative' => true], + ], Usage::TYPE_EVENT)); + + // Net total is +1 (SUM), peak concurrency is 2 (running 1,2,1). + $this->assertEquals(1, $this->usage->getTotal('1', $metric, [], Usage::TYPE_EVENT)); + + $peak = $this->usage->find('1', [ + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 04:00:00'), + Query::lessThanEqual('time', '2026-06-01 04:01:00'), + UsageQuery::aggregate('peak'), + ], Usage::TYPE_EVENT); + + $this->assertCount(1, $peak); + $this->assertEquals(2, $peak[0]->getValue()); + } } From d5abba81371eb2e93906a68c62e32fd3b2c6f79f Mon Sep 17 00:00:00 2001 From: Damodar Lohani Date: Tue, 4 Aug 2026 11:16:26 +0545 Subject: [PATCH 3/8] feat(usage): add cross-tenant read for operator-side aggregation findAcrossTenants() applies no tenant filter so an aggregation job can roll every tenant up in one pass instead of issuing N per-tenant queries. Shared tables only; groupBy('tenant') keeps the rows attributable. parseQueries() now takes a nullable tenant - null is the explicit cross-tenant read, '' still fails fast so an empty scope can never be read silently. --- src/Usage/Adapter.php | 20 ++++++++ src/Usage/Adapter/ClickHouse.php | 83 +++++++++++++++++++++++++++++--- src/Usage/Usage.php | 18 +++++++ 3 files changed, 114 insertions(+), 7 deletions(-) diff --git a/src/Usage/Adapter.php b/src/Usage/Adapter.php index 135f5de2..9fd6feab 100644 --- a/src/Usage/Adapter.php +++ b/src/Usage/Adapter.php @@ -102,6 +102,26 @@ abstract public function purge(string $tenant, array $queries = [], ?string $typ */ abstract public function find(string $tenant, array $queries = [], ?string $type = null): array; + /** + * Find metrics across every tenant in shared-tables mode. + * + * Deliberately crosses the per-tenant isolation every other read enforces, + * so it is reserved for operator-side aggregation jobs that roll many + * tenants up in one pass. Never reachable from a tenant-scoped request + * path. Pair with `groupBy('tenant')` to keep the rows attributable. + * + * Adapters that cannot express an unscoped read leave this unsupported. + * + * @param array<\Utopia\Query\Query> $queries + * @param string|null $type Metric type: 'event', 'gauge', or null (query both) + * @return array + * @throws \Exception + */ + public function findAcrossTenants(array $queries = [], ?string $type = null): array + { + throw new \Exception($this->getName() . ' does not support cross-tenant reads'); + } + /** * Count metrics using Query objects. * diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index 6c6b1657..fb9243b0 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -1053,6 +1053,13 @@ private function validateGroupByAttribute(string $attribute, string $type): bool { $allowed = $type === Usage::TYPE_GAUGE ? Metric::GAUGE_COLUMNS : Metric::EVENT_COLUMNS; + // `tenant` is a real column in shared-tables mode, not a dimension on + // Metric. Grouping by it is what makes a cross-tenant read + // ({@see findAcrossTenants}) attributable, so allow it there. + if ($this->sharedTables) { + $allowed[] = 'tenant'; + } + if (in_array($attribute, $allowed, true)) { return true; } @@ -1386,6 +1393,44 @@ public function find(string $tenant, array $queries = [], ?string $type = null): { $this->setOperationContext('find()'); + return $this->findScoped($tenant, $queries, $type); + } + + /** + * Find metrics across every tenant. Applies no tenant filter, so it is + * restricted to shared-tables mode and reserved for operator-side + * aggregation jobs — see {@see Adapter::findAcrossTenants()}. + * + * Callers should add `groupBy('tenant')` to keep rows attributable; the + * aggregated paths already carry `tenant` through select/group-by in + * shared-tables mode. + * + * @param array $queries + * @param string|null $type + * @return array + * @throws Exception + */ + public function findAcrossTenants(array $queries = [], ?string $type = null): array + { + $this->setOperationContext('findAcrossTenants()'); + + if (!$this->sharedTables) { + throw new Exception('findAcrossTenants() requires shared-tables mode; use find() instead'); + } + + return $this->findScoped(null, $queries, $type); + } + + /** + * Shared body for find()/findAcrossTenants(). A null $tenant means no + * tenant filter (cross-tenant); a string scopes to that tenant. + * + * @param array $queries + * @return array + * @throws Exception + */ + private function findScoped(?string $tenant, array $queries, ?string $type): array + { if ($type !== null) { return $this->findFromTable($tenant, $queries, $type); } @@ -1476,7 +1521,7 @@ private function queriesMatchType(array $queries, string $type): bool * @return array * @throws Exception */ - private function findFromTable(string $tenant, array $queries, string $type): array + private function findFromTable(?string $tenant, array $queries, string $type): array { $tableName = $this->getTableForType($type); $fromTable = $this->buildTableReference($tableName); @@ -1687,7 +1732,7 @@ private function findAggregatedFromTable(array $parsed, string $fromTable, strin * @return array * @throws Exception */ - private function findPeakFromTable(string $tenant, array $parsed, string $fromTable, string $type): array + private function findPeakFromTable(?string $tenant, array $parsed, string $fromTable, string $type): array { $hasInterval = isset($parsed['groupByInterval']); $params = $parsed['params']; @@ -1713,6 +1758,27 @@ private function findPeakFromTable(string $tenant, array $parsed, string $fromTa } $partitionSql = implode(', ', $partitionCols); + // The baseline below is a single non-correlated scalar, but the running + // sum is partitioned per (metric[, tenant][, dims]). Those only agree + // when the window covers exactly one series, i.e. when every partition + // column beyond `metric` is pinned by an equality filter. A + // cross-tenant read or a dimension break-down would add the combined + // baseline of every series to each series. Reject rather than return a + // silently inflated peak. + if ($tenant === null && $peakStartParam !== null) { + throw new Exception( + 'aggregate(peak) is not supported on a cross-tenant read: the window baseline ' + . 'is not correlated per tenant. Query one tenant at a time, or drop the lower ' + . 'time bound.' + ); + } + if (!empty($groupByDims) && $peakStartParam !== null) { + throw new Exception( + 'aggregate(peak) cannot be combined with groupBy(' . implode(', ', $groupByDims) + . '): the window baseline is not correlated per dimension.' + ); + } + // Baseline: concurrency already open at the window start. Non-correlated // scalar; omitted (0) when the query has no lower time bound. $baseline = '0'; @@ -3549,12 +3615,15 @@ private function buildOrderBySql(array $orderAttributes, bool $flip = false): ar * @return array{filters: array, params: array, orderBy?: array, orderAttributes?: array, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array, aggregate?: string, timeFilters?: array, nonTimeFilters?: array, peakStartParam?: string|null, cursor?: array, cursorDirection?: string} * @throws Exception */ - private function parseQueries(string $tenant, array $queries, string $type = 'event'): array + private function parseQueries(?string $tenant, array $queries, string $type = 'event'): array { - if ($this->sharedTables) { - // An empty tenant would compile to `tenant = ''` and silently read - // an empty scope. Fail fast instead, like the write side. ("0" is - // a valid tenant id, so check for '' specifically.) + // A null tenant is the explicit cross-tenant read used by operator-side + // aggregation ({@see findAcrossTenants}) — no tenant filter is applied. + // It is deliberately distinct from '': an empty string would compile to + // `tenant = ''` and silently read an empty scope, so that still fails + // fast, like the write side. ("0" is a valid tenant id, so check for '' + // specifically.) + if ($this->sharedTables && $tenant !== null) { if ($tenant === '') { throw new Exception('Tenant cannot be empty in shared-tables mode'); } diff --git a/src/Usage/Usage.php b/src/Usage/Usage.php index a5f65bc3..c70a96e0 100644 --- a/src/Usage/Usage.php +++ b/src/Usage/Usage.php @@ -154,6 +154,24 @@ public function find(string $tenant, array $queries = [], ?string $type = null): return $this->adapter->find($tenant, $queries, $type); } + /** + * Find metrics across every tenant in shared-tables mode. + * + * Unlike find(), this applies no tenant filter — it is for operator-side + * aggregation jobs that need one pass over all tenants instead of N + * per-tenant queries. Keep it out of request-scoped code paths, and add + * `groupBy('tenant')` so the returned rows stay attributable. + * + * @param array<\Utopia\Query\Query> $queries + * @param string|null $type Metric type: 'event', 'gauge', or null (query both) + * @return array + * @throws \Exception + */ + public function findAcrossTenants(array $queries = [], ?string $type = null): array + { + return $this->adapter->findAcrossTenants($queries, $type); + } + /** * Count metrics using Query objects. * From 2002b279a8b2bf47842b5c6fafe5c3a68e8ae7f6 Mon Sep 17 00:00:00 2001 From: Damodar Lohani Date: Tue, 4 Aug 2026 11:20:13 +0545 Subject: [PATCH 4/8] feat(usage): replace peak aggregation with max for gauges The peak concurrency figure is now pre-computed into a gauge level series by an operator-side job, so nothing reads a running-sum max at request time. Drop findPeakFromTable and its baseline plumbing, and keep the aggregate hint for what does need it: rolling a gauge series up to a coarser interval. Gauges default to argMax(value, time) - the latest reading in the bucket, right for a snapshot but wrong for a sampled level series, where the bucket's highest sample is the answer. aggregate('max') selects that. Removing the peak path also removes its baseline defect: the pre-window baseline was a single non-correlated scalar added to a running sum partitioned by metric[, tenant][, dims], so any dimension break-down had every series inflated by the combined baseline of all of them. --- src/Usage/Adapter/ClickHouse.php | 222 +------------ src/Usage/UsageQuery.php | 27 +- .../Usage/Adapter/ClickHouseGaugeMaxTest.php | 262 +++++++++++++++ tests/Usage/Adapter/ClickHousePeakTest.php | 310 ------------------ tests/Usage/UsageQueryTest.php | 22 +- 5 files changed, 301 insertions(+), 542 deletions(-) create mode 100644 tests/Usage/Adapter/ClickHouseGaugeMaxTest.php delete mode 100644 tests/Usage/Adapter/ClickHousePeakTest.php diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index fb9243b0..032cc3a3 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -1512,9 +1512,8 @@ private function queriesMatchType(array $queries, string $type): bool * - Gauges: SELECT metric, argMax(value, time) as value, toStartOfInterval(time, INTERVAL ...) as time * Results are grouped by metric and time bucket, ordered by time ASC. * - * When an `aggregate('peak')` query is present, switches to the running-sum - * max path ({@see findPeakFromTable}) for delta metrics like realtime - * connections, honouring interval bucketing and dimensions inside it. + * An `aggregate('max')` query overrides the per-type default value + * expression — see {@see findAggregatedFromTable()}. * * @param array $queries * @param string $type 'event' or 'gauge' @@ -1534,13 +1533,6 @@ private function findFromTable(?string $tenant, array $queries, string $type): a throw new Exception('Cursor pagination cannot be combined with groupByInterval'); } - // Peak aggregation (running-sum max) needs its own window-function - // query. Route it here — before the SUM-based aggregated path — while - // still honouring interval bucketing and dimension breakdowns inside. - if (($parsed['aggregate'] ?? null) === 'peak') { - return $this->findPeakFromTable($tenant, $parsed, $fromTable, $type); - } - // Route through the aggregated path whenever any aggregation // hint is present — time bucketing, dimension breakdown, or both. if (isset($parsed['groupByInterval']) || !empty($parsed['groupBy'])) { @@ -1614,11 +1606,18 @@ private function findAggregatedFromTable(array $parsed, string $fromTable, strin { $hasInterval = isset($parsed['groupByInterval']); - // Choose aggregation function based on metric type + // Choose aggregation function based on metric type. `aggregate('max')` + // overrides both defaults: for gauges it takes the highest reading in + // the bucket rather than the latest one (argMax), which is what rolling + // a sampled level series up to a coarser interval needs. $valueExpr = $type === Usage::TYPE_GAUGE ? 'argMax(value, time) as value' : 'SUM(value) as value'; + if (($parsed['aggregate'] ?? null) === 'max') { + $valueExpr = 'max(value) as value'; + } + // Bucket column is only emitted when time bucketing is requested. // Without it the result is a flat aggregate per (metric, …dims). $bucketSelect = ''; @@ -1694,173 +1693,6 @@ private function findAggregatedFromTable(array $parsed, string $fromTable, strin return $this->parseAggregatedResults($result, $type); } - /** - * Find peak (running-sum maximum) metrics from a table. - * - * Delta metrics — emitted as `+1` on open / `-1` on close (e.g. realtime - * connections) and stored as events — carry no meaningful `MAX(value)` - * (that is just `1`). The true peak concurrent value is the maximum of the - * cumulative sum ordered by time, computed here cross-producer with a - * single window function so interleaved rows from many pods sum before the - * max is taken. - * - * A baseline scalar subquery adds connections opened strictly before the - * window start (still-open at `start`) so an in-window peak is not - * understated. The baseline is non-correlated — it is scoped to the - * non-time filters (tenant, metric) that pin a single series, matching the - * realtime-connections use case. - * - * Produces SQL like (interval variant): - * SELECT metric, max(running) AS value, - * toStartOfInterval(time, INTERVAL 1 HOUR) AS bucket - * FROM ( - * SELECT metric, time, - * ifNull((SELECT sum(value) FROM t - * WHERE `metric` = {m} AND time < {start}), 0) - * + sum(value) OVER (PARTITION BY metric ORDER BY time ASC - * ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running - * FROM t WHERE `metric` = {m} AND time >= {start} AND time <= {end} - * ) - * GROUP BY metric, bucket ORDER BY bucket ASC - * - * Without `groupByInterval` the bucket column is dropped, yielding a single - * `max(running)` row per metric (flat peak over the window). - * - * @param array{filters: array, params: array, orderBy?: array, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array, aggregate?: string, timeFilters?: array, nonTimeFilters?: array, peakStartParam?: string|null} $parsed - * @param string $fromTable Fully qualified table reference - * @param string $type 'event' or 'gauge' - * @return array - * @throws Exception - */ - private function findPeakFromTable(?string $tenant, array $parsed, string $fromTable, string $type): array - { - $hasInterval = isset($parsed['groupByInterval']); - $params = $parsed['params']; - - $nonTimeFilters = $parsed['nonTimeFilters'] ?? []; - $timeFilters = $parsed['timeFilters'] ?? []; - $peakStartParam = $parsed['peakStartParam'] ?? null; - - // Dimension columns join the window PARTITION so the running sum is - // computed per (metric[, tenant][, dims]) series. - $groupByDims = $parsed['groupBy'] ?? []; - $escapedDims = array_map( - fn (string $dim): string => $this->escapeIdentifier($dim), - $groupByDims - ); - - $partitionCols = ['metric']; - if ($this->sharedTables) { - $partitionCols[] = 'tenant'; - } - foreach ($escapedDims as $escapedDim) { - $partitionCols[] = $escapedDim; - } - $partitionSql = implode(', ', $partitionCols); - - // The baseline below is a single non-correlated scalar, but the running - // sum is partitioned per (metric[, tenant][, dims]). Those only agree - // when the window covers exactly one series, i.e. when every partition - // column beyond `metric` is pinned by an equality filter. A - // cross-tenant read or a dimension break-down would add the combined - // baseline of every series to each series. Reject rather than return a - // silently inflated peak. - if ($tenant === null && $peakStartParam !== null) { - throw new Exception( - 'aggregate(peak) is not supported on a cross-tenant read: the window baseline ' - . 'is not correlated per tenant. Query one tenant at a time, or drop the lower ' - . 'time bound.' - ); - } - if (!empty($groupByDims) && $peakStartParam !== null) { - throw new Exception( - 'aggregate(peak) cannot be combined with groupBy(' . implode(', ', $groupByDims) - . '): the window baseline is not correlated per dimension.' - ); - } - - // Baseline: concurrency already open at the window start. Non-correlated - // scalar; omitted (0) when the query has no lower time bound. - $baseline = '0'; - if ($peakStartParam !== null) { - $baselineConds = $nonTimeFilters; - $baselineConds[] = "time < {{$peakStartParam}:DateTime64(3, 'UTC')}"; - $baseline = "ifNull((SELECT sum(value) FROM {$fromTable} WHERE " - . implode(' AND ', $baselineConds) - . '), 0)'; - } - - $runningExpr = "{$baseline}" - . " + sum(value) OVER (PARTITION BY {$partitionSql} ORDER BY time ASC" - . ' ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)'; - - // Inner query: window WHERE = non-time filters + time bounds. - $innerConds = array_merge($nonTimeFilters, $timeFilters); - $innerWhere = !empty($innerConds) ? ' WHERE ' . implode(' AND ', $innerConds) : ''; - $innerDimSelect = !empty($escapedDims) ? ', ' . implode(', ', $escapedDims) : ''; - $innerSql = "SELECT metric, time{$innerDimSelect}, {$runningExpr} AS running" - . " FROM {$fromTable}{$innerWhere}"; - - // Outer query: max(running) per (metric[, bucket][, dims]). - $bucketSelect = ''; - $bucketGroup = ''; - if ($hasInterval) { - $interval = $parsed['groupByInterval']; - $intervalSql = UsageQuery::VALID_INTERVALS[$interval]; - $bucketSelect = ", toStartOfInterval(time, {$intervalSql}) AS bucket"; - $bucketGroup = ', bucket'; - } - - $dimSelect = !empty($escapedDims) ? ', ' . implode(', ', $escapedDims) : ''; - $dimGroup = !empty($escapedDims) ? ', ' . implode(', ', $escapedDims) : ''; - - // ORDER BY mirrors findAggregatedFromTable: chronological when bucketed, - // value DESC otherwise. `time` in a caller ORDER BY maps to `bucket` - // when bucketing, and is rejected without it (no time column survives). - if ($hasInterval) { - $orderClause = ' ORDER BY bucket ASC'; - if (!empty($parsed['orderBy'])) { - $rewrittenOrderBy = array_map( - fn (string $clause): string => preg_replace( - '/^`time`(\s+(?:ASC|DESC))?$/', - '`bucket`$1', - $clause - ) ?? $clause, - $parsed['orderBy'] - ); - $orderClause = ' ORDER BY ' . implode(', ', $rewrittenOrderBy); - } - } else { - $orderClause = ' ORDER BY value DESC'; - if (!empty($parsed['orderBy'])) { - foreach ($parsed['orderBy'] as $clause) { - if (preg_match('/^`time`/', $clause)) { - throw new Exception( - 'orderBy("time") requires groupByInterval — without time bucketing the result has no time column' - ); - } - } - $orderClause = ' ORDER BY ' . implode(', ', $parsed['orderBy']); - } - } - - $limitClause = isset($parsed['limit']) ? ' LIMIT {limit:UInt64}' : ''; - $offsetClause = isset($parsed['offset']) ? ' OFFSET {offset:UInt64}' : ''; - - $sql = " - SELECT metric, max(running) as value{$bucketSelect}{$dimSelect} - FROM ( - {$innerSql} - ) - GROUP BY metric{$bucketGroup}{$dimGroup}{$orderClause}{$limitClause}{$offsetClause} - FORMAT JSON - "; - - $result = $this->query($sql, $params); - - return $this->parseAggregatedResults($result, $type); - } - /** * Parse ClickHouse JSON results from an aggregated (groupByInterval) query into Metric array. * @@ -3612,7 +3444,7 @@ private function buildOrderBySql(array $orderAttributes, bool $flip = false): ar * @param string $tenant Tenant scope (shared-tables mode) * @param array $queries * @param string $type 'event' or 'gauge' — used for attribute validation - * @return array{filters: array, params: array, orderBy?: array, orderAttributes?: array, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array, aggregate?: string, timeFilters?: array, nonTimeFilters?: array, peakStartParam?: string|null, cursor?: array, cursorDirection?: string} + * @return array{filters: array, params: array, orderBy?: array, orderAttributes?: array, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array, aggregate?: string, cursor?: array, cursorDirection?: string} * @throws Exception */ private function parseQueries(?string $tenant, array $queries, string $type = 'event'): array @@ -3920,36 +3752,8 @@ private function parseQueries(?string $tenant, array $queries, string $type = 'e $result['groupBy'] = $groupBy; } - // Peak aggregation needs the time filters split out from the other - // filters: the running-sum window and the pre-window baseline subquery - // apply different time predicates (window: >= start AND <= end; - // baseline: < start) while sharing the same non-time filters (tenant, - // metric, dims). The `sum`/default path is untouched — it ignores these - // extra keys entirely. - if ($aggregate === 'peak') { - $result['aggregate'] = 'peak'; - - $timeFilters = []; - $nonTimeFilters = []; - $peakStartParam = null; - foreach ($filters as $filter) { - if (str_starts_with($filter, $this->escapeIdentifier('time'))) { - $timeFilters[] = $filter; - // The window lower bound (`time >=`/`time >`, or the first - // arg of a BETWEEN) is the baseline cutoff: connections - // opened strictly before it are the starting concurrency. - if ($peakStartParam === null - && preg_match('/(?:>=|>|BETWEEN)\s*\{([A-Za-z0-9_]+):/', $filter, $matches) === 1) { - $peakStartParam = $matches[1]; - } - } else { - $nonTimeFilters[] = $filter; - } - } - - $result['timeFilters'] = $timeFilters; - $result['nonTimeFilters'] = $nonTimeFilters; - $result['peakStartParam'] = $peakStartParam; + if ($aggregate !== null) { + $result['aggregate'] = $aggregate; } if ($cursor !== null && $cursorDirection !== null) { diff --git a/src/Usage/UsageQuery.php b/src/Usage/UsageQuery.php index dc0bd9ff..44fb2a5a 100644 --- a/src/Usage/UsageQuery.php +++ b/src/Usage/UsageQuery.php @@ -27,10 +27,10 @@ * - Events: SUM(value) per bucket * - Gauges: argMax(value, time) per bucket * - * An `aggregate` hint selects how event values are combined. The default `sum` - * totals values; `peak` computes the running-sum maximum for delta metrics - * emitted as `+1`/`-1` (e.g. realtime connections), yielding the peak concurrent - * value rather than the net total. + * An `aggregate` hint overrides how values are combined. The default `sum` + * totals them; `max` takes the largest value per bucket, which is how a gauge + * level series rolls up to a coarser interval — the default `argMax` would + * return the latest reading rather than the highest. */ class UsageQuery extends Query { @@ -41,12 +41,14 @@ class UsageQuery extends Query /** * Valid aggregation functions. * - * - `sum` — the default; totals the metric value per bucket (current behaviour). - * - `peak` — running-sum maximum, for delta metrics emitted as `+1`/`-1` - * (e.g. realtime connections). Returns the peak concurrent value: the - * maximum of the cumulative sum ordered by time. + * - `sum` — the default; totals the metric value per bucket (current behaviour). + * - `max` — the largest value per bucket. Intended for gauges, where the + * default `argMax(value, time)` returns the *latest* reading rather than + * the highest one. Reading a pre-computed level series (e.g. realtime + * concurrency sampled every 5 minutes) at a coarser interval needs the + * peak of the samples, not the last one. */ - public const VALID_AGGREGATES = ['sum', 'peak']; + public const VALID_AGGREGATES = ['sum', 'max']; /** * Valid interval values and their ClickHouse INTERVAL equivalents. @@ -201,9 +203,10 @@ public static function removeGroupBy(array $queries): array * Create an aggregate query selecting the aggregation function. * * `sum` (the default when no aggregate query is present) totals the metric - * value per bucket. `peak` computes the running-sum maximum — the peak - * concurrent value for delta metrics emitted as `+1`/`-1` (realtime - * connections). See {@see UsageQuery::VALID_AGGREGATES}. + * value per bucket. `max` takes the largest value in the bucket instead — + * the meaningful roll-up for a gauge level series, whose default + * `argMax(value, time)` would return the latest reading. + * See {@see UsageQuery::VALID_AGGREGATES}. * * @param string $function One of {@see UsageQuery::VALID_AGGREGATES}. * @return self diff --git a/tests/Usage/Adapter/ClickHouseGaugeMaxTest.php b/tests/Usage/Adapter/ClickHouseGaugeMaxTest.php new file mode 100644 index 00000000..20608673 --- /dev/null +++ b/tests/Usage/Adapter/ClickHouseGaugeMaxTest.php @@ -0,0 +1,262 @@ +adapter = new ClickHouseAdapter( + $host, + $username, + $password, + $port, + $secure, + namespace: 'utopia_usage_gauge_max', + database: getenv('CLICKHOUSE_DATABASE') ?: 'default', + ); + + $this->usage = new Usage($this->adapter); + $this->usage->setup(); + } + + /** + * Insert gauge samples straight into the gauges table. + * + * @param array $rows + */ + private function insertSamples(string $metric, array $rows): void + { + $table = $this->resolveTableName($this->adapter, 'getGaugesTableName'); + $database = $this->databaseName($this->adapter); + $ref = '`'.$database.'`.`'.$table.'`'; + + $tuples = []; + foreach ($rows as $i => $row) { + $id = $metric.'-'.$i; + $value = (int) $row['value']; + $time = $row['time']; + $tuples[] = "('{$id}', '{$metric}', {$value}, '{$time}')"; + } + + $sql = "INSERT INTO {$ref} (id, metric, value, time) VALUES ".implode(', ', $tuples); + $this->queryRaw($this->adapter, $sql); + } + + public function test_gauge_defaults_to_latest_reading(): void + { + $metric = 'rt-concurrent-default-'.uniqid(); + + // Peak is 9, but the last sample is 4 — the default argMax path + // must still return the latest reading. + $this->insertSamples($metric, [ + ['value' => 2, 'time' => '2026-06-01 00:00:00'], + ['value' => 9, 'time' => '2026-06-01 00:05:00'], + ['value' => 4, 'time' => '2026-06-01 00:10:00'], + ]); + + $results = $this->usage->find('1', [ + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 00:00:00'), + Query::lessThanEqual('time', '2026-06-01 01:00:00'), + UsageQuery::groupByInterval('time', '1h'), + ], Usage::TYPE_GAUGE); + + $this->assertCount(1, $results); + $this->assertEquals(4, $results[0]->getValue()); + } + + public function test_gauge_max_returns_highest_sample(): void + { + $metric = 'rt-concurrent-max-'.uniqid(); + + $this->insertSamples($metric, [ + ['value' => 2, 'time' => '2026-06-01 00:00:00'], + ['value' => 9, 'time' => '2026-06-01 00:05:00'], + ['value' => 4, 'time' => '2026-06-01 00:10:00'], + ]); + + $results = $this->usage->find('1', [ + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 00:00:00'), + Query::lessThanEqual('time', '2026-06-01 01:00:00'), + UsageQuery::groupByInterval('time', '1h'), + UsageQuery::aggregate('max'), + ], Usage::TYPE_GAUGE); + + $this->assertCount(1, $results); + $this->assertEquals(9, $results[0]->getValue()); + } + + public function test_gauge_max_per_bucket(): void + { + $metric = 'rt-concurrent-buckets-'.uniqid(); + + // Hour 00 peaks at 9; hour 01 peaks at 6. Max composes upward, so + // each bucket reports its own highest sample. + $this->insertSamples($metric, [ + ['value' => 2, 'time' => '2026-06-01 00:00:00'], + ['value' => 9, 'time' => '2026-06-01 00:05:00'], + ['value' => 4, 'time' => '2026-06-01 00:10:00'], + ['value' => 6, 'time' => '2026-06-01 01:05:00'], + ['value' => 1, 'time' => '2026-06-01 01:10:00'], + ]); + + $results = $this->usage->find('1', [ + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 00:00:00'), + Query::lessThanEqual('time', '2026-06-01 02:00:00'), + UsageQuery::groupByInterval('time', '1h'), + UsageQuery::aggregate('max'), + ], Usage::TYPE_GAUGE); + + $this->assertCount(2, $results); + $this->assertEquals(9, $results[0]->getValue()); + $this->assertEquals(6, $results[1]->getValue()); + } + + public function test_gauge_max_flat_aggregate_over_window(): void + { + $metric = 'rt-concurrent-flat-'.uniqid(); + + // No interval: one row for the whole window — the billing shape. + $this->insertSamples($metric, [ + ['value' => 2, 'time' => '2026-06-01 00:00:00'], + ['value' => 9, 'time' => '2026-06-01 00:05:00'], + ['value' => 6, 'time' => '2026-06-01 01:05:00'], + ]); + + $results = $this->usage->find('1', [ + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 00:00:00'), + Query::lessThanEqual('time', '2026-06-01 02:00:00'), + UsageQuery::aggregate('max'), + ], Usage::TYPE_GAUGE); + + $this->assertCount(1, $results); + $this->assertEquals(9, $results[0]->getValue()); + } + + /** + * Shared-tables adapter — the multi-tenant shape cloud runs. Built on + * demand so the single-tenant tests above keep the simpler schema. + * + * @return array{0: Usage, 1: ClickHouseAdapter} + */ + private function sharedTablesUsage(): array + { + $adapter = new ClickHouseAdapter( + getenv('CLICKHOUSE_HOST') ?: 'clickhouse', + getenv('CLICKHOUSE_USER') ?: 'default', + getenv('CLICKHOUSE_PASSWORD') ?: 'clickhouse', + (int) (getenv('CLICKHOUSE_PORT') ?: 8123), + (bool) (getenv('CLICKHOUSE_SECURE') ?: false), + namespace: 'utopia_usage_gauge_max_shared', + database: getenv('CLICKHOUSE_DATABASE') ?: 'default', + sharedTables: true, + ); + + $usage = new Usage($adapter); + $usage->setup(); + + return [$usage, $adapter]; + } + + /** + * Two tenants, one table, one metric — 7 for tenant-a and 5 for tenant-b. + */ + private function seedTwoTenants(ClickHouseAdapter $adapter, string $metric): void + { + $table = $this->resolveTableName($adapter, 'getGaugesTableName'); + $database = $this->databaseName($adapter); + $ref = '`'.$database.'`.`'.$table.'`'; + + $sql = "INSERT INTO {$ref} (id, metric, value, time, tenant) VALUES " + ."('{$metric}-a1', '{$metric}', 3, '2026-06-01 00:00:00', 'tenant-a'), " + ."('{$metric}-a2', '{$metric}', 7, '2026-06-01 00:05:00', 'tenant-a'), " + ."('{$metric}-b1', '{$metric}', 2, '2026-06-01 00:00:00', 'tenant-b'), " + ."('{$metric}-b2', '{$metric}', 5, '2026-06-01 00:05:00', 'tenant-b')"; + $this->queryRaw($adapter, $sql); + } + + public function test_gauge_max_stays_scoped_to_one_tenant(): void + { + [$usage, $adapter] = $this->sharedTablesUsage(); + $metric = 'rt-concurrent-scoped-'.uniqid(); + $this->seedTwoTenants($adapter, $metric); + + // tenant-a's own peak — never tenant-b's, never a combined figure. + $results = $usage->find('tenant-a', [ + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 00:00:00'), + Query::lessThanEqual('time', '2026-06-01 01:00:00'), + UsageQuery::aggregate('max'), + ], Usage::TYPE_GAUGE); + + $this->assertCount(1, $results); + $this->assertEquals(7, $results[0]->getValue()); + } + + public function test_find_across_tenants_returns_a_row_per_tenant(): void + { + [$usage, $adapter] = $this->sharedTablesUsage(); + $metric = 'rt-concurrent-crosstenant-'.uniqid(); + $this->seedTwoTenants($adapter, $metric); + + // One pass over every tenant — what the aggregation job needs instead + // of N per-tenant queries. groupBy('tenant') keeps rows attributable. + $results = $usage->findAcrossTenants([ + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', '2026-06-01 00:00:00'), + Query::lessThanEqual('time', '2026-06-01 01:00:00'), + UsageQuery::groupBy('tenant'), + UsageQuery::aggregate('max'), + ], Usage::TYPE_GAUGE); + + $byTenant = []; + foreach ($results as $row) { + $byTenant[$row->getTenant()] = $row->getValue(); + } + + $this->assertCount(2, $byTenant); + $this->assertEquals(7, $byTenant['tenant-a']); + $this->assertEquals(5, $byTenant['tenant-b']); + } + + public function test_find_across_tenants_rejected_without_shared_tables(): void + { + $this->expectException(\Exception::class); + $this->expectExceptionMessage('requires shared-tables mode'); + + $this->usage->findAcrossTenants([ + Query::equal('metric', ['anything']), + ], Usage::TYPE_GAUGE); + } +} diff --git a/tests/Usage/Adapter/ClickHousePeakTest.php b/tests/Usage/Adapter/ClickHousePeakTest.php deleted file mode 100644 index 3dd9000d..00000000 --- a/tests/Usage/Adapter/ClickHousePeakTest.php +++ /dev/null @@ -1,310 +0,0 @@ - addBatch path, which now persists - * negative event deltas. - */ -class ClickHousePeakTest extends ClickHouseTestCase -{ - private Usage $usage; - - private ClickHouseAdapter $adapter; - - protected function setUp(): void - { - $host = getenv('CLICKHOUSE_HOST') ?: 'clickhouse'; - $username = getenv('CLICKHOUSE_USER') ?: 'default'; - $password = getenv('CLICKHOUSE_PASSWORD') ?: 'clickhouse'; - $port = (int) (getenv('CLICKHOUSE_PORT') ?: 8123); - $secure = (bool) (getenv('CLICKHOUSE_SECURE') ?: false); - - $this->adapter = new ClickHouseAdapter( - $host, - $username, - $password, - $port, - $secure, - namespace: 'utopia_usage_peak', - database: getenv('CLICKHOUSE_DATABASE') ?: 'default', - ); - - $this->usage = new Usage($this->adapter); - $this->usage->setup(); - } - - /** - * Insert raw concurrency deltas straight into the events table. - * - * @param array $rows - */ - private function insertDeltas(string $metric, array $rows): void - { - $table = $this->resolveTableName($this->adapter, 'getEventsTableName'); - $database = $this->databaseName($this->adapter); - $ref = '`'.$database.'`.`'.$table.'`'; - - $tuples = []; - foreach ($rows as $i => $row) { - $id = $metric.'-'.$i; - $value = (int) $row['value']; - $time = $row['time']; - $tuples[] = "('{$id}', '{$metric}', {$value}, '{$time}')"; - } - - $sql = "INSERT INTO {$ref} (id, metric, value, time) VALUES ".implode(', ', $tuples); - $this->queryRaw($this->adapter, $sql); - } - - public function test_peak_flat_aggregate(): void - { - $metric = 'rt-peak-flat-'.uniqid(); - - // +1,+1,+1,+1,-1 → running 1,2,3,4,3 → peak = 4. - $this->insertDeltas($metric, [ - ['value' => 1, 'time' => '2026-06-01 00:01:00'], - ['value' => 1, 'time' => '2026-06-01 00:02:00'], - ['value' => 1, 'time' => '2026-06-01 00:03:00'], - ['value' => 1, 'time' => '2026-06-01 00:04:00'], - ['value' => -1, 'time' => '2026-06-01 00:05:00'], - ]); - - $results = $this->usage->find('1', [ - Query::equal('metric', [$metric]), - Query::greaterThanEqual('time', '2026-06-01 00:00:00'), - Query::lessThanEqual('time', '2026-06-01 01:00:00'), - UsageQuery::aggregate('peak'), - ], Usage::TYPE_EVENT); - - $this->assertCount(1, $results); - $this->assertEquals(4, $results[0]->getValue()); - } - - public function test_peak_with_group_by_interval(): void - { - $metric = 'rt-peak-interval-'.uniqid(); - - // Bucket 00: +1,+1 → running 1,2 → peak 2. - // Bucket 01: +1,-1 → running 3,2 → peak 3 (carries the still-open - // connections from bucket 00 forward, as concurrency should). - $this->insertDeltas($metric, [ - ['value' => 1, 'time' => '2026-06-01 00:10:00'], - ['value' => 1, 'time' => '2026-06-01 00:20:00'], - ['value' => 1, 'time' => '2026-06-01 01:10:00'], - ['value' => -1, 'time' => '2026-06-01 01:20:00'], - ]); - - $results = $this->usage->find('1', [ - UsageQuery::groupByInterval('time', '1h'), - Query::equal('metric', [$metric]), - Query::greaterThanEqual('time', '2026-06-01 00:00:00'), - Query::lessThanEqual('time', '2026-06-01 02:00:00'), - UsageQuery::aggregate('peak'), - ], Usage::TYPE_EVENT); - - $this->assertCount(2, $results); - - $byHour = []; - foreach ($results as $row) { - $time = $row->getTime(); - $this->assertNotNull($time); - $hour = (new \DateTime($time))->format('H'); - $byHour[$hour] = $row->getValue(); - } - - $this->assertEquals(2, $byHour['00'] ?? null); - $this->assertEquals(3, $byHour['01'] ?? null); - } - - public function test_peak_includes_pre_window_baseline(): void - { - $metric = 'rt-peak-baseline-'.uniqid(); - - // Two connections opened before the window start (still open at start): - // baseline = 2. In-window deltas +1,+1,-1 → cumulative 1,2,1 → - // running = 2 + [1,2,1] = 3,4,3 → peak = 4. Without the baseline the - // in-window max would understate to 2. - $this->insertDeltas($metric, [ - ['value' => 1, 'time' => '2026-05-31 23:00:00'], - ['value' => 1, 'time' => '2026-05-31 23:30:00'], - ['value' => 1, 'time' => '2026-06-01 00:01:00'], - ['value' => 1, 'time' => '2026-06-01 00:02:00'], - ['value' => -1, 'time' => '2026-06-01 00:03:00'], - ]); - - $results = $this->usage->find('1', [ - Query::equal('metric', [$metric]), - Query::greaterThanEqual('time', '2026-06-01 00:00:00'), - Query::lessThanEqual('time', '2026-06-01 01:00:00'), - UsageQuery::aggregate('peak'), - ], Usage::TYPE_EVENT); - - $this->assertCount(1, $results); - $this->assertEquals(4, $results[0]->getValue()); - } - - public function test_peak_sums_interleaved_producers(): void - { - $metric = 'rt-peak-pods-'.uniqid(); - - // Rows as if emitted by two realtime pods, interleaved in time. The - // running sum combines them (no per-pod partition), so the peak - // reflects the global concurrency, not any single pod's max. - // A:+1(01) B:+1(02) A:+1(03) B:+1(04) A:-1(05) - // running: 1, 2, 3, 4, 3 → peak = 4 - $this->insertDeltas($metric, [ - ['value' => 1, 'time' => '2026-06-01 00:01:00'], - ['value' => 1, 'time' => '2026-06-01 00:02:00'], - ['value' => 1, 'time' => '2026-06-01 00:03:00'], - ['value' => 1, 'time' => '2026-06-01 00:04:00'], - ['value' => -1, 'time' => '2026-06-01 00:05:00'], - ]); - - $results = $this->usage->find('1', [ - Query::equal('metric', [$metric]), - Query::greaterThanEqual('time', '2026-06-01 00:00:00'), - Query::lessThanEqual('time', '2026-06-01 01:00:00'), - UsageQuery::aggregate('peak'), - ], Usage::TYPE_EVENT); - - $this->assertCount(1, $results); - $this->assertEquals(4, $results[0]->getValue()); - } - - public function test_sum_aggregate_path_unchanged(): void - { - $metric = 'rt-peak-sum-'.uniqid(); - - // Same deltas — the default `sum` aggregate must still return the net - // total (1+1+1+1-1 = 3), proving the peak routing does not leak into - // the existing path. - $this->insertDeltas($metric, [ - ['value' => 1, 'time' => '2026-06-01 00:01:00'], - ['value' => 1, 'time' => '2026-06-01 00:02:00'], - ['value' => 1, 'time' => '2026-06-01 00:03:00'], - ['value' => 1, 'time' => '2026-06-01 00:04:00'], - ['value' => -1, 'time' => '2026-06-01 00:05:00'], - ]); - - $results = $this->usage->find('1', [ - UsageQuery::groupByInterval('time', '1h'), - Query::equal('metric', [$metric]), - Query::greaterThanEqual('time', '2026-06-01 00:00:00'), - Query::lessThanEqual('time', '2026-06-01 01:00:00'), - UsageQuery::aggregate('sum'), - ], Usage::TYPE_EVENT); - - $this->assertCount(1, $results); - $this->assertEquals(3, $results[0]->getValue()); - } - - public function testPeakOverPerSecondNetRowsCapturesBurst(): void - { - // A burst that rises to 4 and falls back, all inside one flush window. - // Per-second nets: s0 +2, s1 +2 (running 4 = peak), s2 -2, s3 -1. - // Collected through the Accumulator with foldSeconds=1 so each second - // survives as its own net row; allowNegative lets the -1 closes persist. - $foldMetric = 'rt-peak-persec-' . uniqid(); - $acc = new Accumulator($this->usage); - - $deltas = [ - [1, '2026-06-01 03:00:00.100'], - [1, '2026-06-01 03:00:00.900'], - [1, '2026-06-01 03:00:01.100'], - [1, '2026-06-01 03:00:01.900'], - [-1, '2026-06-01 03:00:02.100'], - [-1, '2026-06-01 03:00:02.900'], - [-1, '2026-06-01 03:00:03.100'], - ]; - foreach ($deltas as [$value, $time]) { - $acc->collect('1', $foldMetric, $value, Usage::TYPE_EVENT, [], new \DateTime($time), 1, allowNegative: true); - } - - // Four per-second net rows (+2, +2, -2, -1), not one folded delta. - $this->assertEquals(4, $acc->count()); - $this->assertTrue($acc->flush()); - - $window = [ - Query::greaterThanEqual('time', '2026-06-01 03:00:00'), - Query::lessThanEqual('time', '2026-06-01 03:01:00'), - ]; - - $peak = $this->usage->find('1', array_merge([ - Query::equal('metric', [$foldMetric]), - ], $window, [UsageQuery::aggregate('peak')]), Usage::TYPE_EVENT); - - $this->assertCount(1, $peak); - $this->assertEquals(4, $peak[0]->getValue()); - - // Contrast: the same deltas folded WITHOUT a per-second key collapse to - // a single net row (+1), whose running-sum peak is only 1 — the burst - // to 4 is invisible. This is exactly what the per-second fold fixes. - $flatMetric = 'rt-peak-perflush-' . uniqid(); - $accFlat = new Accumulator($this->usage); - foreach ($deltas as [$value, $time]) { - $accFlat->collect('1', $flatMetric, $value, Usage::TYPE_EVENT, [], new \DateTime($time), allowNegative: true); - } - $this->assertEquals(1, $accFlat->count()); - $this->assertTrue($accFlat->flush()); - - $flatPeak = $this->usage->find('1', array_merge([ - Query::equal('metric', [$flatMetric]), - ], $window, [UsageQuery::aggregate('peak')]), Usage::TYPE_EVENT); - - $this->assertCount(1, $flatPeak); - $this->assertEquals(1, $flatPeak[0]->getValue()); - } - - public function testAddBatchRejectsNegativeEventByDefault(): void - { - // Strict by default: a negative event without the opt-in is rejected - // at write time, so a buggy negative count never reaches storage. - $this->expectException(\Exception::class); - $this->expectExceptionMessage('Value cannot be negative'); - $this->usage->addBatch([ - ['tenant' => '1', 'metric' => 'rt-neg-' . uniqid(), 'value' => -1, 'tags' => []], - ], Usage::TYPE_EVENT); - } - - public function testAddBatchPersistsNegativeEventWhenOptedIn(): void - { - // With allowNegative on the row, the -1 close persists and nets with - // the +1 opens so the peak path sees true concurrency. - $metric = 'rt-neg-allowed-' . uniqid(); - - $this->assertTrue($this->usage->addBatch([ - ['tenant' => '1', 'metric' => $metric, 'value' => 1, 'time' => new \DateTime('2026-06-01 04:00:01'), 'tags' => [], 'allowNegative' => true], - ['tenant' => '1', 'metric' => $metric, 'value' => 1, 'time' => new \DateTime('2026-06-01 04:00:02'), 'tags' => [], 'allowNegative' => true], - ['tenant' => '1', 'metric' => $metric, 'value' => -1, 'time' => new \DateTime('2026-06-01 04:00:03'), 'tags' => [], 'allowNegative' => true], - ], Usage::TYPE_EVENT)); - - // Net total is +1 (SUM), peak concurrency is 2 (running 1,2,1). - $this->assertEquals(1, $this->usage->getTotal('1', $metric, [], Usage::TYPE_EVENT)); - - $peak = $this->usage->find('1', [ - Query::equal('metric', [$metric]), - Query::greaterThanEqual('time', '2026-06-01 04:00:00'), - Query::lessThanEqual('time', '2026-06-01 04:01:00'), - UsageQuery::aggregate('peak'), - ], Usage::TYPE_EVENT); - - $this->assertCount(1, $peak); - $this->assertEquals(2, $peak[0]->getValue()); - } -} diff --git a/tests/Usage/UsageQueryTest.php b/tests/Usage/UsageQueryTest.php index dd1eb8e0..fa19b86a 100644 --- a/tests/Usage/UsageQueryTest.php +++ b/tests/Usage/UsageQueryTest.php @@ -234,12 +234,12 @@ public function testExtractGroupByFromParsedQuery(): void public function testAggregateCreation(): void { - $query = UsageQuery::aggregate('peak'); + $query = UsageQuery::aggregate('max'); $this->assertInstanceOf(UsageQuery::class, $query); $this->assertEquals(UsageQuery::TYPE_AGGREGATE, $query->getMethod()); - $this->assertEquals(['peak'], $query->getValues()); - $this->assertEquals('peak', $query->getValue()); + $this->assertEquals(['max'], $query->getValues()); + $this->assertEquals('max', $query->getValue()); } public function testAggregateAcceptsAllValidFunctions(): void @@ -253,7 +253,7 @@ public function testAggregateAcceptsAllValidFunctions(): void public function testAggregateRejectsInvalidFunction(): void { $this->expectException(\InvalidArgumentException::class); - $this->expectExceptionMessage("Invalid aggregate 'max'"); + $this->expectExceptionMessage("Invalid aggregate 'peak'"); UsageQuery::aggregate('max'); } @@ -264,7 +264,7 @@ public function testAggregateIsMethod(): void public function testIsAggregate(): void { - $aggregate = UsageQuery::aggregate('peak'); + $aggregate = UsageQuery::aggregate('max'); $regular = Query::equal('metric', ['bandwidth']); $this->assertTrue(UsageQuery::isAggregate($aggregate)); @@ -275,19 +275,19 @@ public function testExtractAggregate(): void { $queries = [ Query::equal('metric', ['realtime.connections']), - UsageQuery::aggregate('peak'), + UsageQuery::aggregate('max'), ]; - $this->assertEquals('peak', UsageQuery::extractAggregate($queries)); + $this->assertEquals('max', UsageQuery::extractAggregate($queries)); } public function testExtractAggregateFromParsedQuery(): void { // Queries created via Query::parse() are base Query objects, not UsageQuery. - $parsedAggregate = new Query(UsageQuery::TYPE_AGGREGATE, 'value', ['peak']); + $parsedAggregate = new Query(UsageQuery::TYPE_AGGREGATE, 'value', ['max']); $equal = Query::equal('metric', ['realtime.connections']); - $this->assertEquals('peak', UsageQuery::extractAggregate([$equal, $parsedAggregate])); + $this->assertEquals('max', UsageQuery::extractAggregate([$equal, $parsedAggregate])); } public function testExtractAggregateReturnsNullWhenMissing(): void @@ -304,7 +304,7 @@ public function testRemoveAggregate(): void { $queries = [ Query::equal('metric', ['realtime.connections']), - UsageQuery::aggregate('peak'), + UsageQuery::aggregate('max'), UsageQuery::groupByInterval('time', '1h'), ]; @@ -319,6 +319,6 @@ public function testRemoveAggregate(): void public function testValidAggregatesConstant(): void { $this->assertContains('sum', UsageQuery::VALID_AGGREGATES); - $this->assertContains('peak', UsageQuery::VALID_AGGREGATES); + $this->assertContains('max', UsageQuery::VALID_AGGREGATES); } } From c90c40ad38bc10dbaa4ae76bf20a165c1676c228 Mon Sep 17 00:00:00 2001 From: Damodar Lohani Date: Wed, 5 Aug 2026 07:57:51 +0545 Subject: [PATCH 5/8] Route a bare aggregate() to the aggregated path findFromTable() only checked for groupByInterval or groupBy, so an aggregate('max') with neither fell through and returned raw rows instead of one aggregated row. That is exactly the flat 'highest value over this window' shape billing reads, which would have taken an arbitrary row's value for the peak. The removed peak branch used to catch this case before the check. Also declare aggregate on the parsed-array shape so PHPStan sees the offset, fix a test that asserted a rejection while passing a valid function, and assert the tenant is present before using it as an array key. --- src/Usage/Adapter/ClickHouse.php | 16 ++++++++++++---- tests/Usage/Adapter/ClickHouseGaugeMaxTest.php | 4 +++- tests/Usage/UsageQueryTest.php | 2 +- 3 files changed, 16 insertions(+), 6 deletions(-) diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index 032cc3a3..8a3501e6 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -1533,9 +1533,17 @@ private function findFromTable(?string $tenant, array $queries, string $type): a throw new Exception('Cursor pagination cannot be combined with groupByInterval'); } - // Route through the aggregated path whenever any aggregation - // hint is present — time bucketing, dimension breakdown, or both. - if (isset($parsed['groupByInterval']) || !empty($parsed['groupBy'])) { + // Route through the aggregated path whenever any aggregation hint is + // present — time bucketing, dimension breakdown, an explicit + // aggregate(), or any combination. An aggregate() on its own still + // counts: `aggregate('max')` with no interval and no dimensions is the + // flat "highest value over this window" shape, and without this it + // would fall through and return raw rows instead. + if ( + isset($parsed['groupByInterval']) + || !empty($parsed['groupBy']) + || isset($parsed['aggregate']) + ) { return $this->findAggregatedFromTable($parsed, $fromTable, $type); } @@ -1596,7 +1604,7 @@ private function findFromTable(?string $tenant, array $queries, string $type): a * toStartOfInterval(time, INTERVAL 1 HOUR) as time * FROM table WHERE ... GROUP BY metric, time ORDER BY time ASC * - * @param array{filters: array, params: array, orderBy?: array, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array} $parsed Parsed query data from parseQueries() + * @param array{filters: array, params: array, orderBy?: array, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array, aggregate?: string} $parsed Parsed query data from parseQueries() * @param string $fromTable Fully qualified table reference * @param string $type 'event' or 'gauge' * @return array diff --git a/tests/Usage/Adapter/ClickHouseGaugeMaxTest.php b/tests/Usage/Adapter/ClickHouseGaugeMaxTest.php index 20608673..882ec4d6 100644 --- a/tests/Usage/Adapter/ClickHouseGaugeMaxTest.php +++ b/tests/Usage/Adapter/ClickHouseGaugeMaxTest.php @@ -242,7 +242,9 @@ public function test_find_across_tenants_returns_a_row_per_tenant(): void $byTenant = []; foreach ($results as $row) { - $byTenant[$row->getTenant()] = $row->getValue(); + $tenant = $row->getTenant(); + $this->assertNotNull($tenant, 'every cross-tenant row must carry its tenant'); + $byTenant[$tenant] = $row->getValue(); } $this->assertCount(2, $byTenant); diff --git a/tests/Usage/UsageQueryTest.php b/tests/Usage/UsageQueryTest.php index fa19b86a..5a0acd4a 100644 --- a/tests/Usage/UsageQueryTest.php +++ b/tests/Usage/UsageQueryTest.php @@ -254,7 +254,7 @@ public function testAggregateRejectsInvalidFunction(): void { $this->expectException(\InvalidArgumentException::class); $this->expectExceptionMessage("Invalid aggregate 'peak'"); - UsageQuery::aggregate('max'); + UsageQuery::aggregate('peak'); } public function testAggregateIsMethod(): void From ef022112d6315ff1f205b9dab651a6933a0bda42 Mon Sep 17 00:00:00 2001 From: Damodar Lohani Date: Wed, 5 Aug 2026 09:36:18 +0545 Subject: [PATCH 6/8] Drop the unused per-second fold from Accumulator foldSeconds was added for a request-time peak that no longer exists: with the concurrency level pre-computed into a gauge and sampled every five minutes, per-second rows cost roughly twenty times the storage to sharpen a resolution the bucketing discards. Nothing passed it. Leaving it in would have meant an unused parameter on the hottest write path of a shared library, on a branch nothing exercises. It is one commit back in history if a caller ever wants it, and then it arrives with the caller that justifies it. allowNegative stays - the concurrency level is the cumulative sum of the +1/-1s, so without it there is nothing to fold. --- src/Usage/Accumulator.php | 41 +++----------------- tests/Usage/AccumulatorTest.php | 66 --------------------------------- 2 files changed, 6 insertions(+), 101 deletions(-) diff --git a/src/Usage/Accumulator.php b/src/Usage/Accumulator.php index 716ad687..159f9687 100644 --- a/src/Usage/Accumulator.php +++ b/src/Usage/Accumulator.php @@ -44,17 +44,10 @@ public function __construct(Usage $usage) * `$allowNegative = true`. The library stays generic — the decision lives * with the caller, not with any metric-name knowledge here. * - * When $foldSeconds is set, events only fold within the same time bucket: - * the bucket start (floor(timestamp / $foldSeconds) * $foldSeconds) joins - * the fold key and becomes the entry's canonical time, so every event in - * that window shares one timestamp and sub-flush peaks survive the fold. - * When $foldSeconds is null the fold key and time handling are unchanged. - * * @param array $tags - * @param int|null $foldSeconds Fold events within this many seconds, or null for the default per-flush fold. * @param bool $allowNegative Permit a negative value for this metric (default: reject). */ - public function collect(string $tenant, string $metric, int $value, string $type, array $tags = [], ?\DateTime $time = null, ?int $foldSeconds = null, bool $allowNegative = false): self + public function collect(string $tenant, string $metric, int $value, string $type, array $tags = [], ?\DateTime $time = null, bool $allowNegative = false): self { // Compare against '' rather than empty(): the string "0" is a valid // tenant/metric id but empty("0") is true in PHP. @@ -70,9 +63,6 @@ public function collect(string $tenant, string $metric, int $value, string $type if ($type !== Usage::TYPE_EVENT && $type !== Usage::TYPE_GAUGE) { throw new \InvalidArgumentException("Invalid metric type '{$type}'. Allowed: " . Usage::TYPE_EVENT . ', ' . Usage::TYPE_GAUGE); } - if ($foldSeconds !== null && $foldSeconds <= 0) { - throw new \InvalidArgumentException('foldSeconds must be a positive integer'); - } // Hash the full identity so distinct (tenant, metric, type, tags) // tuples never collide on the key — a raw `:`-join would let @@ -80,31 +70,12 @@ public function collect(string $tenant, string $metric, int $value, string $type // entry. Tags are sorted first so key order doesn't matter. $canonicalTags = $tags; ksort($canonicalTags); - - // Per-bucket fold: the bucket start joins the key so events only fold - // within the same window, and becomes the canonical entry time so all - // events in that bucket share one timestamp. - $bucketTime = null; - if ($foldSeconds !== null) { - $timestamp = $time !== null ? $time->getTimestamp() : time(); - $bucketStart = intdiv($timestamp, $foldSeconds) * $foldSeconds; - $bucketTime = new \DateTime('@' . $bucketStart); - $key = md5(json_encode([$tenant, $metric, $type, $canonicalTags, $bucketStart], JSON_THROW_ON_ERROR)); - } else { - $key = md5(json_encode([$tenant, $metric, $type, $canonicalTags], JSON_THROW_ON_ERROR)); - } - - // With bucketing, the bucket start is the entry time; otherwise the - // caller-supplied time (if any) is used. - $effectiveTime = $bucketTime ?? $time; + $key = md5(json_encode([$tenant, $metric, $type, $canonicalTags], JSON_THROW_ON_ERROR)); if ($type === Usage::TYPE_EVENT && isset($this->buffer[$key])) { + // earliest time wins on merge $this->buffer[$key]['value'] += $value; - if ($bucketTime !== null) { - // Every event in this bucket shares the bucket-start time. - $this->buffer[$key]['time'] = $bucketTime; - } elseif ($time !== null && (!isset($this->buffer[$key]['time']) || $time < $this->buffer[$key]['time'])) { - // earliest time wins on merge + if ($time !== null && (!isset($this->buffer[$key]['time']) || $time < $this->buffer[$key]['time'])) { $this->buffer[$key]['time'] = $time; } } else { @@ -116,8 +87,8 @@ public function collect(string $tenant, string $metric, int $value, string $type 'tags' => $tags, 'allowNegative' => $allowNegative, ]; - if ($effectiveTime !== null) { - $entry['time'] = $effectiveTime; + if ($time !== null) { + $entry['time'] = $time; } $this->buffer[$key] = $entry; } diff --git a/tests/Usage/AccumulatorTest.php b/tests/Usage/AccumulatorTest.php index d1ea20f9..0dc3cc4a 100644 --- a/tests/Usage/AccumulatorTest.php +++ b/tests/Usage/AccumulatorTest.php @@ -319,72 +319,6 @@ public function testEventMixedSignsNetToDelta(): void $this->assertEquals(1, $this->adapter->batches[0]['metrics'][0]['value']); } - public function testFoldSecondsGroupsEventsInSameSecond(): void - { - // Two events landing in the same 1s bucket fold into one net row - // stamped at the bucket start. - $a = new \DateTime('2026-04-15 12:00:00.200'); - $b = new \DateTime('2026-04-15 12:00:00.900'); - - $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $a, 1); - $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $b, 1); - - $this->assertEquals(1, $this->accumulator->count()); - $this->assertTrue($this->accumulator->flush()); - - $entry = $this->adapter->batches[0]['metrics'][0]; - $this->assertEquals(2, $entry['value']); - $this->assertArrayHasKey('time', $entry); - $time = $entry['time']; - $this->assertInstanceOf(\DateTime::class, $time); - // Stamped at the second boundary, not the sub-second event time. - $this->assertEquals('2026-04-15 12:00:00', $time->format('Y-m-d H:i:s')); - } - - public function testFoldSecondsKeepsDistinctSecondsSeparate(): void - { - // Events in different 1s buckets stay as separate rows so a rise and - // fall inside a flush is not collapsed away. - $s0 = new \DateTime('2026-04-15 12:00:00.500'); - $s1 = new \DateTime('2026-04-15 12:00:01.500'); - - $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $s0, 1); - $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $s1, 1); - - $this->assertEquals(2, $this->accumulator->count()); - $this->assertTrue($this->accumulator->flush()); - - $times = []; - foreach ($this->adapter->batches[0]['metrics'] as $m) { - $time = $m['time'] ?? null; - $this->assertInstanceOf(\DateTime::class, $time); - $times[] = $time->format('Y-m-d H:i:s'); - } - sort($times); - $this->assertEquals(['2026-04-15 12:00:00', '2026-04-15 12:00:01'], $times); - } - - public function testFoldSecondsNetsWithinBucket(): void - { - // Within a single second, +/- deltas net; the per-second row carries - // the net so cross-pod running sums stay exact. - $t = new \DateTime('2026-04-15 12:00:03.100'); - $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $t, 1, allowNegative: true); - $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], $t, 1, allowNegative: true); - $this->accumulator->collect('t1', 'realtime.connections', -1, Usage::TYPE_EVENT, [], $t, 1, allowNegative: true); - - $this->assertEquals(1, $this->accumulator->count()); - $this->assertTrue($this->accumulator->flush()); - $this->assertEquals(1, $this->adapter->batches[0]['metrics'][0]['value']); - } - - public function testFoldSecondsRejectsNonPositive(): void - { - $this->expectException(\InvalidArgumentException::class); - $this->expectExceptionMessage('foldSeconds must be a positive integer'); - $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, [], null, 0); - } - public function testInvalidTypeThrows(): void { $this->expectException(\InvalidArgumentException::class); From 8f4690f1b6dcdae21de35515c4abbf3f1d0ac0a7 Mon Sep 17 00:00:00 2001 From: Damodar Lohani Date: Wed, 5 Aug 2026 09:39:35 +0545 Subject: [PATCH 7/8] Keep the negative opt-in when events fold, drop the sum aggregate Two findings from review. The opt-in was stored only when a buffer entry was created, so folding a signed delta into an entry opened by a plain positive left the net row looking unauthorised. It would be rejected at write time, and because a failed batch keeps its entries buffered, every later flush would retry the same rejection - a stuck buffer rather than one lost row. The flag now ORs across the fold, with tests in both call orders. Unreachable from cloud today, since the flag is derived from the metric name and the fold key includes it. But the fold key uses the name as written while the flag is keyed on the source, so a mirrored metric that was also a delta would mix them. Cheap to make the invariant hold on its own rather than resting on a caller-side coincidence. aggregate('sum') was accepted but left gauges on argMax, returning the latest sample rather than the documented total. Rather than make it sum, it is gone: it is already the default for events, and on gauges it would total point-in-time snapshots, which this library elsewhere refuses to do. max is now the only selectable aggregate - it exists to override the per-type default. --- src/Usage/Accumulator.php | 9 +++++++++ src/Usage/UsageQuery.php | 29 ++++++++++++++-------------- tests/Usage/AccumulatorTest.php | 34 +++++++++++++++++++++++++++++++++ tests/Usage/UsageQueryTest.php | 6 ++++-- 4 files changed, 62 insertions(+), 16 deletions(-) diff --git a/src/Usage/Accumulator.php b/src/Usage/Accumulator.php index 159f9687..dac544df 100644 --- a/src/Usage/Accumulator.php +++ b/src/Usage/Accumulator.php @@ -75,6 +75,15 @@ public function collect(string $tenant, string $metric, int $value, string $type if ($type === Usage::TYPE_EVENT && isset($this->buffer[$key])) { // earliest time wins on merge $this->buffer[$key]['value'] += $value; + + // The opt-in is a property of the folded row, not of whichever + // call happened to create it. Folding a signed delta into an entry + // opened by a plain positive must not leave the net row looking + // unauthorised: it would be rejected at write time, and since a + // failed batch keeps its entries buffered, every later flush would + // retry the same rejection. + $this->buffer[$key]['allowNegative'] = $this->buffer[$key]['allowNegative'] || $allowNegative; + if ($time !== null && (!isset($this->buffer[$key]['time']) || $time < $this->buffer[$key]['time'])) { $this->buffer[$key]['time'] = $time; } diff --git a/src/Usage/UsageQuery.php b/src/Usage/UsageQuery.php index 44fb2a5a..cb4b5072 100644 --- a/src/Usage/UsageQuery.php +++ b/src/Usage/UsageQuery.php @@ -27,10 +27,10 @@ * - Events: SUM(value) per bucket * - Gauges: argMax(value, time) per bucket * - * An `aggregate` hint overrides how values are combined. The default `sum` - * totals them; `max` takes the largest value per bucket, which is how a gauge - * level series rolls up to a coarser interval — the default `argMax` would - * return the latest reading rather than the highest. + * An `aggregate` hint overrides how values are combined: `max` takes the + * largest value per bucket, which is how a gauge level series rolls up to a + * coarser interval — the default `argMax` would return the latest reading + * rather than the highest. */ class UsageQuery extends Query { @@ -41,14 +41,16 @@ class UsageQuery extends Query /** * Valid aggregation functions. * - * - `sum` — the default; totals the metric value per bucket (current behaviour). - * - `max` — the largest value per bucket. Intended for gauges, where the - * default `argMax(value, time)` returns the *latest* reading rather than - * the highest one. Reading a pre-computed level series (e.g. realtime - * concurrency sampled every 5 minutes) at a coarser interval needs the - * peak of the samples, not the last one. + * - `max` — the largest value per bucket, overriding the per-type default. + * Intended for gauges, where the default `argMax(value, time)` returns the + * *latest* reading rather than the highest one. Reading a pre-computed + * level series (e.g. realtime concurrency sampled every 5 minutes) at a + * coarser interval needs the peak of the samples, not the last one. + * + * There is deliberately no `sum`: it is already the default for events, and + * on gauges it would total point-in-time snapshots, which means nothing. */ - public const VALID_AGGREGATES = ['sum', 'max']; + public const VALID_AGGREGATES = ['max']; /** * Valid interval values and their ClickHouse INTERVAL equivalents. @@ -202,9 +204,8 @@ public static function removeGroupBy(array $queries): array /** * Create an aggregate query selecting the aggregation function. * - * `sum` (the default when no aggregate query is present) totals the metric - * value per bucket. `max` takes the largest value in the bucket instead — - * the meaningful roll-up for a gauge level series, whose default + * `max` takes the largest value in the bucket, overriding the per-type + * default — the meaningful roll-up for a gauge level series, whose default * `argMax(value, time)` would return the latest reading. * See {@see UsageQuery::VALID_AGGREGATES}. * diff --git a/tests/Usage/AccumulatorTest.php b/tests/Usage/AccumulatorTest.php index 0dc3cc4a..a2417dfe 100644 --- a/tests/Usage/AccumulatorTest.php +++ b/tests/Usage/AccumulatorTest.php @@ -319,6 +319,40 @@ public function testEventMixedSignsNetToDelta(): void $this->assertEquals(1, $this->adapter->batches[0]['metrics'][0]['value']); } + public function testFoldedEntryKeepsTheNegativeOptIn(): void + { + // The opt-in belongs to the folded row, not to whichever call opened + // it. Here the entry is created by a plain positive with the flag off, + // then a signed delta folds in and takes the net negative. + $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT); + $this->accumulator->collect('t1', 'realtime.connections', -3, Usage::TYPE_EVENT, allowNegative: true); + + $this->assertEquals(1, $this->accumulator->count()); + $this->assertTrue($this->accumulator->flush()); + + $entry = $this->adapter->batches[0]['metrics'][0]; + $this->assertEquals(-2, $entry['value']); + + // Without this the net row is written unauthorised and rejected at + // validation. A failed batch keeps its entries buffered, so every + // later flush would retry the same rejection. + $this->assertTrue($entry['allowNegative']); + } + + public function testFoldRetainsOptInRegardlessOfCallOrder(): void + { + // Same fold, opposite order: the entry is opened by the opted-in + // delta and a plain positive folds in afterwards. + $this->accumulator->collect('t1', 'realtime.connections', -3, Usage::TYPE_EVENT, allowNegative: true); + $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT); + + $this->assertTrue($this->accumulator->flush()); + + $entry = $this->adapter->batches[0]['metrics'][0]; + $this->assertEquals(-2, $entry['value']); + $this->assertTrue($entry['allowNegative']); + } + public function testInvalidTypeThrows(): void { $this->expectException(\InvalidArgumentException::class); diff --git a/tests/Usage/UsageQueryTest.php b/tests/Usage/UsageQueryTest.php index 5a0acd4a..3b411069 100644 --- a/tests/Usage/UsageQueryTest.php +++ b/tests/Usage/UsageQueryTest.php @@ -318,7 +318,9 @@ public function testRemoveAggregate(): void public function testValidAggregatesConstant(): void { - $this->assertContains('sum', UsageQuery::VALID_AGGREGATES); - $this->assertContains('max', UsageQuery::VALID_AGGREGATES); + // `max` is the only selectable aggregate: it overrides the per-type + // default. `sum` is absent on purpose - already the default for events, + // and on gauges it would total point-in-time snapshots. + $this->assertSame(['max'], UsageQuery::VALID_AGGREGATES); } } From 51d11b628f1b8fc9382336eecbcf6e86b82ce66e Mon Sep 17 00:00:00 2001 From: Damodar Lohani Date: Wed, 5 Aug 2026 09:52:22 +0545 Subject: [PATCH 8/8] Honour the negative opt-in in the Database adapter too Accumulator marks a signed-delta row allowNegative and the ClickHouse adapter honours it, but Database::addBatch() rejected the value regardless. The opt-in is part of the row contract, so an adapter that ignores it turns a valid metric into a permanently stuck buffer: the batch throws, flush() keeps unwritten entries, and every later flush retries the same rejection. Not reachable from cloud, which runs ClickHouse, but the two adapters have to agree about what a row means. --- src/Usage/Adapter/Database.php | 10 ++++++-- tests/Usage/Adapter/DatabaseTest.php | 37 ++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 2 deletions(-) diff --git a/src/Usage/Adapter/Database.php b/src/Usage/Adapter/Database.php index dae9526c..866aeb52 100644 --- a/src/Usage/Adapter/Database.php +++ b/src/Usage/Adapter/Database.php @@ -142,7 +142,7 @@ protected function getColumnDefinition(string $id, string $type = 'event'): stri * Each metric carries its own `tenant` (shared-tables mode), so a single * batch may span multiple tenants. * - * @param array}> $metrics + * @param array, allowNegative?: bool}> $metrics * @param string $type Metric type: 'event' or 'gauge' * @param int $batchSize * @return bool @@ -157,7 +157,13 @@ public function addBatch(array $metrics, string $type, int $batchSize = 1000): b throw new \InvalidArgumentException("Invalid type '{$type}'. Allowed: event, gauge"); } - if ($metric['value'] < 0) { + // Negatives are rejected by default so a buggy count or + // bandwidth figure is still caught. A row carrying a genuine + // signed delta opts in, the same contract the accumulator sets + // and the ClickHouse adapter honours - an adapter that ignored + // it would reject the row on every flush, and a failed batch + // keeps its entries buffered. + if ($metric['value'] < 0 && !($metric['allowNegative'] ?? false)) { throw new \InvalidArgumentException('Value cannot be negative'); } diff --git a/tests/Usage/Adapter/DatabaseTest.php b/tests/Usage/Adapter/DatabaseTest.php index 3352b3ad..4dfcb806 100644 --- a/tests/Usage/Adapter/DatabaseTest.php +++ b/tests/Usage/Adapter/DatabaseTest.php @@ -9,6 +9,7 @@ use Utopia\Database\Adapter\MariaDB; use Utopia\Database\Database; use Utopia\Database\Exception\Duplicate; +use Utopia\Query\Query; use Utopia\Tests\Usage\UsageBase; use Utopia\Usage\Adapter\Database as AdapterDatabase; use Utopia\Usage\Usage; @@ -259,4 +260,40 @@ public function testHealthCheckWithNonExistentDatabase(): void $this->assertNotEmpty($health['error']); } } + + public function testNegativeValueRejectedByDefault(): void + { + if (!extension_loaded('pdo_mysql')) { + $this->markTestSkipped('pdo_mysql extension is not installed'); + } + + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage('Value cannot be negative'); + + $this->usage->addBatch([ + ['tenant' => '1', 'metric' => 'db-negative-default', 'value' => -1], + ], Usage::TYPE_EVENT); + } + + public function testNegativeValuePersistsWhenOptedIn(): void + { + if (!extension_loaded('pdo_mysql')) { + $this->markTestSkipped('pdo_mysql extension is not installed'); + } + + $this->usage->purge('1', [], Usage::TYPE_EVENT); + + // A signed delta opts in per row. The accumulator sets this flag and + // the ClickHouse adapter honours it; an adapter that ignored it would + // reject the row on every flush, since a failed batch stays buffered. + $this->assertTrue($this->usage->addBatch([ + ['tenant' => '1', 'metric' => 'db-negative-optin', 'value' => -3, 'allowNegative' => true], + ], Usage::TYPE_EVENT)); + + $this->assertEquals( + -3, + $this->usage->sum('1', [Query::equal('metric', ['db-negative-optin'])], 'value', Usage::TYPE_EVENT), + 'the opted-in negative must be stored, not silently dropped', + ); + } }