fix(telemetry): address OTLP export writer review findings

EhabY · EhabY · commit d8aa4677091e · 2026-05-26T23:37:42.000+03:00
- Stamp coder.event.{extension_version,session_id,deployment_url} on
  every log/span/metric record so multi-session and multi-deployment
  exports preserve the original producer's identity. The resource block
  remains an export-tool snapshot.
- Stage envelopes in os.tmpdir() instead of next to the user's save
  path so cloud-sync agents do not ingest the intermediate uncompressed
  logs.json/traces.json/metrics.json.
- Replace the in-memory fflate.zip with streaming Zip + ZipDeflate so
  multi-GB exports do not hold every envelope plus the zipped output in
  the V8 heap at once.
- Forward staging-cleanup failures via
  OtlpWriteOptions.onStagingCleanupError instead of letting a Windows
  EBUSY in finally mask an otherwise-successful export (which
  writeAtomically would then erase via temp-file cleanup).
- Increment metric channel counts per event rather than per non-empty
  record batch so metric events with all-zero counters stay in the
  user-visible total and JSON/OTLP exports of the same range agree.
- Accept an optional AbortSignal so callers can cancel a long-running
  export between events and between files.
diff --git a/src/telemetry/export/writers/otlp/records.ts b/src/telemetry/export/writers/otlp/records.ts
@@ -69,6 +69,11 @@ export function newCumulativeState(): CumulativeState {
 	return { anchor: undefined, totals: new Map() };
 }
 
+/**
+ * Resource attributes describe the export tool (forwarder), not the original
+ * producer. Per-event identity is stamped on each record by
+ * `eventContextAttributes` so multi-session exports preserve provenance.
+ */
 export function otlpResource(context: TelemetryContext) {
 	return {
 		attributes: keyValues({
@@ -86,6 +91,18 @@ export function otlpResource(context: TelemetryContext) {
 	};
 }
 
+/**
+ * Per-event identity stamped on every record so multi-session exports stay
+ * attributable even when the resource snapshot is from the export tool.
+ */
+function eventContextAttributes(event: TelemetryEvent): Record<string, string> {
+	return {
+		"coder.event.extension_version": event.context.extensionVersion,
+		"coder.event.session_id": event.context.sessionId,
+		"coder.event.deployment_url": event.context.deploymentUrl,
+	};
+}
+
 export function otlpScope(version: string) {
 	return { name: "coder.vscode-coder.telemetry.export", version };
 }
@@ -100,6 +117,7 @@ export function logRecord(event: TelemetryEvent): OtlpLogRecord {
 		severityText: errored ? "ERROR" : "INFO",
 		body: { stringValue: event.eventName },
 		attributes: keyValues({
+			...eventContextAttributes(event),
 			...event.properties,
 			...event.measurements,
 			...(event.error && exceptionAttributes(event.error)),
@@ -126,6 +144,7 @@ export function spanRecord(
 		endTimeUnixNano,
 		attributes: keyValues({
 			"coder.event_name": event.eventName,
+			...eventContextAttributes(event),
 			...event.properties,
 			...measurements,
 		}),
@@ -185,6 +204,7 @@ export function metricRecords(
 		properties: event.properties,
 		attributes: keyValues({
 			"coder.event_name": event.eventName,
+			...eventContextAttributes(event),
 			...event.properties,
 		}),
 		timeNano,
diff --git a/src/telemetry/export/writers/otlp/writer.ts b/src/telemetry/export/writers/otlp/writer.ts
@@ -1,9 +1,10 @@
-import { zip } from "fflate";
+import { Zip, ZipDeflate } from "fflate";
+import { createReadStream, createWriteStream } from "node:fs";
 import * as fs from "node:fs/promises";
+import * as os from "node:os";
 import * as path from "node:path";
-import { promisify } from "node:util";
 
-import { wrapError } from "../../../../error/errorUtils";
+import { toError, wrapError } from "../../../../error/errorUtils";
 import { writeAtomically } from "../../../../util/fs";
 import { describeMetricEvent } from "../../metrics";
 
@@ -30,43 +31,75 @@ export interface OtlpExportCounts {
 	readonly metrics: number;
 }
 
+export interface OtlpWriteOptions {
+	readonly signal?: AbortSignal;
+	/**
+	 * Invoked when removing the staging directory fails after the export
+	 * itself succeeded. Letting the cleanup throw would mask success.
+	 */
+	readonly onStagingCleanupError?: (err: unknown, dir: string) => void;
+}
+
 interface Channel {
 	file: EnvelopeFile;
 	count: number;
 }
 
-const zipAsync = promisify(zip);
-
 /**
  * Writes `events` as an OTLP/JSON zip (`logs.json`, `traces.json`,
- * `metrics.json`) to `outputPath`.
+ * `metrics.json`) to `outputPath`. Staging happens in the OS temp dir so
+ * cloud-sync agents on the user's chosen save location never see the
+ * intermediate uncompressed envelopes.
  */
 export async function writeOtlpZipExport(
 	outputPath: string,
 	events: AsyncIterable<TelemetryEvent>,
 	context: TelemetryContext,
 	onCleanupError: (err: unknown, tempPath: string) => void,
+	options: OtlpWriteOptions = {},
 ): Promise<OtlpExportCounts> {
 	return writeAtomically(
 		outputPath,
 		async (zipPath) => {
-			const stagingDir = await fs.mkdtemp(`${outputPath}.staging-`);
+			const stagingDir = await fs.mkdtemp(
+				path.join(os.tmpdir(), "coder-telemetry-otlp-"),
+			);
+			let counts: OtlpExportCounts;
 			try {
-				const counts = await writeStagedFiles(stagingDir, events, context);
-				await packZip(zipPath, stagingDir);
-				return counts;
-			} finally {
-				await fs.rm(stagingDir, { recursive: true, force: true });
+				counts = await writeStagedFiles(
+					stagingDir,
+					events,
+					context,
+					options.signal,
+				);
+				await packZip(zipPath, stagingDir, options.signal);
+			} catch (err) {
+				await safeRemove(stagingDir, options.onStagingCleanupError);
+				throw err;
 			}
+			await safeRemove(stagingDir, options.onStagingCleanupError);
+			return counts;
 		},
 		onCleanupError,
 	);
 }
 
+async function safeRemove(
+	dir: string,
+	onError?: (err: unknown, dir: string) => void,
+): Promise<void> {
+	try {
+		await fs.rm(dir, { recursive: true, force: true });
+	} catch (err) {
+		onError?.(err, dir);
+	}
+}
+
 async function writeStagedFiles(
 	dir: string,
 	events: AsyncIterable<TelemetryEvent>,
 	context: TelemetryContext,
+	signal: AbortSignal | undefined,
 ): Promise<OtlpExportCounts> {
 	const resource = JSON.stringify(otlpResource(context));
 	const scope = JSON.stringify(otlpScope(context.extensionVersion));
@@ -76,6 +109,7 @@ async function writeStagedFiles(
 	let succeeded = false;
 	try {
 		for await (const event of events) {
+			throwIfAborted(signal);
 			await routeEvent(event, channels, state);
 		}
 		succeeded = true;
@@ -142,14 +176,20 @@ async function routeEvent(
 	try {
 		const metric = describeMetricEvent(event);
 		if (metric) {
+			// Count metric events even when every measurement is suppressed (e.g.
+			// http.requests window with all counters at zero); otherwise JSON and
+			// OTLP exports of the same range report different event totals.
 			await appendRecords(
 				channels.metrics,
 				metricRecords(event, metric, state),
 			);
+			channels.metrics.count += 1;
 		} else if (hasTraceId(event)) {
 			await appendRecords(channels.traces, [spanRecord(event)]);
+			channels.traces.count += 1;
 		} else {
 			await appendRecords(channels.logs, [logRecord(event)]);
+			channels.logs.count += 1;
 		}
 	} catch (err) {
 		throw wrapError(
@@ -164,29 +204,98 @@ async function appendRecords(
 	channel: Channel,
 	records: Iterable<unknown>,
 ): Promise<void> {
-	let wrote = false;
 	for (const record of records) {
 		await channel.file.append(record);
-		wrote = true;
-	}
-	if (wrote) {
-		channel.count += 1;
 	}
 }
 
-async function packZip(outputPath: string, sourceDir: string): Promise<void> {
+/**
+ * Streams each staged envelope file into a deflate-compressed zip without
+ * holding the file contents in memory. Honors `signal` between files; the
+ * inner read pump only checks before pulling the next chunk.
+ */
+async function packZip(
+	outputPath: string,
+	sourceDir: string,
+	signal: AbortSignal | undefined,
+): Promise<void> {
+	const outStream = createWriteStream(outputPath);
 	try {
-		const entries = await Promise.all(
-			Object.values(ENVELOPES).map(
-				async (envelope) =>
-					[
-						envelope.file,
-						await fs.readFile(path.join(sourceDir, envelope.file)),
-					] as const,
-			),
-		);
-		await fs.writeFile(outputPath, await zipAsync(Object.fromEntries(entries)));
+		await new Promise<void>((resolve, reject) => {
+			let zipEnded = false;
+			let pendingWrites = 0;
+			const onError = (err: unknown) => reject(toError(err));
+
+			outStream.once("error", onError);
+
+			const zip = new Zip((err, chunk, final) => {
+				if (err) {
+					onError(err);
+					return;
+				}
+				pendingWrites += 1;
+				outStream.write(chunk, (writeErr) => {
+					pendingWrites -= 1;
+					if (writeErr) {
+						onError(writeErr);
+						return;
+					}
+					if (final && zipEnded && pendingWrites === 0) {
+						outStream.end(() => resolve());
+					}
+				});
+			});
+
+			void (async () => {
+				try {
+					for (const envelope of Object.values(ENVELOPES)) {
+						throwIfAborted(signal);
+						await streamFileIntoZip(
+							zip,
+							envelope.file,
+							path.join(sourceDir, envelope.file),
+							signal,
+						);
+					}
+					zip.end();
+					zipEnded = true;
+				} catch (err) {
+					zip.terminate();
+					reject(toError(err));
+				}
+			})();
+		});
 	} catch (err) {
+		outStream.destroy();
 		throw wrapError("pack OTLP zip", path.basename(outputPath), err);
 	}
 }
+
+async function streamFileIntoZip(
+	zip: Zip,
+	name: string,
+	filePath: string,
+	signal: AbortSignal | undefined,
+): Promise<void> {
+	const entry = new ZipDeflate(name);
+	zip.add(entry);
+	const readStream = createReadStream(filePath);
+	try {
+		for await (const chunk of readStream) {
+			throwIfAborted(signal);
+			entry.push(chunk as Uint8Array, false);
+		}
+		entry.push(new Uint8Array(0), true);
+	} finally {
+		readStream.destroy();
+	}
+}
+
+function throwIfAborted(signal: AbortSignal | undefined): void {
+	if (signal?.aborted) {
+		const reason: unknown = signal.reason;
+		throw reason instanceof Error
+			? reason
+			: Object.assign(new Error("Aborted"), { name: "AbortError" });
+	}
+}
diff --git a/test/unit/telemetry/export/writers/otlp/records.test.ts b/test/unit/telemetry/export/writers/otlp/records.test.ts
@@ -20,6 +20,13 @@ import { TRACE_ID, attrs } from "./helpers";
 
 const makeEvent = createTelemetryEventFactory();
 
+/** Per-event identity that every record carries, derived from the factory defaults. */
+const EVENT_CONTEXT_ATTRS = {
+	"coder.event.extension_version": "1.14.5",
+	"coder.event.session_id": "session-id",
+	"coder.event.deployment_url": "https://coder.example.com",
+} as const;
+
 const makeSpanEvent = (overrides: Parameters<typeof makeEvent>[0] = {}) => ({
 	...makeEvent({ traceId: TRACE_ID, ...overrides }),
 	traceId: TRACE_ID,
@@ -80,7 +87,11 @@ describe("logRecord", () => {
 			body: { stringValue: "log.info" },
 		});
 		expect(record.timeUnixNano).toBe(record.observedTimeUnixNano);
-		expect(attrs(record.attributes)).toEqual({ source: "unit", count: 3 });
+		expect(attrs(record.attributes)).toEqual({
+			...EVENT_CONTEXT_ATTRS,
+			source: "unit",
+			count: 3,
+		});
 	});
 
 	it("emits ERROR records and omits optional exception fields when unset", () => {
@@ -97,7 +108,29 @@ describe("logRecord", () => {
 			"exception.type": "RangeError",
 			"exception.code": "E_RANGE",
 		});
-		expect(attrs(minimal.attributes)).toEqual({ "exception.message": "boom" });
+		expect(attrs(minimal.attributes)).toEqual({
+			...EVENT_CONTEXT_ATTRS,
+			"exception.message": "boom",
+		});
+	});
+
+	it("stamps each event's own context onto its record", () => {
+		const base = makeEvent();
+		const record = logRecord({
+			...base,
+			context: {
+				...base.context,
+				extensionVersion: "0.9.0",
+				sessionId: "older-session",
+				deploymentUrl: "https://prev.coder.example.com",
+			},
+		});
+
+		expect(attrs(record.attributes)).toMatchObject({
+			"coder.event.extension_version": "0.9.0",
+			"coder.event.session_id": "older-session",
+			"coder.event.deployment_url": "https://prev.coder.example.com",
+		});
 	});
 });
 
@@ -124,6 +157,7 @@ describe("spanRecord", () => {
 		);
 		expect(attrs(span.attributes)).toEqual({
 			"coder.event_name": "remote.setup.workspace_ready",
+			...EVENT_CONTEXT_ATTRS,
 			result: "success",
 			route: "/api",
 			retries: 2,
diff --git a/test/unit/telemetry/export/writers/otlp/writer.test.ts b/test/unit/telemetry/export/writers/otlp/writer.test.ts
