From 920ee1080c465e25e1798b1cbcb2c5c595097bd5 Mon Sep 17 00:00:00 2001 From: Damodar Lohani Date: Wed, 8 Jul 2026 11:13:18 +0545 Subject: [PATCH 1/2] 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/2] 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()); + } }