[miniflare] Capture local observability spans over OTLP

.changeset/local-observability-otlp.md

@@ -0,0 +1,7 @@
+---
+"miniflare": minor
+---
+
+Capture local observability spans over OTLP instead of from streaming tail events
+
+With `unsafeObservability`, workerd now exports every invocation's spans to the local observability collector over OTLP, rather than the collector rebuilding them from streaming tail events. Spans carry precise timestamps rather than the coarse I/O time tail events report. Logs and exceptions are still captured from the tail stream. Requires a workerd release with the `tracing.otlp` config setting.

packages/miniflare/src/index.ts

@@ -86,6 +86,7 @@ import {
 	getUserServiceName,
 	handlePrettyErrorRequest,
 	JsonErrorSchema,
+	OBSERVABILITY_COLLECTOR_SERVICE_NAME,
 	reviveError,
 } from "./plugins/core";
 import { ContainerPrivilegesCache } from "./plugins/core/container";
@@ -97,6 +98,10 @@ import {
 	cfImageLocalFetcher,
 	imagesLocalFetcher,
 } from "./plugins/images/fetcher";
+import {
+	getWorkflowBindingServiceName,
+	getWorkflowsEngineServiceName,
+} from "./plugins/workflows";
 import {
 	HttpOptions_Style,
 	kInspectorSocket,
@@ -152,6 +157,7 @@ import type {
 	QueueConsumers,
 	QueueProducers,
 	ReplaceWorkersTypes,
+	TracedServices,
 } from "./plugins";
 import type {
 	Config,
@@ -527,6 +533,38 @@ function getQueueProducers(
 	return queueProducers;
 }
 
+/**
+ * The services whose spans the observability collector keeps: each user worker,
+ * and the Workflows services that run its workflows (attributed to the
+ * workflow, or for the engine of exported workflows, to the worker). A worker
+ * that isn't registered (`unsafeRegisterWorker: false`) isn't observed, as it
+ * isn't tailed either.
+ */
+function getTracedServices(
+	allWorkerOpts: ParsedWorkerOptions[]
+): TracedServices {
+	const tracedServices: TracedServices = {};
+	for (const { config, dev } of allWorkerOpts) {
+		if (dev?.unsafeRegisterWorker === false) {
+			continue;
+		}
+		tracedServices[getUserServiceName(config.name)] = config.name;
+		for (const [, binding] of getEnvBindingsOfType(config, "workflow")) {
+			tracedServices[getWorkflowBindingServiceName(binding.name)] =
+				binding.name;
+		}
+		const exportedWorkflows = getExportsOfType(config, "workflow");
+		if (exportedWorkflows.length > 0) {
+			tracedServices[getWorkflowsEngineServiceName(config.name)] = config.name;
+		}
+		for (const [, workflow] of exportedWorkflows) {
+			tracedServices[getWorkflowBindingServiceName(workflow.name)] =
+				workflow.name;
+		}
+	}
+	return tracedServices;
+}
+
 function getQueueConsumers(
 	allWorkerOpts: ParsedWorkerOptions[]
 ): QueueConsumers {
@@ -2580,6 +2618,7 @@ export class Miniflare {
 			durableObjectClassNames,
 			workflowExporters,
 			allWorkerOpts,
+			tracedServices: getTracedServices(allWorkerOpts),
 		});
 		for (const service of globalServices) {
 			// Global services should all have unique names
@@ -2597,6 +2636,15 @@ export class Miniflare {
 			// that Miniflare parses and either forwards to a `handleStructuredLogs`
 			// handler or writes to the console by default.
 			structuredLogging: true,
+			// workerd exports the spans of every invocation to the observability
+			// collector, which keeps those of the user's workers.
+			tracing: sharedOpts.unsafeObservability
+				? {
+						otlp: {
+							name: getUserServiceName(OBSERVABILITY_COLLECTOR_SERVICE_NAME),
+						},
+					}
+				: undefined,
 			autogates: process.env.MINIFLARE_WORKERD_AUTOGATES
 				? process.env.MINIFLARE_WORKERD_AUTOGATES.split(" ")
 				: [],

packages/miniflare/src/plugins/core/constants.ts

@@ -11,7 +11,8 @@ export const SERVICE_ENTRY = `${CORE_PLUGIN_NAME}:entry`;
 export const SERVICE_LOCAL_EXPLORER = `${CORE_PLUGIN_NAME}:local-explorer`;
 // Disk service for local explorer UI assets
 export const LOCAL_EXPLORER_DISK = `${CORE_PLUGIN_NAME}:local-explorer-disk`;
-// Local observability collector; user workers' `streamingTails` point here.
+// Local observability collector. workerd exports spans to it over OTLP
+// (`tracing.otlp`), and user workers' `streamingTails` point here for logs.
 // Prefixed via getUserServiceName(), so the name itself must not contain a
 // colon (it collides with the `core:user:` service namespacing).
 export const OBSERVABILITY_COLLECTOR_SERVICE_NAME =
@@ -22,12 +23,9 @@ export const OBSERVABILITY_COLLECTOR_SERVICE_NAME =
 export const EMAIL_STORE_SERVICE_NAME = `email:store`;
 // Disk service backing the EmailStore DO's SQLite storage.
 export const EMAIL_STORE_DISK = `email:store-disk`;
-// Flags that make a user worker stream its tail (incl. user spans) to the
+// Flags that make a user worker stream its tail (logs and exceptions) to the
 // collector; applied to each user worker when observability is enabled
-export const OBSERVABILITY_COMPAT_FLAGS = [
-	"streaming_tail_worker",
-	"tail_worker_user_spans",
-] as const;
+export const OBSERVABILITY_COMPAT_FLAGS = ["streaming_tail_worker"] as const;
 // Service prefix for all regular user workers
 const SERVICE_USER_PREFIX = `${CORE_PLUGIN_NAME}:user`;
 // Service prefix for `workerd`'s builtin services (network, external, disk)

packages/miniflare/src/plugins/core/index.ts

@@ -81,7 +81,7 @@ import {
 	convertManifestModule,
 	withSourceURL,
 } from "./modules";
-import { getObservabilityServices } from "./observability";
+import { getObservabilityServices, type TracedServices } from "./observability";
 import { PROXY_SECRET } from "./proxy";
 import type {
 	Extension,
@@ -105,6 +105,8 @@ import type {
 } from "../shared";
 import type { BindingIdMap } from "./types";
 
+export type { TracedServices };
+
 // `workerd`'s `trustBrowserCas` should probably be named `trustSystemCas`.
 // Rather than using a bundled CA store like Node, it uses
 // `SSL_CTX_set_default_verify_paths()` to use the system CA store:
@@ -565,23 +567,19 @@ export const CORE_PLUGIN: Plugin = {
 		const services: Service[] = [];
 		const extensions: Extension[] = [];
 		// When local observability is on, attach the collector to every user worker
-		// (as a streaming tail consumer) and add the compatibility flags workerd
-		// needs to emit those tail events. This is done here so wrangler and the
-		// Vite plugin don't each have to repeat it.
+		// (as a streaming tail consumer, for logs and exceptions) and add the
+		// compatibility flags workerd needs to emit those tail events. Spans reach
+		// the collector separately, over OTLP (see `#assembleConfig()`). This is
+		// done here so wrangler and the Vite plugin don't each have to repeat it.
 		const observabilityEnabled =
 			sharedOptions.unsafeObservability === true &&
 			dev?.unsafeRegisterWorker !== false;
 		const tailConsumers = observabilityEnabled
 			? [
 					...(config.tailConsumers ?? []),
-					// Pass the worker's name to the collector via binding props. workerd
-					// doesn't populate the tail onset's `scriptName` locally, so this is
-					// how the collector attributes each captured invocation to its worker
-					// (each worker streams to the collector with its own props).
 					{
 						worker: OBSERVABILITY_COLLECTOR_SERVICE_NAME,
 						streaming: true,
-						props: { worker: config.name },
 					},
 				]
 			: (config.tailConsumers ?? []);
@@ -767,6 +765,11 @@ export interface GlobalServicesOptions {
 	workflowExporters: WorkflowExporters;
 	/** All worker options for building per-worker resource bindings */
 	allWorkerOpts?: ParsedWorkerOptions[];
+	/**
+	 * The services whose spans the observability collector keeps, mapped to the
+	 * worker name each one's spans are attributed to
+	 */
+	tracedServices: TracedServices;
 }
 export function getGlobalServices({
 	sharedOptions,
@@ -779,6 +782,7 @@ export function getGlobalServices({
 	durableObjectClassNames,
 	workflowExporters,
 	allWorkerOpts,
+	tracedServices,
 }: GlobalServicesOptions): Service[] {
 	// Collect list of workers we could route to, then parse and sort all routes
 	const workerNames = [...allWorkerRoutes.keys()];
@@ -1048,13 +1052,14 @@ export function getGlobalServices({
 	}
 
 	// Register the trace collector service. It's attached to each user worker's
-	// tail above.
+	// tail above, and receives the spans of the services in `tracedServices`.
 	if (sharedOptions.unsafeObservability) {
 		services.push(
 			...getObservabilityServices(
 				tmpPath,
 				sharedOptions.isolatedResourcePersistencePath,
-				sharedOptions.unsafeRuntimeEnv
+				sharedOptions.unsafeRuntimeEnv,
+				tracedServices
 			)
 		);
 	}

packages/miniflare/src/plugins/core/observability.ts

@@ -7,13 +7,18 @@ import {
 	getUserServiceName,
 	OBSERVABILITY_COLLECTOR_SERVICE_NAME,
 } from "./constants";
+import type { TracedServices } from "../../workers/observability/otlp-to-store";
+
+export type { TracedServices };
 
 /**
- * Builds the trace collector service and the storage behind it. Miniflare core
- * attaches the collector to the user's worker(s) as a tail consumer (see the
- * `unsafeObservability` wiring in this plugin). The collector also hosts the
- * `TraceStore` Durable Object and binds to it, so captured spans and logs are
- * saved there for the Local Explorer / `wrangler observability` to read.
+ * Builds the trace collector service and the storage behind it. workerd exports
+ * every traced invocation's spans to the collector over OTLP (Miniflare sets
+ * `tracing.otlp` to it), and Miniflare core attaches the collector to the user's
+ * worker(s) as a tail consumer for their logs (see the `unsafeObservability`
+ * wiring in this plugin). The collector also hosts the `TraceStore` Durable
+ * Object and binds to it, so captured spans and logs are saved there for the
+ * Local Explorer / `wrangler observability` to read.
  */
 
 /** DO class name — must match the class exported by collector.worker.ts. */
@@ -24,13 +29,16 @@ const TRACE_STORE_BINDING = "TRACE_STORE";
 const OBSERVABILITY_BATCH_SIZE_ENV = "X_LOCAL_OBSERVABILITY_BATCH_SIZE";
 /** Binding name — must match the collector worker's `Env.TRACE_BATCH_SIZE`. */
 const TRACE_BATCH_SIZE_BINDING = "TRACE_BATCH_SIZE";
+/** Binding name — must match the collector worker's `Env.TRACED_SERVICES`. */
+const TRACED_SERVICES_BINDING = "TRACED_SERVICES";
 /** Disk service backing the TraceStore DO's SQLite storage. */
 const OBSERVABILITY_STORAGE_SERVICE_NAME = "obs:storage";
 
 export function getObservabilityServices(
 	tmpPath: string,
 	isolatedResourcePersistencePath: string | undefined,
-	runtimeEnv: Record<string, string> | undefined
+	runtimeEnv: Record<string, string> | undefined,
+	tracedServices: TracedServices
 ): Service[] {
 	// The TraceStore DO is SQLite-backed, so it needs disk-backed storage (the
 	// in-memory option doesn't support SQL). Persist under `.wrangler/state` when
@@ -54,14 +62,15 @@ export function getObservabilityServices(
 			disk: { path: storagePath, writable: true },
 		},
 		{
-			// User workers' `streamingTails` references resolve through
-			// getUserServiceName(), so register the collector under that same
+			// User workers' `streamingTails` references, and `tracing.otlp`, resolve
+			// through getUserServiceName(), so register the collector under that same
 			// prefixed name.
 			name: getUserServiceName(OBSERVABILITY_COLLECTOR_SERVICE_NAME),
 			worker: {
 				compatibilityDate: "2026-01-01",
-				// The collector consumes tails; it isn't tailed itself, so it doesn't
-				// need OBSERVABILITY_COMPAT_FLAGS (those go on the user workers).
+				// The collector consumes tails and spans; it isn't tailed or traced
+				// itself, so it doesn't need OBSERVABILITY_COMPAT_FLAGS (those go on
+				// the user workers).
 				compatibilityFlags: ["nodejs_compat"],
 				modules: [
 					{
@@ -87,6 +96,10 @@ export function getObservabilityServices(
 						name: TRACE_BATCH_SIZE_BINDING,
 						json: JSON.stringify(batchSize),
 					},
+					{
+						name: TRACED_SERVICES_BINDING,
+						json: JSON.stringify(tracedServices),
+					},
 				],
 			},
 		},

packages/miniflare/src/plugins/index.ts

@@ -117,6 +117,7 @@ export type {
 	CompiledModuleRule,
 	GlobalServicesOptions,
 	NodeJSCompatMode,
+	TracedServices,
 } from "./core";
 export type {
 	ContainerPrivileges,

packages/miniflare/src/plugins/workflows/index.ts

@@ -185,8 +185,9 @@ export const WORKFLOWS_PLUGIN: Plugin = {
 		}));
 
 		// The engine service is built here, not through the core plugin's
-		// per-user-worker path, so tail it explicitly or workflow invocations are
-		// invisible in the Local Explorer.
+		// per-user-worker path, so tail it explicitly or workflow logs are missing
+		// from the Local Explorer. Its spans reach the collector over OTLP (see
+		// getTracedServices() in src/index.ts).
 		const observabilityEnabled = sharedOptions.unsafeObservability === true;
 		const engineCompatibilityFlags = [
 			"experimental",
@@ -199,17 +200,12 @@ export const WORKFLOWS_PLUGIN: Plugin = {
 				)
 			);
 		}
-		// Mirrors core's designator shape (prefixed name, JSON props); `worker`
-		// is the name invocations are attributed to.
-		const getStreamingTails = (
-			worker: string
-		): { streamingTails?: ServiceDesignator[] } =>
+		const streamingTails: { streamingTails?: ServiceDesignator[] } =
 			observabilityEnabled
 				? {
 						streamingTails: [
 							{
 								name: getUserServiceName(OBSERVABILITY_COLLECTOR_SERVICE_NAME),
-								props: { json: JSON.stringify({ worker }) },
 							},
 						],
 					}
@@ -234,7 +230,7 @@ export const WORKFLOWS_PLUGIN: Plugin = {
 				name: getWorkflowBindingServiceName(binding.name),
 				worker: {
 					...engineWorker,
-					...getStreamingTails(binding.name),
+					...streamingTails,
 					durableObjectNamespaces: [
 						{
 							className: "Engine",
@@ -316,7 +312,7 @@ export const WORKFLOWS_PLUGIN: Plugin = {
 			name: getWorkflowsEngineServiceName(config.name),
 			worker: {
 				...engineWorker,
-				...getStreamingTails(config.name),
+				...streamingTails,
 				bindings: [
 					WORKER_BINDING_SERVICE_LOOPBACK,
 					{ name: "STEP_LIMITS", json: JSON.stringify(stepLimits) },
@@ -328,7 +324,7 @@ export const WORKFLOWS_PLUGIN: Plugin = {
 				name: getWorkflowBindingServiceName(workflow.name),
 				worker: {
 					...engineWorker,
-					...getStreamingTails(workflow.name),
+					...streamingTails,
 					durableObjectStorage: {
 						localDisk: `${WORKFLOWS_STORAGE_SERVICE_NAME}-${workflow.name}`,
 					},