fix(telemetry): make OTLP export ingestable by Prom-family backends

EhabY · EhabY · commit 7ce5f222964f · 2026-05-21T13:10:12.000+03:00
Wire-format fixes:
- Switch http.requests count_* sums from DELTA to CUMULATIVE temporality
  (Prom/Mimir/Grafana Cloud reject the delta+sum combination).
- Maintain per-series running totals across the export, anchored at the
  first observed window's startTimeUnixNano.
- Suppress zero-cumulative counters so routes that never errored don't
  ship empty data points.

Semantic-convention compliance:
- Use OTel-standard `service.instance.id` (was `coder.session.id`) and
  `host.id` (was `coder.machine.id`) for portability with OTel-aware
  backends. The previous vendor keys were pure aliases.
- Add `schemaUrl` to each ResourceLogs/ResourceSpans/ResourceMetrics
  container and each scope container.
- Stamp scope.version with the extension version.
diff --git a/src/telemetry/export/writers/otlp.ts b/src/telemetry/export/writers/otlp.ts
@@ -29,7 +29,8 @@ export interface OtlpExportCounts {
 // OTLP proto SpanKind reserves 0 for UNSPECIFIED, so api values shift by 1
 // on the wire. AGGREGATION_TEMPORALITY has no enum in @opentelemetry/api.
 const otlpSpanKind = (kind: SpanKind): number => kind + 1;
-const AGGREGATION_TEMPORALITY_DELTA = 1;
+const AGGREGATION_TEMPORALITY_CUMULATIVE = 2;
+const OTLP_SCHEMA_URL = "https://opentelemetry.io/schemas/1.24.0";
 const zipAsync = promisify(zip);
 
 /**
@@ -60,6 +61,14 @@ export async function writeOtlpZipExport(
 	});
 }
 
+/** Per-export state threaded through the metric mapper for cumulative sums. */
+interface ExportState {
+	readonly cumulative: {
+		startTimeUnixNano: string | undefined;
+		readonly totals: Map<string, bigint>;
+	};
+}
+
 // Per-signal layout driving file names, envelope JSON keys, routing, and counts.
 const ENVELOPES = [
 	{
@@ -69,7 +78,7 @@ const ENVELOPES = [
 		resourceKey: "resourceLogs",
 		scopeKey: "scopeLogs",
 		recordsKey: "logRecords",
-		toRecords: (e: TelemetryEvent) => [toOtlpLogRecord(e)],
+		toRecords: (e: TelemetryEvent, _state: ExportState) => [toOtlpLogRecord(e)],
 	},
 	{
 		signal: "trace",
@@ -78,7 +87,7 @@ const ENVELOPES = [
 		resourceKey: "resourceSpans",
 		scopeKey: "scopeSpans",
 		recordsKey: "spans",
-		toRecords: (e: TelemetryEvent) => [toOtlpSpan(e)],
+		toRecords: (e: TelemetryEvent, _state: ExportState) => [toOtlpSpan(e)],
 	},
 	{
 		signal: "metric",
@@ -96,7 +105,7 @@ const ENVELOPES = [
 	resourceKey: string;
 	scopeKey: string;
 	recordsKey: string;
-	toRecords: (event: TelemetryEvent) => readonly unknown[];
+	toRecords: (event: TelemetryEvent, state: ExportState) => readonly unknown[];
 }>;
 
 async function writeOtlpJsonFiles(
@@ -105,17 +114,20 @@ async function writeOtlpJsonFiles(
 	context: TelemetryContext,
 ): Promise<OtlpExportCounts> {
 	const resource = JSON.stringify(toOtlpResource(context));
-	const scope = JSON.stringify(toOtlpScope());
+	const scope = JSON.stringify(toOtlpScope(context.extensionVersion));
 	const entries = await Promise.all(
 		ENVELOPES.map(async (envelope) => ({
 			...envelope,
 			writer: await openEnvelope(
 				path.join(dir, envelope.file),
-				`{"${envelope.resourceKey}":[{"resource":${resource},"${envelope.scopeKey}":[{"scope":${scope},"${envelope.recordsKey}":[`,
+				`{"${envelope.resourceKey}":[{"resource":${resource},"schemaUrl":"${OTLP_SCHEMA_URL}","${envelope.scopeKey}":[{"scope":${scope},"schemaUrl":"${OTLP_SCHEMA_URL}","${envelope.recordsKey}":[`,
 				"]}]}]}\n",
 			),
 		})),
 	);
+	const state: ExportState = {
+		cumulative: { startTimeUnixNano: undefined, totals: new Map() },
+	};
 	const counts: Record<keyof OtlpExportCounts, number> = {
 		logs: 0,
 		traces: 0,
@@ -130,7 +142,7 @@ async function writeOtlpJsonFiles(
 				continue;
 			}
 			counts[entry.counter] += 1;
-			for (const record of entry.toRecords(event)) {
+			for (const record of entry.toRecords(event, state)) {
 				await entry.writer.write(record);
 			}
 		}
@@ -203,11 +215,11 @@ function toOtlpResource(context: TelemetryContext): JsonObject {
 		attributes: keyValues({
 			"service.name": "coder-vscode-extension",
 			"service.version": context.extensionVersion,
-			"coder.machine.id": context.machineId,
-			"coder.session.id": context.sessionId,
+			"service.instance.id": context.sessionId,
+			"host.id": context.machineId,
+			"host.arch": context.hostArch,
 			"os.type": context.osType,
 			"os.version": context.osVersion,
-			"host.arch": context.hostArch,
 			"vscode.platform.name": context.platformName,
 			"vscode.platform.version": context.platformVersion,
 			"coder.deployment.url": context.deploymentUrl,
@@ -216,8 +228,8 @@ function toOtlpResource(context: TelemetryContext): JsonObject {
 }
 
 /** OTLP `InstrumentationScope` shared by every record. */
-function toOtlpScope(): JsonObject {
-	return { name: "coder.vscode-coder.telemetry.export" };
+function toOtlpScope(version: string): JsonObject {
+	return { name: "coder.vscode-coder.telemetry.export", version };
 }
 
 function toOtlpLogRecord(event: TelemetryEvent): JsonObject {
@@ -285,11 +297,14 @@ function spanStatus(event: TelemetryEvent): JsonObject {
 }
 
 /** A metric event yields one record per measurement (http.requests fans out). */
-function toOtlpMetricRecords(event: TelemetryEvent): JsonObject[] {
+function toOtlpMetricRecords(
+	event: TelemetryEvent,
+	state: ExportState,
+): JsonObject[] {
 	const timeUnixNano = toUnixNano(event.timestamp);
 	const attributes = metricAttributes(event);
 	if (event.eventName === "http.requests") {
-		return toHttpRequestMetrics(event, timeUnixNano, attributes);
+		return toHttpRequestMetrics(event, timeUnixNano, attributes, state);
 	}
 	return toGaugeMetrics(
 		event,
@@ -303,6 +318,7 @@ function toHttpRequestMetrics(
 	event: TelemetryEvent,
 	timeUnixNano: string,
 	attributes: JsonObject[],
+	state: ExportState,
 ): JsonObject[] {
 	const counts: Array<[string, number]> = [];
 	const gauges: Array<[string, number]> = [];
@@ -316,38 +332,64 @@ function toHttpRequestMetrics(
 			gauges.push([name, value]);
 		}
 	}
-	const startTimeUnixNano = String(
+	const windowStartUnixNano = String(
 		BigInt(timeUnixNano) - nanosFromSeconds(windowSeconds ?? 0),
 	);
-
-	return [
-		...counts.map(([name, value]) => ({
+	// Cumulative series anchor: first observed window's start, stable across
+	// every subsequent http.requests data point in this export.
+	state.cumulative.startTimeUnixNano ??= windowStartUnixNano;
+	const cumulativeStart = state.cumulative.startTimeUnixNano;
+
+	const sumRecords: JsonObject[] = [];
+	for (const [name, value] of counts) {
+		const seriesKey = `${name}|${stableSeriesKey(event.properties)}`;
+		const total =
+			(state.cumulative.totals.get(seriesKey) ?? 0n) +
+			BigInt(Math.trunc(value));
+		state.cumulative.totals.set(seriesKey, total);
+		// Suppress unchanged-from-zero counters to save bytes; receivers
+		// interpret an absent series as "no events yet".
+		if (total === 0n) {
+			continue;
+		}
+		sumRecords.push({
 			name: `${event.eventName}.${name}`,
 			description: event.eventName,
 			unit: "{request}",
 			sum: {
-				aggregationTemporality: AGGREGATION_TEMPORALITY_DELTA,
+				aggregationTemporality: AGGREGATION_TEMPORALITY_CUMULATIVE,
 				isMonotonic: true,
 				dataPoints: [
 					{
 						attributes,
-						startTimeUnixNano,
+						startTimeUnixNano: cumulativeStart,
 						timeUnixNano,
-						asInt: String(Math.trunc(value)),
+						asInt: String(total),
 					},
 				],
 			},
-		})),
+		});
+	}
+
+	return [
+		...sumRecords,
 		...toGaugeMetrics(
 			event,
 			gauges,
 			attributes,
 			timeUnixNano,
-			startTimeUnixNano,
+			windowStartUnixNano,
 		),
 	];
 }
 
+function stableSeriesKey(properties: Readonly<Record<string, string>>): string {
+	return Object.entries(properties)
+		.sort(([a], [b]) => a.localeCompare(b))
+		.map(([k, v]) => `${k}=${v}`)
+		.join("|");
+}
+
 function toGaugeMetrics(
 	event: TelemetryEvent,
 	measurements: Array<[string, number]>,
diff --git a/test/unit/telemetry/export/writers/otlp.test.ts b/test/unit/telemetry/export/writers/otlp.test.ts
@@ -52,12 +52,19 @@ interface TracesFile {
 	resourceSpans: [{ scopeSpans: [{ spans: RawRecord[] }] }];
 }
 interface MetricsFile {
-	resourceMetrics: [{ scopeMetrics: [{ metrics: RawRecord[] }] }];
+	resourceMetrics: [
+		{
+			resource: { attributes: unknown };
+			scopeMetrics: [{ scope: RawRecord; metrics: RawRecord[] }];
+		},
+	];
 }
 
 interface Captured {
 	counts: Awaited<ReturnType<typeof writeOtlpZipExport>>;
 	logResource: { attributes: unknown };
+	metricsResource: { attributes: unknown };
+	logsScope: RawRecord;
 	logs: RawRecord[];
 	spans: RawRecord[];
 	metrics: RawRecord[];
@@ -77,27 +84,46 @@ async function capture(events: readonly TelemetryEvent[]): Promise<Captured> {
 	return {
 		counts,
 		logResource: logs.resource,
+		metricsResource: metrics.resource,
+		logsScope: (logs.scopeLogs[0] as unknown as { scope: RawRecord }).scope,
 		logs: logs.scopeLogs[0].logRecords,
 		spans: traces.scopeSpans[0].spans,
 		metrics: metrics.scopeMetrics[0].metrics,
 	};
 }
 
 describe("writeOtlpZipExport: resource", () => {
-	it("emits service, OS, host, vscode, and coder attributes from the context", async () => {
-		const { logResource } = await capture([makeEvent()]);
-
-		expect(attrs(logResource.attributes)).toEqual({
-			"service.name": "coder-vscode-extension",
-			"service.version": "1.14.5",
-			"coder.machine.id": "machine-id",
-			"coder.session.id": "session-id",
-			"os.type": "linux",
-			"os.version": "6.0.0",
-			"host.arch": "x64",
-			"vscode.platform.name": "Visual Studio Code",
-			"vscode.platform.version": "1.106.0",
-			"coder.deployment.url": "https://coder.example.com",
+	const EXPECTED_RESOURCE = {
+		"service.name": "coder-vscode-extension",
+		"service.version": "1.14.5",
+		"service.instance.id": "session-id",
+		"host.id": "machine-id",
+		"host.arch": "x64",
+		"os.type": "linux",
+		"os.version": "6.0.0",
+		"vscode.platform.name": "Visual Studio Code",
+		"vscode.platform.version": "1.106.0",
+		"coder.deployment.url": "https://coder.example.com",
+	};
+
+	it("uses OTel-standard semconv keys on the resource", async () => {
+		const { logResource, metricsResource } = await capture([
+			makeEvent(),
+			makeEvent({
+				eventName: "ssh.network.sampled",
+				measurements: { latencyMs: 1 },
+			}),
+		]);
+
+		expect(attrs(logResource.attributes)).toEqual(EXPECTED_RESOURCE);
+		expect(attrs(metricsResource.attributes)).toEqual(EXPECTED_RESOURCE);
+	});
+
+	it("stamps each scope with the extension version", async () => {
+		const { logsScope } = await capture([makeEvent()]);
+		expect(logsScope).toMatchObject({
+			name: "coder.vscode-coder.telemetry.export",
+			version: "1.14.5",
 		});
 	});
 });
@@ -237,40 +263,83 @@ describe("writeOtlpZipExport: metrics", () => {
 		});
 	});
 
-	it("splits http.requests into monotonic count sums and gauges sharing the window", async () => {
+	it("emits http.requests counts as cumulative monotonic sums with a stable start time", async () => {
+		const { metrics } = await capture([
+			makeEvent({
+				eventName: "http.requests",
+				properties: { method: "GET", route: "/a" },
+				timestamp: "2026-05-04T12:01:00.000Z",
+				measurements: { window_seconds: 60, count_2xx: 2 },
+			}),
+			makeEvent({
+				eventName: "http.requests",
+				properties: { method: "GET", route: "/a" },
+				timestamp: "2026-05-04T12:02:00.000Z",
+				measurements: { window_seconds: 60, count_2xx: 3 },
+			}),
+		]);
+		const counts = metrics.map(
+			(m) =>
+				m.sum as {
+					aggregationTemporality: number;
+					isMonotonic: boolean;
+					dataPoints: [
+						{
+							asInt: string;
+							startTimeUnixNano: string;
+							timeUnixNano: string;
+						},
+					];
+				},
+		);
+
+		expect(counts.map((c) => c.dataPoints[0].asInt)).toEqual(["2", "5"]);
+		expect(counts.every((c) => c.aggregationTemporality === 2)).toBe(true);
+		expect(counts.every((c) => c.isMonotonic)).toBe(true);
+		// startTimeUnixNano is set once on the first event and stays fixed.
+		expect(counts[1].dataPoints[0].startTimeUnixNano).toBe(
+			counts[0].dataPoints[0].startTimeUnixNano,
+		);
+	});
+
+	it("keeps gauges windowed alongside the cumulative count sums", async () => {
 		const { metrics } = await capture([
 			makeEvent({
 				eventName: "http.requests",
 				measurements: { window_seconds: 60, count_2xx: 2, p95_duration_ms: 42 },
 			}),
 		]);
-		const count = metrics[0].sum as {
-			aggregationTemporality: number;
-			isMonotonic: boolean;
-			dataPoints: [
-				{ asInt: string; startTimeUnixNano: string; timeUnixNano: string },
-			];
-		};
-		const p95Point = (
-			metrics[1].gauge as { dataPoints: [{ startTimeUnixNano: string }] }
-		).dataPoints[0];
 
 		expect(metrics.map((m) => [m.name, m.unit])).toEqual([
 			["http.requests.count_2xx", "{request}"],
 			["http.requests.p95_duration_ms", "ms"],
 		]);
-		expect(count).toMatchObject({
-			aggregationTemporality: 1,
-			isMonotonic: true,
-		});
-		expect(count.dataPoints[0].asInt).toBe("2");
+		const p95Point = (
+			metrics[1].gauge as {
+				dataPoints: [{ startTimeUnixNano: string; timeUnixNano: string }];
+			}
+		).dataPoints[0];
 		expect(
-			BigInt(count.dataPoints[0].timeUnixNano) -
-				BigInt(count.dataPoints[0].startTimeUnixNano),
+			BigInt(p95Point.timeUnixNano) - BigInt(p95Point.startTimeUnixNano),
 		).toBe(60_000_000_000n);
-		expect(p95Point.startTimeUnixNano).toBe(
-			count.dataPoints[0].startTimeUnixNano,
-		);
+	});
+
+	it("suppresses zero-valued cumulative counters", async () => {
+		const { metrics } = await capture([
+			makeEvent({
+				eventName: "http.requests",
+				measurements: {
+					window_seconds: 60,
+					count_2xx: 0,
+					count_5xx: 0,
+					p95_duration_ms: 10,
+				},
+			}),
+		]);
+
+		expect(metrics.map((m) => m.name)).toEqual([
+			"http.requests.p95_duration_ms",
+		]);
 	});
 
 	it("treats http.requests without window_seconds as a zero-width window", async () => {
