diff --git a/CHANGELOG.md b/CHANGELOG.md index 776ea8b..6fd82e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ ## Unreleased — query 0.6.x builder +### Added + +- Added a separate immutable ClickHouse sample ledger for billable usage. A + canonical identity covers environment, region, project/database internal + IDs, member, generation, sequence and metric. Identical retries are + deduplicated at read time; conflicting payloads, sequence gaps, bounded-read + truncation and exact ingestion-ID watermark exclusions are explicit in + `SampleResult`. Conflict representatives are selected as one physical tuple, + never assembled from independent column aggregates. +- Added `Usage::addSamples()`, `Usage::getSampleWatermark()` and + `Usage::findSamples()`. Existing events, gauges and daily rollups are + unchanged and are not used as a canonical sample source. + ### Breaking - Bumped `utopia-php/query` from `0.1.*` to `0.6.*`. diff --git a/README.md b/README.md index 9272193..cd23928 100644 --- a/README.md +++ b/README.md @@ -185,6 +185,82 @@ $usage->addBatch([ ], Usage::TYPE_GAUGE); ``` +### Canonical Samples + +Billable inputs that must survive request retries use the separate immutable +sample ledger. A sample's identity is derived from its environment, region, +project and database internal IDs, member, generation, sequence and metric. +Retrying the same identity and payload is safe. Reusing an identity with a +different interval, value or event version is returned as a conflict. +The event ID is SHA-256 over those identity fields in the documented order, +each encoded as its decimal byte length, `:`, then its UTF-8 value. The payload +hash uses the same encoding over event ID, UTC millisecond interval bounds, +value and event version. + +```php +use Utopia\Usage\Sample; +use Utopia\Usage\SampleRange; + +$sample = new Sample( + environment: 'production', + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: '01J...', + sequence: 42, + metric: 'bandwidth.inbound', + intervalStart: new DateTimeImmutable('2026-08-01T00:42:00Z'), + intervalEnd: new DateTimeImmutable('2026-08-01T00:43:00Z'), + value: 4096, + eventVersion: 1, +); + +$range = new SampleRange( + environment: 'production', + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: '01J...', + metric: 'bandwidth.inbound', + firstSequence: 42, + lastSequence: 42, + intervalStart: new DateTimeImmutable('2026-08-01T00:42:00Z'), + intervalEnd: new DateTimeImmutable('2026-08-01T00:43:00Z'), +); + +$usage->addSamples([$sample]); +$watermark = $usage->getSampleWatermark($range, limit: 100); +$result = $usage->findSamples( + $range, + $watermark, + limit: 100, +); + +if (!$result->isComplete()) { + // Refuse billing. Inspect conflicts, compact gap ranges and truncation. +} +``` + +Each supplied sample row receives an adapter-owned random ingestion ID before +its request is sent. `getSampleWatermark()` performs one bounded ClickHouse +snapshot read and captures each visible ingestion ID bound to its canonical ID +and payload hash for that stream and range. `findSamples()` admits only those +exact entry fingerprints, so later inserts or changed rows cannot cross the +boundary even when their server timestamps would be identical. A transport +retry of the same request retains its fingerprint and is counted once; a new +logical retry gets a new ingestion ID and is included only when visible to the +watermark query. + +Both the watermark evidence and `findSamples()` result are explicitly bounded. +A result is complete only when neither bound is truncated and there are no +conflicts, sequence gaps or interval-boundary discontinuities. Conflicting +physical rows are never combined into a synthetic sample. The sample ledger +does not make HTTP delivery or a producer's local spool durable; callers must +retain a sample until the write is acknowledged and retry the identical +payload. + ## Querying Metrics ### Find with Query Objects @@ -296,6 +372,7 @@ $usage->purge('project_123', [], Usage::TYPE_GAUGE); | `{ns}_usage_gauges` | MergeTree | Resource snapshot gauges | | `{ns}_usage_events_daily` | SummingMergeTree | Pre-aggregated daily event totals | | `{ns}_usage_events_daily_mv` | Materialized View | Auto-populates daily table on insert | +| `{ns}_usage_samples` | MergeTree | Immutable canonical samples with retry/conflict evidence | ### Events Table Schema @@ -378,7 +455,7 @@ coroutines. ## System Requirements -Utopia Framework requires PHP 8.0 or later. We recommend using the latest PHP version whenever possible. +Utopia Framework requires PHP 8.4 or later. We recommend using the latest PHP version whenever possible. ## Copyright and license diff --git a/src/Usage/Adapter.php b/src/Usage/Adapter.php index 9fd6fea..b3473e1 100644 --- a/src/Usage/Adapter.php +++ b/src/Usage/Adapter.php @@ -37,6 +37,33 @@ abstract public function setup(): void; */ abstract public function addBatch(array $metrics, string $type, int $batchSize = 1000): bool; + /** + * Add immutable, canonically identified usage samples. + * + * Adapters that do not provide canonical sample storage leave this + * unsupported. Existing telemetry APIs are unaffected. + * + * @param list $samples + */ + public function addSamples(array $samples, int $batchSize = 1000): bool + { + throw new \Exception($this->getName() . ' does not support canonical samples'); + } + + /** + * Capture at most $limit logical ingestion IDs from one exact range. + * Truncation is carried by the returned watermark and fails completeness. + */ + public function getSampleWatermark(SampleRange $range, int $limit): SampleWatermark + { + throw new \Exception($this->getName() . ' does not support canonical samples'); + } + + public function findSamples(SampleRange $range, SampleWatermark $watermark, int $limit): SampleResult + { + throw new \Exception($this->getName() . ' does not support canonical samples'); + } + /** * Get time series data for metrics with query-time aggregation. * diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index c905006..59316c6 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -4,6 +4,7 @@ use ArrayObject; use DateTime; +use DateTimeImmutable; use DateTimeZone; use Exception; use Psr\Http\Client\ClientInterface; @@ -19,6 +20,11 @@ use Utopia\Query\Schema\ClickHouse\Engine; use Utopia\Query\Schema\Table\ClickHouse as ClickHouseTable; use Utopia\Usage\Metric; +use Utopia\Usage\Sample; +use Utopia\Usage\SampleGap; +use Utopia\Usage\SampleRange; +use Utopia\Usage\SampleResult; +use Utopia\Usage\SampleWatermark; use Utopia\Usage\Usage; use Utopia\Usage\UsageQuery; use Utopia\Validator\Hostname; @@ -55,6 +61,8 @@ class ClickHouse extends SQL private const INSERT_BATCH_SIZE = 1_000; + private const int SAMPLE_BATCH_SIZE = 1_000; + private const ROUTE_LOG_MAX = 1_000; /** @var array Maps interval strings to ClickHouse time functions */ @@ -433,6 +441,11 @@ private function getEventsDailyTableName(): string return $this->getTableName() . '_events_daily'; } + private function getSamplesTableName(): string + { + return $this->getTableName() . '_samples'; + } + /** * Get the appropriate table name for a given type. * @@ -616,7 +629,7 @@ private function buildHeaders(): array * @param array $data Array of JSON strings (one per row) * @throws Exception */ - private function insert(string $table, string $sql, array $data): void + private function insert(string $table, string $sql, array $data, bool $durable = false): void { if (empty($data)) { return; @@ -632,7 +645,7 @@ private function insert(string $table, string $sql, array $data): void $queryParams = ['query' => $sql]; if ($this->asyncInserts) { $queryParams['async_insert'] = '1'; - $queryParams['wait_for_async_insert'] = $this->asyncInsertWait ? '1' : '0'; + $queryParams['wait_for_async_insert'] = ($durable || $this->asyncInsertWait) ? '1' : '0'; } $url = "{$scheme}://{$this->host}:{$this->port}/?" . http_build_query($queryParams); @@ -920,7 +933,6 @@ public function setup(): void $createDbSql = "CREATE DATABASE IF NOT EXISTS {$escapedDatabase}"; $this->query($createDbSql); - // --- Events table --- $this->createTable( $this->getEventsTableName(), 'event', @@ -931,15 +943,12 @@ public function setup(): void $this->applyRetention($this->getEventsTableName()); - // --- Events daily table (SummingMergeTree) --- $this->createDailyTable(); $this->applyRetention($this->getEventsDailyTableName()); - // --- Events daily materialized view --- $this->createDailyMaterializedView(); - // --- Gauges table --- $this->createTable( $this->getGaugesTableName(), 'gauge', @@ -948,7 +957,9 @@ public function setup(): void $this->ensureGaugeDimColumns(); - // --- Per-dim projections on the events / gauges base tables --- + $this->createSamplesTable(); + $this->ensureSampleColumns(); + $this->setLightweightMutationProjectionMode($this->getEventsTableName()); foreach (self::EVENT_PROJECTIONS as $projection) { $this->addProjection( @@ -969,6 +980,64 @@ public function setup(): void } } + /** + * Create the immutable canonical-sample ledger. Retries remain as raw + * physical rows; findSamples() groups by the canonical identity and + * exposes conflicting payloads instead of allowing them to affect totals. + */ + private function createSamplesTable(): void + { + $tableName = $this->getSamplesTableName(); + $table = $this->newSchema()->table($tableName); + + $table->rawColumn('`id` String CODEC(ZSTD(3))'); + $table->rawColumn('`payloadHash` String CODEC(ZSTD(3))'); + $table->rawColumn('`ingestId` String CODEC(ZSTD(3))'); + $table->rawColumn('`environment` LowCardinality(String)'); + $table->rawColumn('`region` LowCardinality(String)'); + $table->rawColumn('`projectInternalId` String CODEC(ZSTD(3))'); + $table->rawColumn('`databaseInternalId` String CODEC(ZSTD(3))'); + $table->rawColumn('`member` String CODEC(ZSTD(3))'); + $table->rawColumn('`generation` String CODEC(ZSTD(3))'); + $table->rawColumn('`sequence` UInt64'); + $table->rawColumn('`metric` LowCardinality(String)'); + $table->rawColumn("`intervalStart` DateTime64(3, 'UTC') CODEC(Delta(4), LZ4)"); + $table->rawColumn("`intervalEnd` DateTime64(3, 'UTC') CODEC(Delta(4), LZ4)"); + $table->rawColumn('`value` Int64'); + $table->rawColumn('`eventVersion` UInt32'); + + $table->engine(Engine::MergeTree) + ->orderBy([ + 'environment', + 'region', + 'projectInternalId', + 'databaseInternalId', + 'member', + 'generation', + 'metric', + 'sequence', + 'id', + 'payloadHash', + 'ingestId', + ]) + ->partitionBy('toYYYYMM(intervalStart)') + ->settings(['index_granularity' => 8192]); + + $statement = $table->createIfNotExists(); + + $this->query($this->qualifyDdl($statement->query, $tableName)); + } + + /** + * Older pre-release tables have no exact snapshot identifier. Keep their + * default empty so watermark reads omit them and fail closed with gaps. + */ + private function ensureSampleColumns(): void + { + $table = $this->buildTableReference($this->getSamplesTableName()); + $this->query("ALTER TABLE {$table} ADD COLUMN IF NOT EXISTS `ingestId` String DEFAULT '' CODEC(ZSTD(3))"); + } + /** * Apply (or strip) the retention TTL on a table as a separate idempotent * ALTER. CREATE TABLE IF NOT EXISTS won't add a TTL to an existing table, @@ -1679,6 +1748,327 @@ public function addBatch(array $metrics, string $type, int $batchSize = self::IN return true; } + /** + * @param list $samples + */ + #[\Override] + public function addSamples(array $samples, int $batchSize = self::SAMPLE_BATCH_SIZE): bool + { + if ($samples === []) { + return true; + } + + $this->setOperationContext('addSamples()'); + + $batchSize = min(self::SAMPLE_BATCH_SIZE, max(1, $batchSize)); + $tableName = $this->getSamplesTableName(); + $columns = [ + 'id', + 'payloadHash', + 'ingestId', + 'environment', + 'region', + 'projectInternalId', + 'databaseInternalId', + 'member', + 'generation', + 'sequence', + 'metric', + 'intervalStart', + 'intervalEnd', + 'value', + 'eventVersion', + ]; + $escapedColumns = implode(', ', array_map($this->escapeIdentifier(...), $columns)); + $insertSql = 'INSERT INTO ' . $this->buildTableReference($tableName) + . " ({$escapedColumns}) FORMAT JSONEachRow"; + + foreach (array_chunk($samples, $batchSize) as $batch) { + $rows = []; + + foreach ($batch as $sample) { + // The ID belongs to this logical row, not its HTTP attempt. + // A transport retry repeats the encoded body and therefore + // keeps the same ID; a later addSamples() call receives a new + // one and cannot cross an already captured watermark. + $rows[] = json_encode([ + 'id' => $sample->getId(), + 'payloadHash' => $sample->getPayloadHash(), + 'ingestId' => bin2hex(random_bytes(16)), + 'environment' => $sample->environment, + 'region' => $sample->region, + 'projectInternalId' => $sample->projectInternalId, + 'databaseInternalId' => $sample->databaseInternalId, + 'member' => $sample->member, + 'generation' => $sample->generation, + 'sequence' => $sample->sequence, + 'metric' => $sample->metric, + 'intervalStart' => $sample->getFormattedIntervalStart(), + 'intervalEnd' => $sample->getFormattedIntervalEnd(), + 'value' => $sample->value, + 'eventVersion' => $sample->eventVersion, + ], JSON_THROW_ON_ERROR); + } + + $this->insert($tableName, $insertSql, $rows, durable: true); + } + + return true; + } + + #[\Override] + public function getSampleWatermark(SampleRange $range, int $limit): SampleWatermark + { + if ($limit < 1 || $limit === PHP_INT_MAX) { + throw new \InvalidArgumentException('Sample watermark limit must be positive and leave room for truncation detection'); + } + + $this->setOperationContext('getSampleWatermark()'); + + $table = $this->buildTableReference($this->getSamplesTableName()); + $sql = <<= {firstSequence:UInt64} + AND sequence <= {lastSequence:UInt64} + AND ingestId != '' + LIMIT 1 BY entryId + LIMIT {queryLimit:UInt64} + FORMAT JSON + SQL; + + $rows = $this->decodeRows($this->query($sql, [ + 'environment' => $range->environment, + 'region' => $range->region, + 'projectInternalId' => $range->projectInternalId, + 'databaseInternalId' => $range->databaseInternalId, + 'member' => $range->member, + 'generation' => $range->generation, + 'metric' => $range->metric, + 'firstSequence' => $range->firstSequence, + 'lastSequence' => $range->lastSequence, + 'queryLimit' => $limit + 1, + ])); + + $truncated = count($rows) > $limit; + if ($truncated) { + $rows = array_slice($rows, 0, $limit); + } + + $entries = []; + foreach ($rows as $row) { + $entries[] = self::toStr($row['entryId'] ?? null); + } + + return new SampleWatermark($range, $entries, $truncated); + } + + #[\Override] + public function findSamples(SampleRange $range, SampleWatermark $watermark, int $limit): SampleResult + { + if ($limit < 1 || $limit === PHP_INT_MAX) { + throw new \InvalidArgumentException('Sample limit must be positive and leave room for truncation detection'); + } + + if (!$watermark->matches($range)) { + throw new \InvalidArgumentException('Sample watermark does not match the requested range'); + } + + $this->setOperationContext('findSamples()'); + + $table = $this->buildTableReference($this->getSamplesTableName()); + $sql = <<= {firstSequence:UInt64} + AND sequence <= {lastSequence:UInt64} + AND has({entries:Array(String)}, concat(ingestId, ':', id, ':', payloadHash)) + GROUP BY + environment, + region, + projectInternalId, + databaseInternalId, + member, + generation, + sequence, + metric + ORDER BY sequence ASC + LIMIT {queryLimit:UInt64} + FORMAT JSON + SQL; + + $rows = $this->decodeRows($this->query($sql, [ + 'environment' => $range->environment, + 'region' => $range->region, + 'projectInternalId' => $range->projectInternalId, + 'databaseInternalId' => $range->databaseInternalId, + 'member' => $range->member, + 'generation' => $range->generation, + 'metric' => $range->metric, + 'firstSequence' => $range->firstSequence, + 'lastSequence' => $range->lastSequence, + 'entries' => $watermark->getEntries() === [] + ? '[]' + : "['" . implode("','", $watermark->getEntries()) . "']", + 'queryLimit' => $limit + 1, + ])); + + $truncated = count($rows) > $limit; + if ($truncated) { + $rows = array_slice($rows, 0, $limit); + } + + $samples = []; + $conflicts = []; + $duplicates = 0; + + foreach ($rows as $row) { + $sequence = self::toInt($row['sequence'] ?? null); + $copies = self::toInt($row['copies'] ?? null); + $variants = self::toInt($row['variants'] ?? null); + $duplicates += max(0, $copies - max(1, $variants)); + + if ($variants !== 1) { + $conflicts[] = $sequence; + continue; + } + + $observation = $row['observation'] ?? null; + if (!is_array($observation) || count($observation) !== 5) { + $conflicts[] = $sequence; + continue; + } + + try { + $sample = new Sample( + environment: self::toStr($row['environment'] ?? null), + region: self::toStr($row['region'] ?? null), + projectInternalId: self::toStr($row['projectInternalId'] ?? null), + databaseInternalId: self::toStr($row['databaseInternalId'] ?? null), + member: self::toStr($row['member'] ?? null), + generation: self::toStr($row['generation'] ?? null), + sequence: $sequence, + metric: self::toStr($row['metric'] ?? null), + intervalStart: new DateTimeImmutable(self::toStr($observation[0] ?? null), new DateTimeZone('UTC')), + intervalEnd: new DateTimeImmutable(self::toStr($observation[1] ?? null), new DateTimeZone('UTC')), + value: self::toInt($observation[2] ?? null), + eventVersion: self::toInt($observation[3] ?? null), + ); + } catch (\InvalidArgumentException) { + $conflicts[] = $sequence; + continue; + } + + if ( + $sample->getPayloadHash() !== self::toStr($observation[4] ?? null) + || $sample->intervalStart < $range->intervalStart + || $sample->intervalEnd > $range->intervalEnd + ) { + $conflicts[] = $sequence; + continue; + } + + $samples[] = $sample; + } + + $gaps = $this->findSampleGaps($samples, $range->firstSequence, $range->lastSequence); + $discontinuities = $this->findSampleDiscontinuities($samples, $range); + + return new SampleResult( + samples: $samples, + conflicts: array_values(array_unique($conflicts)), + gaps: $gaps, + discontinuities: $discontinuities, + duplicateCount: $duplicates, + truncated: $truncated, + watermark: $watermark, + ); + } + + /** + * @param list $samples + * @return list + */ + private function findSampleGaps(array $samples, int $first, int $last): array + { + $gaps = []; + $expected = $first; + + foreach ($samples as $sample) { + if ($sample->sequence > $expected) { + $gaps[] = new SampleGap($expected, $sample->sequence - 1); + } + + $expected = $sample->sequence + 1; + } + + if ($expected <= $last) { + $gaps[] = new SampleGap($expected, $last); + } + + return $gaps; + } + + /** + * @param list $samples + * @return list + */ + private function findSampleDiscontinuities(array $samples, SampleRange $range): array + { + $discontinuities = []; + $expectedStart = $range->intervalStart; + + foreach ($samples as $sample) { + if ($sample->intervalStart != $expectedStart) { + $discontinuities[] = $sample->sequence; + } + + $expectedStart = $sample->intervalEnd; + } + + if ($samples !== [] && $expectedStart != $range->intervalEnd) { + $last = $samples[array_key_last($samples)]; + $discontinuities[] = $last->sequence; + } + + return array_values(array_unique($discontinuities)); + } + /** * Columns declared in the INSERT envelope for the given type. Matches * the row shape produced by addBatch(): base columns, the type's diff --git a/src/Usage/Sample.php b/src/Usage/Sample.php new file mode 100644 index 0000000..c44a8ac --- /dev/null +++ b/src/Usage/Sample.php @@ -0,0 +1,120 @@ + $environment, + 'region' => $region, + 'projectInternalId' => $projectInternalId, + 'databaseInternalId' => $databaseInternalId, + 'member' => $member, + 'generation' => $generation, + 'metric' => $metric, + ] as $field => $value) { + if ($value === '') { + throw new InvalidArgumentException("{$field} cannot be empty"); + } + } + + if ($sequence < 0) { + throw new InvalidArgumentException('sequence cannot be negative'); + } + + if ($eventVersion < 1 || $eventVersion > 4_294_967_295) { + throw new InvalidArgumentException('eventVersion must fit an unsigned 32-bit integer'); + } + + if ($intervalStart >= $intervalEnd) { + throw new InvalidArgumentException('intervalStart must be before intervalEnd'); + } + } + + /** + * Canonical stream identity. A retry of one observation must retain this + * ID even if a faulty producer changes its payload, so readers can expose + * the conflict rather than counting both values. + * + */ + public function getId(): string + { + return $this->hashParts([ + $this->environment, + $this->region, + $this->projectInternalId, + $this->databaseInternalId, + $this->member, + $this->generation, + $this->sequence, + $this->metric, + ]); + } + + /** + * Hash every money-bearing field. Equal IDs with different payload hashes + * are conflicting observations and make the stream incomplete. + * + */ + public function getPayloadHash(): string + { + return $this->hashParts([ + $this->getId(), + $this->formatDateTime($this->intervalStart), + $this->formatDateTime($this->intervalEnd), + $this->value, + $this->eventVersion, + ]); + } + + public function getFormattedIntervalStart(): string + { + return $this->formatDateTime($this->intervalStart); + } + + public function getFormattedIntervalEnd(): string + { + return $this->formatDateTime($this->intervalEnd); + } + + private function formatDateTime(DateTimeImmutable $time): string + { + return $time->setTimezone(new DateTimeZone('UTC'))->format('Y-m-d H:i:s.v'); + } + + /** + * Length-prefixing makes the digest unambiguous and reproducible by + * producers in other languages without depending on JSON encoding rules. + * + * @param list $parts + */ + private function hashParts(array $parts): string + { + $encoded = ''; + + foreach ($parts as $part) { + $value = (string) $part; + $encoded .= strlen($value) . ':' . $value; + } + + return hash('sha256', $encoded); + } +} diff --git a/src/Usage/SampleGap.php b/src/Usage/SampleGap.php new file mode 100644 index 0000000..95f5cc4 --- /dev/null +++ b/src/Usage/SampleGap.php @@ -0,0 +1,17 @@ + $environment, + 'region' => $region, + 'projectInternalId' => $projectInternalId, + 'databaseInternalId' => $databaseInternalId, + 'member' => $member, + 'generation' => $generation, + 'metric' => $metric, + ] as $field => $value) { + if ($value === '') { + throw new InvalidArgumentException("{$field} cannot be empty"); + } + } + + if ($firstSequence < 0 || $lastSequence < $firstSequence) { + throw new InvalidArgumentException('Invalid sample sequence range'); + } + + if ($intervalStart >= $intervalEnd) { + throw new InvalidArgumentException('intervalStart must be before intervalEnd'); + } + } + + public function getFormattedIntervalStart(): string + { + return $this->formatDateTime($this->intervalStart); + } + + public function getFormattedIntervalEnd(): string + { + return $this->formatDateTime($this->intervalEnd); + } + + private function formatDateTime(DateTimeImmutable $time): string + { + return $time->setTimezone(new DateTimeZone('UTC'))->format('Y-m-d H:i:s.v'); + } +} diff --git a/src/Usage/SampleResult.php b/src/Usage/SampleResult.php new file mode 100644 index 0000000..2a4934f --- /dev/null +++ b/src/Usage/SampleResult.php @@ -0,0 +1,71 @@ + $samples + * @param list $conflicts + * @param list $gaps + * @param list $discontinuities + */ + public function __construct( + private array $samples, + private array $conflicts, + private array $gaps, + private array $discontinuities, + private int $duplicateCount, + private bool $truncated, + private SampleWatermark $watermark, + ) { + } + + /** @return list */ + public function getSamples(): array + { + return $this->samples; + } + + /** @return list */ + public function getConflicts(): array + { + return $this->conflicts; + } + + /** @return list */ + public function getGaps(): array + { + return $this->gaps; + } + + /** @return list */ + public function getDiscontinuities(): array + { + return $this->discontinuities; + } + + public function getDuplicateCount(): int + { + return $this->duplicateCount; + } + + public function isTruncated(): bool + { + return $this->truncated; + } + + public function getWatermark(): SampleWatermark + { + return $this->watermark; + } + + public function isComplete(): bool + { + return !$this->truncated + && !$this->watermark->isTruncated() + && $this->conflicts === [] + && $this->gaps === [] + && $this->discontinuities === []; + } +} diff --git a/src/Usage/SampleWatermark.php b/src/Usage/SampleWatermark.php new file mode 100644 index 0000000..33f824f --- /dev/null +++ b/src/Usage/SampleWatermark.php @@ -0,0 +1,60 @@ + $entries + */ + public function __construct( + private SampleRange $range, + private array $entries, + private bool $truncated, + ) { + if ($entries !== array_values(array_unique($entries))) { + throw new InvalidArgumentException('entries must be a unique list'); + } + + foreach ($entries as $entry) { + if (preg_match('/^[a-f0-9]{32}:[a-f0-9]{64}:[a-f0-9]{64}$/', $entry) !== 1) { + throw new InvalidArgumentException('entries must bind an ingestion ID, canonical ID and payload hash'); + } + } + } + + public function matches(SampleRange $range): bool + { + return $this->range->environment === $range->environment + && $this->range->region === $range->region + && $this->range->projectInternalId === $range->projectInternalId + && $this->range->databaseInternalId === $range->databaseInternalId + && $this->range->member === $range->member + && $this->range->generation === $range->generation + && $this->range->metric === $range->metric + && $this->range->firstSequence === $range->firstSequence + && $this->range->lastSequence === $range->lastSequence + && $this->range->getFormattedIntervalStart() === $range->getFormattedIntervalStart() + && $this->range->getFormattedIntervalEnd() === $range->getFormattedIntervalEnd(); + } + + /** @return list */ + public function getEntries(): array + { + return $this->entries; + } + + public function isTruncated(): bool + { + return $this->truncated; + } +} diff --git a/src/Usage/Usage.php b/src/Usage/Usage.php index c70a96e..9572925 100644 --- a/src/Usage/Usage.php +++ b/src/Usage/Usage.php @@ -77,6 +77,24 @@ public function addBatch(array $metrics, string $type, int $batchSize = 1000): b return $this->adapter->addBatch($metrics, $type, $batchSize); } + /** + * @param list $samples + */ + public function addSamples(array $samples, int $batchSize = 1000): bool + { + return $this->adapter->addSamples($samples, $batchSize); + } + + public function getSampleWatermark(SampleRange $range, int $limit): SampleWatermark + { + return $this->adapter->getSampleWatermark($range, $limit); + } + + public function findSamples(SampleRange $range, SampleWatermark $watermark, int $limit): SampleResult + { + return $this->adapter->findSamples($range, $watermark, $limit); + } + /** * Get time series data for metrics. * diff --git a/tests/Usage/Adapter/ClickHouseSampleTest.php b/tests/Usage/Adapter/ClickHouseSampleTest.php new file mode 100644 index 0000000..667b213 --- /dev/null +++ b/tests/Usage/Adapter/ClickHouseSampleTest.php @@ -0,0 +1,390 @@ +usage = new Usage($adapter); + $this->usage->setup(); + } + + public function testSampleTableHasCanonicalIdentityAndWatermarkColumns(): void + { + $adapter = $this->usage->getAdapter(); + $this->assertInstanceOf(ClickHouseAdapter::class, $adapter); + + $table = $this->resolveTableName($adapter, 'getSamplesTableName'); + $database = $this->databaseName($adapter); + $ddl = $this->queryRaw($adapter, "SHOW CREATE TABLE `{$database}`.`{$table}` FORMAT TabSeparatedRaw"); + + foreach ([ + '`ingestId` String', + '`environment` LowCardinality(String)', + '`region` LowCardinality(String)', + '`projectInternalId` String', + '`databaseInternalId` String', + '`member` String', + '`generation` String', + '`sequence` UInt64', + '`metric` LowCardinality(String)', + "`intervalStart` DateTime64(3, 'UTC')", + "`intervalEnd` DateTime64(3, 'UTC')", + '`value` Int64', + '`eventVersion` UInt32', + ] as $column) { + $this->assertStringContainsString($column, $ddl); + } + } + + public function testCanonicalizesConcurrentDuplicatesAndCrashRetry(): void + { + $key = bin2hex(random_bytes(8)); + $sample = $this->sample($key, sequence: 0); + + $this->assertTrue($this->usage->addSamples([$sample, $sample])); + $this->assertTrue($this->usage->addSamples([$sample])); + + $range = $this->range($key, firstSequence: 0, lastSequence: 0); + $result = $this->usage->findSamples( + $range, + $this->usage->getSampleWatermark($range, 10), + 10, + ); + + $this->assertTrue($result->isComplete()); + $this->assertCount(1, $result->getSamples()); + $this->assertSame(2, $result->getDuplicateCount()); + $this->assertSame([], $result->getConflicts()); + $this->assertSame([], $result->getGaps()); + $this->assertSame([], $result->getDiscontinuities()); + } + + public function testCanonicalSamplesWaitForAsyncInsertDurability(): void + { + $key = bin2hex(random_bytes(8)); + $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_samples_async', + database: getenv('CLICKHOUSE_DATABASE') ?: 'default', + sharedTables: true, + asyncInserts: true, + asyncInsertWait: false, + ); + $usage = new Usage($adapter); + $usage->setup(); + + $this->assertTrue($usage->addSamples([$this->sample($key, sequence: 0)])); + + $range = $this->range($key, firstSequence: 0, lastSequence: 0); + $result = $usage->findSamples( + $range, + $usage->getSampleWatermark($range, 10), + 10, + ); + + $this->assertTrue($result->isComplete()); + $this->assertCount(1, $result->getSamples()); + } + + public function testConflictingDuplicateFailsCompleteness(): void + { + $key = bin2hex(random_bytes(8)); + + $this->assertTrue($this->usage->addSamples([ + $this->sample($key, sequence: 0, value: 10), + $this->sample($key, sequence: 0, value: 11), + ])); + + $range = $this->range($key, firstSequence: 0, lastSequence: 0); + $result = $this->usage->findSamples( + $range, + $this->usage->getSampleWatermark($range, 10), + 10, + ); + + $this->assertFalse($result->isComplete()); + $this->assertSame([0], $result->getConflicts()); + $this->assertSame([], $result->getSamples()); + } + + public function testConflictingIntervalsReturnEvidenceWithoutSyntheticSample(): void + { + $key = bin2hex(random_bytes(8)); + + $this->assertTrue($this->usage->addSamples([ + $this->sample($key, sequence: 0, value: 10, startMinute: 0), + $this->sample($key, sequence: 0, value: 11, startMinute: 2), + ])); + + $range = $this->range($key, firstSequence: 0, lastSequence: 0, endMinute: 3); + $result = $this->usage->findSamples( + $range, + $this->usage->getSampleWatermark($range, 10), + 10, + ); + + $this->assertFalse($result->isComplete()); + $this->assertSame([0], $result->getConflicts()); + $this->assertSame([], $result->getSamples()); + } + + public function testDetectsGapsWithoutExpandingEveryMissingSequence(): void + { + $key = bin2hex(random_bytes(8)); + + $this->assertTrue($this->usage->addSamples([ + $this->sample($key, sequence: 0), + $this->sample($key, sequence: 4), + ])); + + $range = $this->range($key, firstSequence: 0, lastSequence: 4); + $result = $this->usage->findSamples( + $range, + $this->usage->getSampleWatermark($range, 10), + 10, + ); + + $this->assertFalse($result->isComplete()); + $this->assertCount(1, $result->getGaps()); + $this->assertSame(1, $result->getGaps()[0]->first); + $this->assertSame(3, $result->getGaps()[0]->last); + } + + public function testDetectsIntervalDiscontinuityWithContiguousSequences(): void + { + $key = bin2hex(random_bytes(8)); + + $this->assertTrue($this->usage->addSamples([ + $this->sample($key, sequence: 0), + $this->sample($key, sequence: 1, startMinute: 2), + ])); + + $range = $this->range($key, firstSequence: 0, lastSequence: 1, endMinute: 3); + $result = $this->usage->findSamples( + $range, + $this->usage->getSampleWatermark($range, 10), + 10, + ); + + $this->assertFalse($result->isComplete()); + $this->assertSame([], $result->getGaps()); + $this->assertSame([1], $result->getDiscontinuities()); + } + + public function testReportsTruncationAndHonorsAnExactWatermark(): void + { + $key = bin2hex(random_bytes(8)); + + $this->assertTrue($this->usage->addSamples([ + $this->sample($key, sequence: 0), + $this->sample($key, sequence: 1), + $this->sample($key, sequence: 2), + ])); + + $range = $this->range($key, firstSequence: 0, lastSequence: 3); + $watermark = $this->usage->getSampleWatermark($range, 10); + $this->assertTrue($this->usage->addSamples([$this->sample($key, sequence: 3)])); + + $bounded = $this->usage->findSamples( + $range, + $watermark, + 2, + ); + $watermarked = $this->usage->findSamples( + $range, + $watermark, + 10, + ); + + $this->assertTrue($bounded->isTruncated()); + $this->assertCount(2, $bounded->getSamples()); + $this->assertFalse($bounded->isComplete()); + + $this->assertFalse($watermarked->isTruncated()); + $this->assertCount(3, $watermarked->getSamples()); + $this->assertSame(3, $watermarked->getGaps()[0]->first); + $this->assertSame(3, $watermarked->getGaps()[0]->last); + } + + public function testWatermarkEvidenceLimitFailsClosed(): void + { + $key = bin2hex(random_bytes(8)); + $sample = $this->sample($key, sequence: 0); + + $this->assertTrue($this->usage->addSamples([$sample, $sample, $sample])); + + $range = $this->range($key, firstSequence: 0, lastSequence: 0); + $watermark = $this->usage->getSampleWatermark($range, 2); + $result = $this->usage->findSamples($range, $watermark, 10); + + $this->assertTrue($watermark->isTruncated()); + $this->assertFalse($result->isComplete()); + $this->assertTrue($result->getWatermark()->isTruncated()); + } + + public function testTransportRetryAfterWatermarkCannotChangeSnapshot(): void + { + $key = bin2hex(random_bytes(8)); + $sample = $this->sample($key, sequence: 0); + $range = $this->range($key, firstSequence: 0, lastSequence: 0); + + $this->assertTrue($this->usage->addSamples([$sample])); + $watermark = $this->usage->getSampleWatermark($range, 10); + $this->assertCount(1, $watermark->getEntries()); + + $this->insertRawSample($sample, $this->ingestId($watermark->getEntries()[0])); + + $result = $this->usage->findSamples($range, $watermark, 10); + + $this->assertTrue($result->isComplete()); + $this->assertCount(1, $result->getSamples()); + $this->assertSame(0, $result->getDuplicateCount()); + } + + public function testReusedIngestIdWithDifferentPayloadCannotChangeSnapshot(): void + { + $key = bin2hex(random_bytes(8)); + $sample = $this->sample($key, sequence: 0, value: 10); + $range = $this->range($key, firstSequence: 0, lastSequence: 0); + + $this->assertTrue($this->usage->addSamples([$sample])); + $watermark = $this->usage->getSampleWatermark($range, 10); + $this->assertCount(1, $watermark->getEntries()); + + $this->insertRawSample( + $this->sample($key, sequence: 0, value: 11), + $this->ingestId($watermark->getEntries()[0]), + ); + + $result = $this->usage->findSamples($range, $watermark, 10); + + $this->assertTrue($result->isComplete()); + $this->assertSame([], $result->getConflicts()); + $this->assertCount(1, $result->getSamples()); + $this->assertSame(10, $result->getSamples()[0]->value); + } + + public function testRejectsWatermarkFromAnotherRange(): void + { + $key = bin2hex(random_bytes(8)); + $range = $this->range($key, firstSequence: 0, lastSequence: 0); + $watermark = $this->usage->getSampleWatermark($range, 10); + + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage('Sample watermark does not match the requested range'); + + $this->usage->findSamples( + $this->range($key, firstSequence: 0, lastSequence: 1), + $watermark, + 10, + ); + } + + private function insertRawSample(Sample $sample, string $ingestId): void + { + $adapter = $this->usage->getAdapter(); + $this->assertInstanceOf(ClickHouseAdapter::class, $adapter); + $database = $this->databaseName($adapter); + $table = $this->resolveTableName($adapter, 'getSamplesTableName'); + $sql = <<queryRaw($adapter, $sql, [ + 'id' => $sample->getId(), + 'payloadHash' => $sample->getPayloadHash(), + 'ingestId' => $ingestId, + 'environment' => $sample->environment, + 'region' => $sample->region, + 'projectInternalId' => $sample->projectInternalId, + 'databaseInternalId' => $sample->databaseInternalId, + 'member' => $sample->member, + 'generation' => $sample->generation, + 'sequence' => $sample->sequence, + 'metric' => $sample->metric, + 'intervalStart' => $sample->getFormattedIntervalStart(), + 'intervalEnd' => $sample->getFormattedIntervalEnd(), + 'value' => $sample->value, + 'eventVersion' => $sample->eventVersion, + ]); + } + + private function ingestId(string $entry): string + { + return explode(':', $entry, 2)[0]; + } + + private function sample(string $key, int $sequence, int $value = 10, ?int $startMinute = null): Sample + { + $start = new DateTimeImmutable('2026-08-01T00:00:00Z'); + $startMinute ??= $sequence; + + return new Sample( + environment: 'test-' . $key, + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: 'generation-1', + sequence: $sequence, + metric: 'bandwidth.inbound', + intervalStart: $start->modify("+{$startMinute} minutes"), + intervalEnd: $start->modify('+' . ($startMinute + 1) . ' minutes'), + value: $value, + eventVersion: 1, + ); + } + + private function range(string $key, int $firstSequence, int $lastSequence, ?int $endMinute = null): SampleRange + { + $endMinute ??= $lastSequence + 1; + + return new SampleRange( + environment: 'test-' . $key, + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: 'generation-1', + metric: 'bandwidth.inbound', + firstSequence: $firstSequence, + lastSequence: $lastSequence, + intervalStart: new DateTimeImmutable('2026-08-01T00:00:00Z'), + intervalEnd: new DateTimeImmutable("2026-08-01T00:{$endMinute}:00Z"), + ); + } +} diff --git a/tests/Usage/SampleTest.php b/tests/Usage/SampleTest.php new file mode 100644 index 0000000..73e81ee --- /dev/null +++ b/tests/Usage/SampleTest.php @@ -0,0 +1,71 @@ +sample(value: 42); + $retry = $this->sample(value: 42); + $conflict = $this->sample(value: 43); + + $this->assertSame($sample->getId(), $retry->getId()); + $this->assertSame($sample->getPayloadHash(), $retry->getPayloadHash()); + $this->assertSame($sample->getId(), $conflict->getId()); + $this->assertNotSame($sample->getPayloadHash(), $conflict->getPayloadHash()); + $this->assertSame('a8e0eebf28f6eb0e2f632fd59b40734624d5c46a83edd2ba0530b0d83fbf3249', $sample->getId()); + $this->assertSame('dadbc87dbbe5fecba1e96f6c9c608c4ca09447566745c8e3e35bf06f76766568', $sample->getPayloadHash()); + } + + public function testEventVersionChangesPayloadButNotIdentity(): void + { + $first = $this->sample(eventVersion: 1); + $second = $this->sample(eventVersion: 2); + + $this->assertSame($first->getId(), $second->getId()); + $this->assertNotSame($first->getPayloadHash(), $second->getPayloadHash()); + } + + public function testRejectsAnInvalidInterval(): void + { + $this->expectException(\InvalidArgumentException::class); + + new Sample( + environment: 'production', + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: 'generation-1', + sequence: 7, + metric: 'bandwidth.inbound', + intervalStart: new DateTimeImmutable('2026-08-01T00:01:00Z'), + intervalEnd: new DateTimeImmutable('2026-08-01T00:00:00Z'), + value: 42, + eventVersion: 1, + ); + } + + private function sample(int $value = 42, int $eventVersion = 1): Sample + { + return new Sample( + environment: 'production', + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: 'generation-1', + sequence: 7, + metric: 'bandwidth.inbound', + intervalStart: new DateTimeImmutable('2026-08-01T00:00:00Z'), + intervalEnd: new DateTimeImmutable('2026-08-01T00:01:00Z'), + value: $value, + eventVersion: $eventVersion, + ); + } +} diff --git a/tests/Usage/SampleWatermarkTest.php b/tests/Usage/SampleWatermarkTest.php new file mode 100644 index 0000000..5bb1ae5 --- /dev/null +++ b/tests/Usage/SampleWatermarkTest.php @@ -0,0 +1,65 @@ +range(lastSequence: 1); + $watermark = new SampleWatermark( + $range, + [$this->entry()], + false, + ); + + $this->assertTrue($watermark->matches($range)); + $this->assertFalse($watermark->matches($this->range(lastSequence: 2))); + $this->assertSame([$this->entry()], $watermark->getEntries()); + $this->assertFalse($watermark->isTruncated()); + } + + public function testRejectsDuplicateIngestIds(): void + { + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage('entries must be a unique list'); + + new SampleWatermark( + $this->range(lastSequence: 1), + [ + $this->entry(), + $this->entry(), + ], + false, + ); + } + + private function entry(): string + { + return '0123456789abcdef0123456789abcdef' + . ':' . str_repeat('a', 64) + . ':' . str_repeat('b', 64); + } + + private function range(int $lastSequence): SampleRange + { + return new SampleRange( + environment: 'production', + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: 'generation-1', + metric: 'bandwidth.inbound', + firstSequence: 0, + lastSequence: $lastSequence, + intervalStart: new DateTimeImmutable('2026-08-01T00:00:00Z'), + intervalEnd: new DateTimeImmutable('2026-08-01T00:03:00Z'), + ); + } +}