fix(telemetry): address OTLP export writer review findings

- 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.

Author: EhabY

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,19 @@ export function otlpResource(context: TelemetryContext) {
 	};
 }
 
+/**
+ * Per-event identity stamped on every record so multi-session exports stay
+ * attributable. Spread LAST in attribute maps so caller-supplied properties
+ * keyed `coder.event.*` cannot override the canonical provenance.
+ */
+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 };
 }
@@ -103,6 +121,7 @@ export function logRecord(event: TelemetryEvent): OtlpLogRecord {
 			...event.properties,
 			...event.measurements,
 			...(event.error && exceptionAttributes(event.error)),
+			...eventContextAttributes(event),
 		}),
 	};
 }
@@ -125,9 +144,10 @@ export function spanRecord(
 		startTimeUnixNano: String(startNano),
 		endTimeUnixNano,
 		attributes: keyValues({
-			"coder.event_name": event.eventName,
 			...event.properties,
 			...measurements,
+			"coder.event_name": event.eventName,
+			...eventContextAttributes(event),
 		}),
 		status: spanStatus(event),
 		...(event.error && {
@@ -184,8 +204,9 @@ export function metricRecords(
 		eventName: event.eventName,
 		properties: event.properties,
 		attributes: keyValues({
-			"coder.event_name": event.eventName,
 			...event.properties,
+			"coder.event_name": event.eventName,
+			...eventContextAttributes(event),
 		}),
 		timeNano,
 		windowStartNano,

src/telemetry/export/writers/otlp/writer.ts

@@ -1,9 +1,10 @@
-import { zip } from "fflate";
+import { Zip, ZipDeflate } from "fflate";
+import { createReadStream, createWriteStream, type WriteStream } 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 { isAbortError, toError, wrapError } from "../../../../error/errorUtils";
 import { writeAtomically } from "../../../../util/fs";
 import { describeMetricEvent } from "../../metrics";
 
@@ -24,49 +25,90 @@ import {
 
 import type { TelemetryContext, TelemetryEvent } from "../../../event";
 
+/** Event totals by signal — a metric event with all records suppressed still counts as one. */
 export interface OtlpExportCounts {
 	readonly logs: number;
 	readonly traces: number;
 	readonly metrics: number;
 }
 
+export interface OtlpWriteOptions {
+	readonly signal?: AbortSignal;
+	readonly onTempCleanupError?: (err: unknown, tempPath: string) => void;
+	/** Fires on either success or failure path so cleanup errors never mask the export outcome. */
+	readonly onStagingCleanupError?: (err: unknown, dir: string) => void;
+}
+
 interface Channel {
 	file: EnvelopeFile;
 	count: number;
 }
 
-const zipAsync = promisify(zip);
+const READ_HWM_BYTES = 256 * 1024;
 
 /**
  * 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> {
+	throwIfAborted(options.signal);
 	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,
+		options.onTempCleanupError ?? swallowCleanupError,
 	);
 }
 
+function swallowCleanupError(): void {
+	/* Default: temp-cleanup errors from writeAtomically are silently dropped. */
+}
+
+async function safeRemove(
+	dir: string,
+	onError?: (err: unknown, dir: string) => void,
+): Promise<void> {
+	try {
+		await fs.rm(dir, { recursive: true, force: true });
+	} catch (err) {
+		try {
+			onError?.(err, dir);
+		} catch {
+			// Swallow callback throws so they don't displace the export error.
+		}
+	}
+}
+
 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 +118,7 @@ async function writeStagedFiles(
 	let succeeded = false;
 	try {
 		for await (const event of events) {
+			throwIfAborted(signal);
 			await routeEvent(event, channels, state);
 		}
 		succeeded = true;
@@ -146,10 +189,13 @@ async function routeEvent(
 				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 +210,118 @@ 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 the staged envelopes into a deflate-compressed zip; AbortError is rethrown unwrapped. */
+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 streamEnvelopesIntoZip(outStream, sourceDir, signal);
 	} catch (err) {
+		outStream.destroy();
+		if (isAbortError(err)) {
+			throw err;
+		}
 		throw wrapError("pack OTLP zip", path.basename(outputPath), err);
 	}
 }
+
+/** Bridges fflate's Zip callback onto `outStream`; the pump awaits 'drain' on backpressure. */
+function streamEnvelopesIntoZip(
+	outStream: WriteStream,
+	sourceDir: string,
+	signal: AbortSignal | undefined,
+): Promise<void> {
+	return new Promise<void>((resolve, reject) => {
+		const fail = (err: unknown): void => reject(toError(err));
+
+		const waitForDrain = (): Promise<void> =>
+			outStream.writableNeedDrain
+				? new Promise<void>((r) => outStream.once("drain", r))
+				: Promise.resolve();
+
+		outStream.on("error", fail);
+
+		const zip = new Zip((err, chunk, final) => {
+			if (err) {
+				fail(err);
+				return;
+			}
+			if (final) {
+				// end() waits for in-flight writes before 'finish'; no pendingWrites counter needed.
+				outStream.end(chunk, () => resolve());
+			} else {
+				outStream.write(chunk, (writeErr) => {
+					if (writeErr) {
+						fail(writeErr);
+					}
+				});
+			}
+		});
+
+		void pumpEnvelopes(zip, sourceDir, signal, waitForDrain).catch((err) => {
+			zip.terminate();
+			fail(err);
+		});
+	});
+}
+
+async function pumpEnvelopes(
+	zip: Zip,
+	sourceDir: string,
+	signal: AbortSignal | undefined,
+	waitForDrain: () => Promise<void>,
+): Promise<void> {
+	for (const envelope of Object.values(ENVELOPES)) {
+		throwIfAborted(signal);
+		await streamFileIntoZip(
+			zip,
+			envelope.file,
+			path.join(sourceDir, envelope.file),
+			signal,
+			waitForDrain,
+		);
+	}
+	zip.end();
+}
+
+async function streamFileIntoZip(
+	zip: Zip,
+	name: string,
+	filePath: string,
+	signal: AbortSignal | undefined,
+	waitForDrain: () => Promise<void>,
+): Promise<void> {
+	const entry = new ZipDeflate(name);
+	zip.add(entry);
+	const readStream = createReadStream(filePath, {
+		highWaterMark: READ_HWM_BYTES,
+	});
+	try {
+		for await (const chunk of readStream) {
+			throwIfAborted(signal);
+			entry.push(chunk as Uint8Array, false);
+			await waitForDrain();
+		}
+		entry.push(new Uint8Array(0), true);
+	} finally {
+		readStream.destroy();
+	}
+}
+
+/** Like AbortSignal.throwIfAborted() but coerces non-Error reasons to a named AbortError. */
+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" });
+	}
+}