Skip to content

feat(usage): peak concurrent connections aggregation (CLO-4542) #22

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
lohanidamodar wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

feat(usage): peak concurrent connections aggregation (CLO-4542) #22

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
920ee10
feat(usage): add peak running-sum aggregation for delta metrics
lohanidamodar Jul 8, 2026
eafa857
feat(usage): add opt-in negative deltas and per-second fold
lohanidamodar Jul 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 45 additions & 8 deletions src/Usage/Accumulator.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ class Accumulator
private Usage $usage;

/**
* @var array<string, array{tenant: string, metric: string, value: int, type: string, tags: array<string, mixed>, time?: \DateTime}>
* @var array<string, array{tenant: string, metric: string, value: int, type: string, tags: array<string, mixed>, allowNegative: bool, time?: \DateTime}>
*/
private array $buffer = [];

Expand All @@ -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<string,mixed> $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.
Expand All @@ -50,25 +64,47 @@ 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
// e.g. tenant "a"/metric "b:c" and tenant "a:b"/metric "c" share one
// 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 {
Expand All @@ -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;
}
Expand Down
216 changes: 212 additions & 4 deletions src/Usage/Adapter/ClickHouse.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1205,9 +1205,10 @@ private function getColumnCodec(string $id): string
* @param string $type Metric type ('event' or 'gauge')
* @param array<string,mixed> $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}: " : '';

Expand All @@ -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');
}

Expand Down Expand Up @@ -1257,7 +1262,10 @@ private function validateMetricsBatch(array $metrics, string $type): void

/** @var array<string, mixed> */
$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);

Expand Down Expand Up @@ -1459,6 +1467,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<Query> $queries
* @param string $type 'event' or 'gauge'
* @return array<Metric>
Expand All @@ -1477,6 +1489,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'])) {
Expand Down Expand Up @@ -1630,6 +1649,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<int, string>, params: array<string, mixed>, orderBy?: array<string>, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array<int, string>, aggregate?: string, timeFilters?: array<int, string>, nonTimeFilters?: array<int, string>, peakStartParam?: string|null} $parsed
* @param string $fromTable Fully qualified table reference
* @param string $type 'event' or 'gauge'
* @return array<Metric>
* @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.
*
Expand Down Expand Up @@ -3381,7 +3546,7 @@ private function buildOrderBySql(array $orderAttributes, bool $flip = false): ar
* @param string $tenant Tenant scope (shared-tables mode)
* @param array<Query> $queries
* @param string $type 'event' or 'gauge' — used for attribute validation
* @return array{filters: array<int, string>, params: array<string, mixed>, orderBy?: array<string>, orderAttributes?: array<int, array{attribute: string, direction: string}>, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array<int, string>, cursor?: array<string, mixed>, cursorDirection?: string}
* @return array{filters: array<int, string>, params: array<string, mixed>, orderBy?: array<string>, orderAttributes?: array<int, array{attribute: string, direction: string}>, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array<int, string>, aggregate?: string, timeFilters?: array<int, string>, nonTimeFilters?: array<int, string>, peakStartParam?: string|null, cursor?: array<string, mixed>, cursorDirection?: string}
* @throws Exception
*/
private function parseQueries(string $tenant, array $queries, string $type = 'event'): array
Expand All @@ -3404,6 +3569,7 @@ private function parseQueries(string $tenant, array $queries, string $type = 'ev
$offset = null;
$groupByInterval = null;
$groupBy = [];
$aggregate = null;
$cursor = null;
$cursorDirection = null;
$paramCounter = 0;
Expand Down Expand Up @@ -3646,6 +3812,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;
}
}

Expand Down Expand Up @@ -3675,6 +3851,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;
Expand Down
Loading
Loading