From 3c5470e3a674ae34d1224f1426f170051d593327 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Sun, 19 Jul 2026 14:17:37 +0200 Subject: [PATCH] Add service operations minutely rollup --- apps/api/src/routes/query-engine.http.ts | 15 +- apps/cli/src/server/schema/local-inserts.json | 2 +- apps/cli/src/server/schema/local-schema.sql | 36 +++- apps/ingest/src/clickhouse_insert_mappings.rs | 4 +- .../services/service-operations.test.ts | 10 +- .../components/services/service-operations.ts | 5 +- bun.lock | 1 + packages/domain/package.json | 2 + .../domain/src/clickhouse/ddl-emitter.test.ts | 21 +++ .../0008_service_operations_minutely.ts | 84 ++++++++++ .../src/clickhouse/migrations/index.test.ts | 30 +++- .../domain/src/clickhouse/migrations/index.ts | 2 + .../domain/src/generated/clickhouse-schema.ts | 4 +- .../generated/tinybird-project-manifest.ts | 10 +- packages/domain/src/tinybird/datasources.ts | 37 ++++ .../domain/src/tinybird/materializations.ts | 34 ++++ .../src/tinybird/project-manifest.test.ts | 6 + .../domain/src/tinybird/span-display-name.ts | 38 +++++ packages/query-engine/src/ch/index.ts | 2 + .../src/ch/queries/service-operations.test.ts | 125 ++++++++++++-- .../src/ch/queries/service-operations.ts | 158 ++++++++++++++++-- packages/query-engine/src/ch/tables.ts | 15 ++ packages/query-engine/src/traces-shared.ts | 14 +- 23 files changed, 603 insertions(+), 52 deletions(-) create mode 100644 packages/domain/src/clickhouse/migrations/0008_service_operations_minutely.ts create mode 100644 packages/domain/src/tinybird/span-display-name.ts diff --git a/apps/api/src/routes/query-engine.http.ts b/apps/api/src/routes/query-engine.http.ts index 7f91c26d5..7cb26373f 100644 --- a/apps/api/src/routes/query-engine.http.ts +++ b/apps/api/src/routes/query-engine.http.ts @@ -1511,8 +1511,11 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", startTime: payload.startTime, endTime: payload.endTime, } + // Keep production reads on the raw rollback path until the rollup has + // been deployed, backfilled under a write pause, and parity-verified on + // both managed Tinybird and every BYO ClickHouse backend. const summaryCompiled = CH.compile( - CH.serviceOperationsSummaryQuery({ + CH.serviceOperationsSummaryRawQuery({ serviceName: payload.serviceName, environments: payload.environments, limit: payload.limit, @@ -1532,18 +1535,18 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", } const spanNames = summaryRows.map((row) => String(row.spanName)) - // Fallback bucket sizing mirrors the client's chart-grid density - // (~50 buckets), floored at 1 minute. + // The rollup is minute-grain, so every sparkline interval must be + // a whole-minute multiple. Nearest-minute rounding keeps ~50 points. const windowSeconds = Math.max( 0, (Date.parse(`${payload.endTime.replace(" ", "T")}Z`) - Date.parse(`${payload.startTime.replace(" ", "T")}Z`)) / 1000, ) - const bucketSeconds = - payload.bucketSeconds ?? Math.max(60, Math.floor(windowSeconds / 50)) + const requestedBucketSeconds = payload.bucketSeconds ?? windowSeconds / 50 + const bucketSeconds = Math.max(1, Math.round(requestedBucketSeconds / 60)) * 60 const timeseriesCompiled = CH.compile( - CH.serviceOperationsTimeseriesQuery({ + CH.serviceOperationsTimeseriesRawQuery({ serviceName: payload.serviceName, environments: payload.environments, spanNames, diff --git a/apps/cli/src/server/schema/local-inserts.json b/apps/cli/src/server/schema/local-inserts.json index 8eae6a96b..4e241ee8c 100644 --- a/apps/cli/src/server/schema/local-inserts.json +++ b/apps/cli/src/server/schema/local-inserts.json @@ -1,5 +1,5 @@ { - "projectRevision": "ffade4bccc59af00fd33a561c4c919fd0229e0505f659d3242081f670f034a41", + "projectRevision": "c3f2be342187b6f6cf09010043e775b68f0901081a9b861803e55f5fe299e4c6", "orgPlaceholder": "__ORG__", "datasources": { "traces": { diff --git a/apps/cli/src/server/schema/local-schema.sql b/apps/cli/src/server/schema/local-schema.sql index 248205b3a..e0e92d96d 100644 --- a/apps/cli/src/server/schema/local-schema.sql +++ b/apps/cli/src/server/schema/local-schema.sql @@ -1,6 +1,6 @@ -- This file is generated by scripts/generate-clickhouse-schema-sql.ts -- Do not edit manually. Run `bun run clickhouse:schema` to regenerate. --- projectRevision: ffade4bccc59af00fd33a561c4c919fd0229e0505f659d3242081f670f034a41 +-- projectRevision: c3f2be342187b6f6cf09010043e775b68f0901081a9b861803e55f5fe299e4c6 CREATE TABLE IF NOT EXISTS alert_checks ( OrgId LowCardinality(String), @@ -431,6 +431,24 @@ PARTITION BY toDate(Timestamp) ORDER BY (OrgId, TraceId, SpanId, Timestamp) TTL Timestamp + INTERVAL 30 DAY; +CREATE TABLE IF NOT EXISTS service_operations_minutely ( + OrgId LowCardinality(String), + Minute DateTime, + ServiceName LowCardinality(String), + DeploymentEnv LowCardinality(String), + SpanName String, + SpanCount SimpleAggregateFunction(sum, UInt64), + EstimatedSpanCount SimpleAggregateFunction(sum, Float64), + ErrorCount SimpleAggregateFunction(sum, UInt64), + EstimatedErrorCount SimpleAggregateFunction(sum, Float64), + DurationSum SimpleAggregateFunction(sum, Float64), + DurationQuantiles AggregateFunction(quantilesTDigest(0.5, 0.95), UInt64) +) +ENGINE = AggregatingMergeTree +PARTITION BY toDate(Minute) +ORDER BY (OrgId, ServiceName, DeploymentEnv, Minute, SpanName) +TTL toDate(Minute) + INTERVAL 90 DAY; + CREATE TABLE IF NOT EXISTS service_overview_spans ( OrgId LowCardinality(String), Timestamp DateTime, @@ -1153,6 +1171,22 @@ SELECT FROM traces WHERE SpanKind IN ('Client', 'Producer', 'Server', 'Consumer'); +CREATE MATERIALIZED VIEW IF NOT EXISTS service_operations_minutely_mv TO service_operations_minutely AS +SELECT + OrgId, + toStartOfMinute(toDateTime(Timestamp)) AS Minute, + ServiceName, + ResourceAttributes['deployment.environment'] AS DeploymentEnv, + if(((SpanName LIKE 'http.server %' OR SpanName IN ('GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS')) AND (SpanAttributes['http.route'] != '' OR SpanAttributes['url.path'] != '')), concat(if(SpanName LIKE 'http.server %', replaceOne(SpanName, 'http.server ', ''), SpanName), ' ', if(SpanAttributes['http.route'] != '', SpanAttributes['http.route'], SpanAttributes['url.path'])), SpanName) AS SpanName, + count() AS SpanCount, + sum(SampleRate) AS EstimatedSpanCount, + countIf(StatusCode = 'Error') AS ErrorCount, + sumIf(SampleRate, StatusCode = 'Error') AS EstimatedErrorCount, + sum(toFloat64(Duration)) AS DurationSum, + quantilesTDigestState(0.5, 0.95)(Duration) AS DurationQuantiles + FROM traces + GROUP BY OrgId, Minute, ServiceName, DeploymentEnv, SpanName; + CREATE MATERIALIZED VIEW IF NOT EXISTS service_overview_spans_mv TO service_overview_spans AS SELECT OrgId, diff --git a/apps/ingest/src/clickhouse_insert_mappings.rs b/apps/ingest/src/clickhouse_insert_mappings.rs index dd4f94280..cd7e891ba 100644 --- a/apps/ingest/src/clickhouse_insert_mappings.rs +++ b/apps/ingest/src/clickhouse_insert_mappings.rs @@ -1,12 +1,12 @@ // This file is generated by scripts/generate-clickhouse-insert-mappings.ts // Do not edit manually. -pub const PROJECT_REVISION: &str = "ffade4bccc59af00fd33a561c4c919fd0229e0505f659d3242081f670f034a41"; +pub const PROJECT_REVISION: &str = "c3f2be342187b6f6cf09010043e775b68f0901081a9b861803e55f5fe299e4c6"; // Gate for BYO-ClickHouse ingest readiness — the migration version, NOT the // Tinybird-coupled PROJECT_REVISION. Compared against // org_clickhouse_settings.schema_version. See @maple/domain/clickhouse // clickHouseSchemaVersion. -pub const SCHEMA_VERSION: &str = "7"; +pub const SCHEMA_VERSION: &str = "8"; pub const ORG_PLACEHOLDER: &str = "__ORG__"; #[derive(Debug)] diff --git a/apps/web/src/components/services/service-operations.test.ts b/apps/web/src/components/services/service-operations.test.ts index 308cc8fa6..fdd50d6d8 100644 --- a/apps/web/src/components/services/service-operations.test.ts +++ b/apps/web/src/components/services/service-operations.test.ts @@ -24,7 +24,13 @@ describe("windowSeconds", () => { describe("operationsBucketSeconds", () => { it("targets ~50 buckets across the window", () => { - expect(operationsBucketSeconds("2024-01-01 00:00:00", "2024-01-01 12:00:00")).toBe(864) + expect(operationsBucketSeconds("2024-01-01 00:00:00", "2024-01-01 12:00:00")).toBe(840) + }) + + it("always rounds to a whole-minute multiple", () => { + const bucket = operationsBucketSeconds("2024-01-01 00:00:00", "2024-01-02 00:00:00") + expect(bucket).toBe(1740) + expect(bucket % 60).toBe(0) }) it("floors at one minute for short windows", () => { @@ -55,7 +61,7 @@ describe("serviceOperationsQueryInput", () => { startTime: "2024-01-01 00:00:00", endTime: "2024-01-01 12:00:00", environments: ["production"], - bucketSeconds: 864, + bucketSeconds: 840, limit: 25, }) }) diff --git a/apps/web/src/components/services/service-operations.ts b/apps/web/src/components/services/service-operations.ts index 7ea974064..083092590 100644 --- a/apps/web/src/components/services/service-operations.ts +++ b/apps/web/src/components/services/service-operations.ts @@ -1,7 +1,7 @@ import type { GetServiceOperationsInput } from "@/api/warehouse/service-operations" import { normalizeTimestampInput } from "@/lib/timezone-format" -/** Matches the chart grid's density: ~50 buckets across the window, ≥1 minute. */ +/** Matches the chart grid's density: ~50 whole-minute buckets, ≥1 minute. */ export const OPERATIONS_SPARKLINE_BUCKETS = 50 export const OPERATIONS_LIMIT = 25 @@ -13,7 +13,8 @@ export function windowSeconds(startTime: string, endTime: string): number { } export function operationsBucketSeconds(startTime: string, endTime: string): number { - return Math.max(60, Math.floor(windowSeconds(startTime, endTime) / OPERATIONS_SPARKLINE_BUCKETS)) + const targetMinutes = windowSeconds(startTime, endTime) / OPERATIONS_SPARKLINE_BUCKETS / 60 + return Math.max(1, Math.round(targetMinutes)) * 60 } export function callsPerSecond(estimatedSpanCount: number, seconds: number): number { diff --git a/bun.lock b/bun.lock index d804de2c6..fb8224c66 100644 --- a/bun.lock +++ b/bun.lock @@ -494,6 +494,7 @@ "packages/domain": { "name": "@maple/domain", "dependencies": { + "@maple-dev/clickhouse-builder": "workspace:*", "@tinybirdco/sdk": "catalog:tinybird", "effect": "catalog:effect", }, diff --git a/packages/domain/package.json b/packages/domain/package.json index bbfd01ee2..d70d9e396 100644 --- a/packages/domain/package.json +++ b/packages/domain/package.json @@ -17,6 +17,7 @@ "./warehouse-queries": "./src/warehouse-queries.ts", "./tinybird": "./src/tinybird/index.ts", "./tinybird/db-query-shape-sql": "./src/tinybird/db-query-shape-sql.ts", + "./tinybird/span-display-name": "./src/tinybird/span-display-name.ts", "./clickhouse": "./src/clickhouse/index.ts", "./where-clause": "./src/where-clause.ts", "./http/observability": "./src/http/observability.ts", @@ -27,6 +28,7 @@ "typecheck": "tsc --noEmit" }, "dependencies": { + "@maple-dev/clickhouse-builder": "workspace:*", "@tinybirdco/sdk": "catalog:tinybird", "effect": "catalog:effect" }, diff --git a/packages/domain/src/clickhouse/ddl-emitter.test.ts b/packages/domain/src/clickhouse/ddl-emitter.test.ts index 6b054a5e1..6ffb32cbf 100644 --- a/packages/domain/src/clickhouse/ddl-emitter.test.ts +++ b/packages/domain/src/clickhouse/ddl-emitter.test.ts @@ -100,6 +100,27 @@ describe("ClickHouse DDL emitter", () => { const aggDdl = emitCreateTable(aggDs!, { engineFlavor: "ReplicatedMergeTree" }) expect(aggDdl).toContain("ENGINE = AggregatingMergeTree") }) + + it("emits the service-operation rollup key, lifecycle, and MV projection", async () => { + const manifest = await buildTinybirdProjectManifest() + const rollup = manifest.datasources.find((ds) => ds.name === "service_operations_minutely") + const mv = manifest.pipes.find((pipe) => pipe.name === "service_operations_minutely_mv") + expect(rollup).toBeDefined() + expect(mv).toBeDefined() + + const tableDdl = emitCreateTable(rollup!) + expect(tableDdl).toContain("ENGINE = AggregatingMergeTree") + expect(tableDdl).toContain("PARTITION BY toDate(Minute)") + expect(tableDdl).toContain("ORDER BY (OrgId, ServiceName, DeploymentEnv, Minute, SpanName)") + expect(tableDdl).toContain("TTL toDate(Minute) + INTERVAL 90 DAY") + expect(tableDdl).toContain("DurationQuantiles AggregateFunction(quantilesTDigest(0.5, 0.95), UInt64)") + + const mvDdl = emitCreateMaterializedView(mv!) + expect(mvDdl).toContain("FROM traces") + expect(mvDdl).toContain("toStartOfMinute(toDateTime(Timestamp)) AS Minute") + expect(mvDdl).toContain("http.route") + expect(mvDdl).toContain("GROUP BY OrgId, Minute, ServiceName, DeploymentEnv, SpanName") + }) }) describe("extractColumnDefinition", () => { diff --git a/packages/domain/src/clickhouse/migrations/0008_service_operations_minutely.ts b/packages/domain/src/clickhouse/migrations/0008_service_operations_minutely.ts new file mode 100644 index 000000000..01eae1c5d --- /dev/null +++ b/packages/domain/src/clickhouse/migrations/0008_service_operations_minutely.ts @@ -0,0 +1,84 @@ +import { NORMALIZED_SPAN_NAME_SQL } from "../../tinybird/span-display-name" +import type { BackfillSpec } from "../backfill" + +/** + * Migration 0008 — minutely service-operation rollup. + * + * The migration installs the empty target and its live-write MV only. Historical + * backfill is deliberately a separate, chunkable operation: customer collectors + * can write directly to BYO ClickHouse during schema apply, so an automatic + * backfill/MV cutover cannot be gap-free. Rollout must pause writes, truncate the + * target, execute {@link serviceOperationsMinutelyBackfill}, verify parity, and + * then resume writes before enabling hybrid reads. + */ +export const serviceOperationsMinutelyBackfill: BackfillSpec = { + kind: "backfill", + target: "service_operations_minutely", + columns: [ + "OrgId", + "Minute", + "ServiceName", + "DeploymentEnv", + "SpanName", + "SpanCount", + "EstimatedSpanCount", + "ErrorCount", + "EstimatedErrorCount", + "DurationSum", + "DurationQuantiles", + ], + from: "traces", + tsColumn: "Timestamp", + select: `OrgId, + toStartOfMinute(toDateTime(Timestamp)) AS Minute, + ServiceName, + ResourceAttributes['deployment.environment'] AS DeploymentEnv, + ${NORMALIZED_SPAN_NAME_SQL} AS SpanName, + count() AS SpanCount, + sum(SampleRate) AS EstimatedSpanCount, + countIf(StatusCode = 'Error') AS ErrorCount, + sumIf(SampleRate, StatusCode = 'Error') AS EstimatedErrorCount, + sum(toFloat64(Duration)) AS DurationSum, + quantilesTDigestState(0.5, 0.95)(Duration) AS DurationQuantiles`, + groupBy: "OrgId, Minute, ServiceName, DeploymentEnv, SpanName", +} + +export const migration_0008_service_operations_minutely = { + version: 8, + description: "Add the minutely service-operation rollup and its live-write materialized view", + statements: [ + "DROP VIEW IF EXISTS service_operations_minutely_mv", + `CREATE TABLE IF NOT EXISTS service_operations_minutely ( + OrgId LowCardinality(String), + Minute DateTime, + ServiceName LowCardinality(String), + DeploymentEnv LowCardinality(String), + SpanName String, + SpanCount SimpleAggregateFunction(sum, UInt64), + EstimatedSpanCount SimpleAggregateFunction(sum, Float64), + ErrorCount SimpleAggregateFunction(sum, UInt64), + EstimatedErrorCount SimpleAggregateFunction(sum, Float64), + DurationSum SimpleAggregateFunction(sum, Float64), + DurationQuantiles AggregateFunction(quantilesTDigest(0.5, 0.95), UInt64) +) +ENGINE = AggregatingMergeTree +PARTITION BY toDate(Minute) +ORDER BY (OrgId, ServiceName, DeploymentEnv, Minute, SpanName) +TTL toDate(Minute) + INTERVAL 90 DAY`, + `CREATE MATERIALIZED VIEW IF NOT EXISTS service_operations_minutely_mv TO service_operations_minutely AS +SELECT + OrgId, + toStartOfMinute(toDateTime(Timestamp)) AS Minute, + ServiceName, + ResourceAttributes['deployment.environment'] AS DeploymentEnv, + ${NORMALIZED_SPAN_NAME_SQL} AS SpanName, + count() AS SpanCount, + sum(SampleRate) AS EstimatedSpanCount, + countIf(StatusCode = 'Error') AS ErrorCount, + sumIf(SampleRate, StatusCode = 'Error') AS EstimatedErrorCount, + sum(toFloat64(Duration)) AS DurationSum, + quantilesTDigestState(0.5, 0.95)(Duration) AS DurationQuantiles +FROM traces +GROUP BY OrgId, Minute, ServiceName, DeploymentEnv, SpanName`, + ], +} as const diff --git a/packages/domain/src/clickhouse/migrations/index.test.ts b/packages/domain/src/clickhouse/migrations/index.test.ts index 7a747953d..34471b083 100644 --- a/packages/domain/src/clickhouse/migrations/index.test.ts +++ b/packages/domain/src/clickhouse/migrations/index.test.ts @@ -3,7 +3,10 @@ import { isBackfill, renderStatementFull, type BackfillSpec } from "../backfill" import { migration_0004_service_namespace_projections } from "./0004_service_namespace_projections" import { migration_0005_alert_checks_error_columns } from "./0005_alert_checks_error_columns" import { migration_0006_db_edge_namespace } from "./0006_db_edge_namespace" -import { migration_0007_db_namespace_hyperdrive } from "./0007_db_namespace_hyperdrive" +import { + migration_0008_service_operations_minutely, + serviceOperationsMinutelyBackfill, +} from "./0008_service_operations_minutely" import { migrations } from "./index" const backfills = migration_0004_service_namespace_projections.statements.filter( @@ -18,8 +21,29 @@ const renderedSql = migration_0004_service_namespace_projections.statements describe("ClickHouse migrations", () => { it("keeps migrations ordered by version", () => { - expect(migrations.map((m) => m.version)).toEqual([1, 2, 3, 4, 5, 6, 7]) - expect(migrations.at(-1)).toBe(migration_0007_db_namespace_hyperdrive) + expect(migrations.map((m) => m.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8]) + expect(migrations.at(-1)).toBe(migration_0008_service_operations_minutely) + }) + + it("adds the service-operation rollup and exposes a coordinated chunkable backfill", () => { + const statements = migration_0008_service_operations_minutely.statements + const sql = statements.map((statement) => renderStatementFull(statement, "default")).join("\n\n") + + expect(sql).toContain("PARTITION BY toDate(Minute)") + expect(sql).toContain("ORDER BY (OrgId, ServiceName, DeploymentEnv, Minute, SpanName)") + expect(sql).toContain("TTL toDate(Minute) + INTERVAL 90 DAY") + expect(sql).toContain("SpanName String") + expect(sql).toContain("quantilesTDigestState(0.5, 0.95)(Duration)") + expect(sql).toContain("http.route") + expect(sql).toContain("http.server %") + expect(statements.filter(isBackfill)).toHaveLength(0) + expect(sql).not.toContain("TRUNCATE TABLE service_operations_minutely") + expect(serviceOperationsMinutelyBackfill.target).toBe("service_operations_minutely") + expect(serviceOperationsMinutelyBackfill.from).toBe("traces") + expect(serviceOperationsMinutelyBackfill.tsColumn).toBe("Timestamp") + expect(serviceOperationsMinutelyBackfill.groupBy).toBe( + "OrgId, Minute, ServiceName, DeploymentEnv, SpanName", + ) }) it("adds alert_checks error columns as idempotent ALTERs", () => { diff --git a/packages/domain/src/clickhouse/migrations/index.ts b/packages/domain/src/clickhouse/migrations/index.ts index f37329a3b..ec7a46492 100644 --- a/packages/domain/src/clickhouse/migrations/index.ts +++ b/packages/domain/src/clickhouse/migrations/index.ts @@ -6,6 +6,7 @@ import { migration_0004_service_namespace_projections } from "./0004_service_nam import { migration_0005_alert_checks_error_columns } from "./0005_alert_checks_error_columns" import { migration_0006_db_edge_namespace } from "./0006_db_edge_namespace" import { migration_0007_db_namespace_hyperdrive } from "./0007_db_namespace_hyperdrive" +import { migration_0008_service_operations_minutely } from "./0008_service_operations_minutely" /** * A migration statement is either a raw SQL string (structural DDL) or a @@ -42,6 +43,7 @@ export const migrations: ReadonlyArray = [ migration_0005_alert_checks_error_columns, migration_0006_db_edge_namespace, migration_0007_db_namespace_hyperdrive, + migration_0008_service_operations_minutely, ] as const /** Highest migration `version` bundled — i.e. the schema level a fully-applied diff --git a/packages/domain/src/generated/clickhouse-schema.ts b/packages/domain/src/generated/clickhouse-schema.ts index 7607a77f5..47b50b9bb 100644 --- a/packages/domain/src/generated/clickhouse-schema.ts +++ b/packages/domain/src/generated/clickhouse-schema.ts @@ -1,7 +1,7 @@ // This file is generated by scripts/generate-clickhouse-schema.ts // Do not edit manually. -export const projectRevision = "ffade4bccc59af00fd33a561c4c919fd0229e0505f659d3242081f670f034a41" as const +export const projectRevision = "c3f2be342187b6f6cf09010043e775b68f0901081a9b861803e55f5fe299e4c6" as const export const latestSnapshotStatements: ReadonlyArray = [ "CREATE TABLE IF NOT EXISTS alert_checks (\n OrgId LowCardinality(String),\n RuleId String,\n GroupKey String,\n Timestamp DateTime64(3),\n Status LowCardinality(String),\n SignalType LowCardinality(String),\n Comparator LowCardinality(String),\n Threshold Float64,\n ObservedValue Nullable(Float64),\n SampleCount UInt32,\n WindowMinutes UInt16,\n WindowStart DateTime64(3),\n WindowEnd DateTime64(3),\n ConsecutiveBreaches UInt16,\n ConsecutiveHealthy UInt16,\n IncidentId Nullable(String),\n IncidentTransition LowCardinality(String),\n EvaluationDurationMs UInt32,\n ErrorMessage Nullable(String),\n ErrorCategory LowCardinality(String)\n)\nENGINE = MergeTree\nPARTITION BY toDate(Timestamp)\nORDER BY (OrgId, RuleId, GroupKey, Timestamp)\nTTL toDate(Timestamp) + INTERVAL 90 DAY", @@ -24,6 +24,7 @@ export const latestSnapshotStatements: ReadonlyArray = [ "CREATE TABLE IF NOT EXISTS service_map_db_query_shapes_hourly (\n OrgId LowCardinality(String),\n Hour DateTime,\n ServiceName LowCardinality(String),\n DbSystem LowCardinality(String),\n DeploymentEnv LowCardinality(String),\n QueryKey String,\n QueryLabel SimpleAggregateFunction(any, String),\n SampleStatement SimpleAggregateFunction(any, String),\n CallCount SimpleAggregateFunction(sum, UInt64),\n ErrorCount SimpleAggregateFunction(sum, UInt64),\n EstimatedCount SimpleAggregateFunction(sum, Float64),\n EstimatedErrorCount SimpleAggregateFunction(sum, Float64),\n WeightedDurationSumMs SimpleAggregateFunction(sum, Float64),\n DurationQuantiles AggregateFunction(quantilesTDigestWeighted(0.5, 0.95), UInt64, UInt32),\n DbNamespace LowCardinality(String)\n)\nENGINE = AggregatingMergeTree\nPARTITION BY toDate(Hour)\nORDER BY (OrgId, Hour, DeploymentEnv, ServiceName, DbSystem, DbNamespace, QueryKey)\nTTL toDate(Hour) + INTERVAL 90 DAY", "CREATE TABLE IF NOT EXISTS service_map_edges_hourly (\n OrgId LowCardinality(String),\n Hour DateTime,\n SourceService LowCardinality(String),\n TargetService String,\n DeploymentEnv LowCardinality(String),\n CallCount SimpleAggregateFunction(sum, UInt64),\n ErrorCount SimpleAggregateFunction(sum, UInt64),\n DurationSumMs SimpleAggregateFunction(sum, Float64),\n MaxDurationMs SimpleAggregateFunction(max, Float64),\n SampledSpanCount SimpleAggregateFunction(sum, UInt64),\n UnsampledSpanCount SimpleAggregateFunction(sum, UInt64),\n SampleRateSum SimpleAggregateFunction(sum, Float64)\n)\nENGINE = AggregatingMergeTree\nPARTITION BY toDate(Hour)\nORDER BY (OrgId, Hour, DeploymentEnv, SourceService, TargetService)\nTTL toDate(Hour) + INTERVAL 90 DAY", "CREATE TABLE IF NOT EXISTS service_map_spans (\n OrgId LowCardinality(String),\n Timestamp DateTime,\n TraceId String,\n SpanId String,\n ParentSpanId String,\n ServiceName LowCardinality(String),\n SpanKind LowCardinality(String),\n Duration UInt64,\n StatusCode LowCardinality(String),\n TraceState String,\n DeploymentEnv LowCardinality(String)\n)\nENGINE = MergeTree\nPARTITION BY toDate(Timestamp)\nORDER BY (OrgId, TraceId, SpanId, Timestamp)\nTTL Timestamp + INTERVAL 30 DAY", + "CREATE TABLE IF NOT EXISTS service_operations_minutely (\n OrgId LowCardinality(String),\n Minute DateTime,\n ServiceName LowCardinality(String),\n DeploymentEnv LowCardinality(String),\n SpanName String,\n SpanCount SimpleAggregateFunction(sum, UInt64),\n EstimatedSpanCount SimpleAggregateFunction(sum, Float64),\n ErrorCount SimpleAggregateFunction(sum, UInt64),\n EstimatedErrorCount SimpleAggregateFunction(sum, Float64),\n DurationSum SimpleAggregateFunction(sum, Float64),\n DurationQuantiles AggregateFunction(quantilesTDigest(0.5, 0.95), UInt64)\n)\nENGINE = AggregatingMergeTree\nPARTITION BY toDate(Minute)\nORDER BY (OrgId, ServiceName, DeploymentEnv, Minute, SpanName)\nTTL toDate(Minute) + INTERVAL 90 DAY", "CREATE TABLE IF NOT EXISTS service_overview_spans (\n OrgId LowCardinality(String),\n Timestamp DateTime,\n ServiceName LowCardinality(String),\n Duration UInt64,\n StatusCode LowCardinality(String),\n TraceState String,\n DeploymentEnv LowCardinality(String),\n CommitSha LowCardinality(String),\n SampleRate Float64 DEFAULT 1,\n ServiceNamespace LowCardinality(String),\n INDEX idx_service_namespace ServiceNamespace TYPE set(1000) GRANULARITY 4\n)\nENGINE = MergeTree\nPARTITION BY toDate(Timestamp)\nORDER BY (OrgId, ServiceName, Timestamp)\nTTL Timestamp + INTERVAL 30 DAY", "CREATE TABLE IF NOT EXISTS service_platforms_hourly (\n OrgId LowCardinality(String),\n Hour DateTime,\n ServiceName LowCardinality(String),\n DeploymentEnv LowCardinality(String),\n K8sCluster SimpleAggregateFunction(max, String),\n K8sPodName SimpleAggregateFunction(max, String),\n K8sDeploymentName SimpleAggregateFunction(max, String),\n K8sStatefulSetName SimpleAggregateFunction(max, String),\n K8sDaemonSetName SimpleAggregateFunction(max, String),\n K8sNamespaceName SimpleAggregateFunction(max, String),\n CloudPlatform SimpleAggregateFunction(max, String),\n CloudProvider SimpleAggregateFunction(max, String),\n FaasName SimpleAggregateFunction(max, String),\n MapleSdkType SimpleAggregateFunction(max, String),\n ProcessRuntimeName SimpleAggregateFunction(max, String),\n SpanCount SimpleAggregateFunction(sum, UInt64)\n)\nENGINE = AggregatingMergeTree\nPARTITION BY toDate(Hour)\nORDER BY (OrgId, Hour, ServiceName, DeploymentEnv)\nTTL toDate(Hour) + INTERVAL 90 DAY", "CREATE TABLE IF NOT EXISTS service_usage (\n OrgId LowCardinality(String),\n ServiceName LowCardinality(String),\n Hour DateTime,\n LogCount UInt64,\n LogSizeBytes UInt64,\n TraceCount UInt64,\n TraceSizeBytes UInt64,\n SumMetricCount UInt64,\n SumMetricSizeBytes UInt64,\n GaugeMetricCount UInt64,\n GaugeMetricSizeBytes UInt64,\n HistogramMetricCount UInt64,\n HistogramMetricSizeBytes UInt64,\n ExpHistogramMetricCount UInt64,\n ExpHistogramMetricSizeBytes UInt64\n)\nENGINE = SummingMergeTree\nORDER BY (OrgId, ServiceName, Hour)\nTTL Hour + INTERVAL 365 DAY", @@ -52,6 +53,7 @@ export const latestSnapshotStatements: ReadonlyArray = [ "CREATE MATERIALIZED VIEW IF NOT EXISTS service_map_db_edges_hourly_mv TO service_map_db_edges_hourly AS\nSELECT\n OrgId,\n toStartOfHour(toDateTime(Timestamp)) AS Hour,\n ServiceName,\n coalesce(nullIf(SpanAttributes['db.system.name'], ''), SpanAttributes['db.system']) AS DbSystem,\n if(match(coalesce(nullIf(SpanAttributes['db.namespace'], ''), nullIf(SpanAttributes['db.name'], ''), nullIf(SpanAttributes['server.address'], ''), SpanAttributes['net.peer.name']), '^([0-9a-fA-F]{32}|.*[.]hyperdrive[.]local)$'), 'hyperdrive', coalesce(nullIf(SpanAttributes['db.namespace'], ''), nullIf(SpanAttributes['db.name'], ''), nullIf(SpanAttributes['server.address'], ''), SpanAttributes['net.peer.name'])) AS DbNamespace,\n ResourceAttributes['deployment.environment'] AS DeploymentEnv,\n count() AS CallCount,\n countIf(StatusCode = 'Error') AS ErrorCount,\n sum(Duration / 1000000) AS DurationSumMs,\n max(Duration / 1000000) AS MaxDurationMs,\n countIf(TraceState LIKE '%th:%') AS SampledSpanCount,\n countIf(TraceState = '' OR TraceState NOT LIKE '%th:%') AS UnsampledSpanCount,\n sum(SampleRate) AS SampleRateSum\n FROM traces\n WHERE SpanKind IN ('Client', 'Producer')\n AND coalesce(nullIf(SpanAttributes['db.system.name'], ''), SpanAttributes['db.system']) != ''\n AND ServiceName != ''\n GROUP BY OrgId, Hour, ServiceName, DbSystem, DbNamespace, DeploymentEnv", "CREATE MATERIALIZED VIEW IF NOT EXISTS service_map_db_query_shapes_hourly_mv TO service_map_db_query_shapes_hourly AS\nSELECT\n OrgId,\n toStartOfHour(toDateTime(Timestamp)) AS Hour,\n ServiceName,\n coalesce(nullIf(SpanAttributes['db.system.name'], ''), SpanAttributes['db.system']) AS DbSystem,\n if(match(coalesce(nullIf(SpanAttributes['db.namespace'], ''), nullIf(SpanAttributes['db.name'], ''), nullIf(SpanAttributes['server.address'], ''), SpanAttributes['net.peer.name']), '^([0-9a-fA-F]{32}|.*[.]hyperdrive[.]local)$'), 'hyperdrive', coalesce(nullIf(SpanAttributes['db.namespace'], ''), nullIf(SpanAttributes['db.name'], ''), nullIf(SpanAttributes['server.address'], ''), SpanAttributes['net.peer.name'])) AS DbNamespace,\n ResourceAttributes['deployment.environment'] AS DeploymentEnv,\n coalesce(\n nullIf(SpanAttributes['db.query.fingerprint'], ''),\n nullIf(SpanAttributes['db.statement.fingerprint'], ''),\n nullIf(if(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']) != '', toString(cityHash64(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(lower(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement'])), '\\'[^\\']*\\'', '?'), '\\\\bin\\\\s*\\\\([^)]*\\\\)', 'in (?)'), '[0-9]+(\\\\.[0-9]+)?', '?'), '\\\\s+', ' '), '^\\\\s+|\\\\s+$', ''))), ''), ''),\n toString(cityHash64(coalesce(\n nullIf(SpanAttributes['db.query.summary'], ''),\n nullIf(if(SpanAttributes['db.operation.name'] != '', trimBoth(concat(SpanAttributes['db.operation.name'], if(coalesce(nullIf(SpanAttributes['db.collection.name'], ''), SpanAttributes['db.namespace']) != '', concat(' ', coalesce(nullIf(SpanAttributes['db.collection.name'], ''), SpanAttributes['db.namespace'])), ''))), ''), ''),\n nullIf(if(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']) != '', trimBoth(concat(upper(extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '^\\\\s*(\\\\w+)')), if(extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '(?i)(?:from|into|update|join|table)\\\\s+\\\\W?([\\\\w.]+)') != '', concat(' ', extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '(?i)(?:from|into|update|join|table)\\\\s+\\\\W?([\\\\w.]+)')), ''))), ''), ''),\n nullIf(SpanAttributes['query.context'], ''),\n nullIf(SpanAttributes['db.operation.name'], ''),\n nullIf(SpanAttributes['db.operation'], ''),\n SpanName\n)))\n) AS QueryKey,\n any(substring(coalesce(\n nullIf(SpanAttributes['db.query.summary'], ''),\n nullIf(if(SpanAttributes['db.operation.name'] != '', trimBoth(concat(SpanAttributes['db.operation.name'], if(coalesce(nullIf(SpanAttributes['db.collection.name'], ''), SpanAttributes['db.namespace']) != '', concat(' ', coalesce(nullIf(SpanAttributes['db.collection.name'], ''), SpanAttributes['db.namespace'])), ''))), ''), ''),\n nullIf(if(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']) != '', trimBoth(concat(upper(extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '^\\\\s*(\\\\w+)')), if(extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '(?i)(?:from|into|update|join|table)\\\\s+\\\\W?([\\\\w.]+)') != '', concat(' ', extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '(?i)(?:from|into|update|join|table)\\\\s+\\\\W?([\\\\w.]+)')), ''))), ''), ''),\n nullIf(SpanAttributes['query.context'], ''),\n nullIf(SpanAttributes['db.operation.name'], ''),\n nullIf(SpanAttributes['db.operation'], ''),\n SpanName\n), 1, 220)) AS QueryLabel,\n any(substring(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), 1, 1000)) AS SampleStatement,\n count() AS CallCount,\n countIf(StatusCode = 'Error') AS ErrorCount,\n sum(SampleRate) AS EstimatedCount,\n sumIf(SampleRate, StatusCode = 'Error') AS EstimatedErrorCount,\n sum(toFloat64(Duration) * SampleRate / 1000000) AS WeightedDurationSumMs,\n quantilesTDigestWeightedState(0.5, 0.95)(Duration, toUInt32(greatest(SampleRate, 1.0))) AS DurationQuantiles\n FROM traces\n WHERE SpanKind IN ('Client', 'Producer')\n AND coalesce(nullIf(SpanAttributes['db.system.name'], ''), SpanAttributes['db.system']) != ''\n AND ServiceName != ''\n GROUP BY OrgId, Hour, ServiceName, DbSystem, DbNamespace, DeploymentEnv, QueryKey", "CREATE MATERIALIZED VIEW IF NOT EXISTS service_map_spans_mv TO service_map_spans AS\nSELECT\n OrgId,\n toDateTime(Timestamp) AS Timestamp,\n TraceId,\n SpanId,\n ParentSpanId,\n ServiceName,\n SpanKind,\n Duration,\n StatusCode,\n TraceState,\n ResourceAttributes['deployment.environment'] AS DeploymentEnv\n FROM traces\n WHERE SpanKind IN ('Client', 'Producer', 'Server', 'Consumer')", + "CREATE MATERIALIZED VIEW IF NOT EXISTS service_operations_minutely_mv TO service_operations_minutely AS\nSELECT\n OrgId,\n toStartOfMinute(toDateTime(Timestamp)) AS Minute,\n ServiceName,\n ResourceAttributes['deployment.environment'] AS DeploymentEnv,\n if(((SpanName LIKE 'http.server %' OR SpanName IN ('GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS')) AND (SpanAttributes['http.route'] != '' OR SpanAttributes['url.path'] != '')), concat(if(SpanName LIKE 'http.server %', replaceOne(SpanName, 'http.server ', ''), SpanName), ' ', if(SpanAttributes['http.route'] != '', SpanAttributes['http.route'], SpanAttributes['url.path'])), SpanName) AS SpanName,\n count() AS SpanCount,\n sum(SampleRate) AS EstimatedSpanCount,\n countIf(StatusCode = 'Error') AS ErrorCount,\n sumIf(SampleRate, StatusCode = 'Error') AS EstimatedErrorCount,\n sum(toFloat64(Duration)) AS DurationSum,\n quantilesTDigestState(0.5, 0.95)(Duration) AS DurationQuantiles\n FROM traces\n GROUP BY OrgId, Minute, ServiceName, DeploymentEnv, SpanName", "CREATE MATERIALIZED VIEW IF NOT EXISTS service_overview_spans_mv TO service_overview_spans AS\nSELECT\n OrgId,\n toDateTime(Timestamp) AS Timestamp,\n ServiceName,\n Duration,\n StatusCode,\n TraceState,\n ResourceAttributes['deployment.environment'] AS DeploymentEnv,\n ResourceAttributes['deployment.commit_sha'] AS CommitSha,\n SampleRate,\n ResourceAttributes['service.namespace'] AS ServiceNamespace\n FROM traces\n WHERE SpanKind IN ('Server', 'Consumer') OR ParentSpanId = ''", "CREATE MATERIALIZED VIEW IF NOT EXISTS service_platforms_hourly_mv TO service_platforms_hourly AS\nSELECT\n OrgId,\n toStartOfHour(toDateTime(Timestamp)) AS Hour,\n ServiceName,\n ResourceAttributes['deployment.environment'] AS DeploymentEnv,\n max(ResourceAttributes['k8s.cluster.name']) AS K8sCluster,\n max(ResourceAttributes['k8s.pod.name']) AS K8sPodName,\n max(ResourceAttributes['k8s.deployment.name']) AS K8sDeploymentName,\n max(ResourceAttributes['k8s.statefulset.name']) AS K8sStatefulSetName,\n max(ResourceAttributes['k8s.daemonset.name']) AS K8sDaemonSetName,\n max(ResourceAttributes['k8s.namespace.name']) AS K8sNamespaceName,\n max(ResourceAttributes['cloud.platform']) AS CloudPlatform,\n max(ResourceAttributes['cloud.provider']) AS CloudProvider,\n max(ResourceAttributes['faas.name']) AS FaasName,\n max(ResourceAttributes['maple.sdk.type']) AS MapleSdkType,\n max(ResourceAttributes['process.runtime.name']) AS ProcessRuntimeName,\n count() AS SpanCount\n FROM traces\n WHERE ServiceName != ''\n GROUP BY OrgId, Hour, ServiceName, DeploymentEnv", "CREATE MATERIALIZED VIEW IF NOT EXISTS service_usage_logs_mv TO service_usage AS\nSELECT\n OrgId,\n ServiceName,\n toStartOfHour(TimestampTime) AS Hour,\n count() AS LogCount,\n sum(length(Body) + 200) AS LogSizeBytes,\n 0 AS TraceCount,\n 0 AS TraceSizeBytes,\n 0 AS SumMetricCount,\n 0 AS SumMetricSizeBytes,\n 0 AS GaugeMetricCount,\n 0 AS GaugeMetricSizeBytes,\n 0 AS HistogramMetricCount,\n 0 AS HistogramMetricSizeBytes,\n 0 AS ExpHistogramMetricCount,\n 0 AS ExpHistogramMetricSizeBytes\n FROM logs\n GROUP BY OrgId, ServiceName, Hour", diff --git a/packages/domain/src/generated/tinybird-project-manifest.ts b/packages/domain/src/generated/tinybird-project-manifest.ts index 9f1e02d12..571e341e4 100644 --- a/packages/domain/src/generated/tinybird-project-manifest.ts +++ b/packages/domain/src/generated/tinybird-project-manifest.ts @@ -1,7 +1,7 @@ // This file is generated by scripts/generate-tinybird-project-manifest.ts // Do not edit manually. -export const projectRevision = "ffade4bccc59af00fd33a561c4c919fd0229e0505f659d3242081f670f034a41" as const +export const projectRevision = "c3f2be342187b6f6cf09010043e775b68f0901081a9b861803e55f5fe299e4c6" as const export const datasources = [ { @@ -84,6 +84,10 @@ export const datasources = [ "name": "service_map_spans", "content": "DESCRIPTION >\n Lightweight projection of traces for service map JOIN queries. Pre-extracts deployment.environment from Map columns. Populated by materialized view.\n\nSCHEMA >\n OrgId LowCardinality(String),\n Timestamp DateTime,\n TraceId String,\n SpanId String,\n ParentSpanId String,\n ServiceName LowCardinality(String),\n SpanKind LowCardinality(String),\n Duration UInt64,\n StatusCode LowCardinality(String),\n TraceState String,\n DeploymentEnv LowCardinality(String)\n\nENGINE \"MergeTree\"\nENGINE_PARTITION_KEY \"toDate(Timestamp)\"\nENGINE_SORTING_KEY \"OrgId, TraceId, SpanId, Timestamp\"\nENGINE_TTL \"Timestamp + INTERVAL 30 DAY\"" }, + { + "name": "service_operations_minutely", + "content": "DESCRIPTION >\n Minute-grain service operation metrics with normalized HTTP names, exact and sampling-weighted counts, and unweighted duration t-digest state.\n\nSCHEMA >\n OrgId LowCardinality(String),\n Minute DateTime,\n ServiceName LowCardinality(String),\n DeploymentEnv LowCardinality(String),\n SpanName String,\n SpanCount SimpleAggregateFunction(sum, UInt64),\n EstimatedSpanCount SimpleAggregateFunction(sum, Float64),\n ErrorCount SimpleAggregateFunction(sum, UInt64),\n EstimatedErrorCount SimpleAggregateFunction(sum, Float64),\n DurationSum SimpleAggregateFunction(sum, Float64),\n DurationQuantiles AggregateFunction(quantilesTDigest(0.5, 0.95), UInt64)\n\nENGINE \"AggregatingMergeTree\"\nENGINE_PARTITION_KEY \"toDate(Minute)\"\nENGINE_SORTING_KEY \"OrgId, ServiceName, DeploymentEnv, Minute, SpanName\"\nENGINE_TTL \"toDate(Minute) + INTERVAL 90 DAY\"" + }, { "name": "service_overview_spans", "content": "DESCRIPTION >\n Lightweight projection of service entry point spans (Server/Consumer + root) for service overview queries. Pre-extracts deployment attributes from ResourceAttributes. Populated by materialized view.\n\nSCHEMA >\n OrgId LowCardinality(String),\n Timestamp DateTime,\n ServiceName LowCardinality(String),\n Duration UInt64,\n StatusCode LowCardinality(String),\n TraceState String,\n DeploymentEnv LowCardinality(String),\n CommitSha LowCardinality(String),\n SampleRate Float64 DEFAULT 1,\n ServiceNamespace LowCardinality(String)\n\nENGINE \"MergeTree\"\nENGINE_PARTITION_KEY \"toDate(Timestamp)\"\nENGINE_SORTING_KEY \"OrgId, ServiceName, Timestamp\"\nENGINE_TTL \"Timestamp + INTERVAL 30 DAY\"\n\nINDEXES >\n idx_service_namespace ServiceNamespace TYPE set(1000) GRANULARITY 4" @@ -199,6 +203,10 @@ export const pipes = [ "name": "service_map_spans_mv", "content": "DESCRIPTION >\n Materialized view projecting trace spans needed for service dependency map. Extracts deployment.environment from Map columns at write time.\n\nNODE service_map_spans_mv_node\nSQL >\n SELECT\n OrgId,\n toDateTime(Timestamp) AS Timestamp,\n TraceId,\n SpanId,\n ParentSpanId,\n ServiceName,\n SpanKind,\n Duration,\n StatusCode,\n TraceState,\n ResourceAttributes['deployment.environment'] AS DeploymentEnv\n FROM traces\n WHERE SpanKind IN ('Client', 'Producer', 'Server', 'Consumer')\n\nTYPE MATERIALIZED\nDATASOURCE service_map_spans" }, + { + "name": "service_operations_minutely_mv", + "content": "DESCRIPTION >\n Pre-aggregates every span by service operation and minute with normalized HTTP names, exact/estimated counts, errors, duration sum, and unweighted t-digest state.\n\nNODE service_operations_minutely_mv_node\nSQL >\n SELECT\n OrgId,\n toStartOfMinute(toDateTime(Timestamp)) AS Minute,\n ServiceName,\n ResourceAttributes['deployment.environment'] AS DeploymentEnv,\n if(((SpanName LIKE 'http.server %' OR SpanName IN ('GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS')) AND (SpanAttributes['http.route'] != '' OR SpanAttributes['url.path'] != '')), concat(if(SpanName LIKE 'http.server %', replaceOne(SpanName, 'http.server ', ''), SpanName), ' ', if(SpanAttributes['http.route'] != '', SpanAttributes['http.route'], SpanAttributes['url.path'])), SpanName) AS SpanName,\n count() AS SpanCount,\n sum(SampleRate) AS EstimatedSpanCount,\n countIf(StatusCode = 'Error') AS ErrorCount,\n sumIf(SampleRate, StatusCode = 'Error') AS EstimatedErrorCount,\n sum(toFloat64(Duration)) AS DurationSum,\n quantilesTDigestState(0.5, 0.95)(Duration) AS DurationQuantiles\n FROM traces\n GROUP BY OrgId, Minute, ServiceName, DeploymentEnv, SpanName\n\nTYPE MATERIALIZED\nDATASOURCE service_operations_minutely" + }, { "name": "service_overview_spans_mv", "content": "DESCRIPTION >\n Materialized view projecting service entry point spans (Server/Consumer + root) for service overview queries. Pre-extracts deployment attributes from ResourceAttributes at write time.\n\nNODE service_overview_spans_mv_node\nSQL >\n SELECT\n OrgId,\n toDateTime(Timestamp) AS Timestamp,\n ServiceName,\n Duration,\n StatusCode,\n TraceState,\n ResourceAttributes['deployment.environment'] AS DeploymentEnv,\n ResourceAttributes['deployment.commit_sha'] AS CommitSha,\n SampleRate,\n ResourceAttributes['service.namespace'] AS ServiceNamespace\n FROM traces\n WHERE SpanKind IN ('Server', 'Consumer') OR ParentSpanId = ''\n\nTYPE MATERIALIZED\nDATASOURCE service_overview_spans" diff --git a/packages/domain/src/tinybird/datasources.ts b/packages/domain/src/tinybird/datasources.ts index 3625426b8..63c224df3 100644 --- a/packages/domain/src/tinybird/datasources.ts +++ b/packages/domain/src/tinybird/datasources.ts @@ -1262,6 +1262,43 @@ export const alertChecks = defineDatasource("alert_checks", { export type AlertChecksRow = InferRow +/** + * Minute-grain operation metrics used by the service-detail Operations panel. + * The operation name is normalized once by the write-side MV, while exact and + * sampling-weighted counts are retained side-by-side. Duration aggregates are + * deliberately unweighted to preserve the existing service-operations API + * semantics. + * + * Populated by materialized view, not direct ingestion. + */ +export const serviceOperationsMinutely = defineDatasource("service_operations_minutely", { + description: + "Minute-grain service operation metrics with normalized HTTP names, exact and sampling-weighted counts, and unweighted duration t-digest state.", + jsonPaths: false, + schema: { + OrgId: t.string().lowCardinality(), + Minute: t.dateTime(), + ServiceName: t.string().lowCardinality(), + DeploymentEnv: t.string().lowCardinality(), + // Falls back to url.path when no route template exists; keep this a plain + // String so high-cardinality paths do not churn a LowCardinality dictionary. + SpanName: t.string(), + SpanCount: t.simpleAggregateFunction("sum", t.uint64()), + EstimatedSpanCount: t.simpleAggregateFunction("sum", t.float64()), + ErrorCount: t.simpleAggregateFunction("sum", t.uint64()), + EstimatedErrorCount: t.simpleAggregateFunction("sum", t.float64()), + DurationSum: t.simpleAggregateFunction("sum", t.float64()), + DurationQuantiles: t.aggregateFunction("quantilesTDigest(0.5, 0.95)", t.uint64()), + }, + engine: engine.aggregatingMergeTree({ + partitionKey: "toDate(Minute)", + sortingKey: ["OrgId", "ServiceName", "DeploymentEnv", "Minute", "SpanName"], + ttl: "toDate(Minute) + INTERVAL 90 DAY", + }), +}) + +export type ServiceOperationsMinutelyRow = InferRow + /** * Generalized hourly aggregating MV target for traces. Stores partial state * (`-State` aggregates) keyed on the dimensions that show up in 90%+ of diff --git a/packages/domain/src/tinybird/materializations.ts b/packages/domain/src/tinybird/materializations.ts index cddfb2eef..15d5f6c41 100644 --- a/packages/domain/src/tinybird/materializations.ts +++ b/packages/domain/src/tinybird/materializations.ts @@ -19,6 +19,7 @@ import { logsAggregatesHourly, metricCatalog, spanMetricsCallsHourly, + serviceOperationsMinutely, } from "./datasources" import { DB_NAMESPACE_ATTR_SQL, @@ -27,6 +28,7 @@ import { DB_STATEMENT_SQL, DB_SYSTEM_ATTR_SQL, } from "./db-query-shape-sql" +import { NORMALIZED_SPAN_NAME_SQL } from "./span-display-name" /** * Materialized view to aggregate log usage statistics per service per hour @@ -1183,6 +1185,38 @@ export const tracesAggregatesHourlyMv = defineMaterializedView("traces_aggregate ], }) +/** + * Precomputes the operation display name and minute-grain aggregates used by + * the service detail page. All spans are included: internal operations are a + * deliberate part of the ranking, matching the previous raw query. + */ +export const serviceOperationsMinutelyMv = defineMaterializedView("service_operations_minutely_mv", { + description: + "Pre-aggregates every span by service operation and minute with normalized HTTP names, exact/estimated counts, errors, duration sum, and unweighted t-digest state.", + datasource: serviceOperationsMinutely, + nodes: [ + node({ + name: "service_operations_minutely_mv_node", + sql: ` + SELECT + OrgId, + toStartOfMinute(toDateTime(Timestamp)) AS Minute, + ServiceName, + ResourceAttributes['deployment.environment'] AS DeploymentEnv, + ${NORMALIZED_SPAN_NAME_SQL} AS SpanName, + count() AS SpanCount, + sum(SampleRate) AS EstimatedSpanCount, + countIf(StatusCode = 'Error') AS ErrorCount, + sumIf(SampleRate, StatusCode = 'Error') AS EstimatedErrorCount, + sum(toFloat64(Duration)) AS DurationSum, + quantilesTDigestState(0.5, 0.95)(Duration) AS DurationQuantiles + FROM traces + GROUP BY OrgId, Minute, ServiceName, DeploymentEnv, SpanName + `, + }), + ], +}) + /** * Populates logs_aggregates_hourly. Severity-aware drop-in for * severity-distribution and log-volume dashboards. diff --git a/packages/domain/src/tinybird/project-manifest.test.ts b/packages/domain/src/tinybird/project-manifest.test.ts index 794545068..7ae5dd6c8 100644 --- a/packages/domain/src/tinybird/project-manifest.test.ts +++ b/packages/domain/src/tinybird/project-manifest.test.ts @@ -11,6 +11,12 @@ describe("Tinybird project manifest", () => { expect(manifest.datasources.some((resource) => resource.name === "logs")).toBe(true) expect(manifest.pipes.some((resource) => resource.name === "trace_list_mv_mv")).toBe(true) + expect(manifest.datasources.some((resource) => resource.name === "service_operations_minutely")).toBe( + true, + ) + expect(manifest.pipes.some((resource) => resource.name === "service_operations_minutely_mv")).toBe( + true, + ) expect(manifest.projectRevision).toBe( createTinybirdProjectRevision(manifest.datasources, manifest.pipes), ) diff --git a/packages/domain/src/tinybird/span-display-name.ts b/packages/domain/src/tinybird/span-display-name.ts new file mode 100644 index 000000000..67c0928e4 --- /dev/null +++ b/packages/domain/src/tinybird/span-display-name.ts @@ -0,0 +1,38 @@ +import type { Expr } from "@maple-dev/clickhouse-builder/expr" +import * as CH from "@maple-dev/clickhouse-builder/expr" +import { compile } from "@maple-dev/clickhouse-builder/sql" + +/** + * Canonical operation-name expression shared by runtime queries and generated + * trace rollup SQL. HTTP server spans become `METHOD /route`; every other span, + * including internal operations, keeps its original name. + */ +export function normalizedSpanNameExpr( + spanName: Expr, + route: Expr, + urlPath: Expr, +): Expr { + return CH.if_( + spanName + .like("http.server %") + .or(spanName.in_("GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS")) + .and(route.neq("").or(urlPath.neq(""))), + CH.concat( + CH.if_(spanName.like("http.server %"), CH.replaceOne(spanName, "http.server ", ""), spanName), + CH.lit(" "), + CH.if_(route.neq(""), route, urlPath), + ), + spanName, + ) +} + +const spanAttributes = CH.dynamicColumn>("SpanAttributes") + +/** SQL text required by Tinybird materialization and ClickHouse migration DDL. */ +export const NORMALIZED_SPAN_NAME_SQL = compile( + normalizedSpanNameExpr( + CH.dynamicColumn("SpanName"), + CH.mapGet(spanAttributes, "http.route"), + CH.mapGet(spanAttributes, "url.path"), + ).toFragment(), +) diff --git a/packages/query-engine/src/ch/index.ts b/packages/query-engine/src/ch/index.ts index f0f8339bf..ba535cdaa 100644 --- a/packages/query-engine/src/ch/index.ts +++ b/packages/query-engine/src/ch/index.ts @@ -294,8 +294,10 @@ export { // Queries — Service Operations (per-SpanName breakdown for the service detail page) export { serviceOperationsSummaryQuery, + serviceOperationsSummaryRawQuery, serviceOperationsSummaryRowSchema, serviceOperationsTimeseriesQuery, + serviceOperationsTimeseriesRawQuery, serviceOperationsTimeseriesRowSchema, type ServiceOperationsSummaryOpts, type ServiceOperationsSummaryOutput, diff --git a/packages/query-engine/src/ch/queries/service-operations.test.ts b/packages/query-engine/src/ch/queries/service-operations.test.ts index 8e1910285..16e7f3fc5 100644 --- a/packages/query-engine/src/ch/queries/service-operations.test.ts +++ b/packages/query-engine/src/ch/queries/service-operations.test.ts @@ -1,10 +1,13 @@ import { describe, expect, it } from "vitest" import { Schema } from "effect" import { compileCH } from "@maple-dev/clickhouse-builder" +import { NORMALIZED_SPAN_NAME_SQL } from "@maple/domain/tinybird/span-display-name" import { serviceOperationsSummaryQuery, + serviceOperationsSummaryRawQuery, serviceOperationsSummaryRowSchema, serviceOperationsTimeseriesQuery, + serviceOperationsTimeseriesRawQuery, serviceOperationsTimeseriesRowSchema, } from "./service-operations" @@ -15,14 +18,29 @@ const baseParams = { } describe("serviceOperationsSummaryQuery", () => { - it("compiles an OrgId-scoped per-operation aggregation over raw traces", () => { + it("retains the previous raw implementations as rollout rollback paths", () => { + const summary = compileCH(serviceOperationsSummaryRawQuery({ serviceName: "api" }), baseParams) + const timeseries = compileCH( + serviceOperationsTimeseriesRawQuery({ serviceName: "api", spanNames: ["GET /users"] }), + { ...baseParams, bucketSeconds: 300 }, + ) + expect(summary.sql).toContain("FROM traces") + expect(summary.sql).not.toContain("service_operations_minutely") + expect(timeseries.sql).toContain("FROM traces") + expect(timeseries.sql).not.toContain("service_operations_minutely") + }) + + it("combines raw edge fragments with complete rollup minutes", () => { const q = serviceOperationsSummaryQuery({ serviceName: "api" }) const { sql } = compileCH(q, baseParams) expect(sql).toContain("FROM traces") + expect(sql).toContain("UNION ALL") + expect(sql).toContain("FROM service_operations_minutely") expect(sql).toContain("OrgId = 'org_1'") expect(sql).toContain("ServiceName = 'api'") expect(sql).toContain("Timestamp >= '2024-01-01 00:00:00'") - expect(sql).toContain("GROUP BY spanName") + expect(sql).toContain("Minute >= if(toDateTime('2024-01-01 00:00:00')") + expect(sql).toContain("Minute < toStartOfMinute(toDateTime('2024-01-02 00:00:00'))") expect(sql).toContain("ORDER BY estimatedSpanCount DESC") expect(sql).toContain("LIMIT 25") expect(sql).toContain("FORMAT JSON") @@ -35,25 +53,35 @@ describe("serviceOperationsSummaryQuery", () => { expect(sql).toContain("http.route") expect(sql).toContain("url.path") expect(sql).toContain("AS spanName") + expect(sql).toContain(`${NORMALIZED_SPAN_NAME_SQL} AS bSpanName`) }) - it("weights counts and error rate by SampleRate", () => { + it("merges exact, sampling-weighted, error, duration, and t-digest state", () => { const q = serviceOperationsSummaryQuery({ serviceName: "api" }) const { sql } = compileCH(q, baseParams) - expect(sql).toContain("sum(SampleRate) AS estimatedSpanCount") - expect(sql).toContain("sumIf(SampleRate, StatusCode = 'Error') AS estimatedErrorCount") - expect(sql).toContain("countIf(StatusCode = 'Error') AS errorCount") + expect(sql).toContain("sum(SampleRate) AS bEstimatedSpanCount") + expect(sql).toContain("sumIf(SampleRate, StatusCode = 'Error') AS bEstimatedErrorCount") + expect(sql).toContain("countIf(StatusCode = 'Error') AS bErrorCount") + expect(sql).toContain("quantilesTDigestState(0.5, 0.95)(Duration)") + expect(sql).toContain("quantilesTDigestMergeState(0.5, 0.95)(DurationQuantiles)") expect(sql).toContain( - "if(sum(SampleRate) > 0, sumIf(SampleRate, StatusCode = 'Error') / sum(SampleRate), 0) AS errorRate", + "if(sum(bEstimatedSpanCount) > 0, sum(bEstimatedErrorCount) / sum(bEstimatedSpanCount), 0) AS errorRate", ) - expect(sql).toContain("quantile(0.5)(Duration) / 1000000 AS p50DurationMs") - expect(sql).toContain("quantile(0.95)(Duration) / 1000000 AS p95DurationMs") + expect(sql).toContain("quantilesTDigestMerge(0.5, 0.95)(bDurationQuantiles), 1") + expect(sql).toContain("quantilesTDigestMerge(0.5, 0.95)(bDurationQuantiles), 2") }) it("applies environment filter via ResourceAttributes", () => { const q = serviceOperationsSummaryQuery({ serviceName: "api", environments: ["production"] }) const { sql } = compileCH(q, baseParams) expect(sql).toContain("ResourceAttributes['deployment.environment'] IN ('production')") + expect(sql).toContain("DeploymentEnv IN ('production')") + }) + + it("preserves internal operations by applying no SpanKind filter", () => { + const q = serviceOperationsSummaryQuery({ serviceName: "api" }) + const { sql } = compileCH(q, baseParams) + expect(sql).not.toContain("SpanKind IN") }) it("respects a custom limit", () => { @@ -61,6 +89,35 @@ describe("serviceOperationsSummaryQuery", () => { const { sql } = compileCH(q, baseParams) expect(sql).toContain("LIMIT 5") }) + + it("uses disjoint raw and rollup boundaries for partial edge minutes", () => { + const q = serviceOperationsSummaryQuery({ serviceName: "api" }) + const { sql } = compileCH(q, { + orgId: "org_1", + startTime: "2024-01-01 00:00:30", + endTime: "2024-01-01 00:02:15", + }) + expect(sql).toContain("Timestamp < if(toDateTime('2024-01-01 00:00:30') = toStartOfMinute") + expect(sql).toContain("Timestamp >= toStartOfMinute(toDateTime('2024-01-01 00:02:15'))") + expect(sql).toContain("Minute >= if(toDateTime('2024-01-01 00:00:30') = toStartOfMinute") + expect(sql).toContain("Minute < toStartOfMinute(toDateTime('2024-01-01 00:02:15'))") + }) + + it("keeps sub-minute and empty windows valid", () => { + for (const [startTime, endTime] of [ + ["2024-01-01 00:00:10", "2024-01-01 00:00:40"], + ["2024-01-01 00:00:10", "2024-01-01 00:00:10"], + ] as const) { + const { sql } = compileCH(serviceOperationsSummaryQuery({ serviceName: "api" }), { + orgId: "org_1", + startTime, + endTime, + }) + expect(sql).toContain("UNION ALL") + expect(sql).toContain(`Timestamp >= '${startTime}'`) + expect(sql).toContain(`Timestamp <= '${endTime}'`) + } + }) }) describe("serviceOperationsTimeseriesQuery", () => { @@ -68,22 +125,68 @@ describe("serviceOperationsTimeseriesQuery", () => { const q = serviceOperationsTimeseriesQuery({ serviceName: "api", spanNames: ["GET /users"] }) const { sql } = compileCH(q, { ...baseParams, bucketSeconds: 300 }) expect(sql).toContain("FROM traces") + expect(sql).toContain("FROM service_operations_minutely") + expect(sql).toContain("UNION ALL") expect(sql).toContain("toStartOfInterval(Timestamp, INTERVAL 300 SECOND)") + expect(sql).toContain("toStartOfInterval(Minute, INTERVAL 300 SECOND)") expect(sql).toContain("sum(SampleRate) AS count") + expect(sql).toContain("sum(EstimatedSpanCount) AS count") expect(sql).toContain("OrgId = 'org_1'") expect(sql).toContain("GROUP BY bucket, spanName") expect(sql).toContain("ORDER BY bucket ASC") }) - it("matches operations on either the raw or display span name", () => { + it("matches rollup rows directly on normalized SpanName", () => { const q = serviceOperationsTimeseriesQuery({ serviceName: "api", spanNames: ["GET /users"] }) const { sql } = compileCH(q, { ...baseParams, bucketSeconds: 300 }) expect(sql).toContain("SpanName IN ('GET /users')") - // The display-name rewrite must appear in the IN matcher too. + expect(sql).not.toContain("SpanName IN ('GET /users') OR") + // The display-name rewrite remains only for the two raw edge fragments. expect(sql).toContain("http.route") }) }) +describe("hybrid minute coverage", () => { + const minute = (value: string) => Math.floor(Date.parse(value) / 60_000) * 60_000 + + it("assigns every requested timestamp to exactly one branch", () => { + const start = Date.parse("2024-01-01T00:00:30Z") + const end = Date.parse("2024-01-01T00:03:15Z") + const firstFullMinute = minute(new Date(start).toISOString()) + 60_000 + const endMinute = minute(new Date(end).toISOString()) + const timestamps = [ + start, + Date.parse("2024-01-01T00:00:59Z"), + Date.parse("2024-01-01T00:01:00Z"), + Date.parse("2024-01-01T00:02:59Z"), + Date.parse("2024-01-01T00:03:00Z"), + end, + ] + + for (const timestamp of timestamps) { + const rawEdge = timestamp < firstFullMinute || timestamp >= endMinute + const rollupInterior = + minute(new Date(timestamp).toISOString()) >= firstFullMinute && + minute(new Date(timestamp).toISOString()) < endMinute + expect(Number(rawEdge) + Number(rollupInterior)).toBe(1) + } + }) + + it("keeps every timestamp raw for a sub-minute window", () => { + const start = Date.parse("2024-01-01T00:00:10Z") + const end = Date.parse("2024-01-01T00:00:40Z") + const firstFullMinute = minute(new Date(start).toISOString()) + 60_000 + const endMinute = minute(new Date(end).toISOString()) + for (const timestamp of [start, Date.parse("2024-01-01T00:00:25Z"), end]) { + expect(timestamp < firstFullMinute || timestamp >= endMinute).toBe(true) + expect( + minute(new Date(timestamp).toISOString()) >= firstFullMinute && + minute(new Date(timestamp).toISOString()) < endMinute, + ).toBe(false) + } + }) +}) + describe("row schemas (BYO-CH UInt64-as-string)", () => { it("decodes numeric strings in summary rows", () => { const decoded = Schema.decodeUnknownSync(serviceOperationsSummaryRowSchema)({ diff --git a/packages/query-engine/src/ch/queries/service-operations.ts b/packages/query-engine/src/ch/queries/service-operations.ts index e5207837d..92a63abe6 100644 --- a/packages/query-engine/src/ch/queries/service-operations.ts +++ b/packages/query-engine/src/ch/queries/service-operations.ts @@ -18,10 +18,10 @@ import { Schema } from "effect" import * as CH from "@maple-dev/clickhouse-builder/expr" import { param } from "@maple-dev/clickhouse-builder" -import { from, type ColumnAccessor } from "@maple-dev/clickhouse-builder" +import { from, fromUnion, unionAll, type ColumnAccessor } from "@maple-dev/clickhouse-builder" import { httpDisplaySpanName } from "../../traces-shared" import { CHNumber } from "../schema" -import { Traces } from "../tables" +import { ServiceOperationsMinutely, Traces } from "../tables" import { tracesBaseWhereConditions } from "./query-helpers" export interface ServiceOperationsSummaryOpts { @@ -61,7 +61,32 @@ export const serviceOperationsSummaryRowSchema = Schema.Struct({ const displaySpanName = ($: ColumnAccessor) => httpDisplaySpanName($.SpanName, $.SpanAttributes.get("http.route"), $.SpanAttributes.get("url.path")) -export function serviceOperationsSummaryQuery(opts: ServiceOperationsSummaryOpts) { +const START_DT = "toDateTime(__PARAM_startTime__)" +const END_DT = "toDateTime(__PARAM_endTime__)" +const START_MINUTE = `toStartOfMinute(${START_DT})` +const END_MINUTE = `toStartOfMinute(${END_DT})` +const FIRST_FULL_MINUTE = `if(${START_DT} = ${START_MINUTE}, ${START_MINUTE}, ${START_MINUTE} + INTERVAL 1 MINUTE)` +const RAW_EDGE_CONDITION = `(Timestamp < ${FIRST_FULL_MINUTE} OR Timestamp >= ${END_MINUTE})` +const RAW_DURATION_STATE = "quantilesTDigestState(0.5, 0.95)(Duration)" +const ROLLUP_DURATION_STATE = "quantilesTDigestMergeState(0.5, 0.95)(DurationQuantiles)" + +function rollupEnvironmentCondition( + $: ColumnAccessor, + environments: readonly string[] | undefined, +) { + return environments?.length ? CH.inList($.DeploymentEnv, environments) : undefined +} + +const mergedDurationQuantile = (index: 1 | 2) => + CH.rawExpr( + `if(sum(bSpanCount) > 0, arrayElement(quantilesTDigestMerge(0.5, 0.95)(bDurationQuantiles), ${index}) / 1000000, 0)`, + ) + +/** + * Previous all-raw implementation retained as an explicit rollout rollback + * path until managed and BYO rollup parity/latency have been observed. + */ +export function serviceOperationsSummaryRawQuery(opts: ServiceOperationsSummaryOpts) { return from(Traces) .select(($) => { const weight = CH.sum($.SampleRate) @@ -69,12 +94,9 @@ export function serviceOperationsSummaryQuery(opts: ServiceOperationsSummaryOpts return { spanName: displaySpanName($), spanCount: CH.count(), - // Per-span weighted sum: SampleRate is 1.0 for unsampled rows or - // 1/acceptanceProbability for sampled ones (see serviceOverviewQuery). estimatedSpanCount: weight, errorCount: CH.countIf($.StatusCode.eq("Error")), estimatedErrorCount: errorWeight, - // 0–1 ratio end-to-end; ×100 only at display. errorRate: CH.if_(weight.gt(0), errorWeight.div(weight), CH.lit(0)), avgDurationMs: CH.avg($.Duration).div(1_000_000), p50DurationMs: CH.quantile(0.5)($.Duration).div(1_000_000), @@ -93,6 +115,76 @@ export function serviceOperationsSummaryQuery(opts: ServiceOperationsSummaryOpts .format("JSON") } +export function serviceOperationsSummaryQuery(opts: ServiceOperationsSummaryOpts) { + const rawEdges = from(Traces) + .select(($) => ({ + bSpanName: displaySpanName($), + bSpanCount: CH.count(), + bEstimatedSpanCount: CH.sum($.SampleRate), + bErrorCount: CH.countIf($.StatusCode.eq("Error")), + bEstimatedErrorCount: CH.sumIf($.SampleRate, $.StatusCode.eq("Error")), + bDurationSum: CH.sum(CH.rawExpr("toFloat64(Duration)")), + bDurationQuantiles: CH.rawExpr(RAW_DURATION_STATE), + })) + .where(($) => [ + ...tracesBaseWhereConditions($, { + serviceName: opts.serviceName, + environments: opts.environments, + }), + CH.rawCond(RAW_EDGE_CONDITION), + ]) + .groupBy("bSpanName") + + const rollupInterior = from(ServiceOperationsMinutely) + .select(($) => ({ + bSpanName: $.SpanName, + bSpanCount: CH.sum($.SpanCount), + bEstimatedSpanCount: CH.sum($.EstimatedSpanCount), + bErrorCount: CH.sum($.ErrorCount), + bEstimatedErrorCount: CH.sum($.EstimatedErrorCount), + bDurationSum: CH.sum($.DurationSum), + bDurationQuantiles: CH.rawExpr(ROLLUP_DURATION_STATE), + })) + .where(($) => [ + $.OrgId.eq(param.string("orgId")), + $.ServiceName.eq(opts.serviceName), + rollupEnvironmentCondition($, opts.environments), + $.Minute.gte(CH.rawExpr(FIRST_FULL_MINUTE)), + $.Minute.lt(CH.rawExpr(END_MINUTE)), + ]) + .groupBy("bSpanName") + + return fromUnion(unionAll(rawEdges, rollupInterior), "operation_minutes") + .select(($) => { + const spanCount = CH.sum($.bSpanCount) + const estimatedSpanCount = CH.sum($.bEstimatedSpanCount) + const estimatedErrorCount = CH.sum($.bEstimatedErrorCount) + return { + spanName: $.bSpanName, + spanCount, + estimatedSpanCount, + errorCount: CH.sum($.bErrorCount), + estimatedErrorCount, + errorRate: CH.if_( + estimatedSpanCount.gt(0), + estimatedErrorCount.div(estimatedSpanCount), + CH.lit(0), + ), + avgDurationMs: CH.if_( + spanCount.gt(0), + CH.sum($.bDurationSum).div(spanCount).div(1_000_000), + CH.lit(0), + ), + p50DurationMs: mergedDurationQuantile(1), + p95DurationMs: mergedDurationQuantile(2), + } + }) + .groupBy("spanName") + .orderBy(["estimatedSpanCount", "desc"]) + .limit(opts.limit ?? 25) + .format("JSON") +} + export interface ServiceOperationsTimeseriesOpts { serviceName: string spanNames: readonly string[] @@ -111,14 +203,35 @@ export const serviceOperationsTimeseriesRowSchema = Schema.Struct({ count: CHNumber, }) +/** All-raw sparkline rollback companion to {@link serviceOperationsSummaryRawQuery}. */ +export function serviceOperationsTimeseriesRawQuery(opts: ServiceOperationsTimeseriesOpts) { + return from(Traces) + .select(($) => ({ + bucket: CH.toStartOfInterval($.Timestamp, param.int("bucketSeconds")), + spanName: displaySpanName($), + count: CH.sum($.SampleRate), + })) + .where(($) => [ + ...tracesBaseWhereConditions($, { + serviceName: opts.serviceName, + environments: opts.environments, + }), + CH.inList(displaySpanName($), opts.spanNames), + ]) + .groupBy("bucket", "spanName") + .orderBy(["bucket", "asc"]) + .limit(10_000) + .format("JSON") +} + /** * Sampling-weighted per-bucket counts for the operations returned by * {@link serviceOperationsSummaryQuery}. `spanNames` carries display names, so - * rows are matched on either the raw or rewritten spelling — mirroring the - * `spanName` matcher in `tracesBaseWhereConditions`. + * complete minutes match directly on the stored normalized name. Only the two + * raw edge fragments compute the display name at read time. */ export function serviceOperationsTimeseriesQuery(opts: ServiceOperationsTimeseriesOpts) { - return from(Traces) + const rawEdges = from(Traces) .select(($) => ({ bucket: CH.toStartOfInterval($.Timestamp, param.int("bucketSeconds")), spanName: displaySpanName($), @@ -129,9 +242,34 @@ export function serviceOperationsTimeseriesQuery(opts: ServiceOperationsTimeseri serviceName: opts.serviceName, environments: opts.environments, }), - CH.inList($.SpanName, opts.spanNames).or(CH.inList(displaySpanName($), opts.spanNames)), + CH.rawCond(RAW_EDGE_CONDITION), + CH.inList(displaySpanName($), opts.spanNames), ]) .groupBy("bucket", "spanName") + + const rollupInterior = from(ServiceOperationsMinutely) + .select(($) => ({ + bucket: CH.toStartOfInterval($.Minute, param.int("bucketSeconds")), + spanName: $.SpanName, + count: CH.sum($.EstimatedSpanCount), + })) + .where(($) => [ + $.OrgId.eq(param.string("orgId")), + $.ServiceName.eq(opts.serviceName), + rollupEnvironmentCondition($, opts.environments), + $.Minute.gte(CH.rawExpr(FIRST_FULL_MINUTE)), + $.Minute.lt(CH.rawExpr(END_MINUTE)), + CH.inList($.SpanName, opts.spanNames), + ]) + .groupBy("bucket", "spanName") + + return fromUnion(unionAll(rawEdges, rollupInterior), "operation_buckets") + .select(($) => ({ + bucket: $.bucket, + spanName: $.spanName, + count: CH.sum($.count), + })) + .groupBy("bucket", "spanName") .orderBy(["bucket", "asc"]) .limit(10_000) .format("JSON") diff --git a/packages/query-engine/src/ch/tables.ts b/packages/query-engine/src/ch/tables.ts index 990c1a7f9..1a317178d 100644 --- a/packages/query-engine/src/ch/tables.ts +++ b/packages/query-engine/src/ch/tables.ts @@ -322,6 +322,21 @@ export const TracesAggregatesHourly = table("traces_aggregates_hourly", { DurationMax: T.uint64, }) +export const ServiceOperationsMinutely = table("service_operations_minutely", { + OrgId: T.string, + Minute: T.dateTime, + ServiceName: T.string, + DeploymentEnv: T.string, + SpanName: T.string, + SpanCount: T.uint64, + EstimatedSpanCount: T.float64, + ErrorCount: T.uint64, + EstimatedErrorCount: T.float64, + DurationSum: T.float64, + // AggregateFunction(quantilesTDigest(0.5, 0.95), UInt64) — opaque state. + DurationQuantiles: T.string, +}) + export const LogsAggregatesHourly = table("logs_aggregates_hourly", { OrgId: T.string, Hour: T.dateTime, diff --git a/packages/query-engine/src/traces-shared.ts b/packages/query-engine/src/traces-shared.ts index 8e721e1eb..aa576068c 100644 --- a/packages/query-engine/src/traces-shared.ts +++ b/packages/query-engine/src/traces-shared.ts @@ -43,6 +43,7 @@ export const TRACE_LIST_MV_RESOURCE_MAP: Record = { // --------------------------------------------------------------------------- import * as CH from "@maple-dev/clickhouse-builder/expr" +import { normalizedSpanNameExpr } from "@maple/domain/tinybird/span-display-name" // --------------------------------------------------------------------------- // HTTP semconv coalescing @@ -94,18 +95,7 @@ export function httpDisplaySpanName( route: CH.Expr, urlPath: CH.Expr, ): CH.Expr { - return CH.if_( - spanName - .like("http.server %") - .or(spanName.in_("GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS")) - .and(route.neq("").or(urlPath.neq(""))), - CH.concat( - CH.if_(spanName.like("http.server %"), CH.replaceOne(spanName, "http.server ", ""), spanName), - CH.lit(" "), - CH.if_(route.neq(""), route, urlPath), - ), - spanName, - ) + return normalizedSpanNameExpr(spanName, route, urlPath) } export function buildAttrFilterCondition(