From f3f9d2816d2efefb2ca131314e911beb35654b1b Mon Sep 17 00:00:00 2001 From: Dan Lapid Date: Sun, 11 Oct 2026 01:03:14 -0400 Subject: [PATCH] [miniflare] Capture local observability spans over OTLP Spans are going away from streaming tail events: workerd can now export each invocation's spans itself, as OTLP (`tracing.otlp`, from cloudflare/workerd#7686). With `unsafeObservability`, Miniflare makes the observability collector workerd's OTLP collector. Its new `/v1/traces` endpoint decodes the protobuf export with a small reader for the fields workerd sets and stores the spans; the streaming tail now supplies only logs and exceptions, so `tail_worker_user_spans` is no longer enabled. `tracing.otlp` traces every service in the config, including Miniflare's own Workers. The collector keeps only the spans of the services Miniflare passes it (user workers and the Workflows services that run their workflows, with the worker names the tail props used to carry) and drops the rest. An invocation started by one of those dropped spans, such as the entry Worker's, would then point at a parent that is never stored, so the store keeps a span whose parent is missing as a root and parks the parent in `pending_parents`. If the parent arrives later, as a calling user worker's span does after its callee has been exported, the span is attached to it again. The config types are regenerated from the workerd schema that adds `tracing`, so this needs a workerd release that includes it. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/local-observability-otlp.md | 7 + packages/miniflare/src/index.ts | 48 ++ .../miniflare/src/plugins/core/constants.ts | 10 +- packages/miniflare/src/plugins/core/index.ts | 27 +- .../src/plugins/core/observability.ts | 33 +- packages/miniflare/src/plugins/index.ts | 1 + .../miniflare/src/plugins/workflows/index.ts | 18 +- .../src/runtime/config/generated/workerd.ts | 72 ++- .../miniflare/src/runtime/config/workerd.ts | 5 + .../workers/observability/collector.worker.ts | 41 +- .../workers/observability/otlp-to-store.ts | 442 +++++++++++++++++ .../workers/observability/tail-to-store.ts | 452 ++---------------- .../src/workers/observability/trace-store.ts | 71 ++- .../test/plugins/core/observability.spec.ts | 98 +++- 14 files changed, 848 insertions(+), 477 deletions(-) create mode 100644 .changeset/local-observability-otlp.md create mode 100644 packages/miniflare/src/workers/observability/otlp-to-store.ts diff --git a/.changeset/local-observability-otlp.md b/.changeset/local-observability-otlp.md new file mode 100644 index 00000000000..d7a01bf6659 --- /dev/null +++ b/.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. diff --git a/packages/miniflare/src/index.ts b/packages/miniflare/src/index.ts index 1284d267928..6f9fd52a22d 100644 --- a/packages/miniflare/src/index.ts +++ b/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(" ") : [], diff --git a/packages/miniflare/src/plugins/core/constants.ts b/packages/miniflare/src/plugins/core/constants.ts index d68cb230ec8..de4fa93e01f 100644 --- a/packages/miniflare/src/plugins/core/constants.ts +++ b/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) diff --git a/packages/miniflare/src/plugins/core/index.ts b/packages/miniflare/src/plugins/core/index.ts index 6f54fcedace..6918038c444 100644 --- a/packages/miniflare/src/plugins/core/index.ts +++ b/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 ) ); } diff --git a/packages/miniflare/src/plugins/core/observability.ts b/packages/miniflare/src/plugins/core/observability.ts index 9d7a6bfcc98..b18cff053d1 100644 --- a/packages/miniflare/src/plugins/core/observability.ts +++ b/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 | undefined + runtimeEnv: Record | 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), + }, ], }, }, diff --git a/packages/miniflare/src/plugins/index.ts b/packages/miniflare/src/plugins/index.ts index 6c51a302b71..f2a9eecffe1 100644 --- a/packages/miniflare/src/plugins/index.ts +++ b/packages/miniflare/src/plugins/index.ts @@ -117,6 +117,7 @@ export type { CompiledModuleRule, GlobalServicesOptions, NodeJSCompatMode, + TracedServices, } from "./core"; export type { ContainerPrivileges, diff --git a/packages/miniflare/src/plugins/workflows/index.ts b/packages/miniflare/src/plugins/workflows/index.ts index 6fc3f0bae22..c6adf9fe35e 100644 --- a/packages/miniflare/src/plugins/workflows/index.ts +++ b/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}`, }, diff --git a/packages/miniflare/src/runtime/config/generated/workerd.ts b/packages/miniflare/src/runtime/config/generated/workerd.ts index 0c17d36b8cd..cabe7a4d2b4 100644 --- a/packages/miniflare/src/runtime/config/generated/workerd.ts +++ b/packages/miniflare/src/runtime/config/generated/workerd.ts @@ -9,7 +9,7 @@ export class Config extends $.Struct { static readonly _capnp = { displayName: "Config", id: "8794486c76aaa7d6", - size: new $.ObjectSize(8, 6), + size: new $.ObjectSize(8, 7), defaultStructuredLogging: $.getBitMask(false, 0), }; static _Services: $.ListCtor; @@ -187,10 +187,80 @@ export class Config extends $.Struct { set logging(value: LoggingOptions) { $.utils.copyFrom(value, $.utils.getPointer(5, this)); } + _adoptTracing(value: $.Orphan): void { + $.utils.adopt(value, $.utils.getPointer(6, this)); + } + _disownTracing(): $.Orphan { + return $.utils.disown(this.tracing); + } + /** + * Where the spans of Worker invocations are exported. + * + */ + get tracing(): TracingOptions { + return $.utils.getStruct(6, TracingOptions, this); + } + _hasTracing(): boolean { + return !$.utils.isNull($.utils.getPointer(6, this)); + } + _initTracing(): TracingOptions { + return $.utils.initStructAt(6, TracingOptions, this); + } + set tracing(value: TracingOptions) { + $.utils.copyFrom(value, $.utils.getPointer(6, this)); + } toString(): string { return "Config_" + super.toString(); } } +export class TracingOptions extends $.Struct { + static readonly _capnp = { + displayName: "TracingOptions", + id: "f6a0cf21c4a0536c", + size: new $.ObjectSize(0, 1), + }; + _adoptOtlp(value: $.Orphan): void { + $.utils.adopt(value, $.utils.getPointer(0, this)); + } + _disownOtlp(): $.Orphan { + return $.utils.disown(this.otlp); + } + /** + * Export spans to an OpenTelemetry collector. Every Worker invocation produces a trace: a span + * for the invocation itself, one for each subrequest and binding operation it performs, and any + * the Worker opens with `ctx.tracing.enterSpan()`. They are POSTed to `/v1/traces` on this + * service in the OTLP/HTTP protobuf encoding, so the service is typically an `ExternalServer` + * pointing at a collector's OTLP/HTTP port. For example, to see traces in a local Jaeger: + * + * services = [ + * ..., + * (name = "jaeger", external = (address = "localhost:4318", http = ())), + * ], + * tracing = (otlp = "jaeger"), + * + * Use `HttpOptions.injectRequestHeaders` to send credentials along. The `service.name` resource + * attribute is the name of the Worker's service. A request arriving on a socket starts its trace + * with an `http_request` span of the `workers-runtime` service, under which the invocation's span + * sits. Invocations of the designated service itself are not traced. While this is set, spans are + * not delivered to streaming tail workers. + * + */ + get otlp(): ServiceDesignator { + return $.utils.getStruct(0, ServiceDesignator, this); + } + _hasOtlp(): boolean { + return !$.utils.isNull($.utils.getPointer(0, this)); + } + _initOtlp(): ServiceDesignator { + return $.utils.initStructAt(0, ServiceDesignator, this); + } + set otlp(value: ServiceDesignator) { + $.utils.copyFrom(value, $.utils.getPointer(0, this)); + } + toString(): string { + return "TracingOptions_" + super.toString(); + } +} export class LoggingOptions extends $.Struct { static readonly _capnp = { displayName: "LoggingOptions", diff --git a/packages/miniflare/src/runtime/config/workerd.ts b/packages/miniflare/src/runtime/config/workerd.ts index 874a4981b15..bfbeee502fa 100644 --- a/packages/miniflare/src/runtime/config/workerd.ts +++ b/packages/miniflare/src/runtime/config/workerd.ts @@ -22,6 +22,11 @@ export interface Config { extensions?: Extension[]; autogates?: string[]; structuredLogging?: boolean; + tracing?: TracingOptions; +} + +export interface TracingOptions { + otlp?: ServiceDesignator; } export type Socket = { diff --git a/packages/miniflare/src/workers/observability/collector.worker.ts b/packages/miniflare/src/workers/observability/collector.worker.ts index a186e60f58f..547f306c599 100644 --- a/packages/miniflare/src/workers/observability/collector.worker.ts +++ b/packages/miniflare/src/workers/observability/collector.worker.ts @@ -1,14 +1,18 @@ /** - * The local observability collector. Miniflare attaches it to the user's - * worker(s) as a tail consumer, so workerd delivers each invocation's tail events - * here. `TailToStoreHandler` converts them into spans and logs, which are stored - * in the `TraceStore` Durable Object hosted by this same worker. The endpoint for - * querying that store is also defined here. + * The local observability collector. Miniflare designates it as workerd's OTLP + * collector (`tracing.otlp`), so workerd POSTs each invocation's spans to its + * `/v1/traces`, and attaches it to the user's worker(s) as a tail consumer, so + * workerd delivers each invocation's tail events here. `otlp-to-store.ts` turns + * the former into spans and `TailToStoreHandler` the latter into logs, which are + * stored in the `TraceStore` Durable Object hosted by this same worker. The + * endpoint for querying that store is also defined here. */ import { WorkerEntrypoint } from "cloudflare:workers"; +import { decodeExportTraceServiceRequest } from "./otlp-to-store"; import { TailToStoreHandler } from "./tail-to-store"; import { TraceStore } from "./trace-store"; import { TraceWriter } from "./trace-writer"; +import type { TracedServices } from "./otlp-to-store"; // Re-export so the embedded worker registers the DO class under its namespace. export { TraceStore }; @@ -16,6 +20,7 @@ export { TraceStore }; interface Env { TRACE_STORE: DurableObjectNamespace; TRACE_BATCH_SIZE: number; + TRACED_SERVICES: TracedServices; } let writer: TraceWriter | undefined; @@ -27,11 +32,7 @@ export default class LocalObservabilityCollector extends WorkerEntrypoint { this.env.TRACE_STORE.idFromName("singleton") ); writer ??= new TraceWriter(this.env.TRACE_BATCH_SIZE); - // Miniflare core passes the source worker's name in binding props (workerd - // doesn't surface it on the tail onset locally), so captured spans can be - // attributed to the right worker. - const worker = (this.ctx.props as { worker?: string } | undefined)?.worker; - return new TailToStoreHandler(writer, store, onset, worker); + return new TailToStoreHandler(writer, store, onset); } /** @@ -46,6 +47,26 @@ export default class LocalObservabilityCollector extends WorkerEntrypoint { const store = this.env.TRACE_STORE.get( this.env.TRACE_STORE.idFromName("singleton") ); + // workerd's OTLP exporter (`tracing.otlp`) posts every traced invocation's + // spans here. + if (url.pathname === "/v1/traces" && request.method === "POST") { + const spans = decodeExportTraceServiceRequest( + new Uint8Array(await request.arrayBuffer()), + this.env.TRACED_SERVICES + ); + if (spans.length > 0) { + writer ??= new TraceWriter(this.env.TRACE_BATCH_SIZE); + const owner = writer.createOwner(store); + for (const span of spans) { + writer.enqueueCompletedSpan(owner, span); + } + await writer.drain(owner); + } + // An empty `ExportTraceServiceResponse`. + return new Response(null, { + headers: { "content-type": "application/x-protobuf" }, + }); + } if (url.pathname === "/query" && request.method === "POST") { const { sql, params } = (await request.json()) as { sql?: string; diff --git a/packages/miniflare/src/workers/observability/otlp-to-store.ts b/packages/miniflare/src/workers/observability/otlp-to-store.ts new file mode 100644 index 00000000000..4db0321288b --- /dev/null +++ b/packages/miniflare/src/workers/observability/otlp-to-store.ts @@ -0,0 +1,442 @@ +/** + * Converts the spans workerd exports over OTLP (`tracing.otlp`) into the spans + * the store keeps. workerd POSTs each traced invocation's spans to the + * collector's `/v1/traces` as an OTLP/HTTP protobuf `ExportTraceServiceRequest`, + * one `ResourceSpans` per service, whose `service.name` resource attribute is the + * workerd service name. + * + * Only the fields workerd sets are decoded, by a small protobuf reader rather + * than an OpenTelemetry dependency. We keep the attribute names workerd sends, + * which are the ones the Workers Observability UI expects (`faas.trigger`, + * `http.request.method`, `cloudflare.outcome`, `cpu_time_ms`, …). The URLs and + * headers belong to the developer, so nothing is redacted. + */ +import type { SpanInput } from "./trace-types"; + +/** + * The workerd services whose spans the collector keeps, mapped to the worker + * name the spans are attributed to. workerd exports the spans of every service, + * including Miniflare's own; the collector drops those of services not listed + * here. + */ +export type TracedServices = Record; + +/** + * The spans in `body` that belong to one of `tracedServices`. Spans of other + * services (Miniflare's own Workers, and workerd's `workers-runtime` request + * span) are dropped. A dropped span can be the parent of a kept one: the store + * keeps such a span as a root (see `TraceStore.persist()`). + */ +export function decodeExportTraceServiceRequest( + body: Uint8Array, + tracedServices: TracedServices +): SpanInput[] { + const rows: SpanInput[] = []; + const request = new Reader(body); + while (request.next()) { + if (request.field !== 1) { + request.skip(); + continue; + } + // ResourceSpans + const resourceSpans = request.message(); + let service: string | undefined; + const scopeSpans: Reader[] = []; + while (resourceSpans.next()) { + if (resourceSpans.field === 1) { + const attributes = decodeAttributes(resourceSpans.message(), 1); + const name = attributes["service.name"]; + if (typeof name === "string" && Object.hasOwn(tracedServices, name)) { + service = tracedServices[name]; + } + } else if (resourceSpans.field === 2) { + scopeSpans.push(resourceSpans.message()); + } else { + resourceSpans.skip(); + } + } + if (service === undefined) { + continue; + } + for (const scope of scopeSpans) { + while (scope.next()) { + if (scope.field === 2) { + rows.push(decodeSpan(scope.message(), service)); + } else { + scope.skip(); + } + } + } + } + return rows; +} + +const STATUS_CODE_ERROR = 2; + +function decodeSpan(span: Reader, service: string): SpanInput { + let traceId = ""; + let spanId = ""; + let parentId: string | null = null; + let name = ""; + let startNs = 0n; + let endNs = 0n; + const attributes: Record = {}; + let exception: string | null = null; + let statusCode = 0; + let statusMessage = ""; + while (span.next()) { + switch (span.field) { + case 1: + traceId = toHex(span.bytes()); + break; + case 2: + spanId = toHex(span.bytes()); + break; + case 4: { + const parent = span.bytes(); + parentId = parent.length > 0 ? toHex(parent) : null; + break; + } + case 5: + name = span.string(); + break; + case 7: + startNs = span.fixed64(); + break; + case 8: + endNs = span.fixed64(); + break; + case 9: + Object.assign(attributes, decodeKeyValue(span.message())); + break; + case 11: { + // Event. workerd records each exception as an `exception` event; the + // last one describes the span's error. + const event = span.message(); + let eventName = ""; + const eventAttributes: Record = {}; + while (event.next()) { + if (event.field === 2) { + eventName = event.string(); + } else if (event.field === 3) { + Object.assign(eventAttributes, decodeKeyValue(event.message())); + } else { + event.skip(); + } + } + if (eventName === "exception") { + const type = eventAttributes["exception.type"] || "Error"; + const message = eventAttributes["exception.message"] ?? ""; + exception = `${type}: ${message}`; + } + break; + } + case 15: { + const status = span.message(); + while (status.next()) { + if (status.field === 2) { + statusMessage = status.string(); + } else if (status.field === 3) { + statusCode = Number(status.varint()); + } else { + status.skip(); + } + } + break; + } + default: + span.skip(); + } + } + + const failed = statusCode === STATUS_CODE_ERROR; + // The invocation's span carries the invocation's outcome (`exception`, + // `exceededCpu`, …); any other span is "ok" or "error". + const outcome = + typeof attributes["cloudflare.outcome"] === "string" + ? attributes["cloudflare.outcome"] + : failed + ? "error" + : "ok"; + const startMs = nsToMs(startNs); + return { + traceId, + spanId, + parentId, + service, + name, + kind: friendlyKind(attributes["faas.trigger"], name), + startMs, + durationMs: Math.max(0, nsToMs(endNs) - startMs), + outcome, + error: failed + ? (exception ?? errorMessage(statusMessage, attributes)) + : null, + attributes, + }; +} + +/** + * What went wrong in a failed span with no exception event: the status message, + * or failing that the `error.type` workerd derives from the outcome or the HTTP + * status. + */ +function errorMessage( + statusMessage: string, + attributes: Record +): string | null { + if (statusMessage !== "") { + return statusMessage; + } + const errorType = attributes["error.type"]; + return errorType === undefined || errorType === null + ? null + : String(errorType); +} + +/** Friendly `kind` for the UI: root span from its trigger, children from name. */ +function friendlyKind(faasTrigger: unknown, name: string): string { + if (typeof faasTrigger === "string") { + switch (faasTrigger) { + case "http": + return "http"; + case "timer": + return "scheduled"; + case "pubsub": + return "queue"; + case "email": + return "email"; + case "jsrpc": + return "jsrpc"; + case "websocket": + return "websocket"; + case "trace": + return "trace"; + default: + return "worker"; + } + } + const n = name.toLowerCase(); + if (n.includes("kv")) { + return "kv"; + } + if (n.includes("d1")) { + return "d1"; + } + if (n.includes("r2")) { + return "r2"; + } + if (n.includes("queue")) { + return "queue"; + } + if (n.includes("durable") || n.includes("do_")) { + return "do"; + } + if (n.includes("cache")) { + return "cache"; + } + if (n.includes("fetch")) { + return "fetch"; + } + return "span"; +} + +/** The `KeyValue`s at `field` of `message`, as a plain object. */ +function decodeAttributes( + message: Reader, + field: number +): Record { + const attributes: Record = {}; + while (message.next()) { + if (message.field === field) { + Object.assign(attributes, decodeKeyValue(message.message())); + } else { + message.skip(); + } + } + return attributes; +} + +function decodeKeyValue(keyValue: Reader): Record { + let key = ""; + let value: unknown = null; + while (keyValue.next()) { + if (keyValue.field === 1) { + key = keyValue.string(); + } else if (keyValue.field === 2) { + value = decodeAnyValue(keyValue.message()); + } else { + keyValue.skip(); + } + } + return { [key]: value }; +} + +function decodeAnyValue(anyValue: Reader): unknown { + let value: unknown = null; + while (anyValue.next()) { + switch (anyValue.field) { + case 1: + value = anyValue.string(); + break; + case 2: + value = anyValue.varint() !== 0n; + break; + case 3: + value = int64ToJson(BigInt.asIntN(64, anyValue.varint())); + break; + case 4: + value = anyValue.double(); + break; + case 5: { + // ArrayValue + const array = anyValue.message(); + const values: unknown[] = []; + while (array.next()) { + if (array.field === 1) { + values.push(decodeAnyValue(array.message())); + } else { + array.skip(); + } + } + value = values; + break; + } + default: + anyValue.skip(); + } + } + return value; +} + +/** Downcast to number where safe, else string, so it survives `JSON.stringify`. */ +function int64ToJson(value: bigint): number | string { + return value >= Number.MIN_SAFE_INTEGER && value <= Number.MAX_SAFE_INTEGER + ? Number(value) + : value.toString(); +} + +/** Epoch nanoseconds to epoch milliseconds, keeping microseconds. */ +function nsToMs(ns: bigint): number { + return Number(ns / 1000n) / 1000; +} + +function toHex(bytes: Uint8Array): string { + let hex = ""; + for (const byte of bytes) { + hex += byte.toString(16).padStart(2, "0"); + } + return hex; +} + +const WIRE_VARINT = 0; +const WIRE_FIXED64 = 1; +const WIRE_LENGTH_DELIMITED = 2; +const WIRE_FIXED32 = 5; + +const textDecoder = new TextDecoder(); + +/** + * Reads the fields of one protobuf message: call `next()` to advance to the + * next field, then read its value with the method for its type, or `skip()` it. + */ +class Reader { + field = 0; + #wireType = 0; + #pos: number; + #end: number; + #view: DataView; + + constructor( + private readonly buffer: Uint8Array, + start = 0, + end = buffer.length + ) { + this.#pos = start; + this.#end = end; + this.#view = new DataView( + buffer.buffer, + buffer.byteOffset, + buffer.byteLength + ); + } + + next(): boolean { + if (this.#pos >= this.#end) { + return false; + } + const tag = Number(this.varint()); + this.field = tag >>> 3; + this.#wireType = tag & 7; + return true; + } + + varint(): bigint { + let result = 0n; + let shift = 0n; + for (;;) { + if (this.#pos >= this.#end) { + throw new Error("truncated OTLP message"); + } + const byte = this.buffer[this.#pos++]; + result |= BigInt(byte & 0x7f) << shift; + if (byte < 0x80) { + return result; + } + shift += 7n; + } + } + + fixed64(): bigint { + return this.#view.getBigUint64(this.#take(8), true); + } + + double(): number { + return this.#view.getFloat64(this.#take(8), true); + } + + bytes(): Uint8Array { + const length = Number(this.varint()); + const start = this.#take(length); + return this.buffer.subarray(start, start + length); + } + + string(): string { + return textDecoder.decode(this.bytes()); + } + + message(): Reader { + const length = Number(this.varint()); + const start = this.#take(length); + return new Reader(this.buffer, start, start + length); + } + + skip(): void { + switch (this.#wireType) { + case WIRE_VARINT: + this.varint(); + break; + case WIRE_FIXED64: + this.#take(8); + break; + case WIRE_LENGTH_DELIMITED: + this.#take(Number(this.varint())); + break; + case WIRE_FIXED32: + this.#take(4); + break; + default: + throw new Error(`unsupported protobuf wire type ${this.#wireType}`); + } + } + + /** Advances past `length` bytes, returning where they start. */ + #take(length: number): number { + const start = this.#pos; + if (start + length > this.#end) { + throw new Error("truncated OTLP message"); + } + this.#pos += length; + return start; + } +} diff --git a/packages/miniflare/src/workers/observability/tail-to-store.ts b/packages/miniflare/src/workers/observability/tail-to-store.ts index 7077510b9b0..16c9f627365 100644 --- a/packages/miniflare/src/workers/observability/tail-to-store.ts +++ b/packages/miniflare/src/workers/observability/tail-to-store.ts @@ -1,17 +1,14 @@ /** - * Converts one invocation's workerd tail events into the spans and logs the - * store keeps. Open spans and streaming attributes remain in this invocation's - * handler. Each completed snapshot is submitted to the collector-wide bounded - * persistence batch as soon as the span closes. + * Converts one invocation's workerd tail events into the logs the store keeps: + * its `console` calls, its exceptions, and one log for the invocation itself. + * The invocation's spans reach the collector over OTLP instead (see + * `otlp-to-store.ts`). Logs reference spans by the ids the tail events carry, + * which are the ids workerd exports the spans with. * - * There is no OpenTelemetry layer in between. Following the OTLP format would only - * matter for sending traces to a real backend, but here the same process both - * produces and reads them. So we keep just the attribute names the Workers - * Observability UI expects (`faas.trigger`, `http.request.method`, - * `cloudflare.outcome`, `cpu_time_ms`, …) and skip the SDK and wire format. The - * URLs and headers belong to the developer, so nothing is redacted. + * Each log is submitted to the collector-wide bounded persistence batch as soon + * as it arrives. */ -import type { LogInput, SpanInput } from "./trace-types"; +import type { LogInput } from "./trace-types"; import type { BatchStore, TraceWriter, TraceWriterOwner } from "./trace-writer"; /** A tail event's `timestamp` is a `Date` (or ms number); normalise to epoch ms. */ @@ -19,311 +16,49 @@ function toMs(timestamp: Date | number): number { return typeof timestamp === "number" ? timestamp : timestamp.getTime(); } -/** - * The span context for an event. For `onset` / `spanOpen` the new span's id is on - * the event; `spanContext.spanId` is then its parent. For every other event the - * `spanContext` already points at the span the event acts on. - */ -function ids(event: TailStream.TailEvent): { - traceId: string; - spanId?: string; - parentId?: string; -} { - if (event.event.type === "onset" || event.event.type === "spanOpen") { - return { - traceId: event.spanContext.traceId, - spanId: event.event.spanId, - parentId: event.spanContext.spanId, - }; - } - return { - traceId: event.spanContext.traceId, - spanId: event.spanContext.spanId, - }; -} - -/** Friendly `kind` for the UI: root span from its trigger, children from name. */ -function friendlyKind(faasTrigger: string | undefined, name: string): string { - if (faasTrigger) { - switch (faasTrigger) { - case "http": - return "http"; - case "timer": - return "scheduled"; - case "pubsub": - return "queue"; - case "email": - return "email"; - case "jsrpc": - return "jsrpc"; - case "websocket": - return "websocket"; - case "trace": - return "trace"; - default: - return "worker"; - } - } - const n = name.toLowerCase(); - if (n.includes("kv")) { - return "kv"; - } - if (n.includes("d1")) { - return "d1"; - } - if (n.includes("r2")) { - return "r2"; - } - if (n.includes("queue")) { - return "queue"; - } - if (n.includes("durable") || n.includes("do_")) { - return "do"; - } - if (n.includes("cache")) { - return "cache"; - } - if (n.includes("fetch")) { - return "fetch"; - } - return "span"; -} - -/** Root-span name + attributes derived from the onset trigger info. */ -function describeTrigger(info: TailStream.Onset["info"]): { - name: string; - attributes: Record; -} { - switch (info.type) { - case "fetch": - return { - name: info.method, - attributes: { - "faas.trigger": "http", - "http.request.method": info.method, - "url.full": info.url, - }, - }; - case "jsrpc": - return { name: "jsrpc", attributes: { "faas.trigger": "jsrpc" } }; - case "scheduled": - return { - name: "scheduled", - attributes: { "faas.trigger": "timer", "faas.cron": info.cron }, - }; - case "alarm": - return { name: "alarm", attributes: { "faas.trigger": "timer" } }; - case "queue": - return { - name: "queue", - attributes: { - "faas.trigger": "pubsub", - "cloudflare.queue.name": info.queueName, - }, - }; - case "email": - return { - name: "email", - attributes: { - "faas.trigger": "email", - "cloudflare.email.to": info.rcptTo, - }, - }; - case "trace": - return { name: "trace", attributes: { "faas.trigger": "trace" } }; - case "hibernatableWebSocket": - return { - name: "hibernatableWebSocket", - attributes: { "faas.trigger": "websocket" }, - }; - default: - return { name: info.type, attributes: { "faas.trigger": "other" } }; - } -} - -/** - * Every `EventOutcome` other than "ok" is a failure (canceled, exception, - * exceededCpu, exceededWallTime, internalError, …). Matching on "not ok" keeps - * capture in step with the read side (`SPAN_IS_ERROR` uses `outcome != 'ok'`) and - * covers any future outcomes without another allowlist to maintain. - */ -function isErrorOutcome(outcome: TailStream.EventOutcome): boolean { - return outcome !== "ok"; -} - -/** - * Per-span bookkeeping kept between events so we can compute a duration and fold - * in error info when the span closes. The full row, including streamed - * attributes, remains invocation-local until then. - */ -interface PendingSpan { - traceId: string; - startMs: number; - name: string | null; - outcome: string | null; - error: string | null; - errored: boolean; - closed: boolean; +/** The name the invocation's span gets from its trigger, as workerd names it. */ +function triggerName(info: TailStream.Onset["info"]): string { + return info.type === "fetch" ? info.method : info.type; } /** * Handles the tail events for a single invocation. The collector-wide writer - * batches completed rows, bounds pending persistence, and serialises store RPCs. + * batches rows, bounds pending persistence, and serialises store RPCs. */ export class TailToStoreHandler implements TailStream.TailEventHandlerObject { - #spans = new Map(); - #rootSpanId: string | null = null; - #traceId: string | null = null; - #startMs: number | null = null; - #invocationBody: string | null = null; - #rows = new Map(); + #traceId: string; + #rootSpanId: string; + #rootName: string; + #startMs: number; + #invocationBody: string; + #errored = false; #owner: TraceWriterOwner; constructor( private readonly writer: TraceWriter, store: BatchStore, - onset: TailStream.TailEvent, - /** Owning worker name (from miniflare core), for multi-worker attribution. */ - private readonly worker?: string + onset: TailStream.TailEvent ) { this.#owner = this.writer.createOwner(store); - const { traceId, spanId, parentId } = ids(onset); - if (!spanId) { - return; - } - this.#traceId = traceId; + this.#traceId = onset.spanContext.traceId; + this.#rootSpanId = onset.event.spanId; + this.#rootName = triggerName(onset.event.info); this.#startMs = toMs(onset.timestamp); - - const { name, attributes: triggerAttributes } = describeTrigger( - onset.event.info - ); - // Start from any attributes carried directly on the onset (not guaranteed - // to arrive later via the `attributes` callback), then let the canonical - // trigger-derived keys (`faas.trigger`, `http.request.method`, …) win since - // the UI keys off those. - const attributes: Record = {}; - // `attributes` is typed as required but isn't always populated on the onset - // in local dev (events come through a cf-to-otel shim), so guard the loop. - for (const attr of onset.event.attributes ?? []) { - attributes[attr.name] = normalizeAttr(attr.value); - } - Object.assign(attributes, triggerAttributes); - if (onset.invocationId) { - attributes["faas.invocation_id"] = onset.invocationId; - } // Invocation-log body: for fetch, method + URL (mirrors the prod Logs view); // otherwise the trigger name. this.#invocationBody = onset.event.info.type === "fetch" ? `${onset.event.info.method} ${onset.event.info.url}` - : name; - const pending: PendingSpan = { - traceId, - startMs: this.#startMs, - name, - outcome: null, - error: null, - errored: false, - closed: false, - }; - this.#spans.set(spanId, pending); - this.#open({ - traceId, - spanId, - // A sub-invocation (e.g. downstream of a service binding) carries the - // caller's span as its parent, so it nests into one distributed trace; - // a true top-level invocation has no inherited parent (→ null root). - parentId: parentId ?? null, - service: this.worker ?? null, - name, - kind: friendlyKind(attributes["faas.trigger"] as string, name), - startMs: this.#startMs, - durationMs: null, - outcome: null, - error: null, - attributes, - }); - this.#rootSpanId = spanId; - } - - spanOpen(event: TailStream.TailEvent) { - const { traceId, spanId, parentId } = ids(event); - if (!spanId) { - return; - } - const startMs = toMs(event.timestamp); - const pending: PendingSpan = { - traceId, - startMs, - name: event.event.name, - outcome: null, - error: null, - errored: false, - closed: false, - }; - this.#spans.set(spanId, pending); - this.#open({ - traceId, - spanId, - parentId: parentId ?? null, - service: this.worker ?? null, - name: event.event.name, - kind: friendlyKind(undefined, event.event.name), - startMs, - durationMs: null, - outcome: null, - error: null, - attributes: null, - }); - } - - spanClose(event: TailStream.TailEvent) { - const { traceId, spanId } = ids(event); - const pending = spanId ? this.#spans.get(spanId) : undefined; - if (spanId && pending) { - // A span can close with a failure outcome (exception, exceededCpu, …). - // Fold it in so the span isn't recorded as "ok". A preceding exception - // event already set a richer outcome/error, so don't overwrite that. - if (isErrorOutcome(event.event.outcome)) { - pending.errored = true; - pending.outcome ??= event.event.outcome; - } - this.#close(traceId, spanId, pending, toMs(event.timestamp), null); - } - } - - attributes(event: TailStream.TailEvent) { - const { traceId, spanId } = ids(event); - if (!spanId || !this.#spans.has(spanId)) { - return; - } - const attrs: Record = {}; - for (const attr of event.event.info) { - attrs[attr.name] = normalizeAttr(attr.value); - } - if (Object.keys(attrs).length > 0) { - this.#merge(traceId, spanId, attrs); - } - } - - return(event: TailStream.TailEvent) { - const { traceId, spanId } = ids(event); - const pending = spanId ? this.#spans.get(spanId) : undefined; - if (spanId && pending && event.event.info?.type === "fetch") { - this.#merge(traceId, spanId, { - "http.response.status_code": event.event.info.statusCode, - }); - } + : this.#rootName; } log(event: TailStream.TailEvent) { - const { traceId, spanId } = ids(event); // `console.log` surfaces as level "log"; fold into "info" so the stored set // stays {debug, info, warn, error}. const level = event.event.level === "log" ? "info" : event.event.level; this.#append({ - traceId, - spanId: spanId ?? null, + traceId: event.spanContext.traceId, + spanId: event.spanContext.spanId ?? null, tsMs: toMs(event.timestamp), level, message: serialize(event.event.message), @@ -332,143 +67,48 @@ export class TailToStoreHandler implements TailStream.TailEventHandlerObject { } exception(event: TailStream.TailEvent) { - const { traceId, spanId } = ids(event); - const pending = spanId ? this.#spans.get(spanId) : undefined; + const spanId = event.spanContext.spanId ?? null; const type = event.event.name || "Error"; const message = event.event.message ?? ""; const head = `${type}: ${message}`; const text = event.event.stack ? `${head}\n${event.event.stack}` : head; - if (pending) { - // Recorded on the span so it lands as `error`/outcome when it closes. - pending.errored = true; - pending.outcome = "error"; - pending.error = head; + const onRoot = spanId === this.#rootSpanId; + if (onRoot) { + this.#errored = true; } - // Surface exceptions as error-level logs too, so failures show in the Logs - // view even when the worker never called console.error. + // Surface exceptions as error-level logs, so failures show in the Logs view + // even when the worker never called console.error. this.#append({ - traceId, - spanId: spanId ?? null, + traceId: event.spanContext.traceId, + spanId, tsMs: toMs(event.timestamp), level: "error", message: serialize(text), - operation: pending?.name ?? null, + operation: onRoot ? this.#rootName : null, }); } async outcome(event: TailStream.TailEvent) { - const endMs = toMs(event.timestamp); - const traceId = this.#traceId ?? event.spanContext.traceId; - const root = this.#rootSpanId - ? this.#spans.get(this.#rootSpanId) - : undefined; - let rootAccepted = false; - if (root && this.#rootSpanId) { - if (isErrorOutcome(event.event.outcome)) { - root.errored = true; - } - root.outcome = event.event.outcome; - // The root closes with the invocation: fold in the final outcome and - // resource attributes at the same time. - rootAccepted = this.#close(traceId, this.#rootSpanId, root, endMs, { - "cloudflare.outcome": event.event.outcome, - cpu_time_ms: event.event.cpuTime, - wall_time_ms: event.event.wallTime, - }); + // Every `EventOutcome` other than "ok" is a failure (canceled, exception, + // exceededCpu, …), as the read side's `outcome != 'ok'` has it. + if (event.event.outcome !== "ok") { + this.#errored = true; } - - // Close any span left open (error/bug) so it still shows in the waterfall. - for (const [spanId, pending] of this.#spans) { - if (!pending.closed) { - this.#close(pending.traceId, spanId, pending, endMs, null); - } - } - // One synthetic invocation log so silent workers still appear. - if (this.#rootSpanId && this.#invocationBody !== null) { - this.#append({ - traceId, - spanId: rootAccepted ? this.#rootSpanId : null, - tsMs: this.#startMs ?? endMs, - level: root?.errored ? "error" : "info", - message: serialize(this.#invocationBody), - operation: null, - }); - } - + this.#append({ + traceId: this.#traceId, + spanId: this.#rootSpanId, + tsMs: this.#startMs, + level: this.#errored ? "error" : "info", + message: serialize(this.#invocationBody), + operation: null, + }); await this.writer.drain(this.#owner); } - #open(input: SpanInput): void { - const key = rowKey(input.traceId, input.spanId); - this.#rows.set(key, input); - } - - /** Fold attributes into a buffered row, so merging costs no round-trip. */ - #merge(traceId: string, spanId: string, attributes: Record) { - const key = rowKey(traceId, spanId); - const row = this.#rows.get(key); - if (!row) { - return; - } - row.attributes = { ...(row.attributes ?? {}), ...attributes }; - } - #append(log: LogInput) { this.writer.enqueueLog(this.#owner, log); } - - /** Finish a span, setting its duration and outcome and adding any final - * attributes. Runs at most once per span (guarded by `closed`). */ - #close( - traceId: string, - spanId: string, - pending: PendingSpan, - endMs: number, - attributes: Record | null - ): boolean { - if (pending.closed) { - return true; - } - pending.closed = true; - const key = rowKey(traceId, spanId); - const row = this.#rows.get(key); - if (!row) { - return false; - } - row.durationMs = Math.max(0, endMs - pending.startMs); - row.outcome = pending.outcome ?? (pending.errored ? "error" : "ok"); - row.error = pending.error; - if (attributes) { - row.attributes = { ...(row.attributes ?? {}), ...attributes }; - } - const accepted = this.writer.enqueueCompletedSpan(this.#owner, row); - this.#rows.delete(key); - this.#spans.delete(spanId); - return accepted; - } -} - -function rowKey(traceId: string, spanId: string): string { - return `${traceId}\u0000${spanId}`; -} - -/** - * Tail attribute values can be bigint (or bigint arrays); `JSON.stringify` throws - * on those. Downcast to number where safe, else string, so the value survives - * serialization into the store's `attributes` blob. - */ -function normalizeAttr(value: TailStream.Attribute["value"]): unknown { - if (Array.isArray(value)) { - return value.map((v) => (typeof v === "bigint" ? bigintToJson(v) : v)); - } - return typeof value === "bigint" ? bigintToJson(value) : value; -} - -function bigintToJson(value: bigint): number | string { - return value >= Number.MIN_SAFE_INTEGER && value <= Number.MAX_SAFE_INTEGER - ? Number(value) - : value.toString(); } function serialize(value: unknown): string { diff --git a/packages/miniflare/src/workers/observability/trace-store.ts b/packages/miniflare/src/workers/observability/trace-store.ts index 3fb90699bd1..231c77b09e0 100644 --- a/packages/miniflare/src/workers/observability/trace-store.ts +++ b/packages/miniflare/src/workers/observability/trace-store.ts @@ -7,7 +7,12 @@ * * A "trace" is the root span (the one with no parent) plus everything below it. * When a request calls other workers, those sub-invocations share its trace_id - * and attach to the calling span through parent_id. Request-level data (HTTP + * and attach to the calling span through parent_id. A span whose parent is not + * stored is a root: its parent is kept in `pending_parents` instead, and moved + * back to `parent_id` if the parent arrives later. That is how an invocation + * started by one of Miniflare's own Workers (whose spans the collector drops) + * starts a trace, while one called by another user worker, which usually + * finishes and is exported first, still joins its caller's. Request-level data (HTTP * status, CPU/wall time, trigger, worker name) is stored on the root span's * `attributes`. Times are absolute (epoch ms); the read API shifts them so each * trace starts at zero. @@ -46,6 +51,13 @@ const SCHEMA = [ PRIMARY KEY (trace_id, seq) )`, `CREATE INDEX IF NOT EXISTS logs_by_level ON logs (level)`, + `CREATE TABLE IF NOT EXISTS pending_parents ( + trace_id TEXT NOT NULL, + span_id TEXT NOT NULL, + parent_id TEXT NOT NULL, + PRIMARY KEY (trace_id, span_id) + )`, + `CREATE INDEX IF NOT EXISTS pending_parents_by_parent ON pending_parents (trace_id, parent_id)`, ]; /** @@ -123,6 +135,62 @@ export class TraceStore extends DurableObject { ); } + // With every span of the batch stored, settle parents (see the file + // comment): a span whose parent is missing becomes a root, and spans + // waiting for one of this batch's spans are attached to it. + for ( + let offset = 0; + offset < spans.length; + offset += SQL_WRITE_CHUNK_ROWS + ) { + const ids = JSON.stringify( + spans + .slice(offset, offset + SQL_WRITE_CHUNK_ROWS) + .map(({ traceId, spanId }) => ({ traceId, spanId })) + ); + this.sql.exec( + `INSERT INTO pending_parents (trace_id, span_id, parent_id) + SELECT s.trace_id, s.span_id, s.parent_id + FROM json_each(?) j + JOIN spans s ON s.trace_id = json_extract(j.value, '$.traceId') + AND s.span_id = json_extract(j.value, '$.spanId') + WHERE s.parent_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM spans p + WHERE p.trace_id = s.trace_id AND p.span_id = s.parent_id + ) + ON CONFLICT (trace_id, span_id) DO UPDATE SET + parent_id = excluded.parent_id`, + ids + ); + this.sql.exec( + `UPDATE spans SET parent_id = NULL + FROM pending_parents pp + JOIN json_each(?) j ON pp.trace_id = json_extract(j.value, '$.traceId') + AND pp.span_id = json_extract(j.value, '$.spanId') + WHERE spans.trace_id = pp.trace_id + AND spans.span_id = pp.span_id + AND spans.parent_id = pp.parent_id`, + ids + ); + this.sql.exec( + `UPDATE spans SET parent_id = pp.parent_id + FROM pending_parents pp + JOIN json_each(?) j ON pp.trace_id = json_extract(j.value, '$.traceId') + AND pp.parent_id = json_extract(j.value, '$.spanId') + WHERE spans.trace_id = pp.trace_id AND spans.span_id = pp.span_id`, + ids + ); + this.sql.exec( + `DELETE FROM pending_parents + WHERE (trace_id, parent_id) IN ( + SELECT json_extract(value, '$.traceId'), json_extract(value, '$.spanId') + FROM json_each(?) + )`, + ids + ); + } + // Sequence allocation and inserts share this transaction, so concurrent // invocations for one distributed trace cannot collide. const nextSeq = new Map(); @@ -336,5 +404,6 @@ export class TraceStore extends DurableObject { clear(): void { this.sql.exec("DELETE FROM logs"); this.sql.exec("DELETE FROM spans"); + this.sql.exec("DELETE FROM pending_parents"); } } diff --git a/packages/miniflare/test/plugins/core/observability.spec.ts b/packages/miniflare/test/plugins/core/observability.spec.ts index a664d58c17f..178e06387a8 100644 --- a/packages/miniflare/test/plugins/core/observability.spec.ts +++ b/packages/miniflare/test/plugins/core/observability.spec.ts @@ -6,13 +6,14 @@ import { singleModuleManifest, useDispose } from "../../test-shared"; // Local observability wiring. // -// When `unsafeObservability` is enabled, Miniflare core attaches the collector -// to each user worker (as a tail consumer) and adds the compatibility flags -// workerd needs to emit those tail events. The caller sets nothing per worker; -// wrangler and the Vite plugin just turn on the one option. The wiring tests -// below check that this produces a valid, working config. The store tests call -// the `TraceStore` Durable Object and the collector's read API directly over -// RPC, separately from tail capture. +// When `unsafeObservability` is enabled, Miniflare makes the collector workerd's +// OTLP collector (`tracing.otlp`), so it receives every invocation's spans, and +// attaches it to each user worker as a tail consumer for its logs, adding the +// compatibility flags workerd needs to emit those tail events. The caller sets +// nothing per worker; wrangler and the Vite plugin just turn on the one option. +// The wiring tests below check that this produces a valid, working config. The +// store tests call the `TraceStore` Durable Object and the collector's read API +// directly over RPC, separately from capture. function plainWorker(script: string): WorkerOptions { return { @@ -230,6 +231,23 @@ export default { }], []); return new Response("repersisted"); } + if (url.pathname === "/pending") { + const store = env.TRACE_STORE.get(env.TRACE_STORE.idFromName("singleton")); + const span = (spanId, parentId, startMs) => ({ + traceId: "trace-pending", spanId, parentId, service: "user", + name: spanId, kind: "http", startMs, durationMs: 1, outcome: "ok", + error: null, attributes: { "faas.trigger": "http" }, + }); + // First a callee, whose caller is exported later, and an invocation + // started by a span that is never stored; then the caller. + await store.persist( + url.searchParams.get("caller") + ? [span("caller", null, 9000)] + : [span("callee", "caller", 9001), span("started", "dropped", 9002)], + [] + ); + return new Response("persisted"); + } if (url.pathname.startsWith("/wobs/")) { return env.WOBS.fetch( new Request("http://collector" + url.pathname.slice("/wobs".length) + url.search, request) @@ -393,6 +411,37 @@ interface WtSpan { attributes: string | null; } +describe("unsafeObservability (parents)", () => { + test("keeps a span whose parent isn't stored as a root until the parent arrives", async ({ + expect, + }) => { + const mf = new Miniflare({ + unsafeObservability: true, + workers: [storeWorker()], + }); + useDispose(mf); + const parents = async () => + Object.fromEntries( + ( + await queryStore( + mf, + "SELECT span_id, parent_id FROM spans WHERE trace_id = 'trace-pending'" + ) + ).map((row) => [row.span_id, row.parent_id]) + ); + + await mf.dispatchFetch("http://localhost/pending"); + expect(await parents()).toEqual({ callee: null, started: null }); + + await mf.dispatchFetch("http://localhost/pending?caller=1"); + expect(await parents()).toEqual({ + caller: null, + callee: "caller", + started: null, + }); + }); +}); + describe("unsafeObservability (write-through capture)", () => { test("opens spans in-flight, merges streamed attributes, and finalises on close", async ({ expect, @@ -508,10 +557,10 @@ describe("unsafeObservability (write-through capture)", () => { }); // A plain user worker (no manual seeding): it does some work, logs, and proxies -// `/wobs/*` to the collector's read API. With `unsafeObservability`, core attaches -// the collector as a streaming-tail consumer, so the worker's invocation is folded -// (via cf-to-otel) into spans/logs and persisted — nothing observability-specific -// is wired here by hand. +// `/wobs/*` to the collector's read API. With `unsafeObservability`, workerd +// exports the worker's spans to the collector and streams its tail there, so the +// invocation is folded into spans/logs and persisted — nothing +// observability-specific is wired here by hand. const CAPTURE_WORKER = `export default { async fetch(request, env) { const url = new URL(request.url); @@ -551,7 +600,7 @@ async function flush() { await new Promise((resolve) => setTimeout(resolve, 1000)); } -describe("unsafeObservability (capture via cf-to-otel)", () => { +describe("unsafeObservability (capture via OTLP)", () => { test("captures a worker's trace, child spans, and logs", async ({ expect, }) => { @@ -573,7 +622,7 @@ describe("unsafeObservability (capture via cf-to-otel)", () => { mf, TRACE_LIST_SQL )) as unknown as TraceRow[]; - // cf-to-otel names the root fetch span after the method. + // workerd names the invocation's fetch span after the method. const completed = traces.filter( (t) => t.name === "GET" && t.outcome !== null ); @@ -599,7 +648,7 @@ describe("unsafeObservability (capture via cf-to-otel)", () => { expect(logs.some((l) => l.message.includes("handled request"))).toBe(true); expect(logs.some((l) => l.message.includes("/orders"))).toBe(true); - // The root span carries the cf-to-otel-shaped attributes, incl. the HTTP + // The root span carries the attributes workerd exports, incl. the HTTP // status folded in as an attribute (not a column) per the schema. const root = spans.find((s) => { const a = JSON.parse(s.attributes ?? "{}"); @@ -611,6 +660,13 @@ describe("unsafeObservability (capture via cf-to-otel)", () => { expect(attrs["http.response.status_code"]).toBe(200); expect(attrs["cloudflare.outcome"]).toBe("ok"); expect(typeof attrs["cpu_time_ms"]).toBe("number"); + + // Only the user worker's spans are kept: workerd also exports those of + // Miniflare's own Workers (such as the entry Worker the request came + // through), so the invocation, which they started, is the trace's root. + const services = await queryStore(mf, "SELECT DISTINCT service FROM spans"); + expect(services).toEqual([{ service: "user" }]); + expect(root).toMatchObject({ parent_id: null }); }); test("captures an uncaught exception as a span error and error log", async ({ @@ -643,9 +699,9 @@ describe("unsafeObservability (capture via cf-to-otel)", () => { }); }); -// Two user workers with a service binding between them. The collector is attached -// to *every* user worker, so both invocations are captured; each span records its -// owning worker (`service`) for multi-worker attribution. +// Two user workers with a service binding between them. The collector keeps the +// spans of *every* user worker, so both invocations are captured; each span +// records its owning worker (`service`) for multi-worker attribution. const DOWNSTREAM_WORKER = `export default { async fetch(request, env) { await env.CACHE.get("b-key"); @@ -710,8 +766,9 @@ interface DetailSpan { // A Workflow, plus a fetch handler that kicks off an instance and proxies // `/wobs/*` to the collector. The Workflows engine runs as its own service (not -// through the core plugin's per-user-worker path), so it's wired for tailing -// separately; without that, a workflow run leaves no trace in the store. +// through the core plugin's per-user-worker path), so the collector is told to +// keep its spans separately; without that, a workflow run leaves no trace in the +// store. const WORKFLOW_CAPTURE_WORKER = ` import { WorkflowEntrypoint } from "cloudflare:workers"; export class CaptureWorkflow extends WorkflowEntrypoint { @@ -766,7 +823,7 @@ describe("unsafeObservability (workflows)", () => { await (await mf.dispatchFetch("http://localhost/start")).text(); // The engine runs the workflow asynchronously; poll until its spans land - // (attributed to the workflow `name` we pass through the collector props). + // (attributed to the workflow `name`). const begin = performance.now(); let workflowSpans: Record[] = []; while (performance.now() - begin < 5000) { @@ -781,7 +838,6 @@ describe("unsafeObservability (workflows)", () => { await new Promise((resolve) => setTimeout(resolve, 100)); } - // Before the engine service was wired for tailing this was always empty. expect(workflowSpans.length).toBeGreaterThan(0); expect(workflowSpans.every((s) => s.service === "capture-workflow")).toBe( true