presets(cloudflare): bridge tracing channel events to observability custom spans

docs/2.deploy/20.providers/cloudflare.md

@@ -103,6 +103,33 @@ export default defineConfig({
 
 No manual Wrangler configuration is needed - Nitro handles it for you.
 
+### Tracing
+
+**🧪 Experimental!**
+
+When the experimental [`tracingChannel`](/config#tracingchannel) option is enabled, the Cloudflare presets report Nitro's tracing-channel events (h3 routes and middleware, srvx, unstorage operations, …) as [custom spans](https://developers.cloudflare.com/workers/observability/traces/custom-spans/), alongside Cloudflare's automatic instrumentation (fetch calls, KV reads, D1 queries, …) — no OpenTelemetry SDK required.
+
+```ts [nitro.config.ts]
+import { defineConfig } from "nitro";
+
+export default defineConfig({
+  preset: "cloudflare_module",
+  tracingChannel: true,
+});
+```
+
+Tracing must be enabled on the Worker for spans to be recorded:
+
+```jsonc [wrangler.jsonc]
+{
+  "observability": {
+    "traces": {
+      "enabled": true
+    }
+  }
+}
+```
+
 ## Cloudflare Pages
 
 **Preset:** `cloudflare_pages`

src/presets/cloudflare/preset.ts

@@ -2,7 +2,8 @@ import { defineNitroPreset } from "../_utils/preset.ts";
 import { writeFile } from "../_utils/fs.ts";
 import type { Nitro } from "nitro/types";
 import type { Plugin } from "rollup";
-import { resolve } from "pathe";
+import { join, resolve } from "pathe";
+import { presetsDir } from "nitro/meta";
 import { unenvCfExternals } from "./unenv/preset.ts";
 import {
   enableNodeCompat,
@@ -141,6 +142,14 @@ export const cloudflareDev = defineNitroPreset(
     devServer: {
       runner: "miniflare",
     },
+    hooks: {
+      "build:before": (nitro) => {
+        // The bridge imports `cloudflare:workers`, only available in workerd
+        if (nitro.options.devServer.runner === "miniflare") {
+          setupTracingBridge(nitro);
+        }
+      },
+    },
   },
   {
     name: "cloudflare-dev" as const,
@@ -180,6 +189,7 @@ const cloudflareModule = defineNitroPreset(
         nitro.options.unenv.push(unenvCfExternals);
         await enableNodeCompat(nitro);
         await setupEntryExports(nitro);
+        setupTracingBridge(nitro);
       },
       async compiled(nitro: Nitro) {
         await writeWranglerConfig(nitro, "module");
@@ -219,3 +229,17 @@ export default [
   cloudflareDurable,
   cloudflareDev,
 ];
+
+/**
+ * Export tracing-channel spans as Cloudflare custom spans (`tracing.enterSpan`)
+ * Registered first (unshift) so the bridge subscribes to the traced channels at
+ * startup, before any request is handled.
+ */
+function setupTracingBridge(nitro: Nitro) {
+  if (!nitro.options.tracingChannel) {
+    return;
+  }
+  nitro.options.plugins ??= [];
+
+  nitro.options.plugins.unshift(join(presetsDir, "cloudflare/runtime/telemetry/plugin"));
+}

src/presets/cloudflare/runtime/telemetry/plugin.ts

@@ -0,0 +1,84 @@
+import { definePlugin } from "nitro";
+// Handle older compatibility dates without the custom-spans API
+import * as cloudflare from "cloudflare:workers";
+import type { Span, Tracing } from "@cloudflare/workers-types";
+import type { IAnyValue } from "#nitro/runtime/telemetry/types";
+import { subscribeTracedChannels } from "#nitro/runtime/telemetry/subscribe";
+
+// https://developers.cloudflare.com/workers/observability/traces/custom-spans/
+const tracing = (cloudflare as { tracing?: Tracing }).tracing;
+
+interface PendingSpan {
+  span: Span;
+  close: () => void;
+}
+
+/**
+ * Exports Nitro tracing-channel events as Cloudflare Workers custom spans,
+ * alongside Cloudflare's automatic instrumentation (fetch, KV, D1, …)
+ */
+export default definePlugin(() => {
+  if (typeof tracing?.enterSpan !== "function") return;
+
+  subscribeTracedChannels<PendingSpan>(
+    (info, _startTimeUnixNano, error, entry) => {
+      if (!entry) return;
+      try {
+        // Skip attribute work for unsampled requests (`head_sampling_rate`).
+        if (entry.span.isTraced) {
+          for (const { key, value } of info.attributes) {
+            const attribute = attributeValue(value);
+            if (attribute !== undefined) {
+              entry.span.setAttribute(key, attribute);
+            }
+          }
+          if (error !== undefined) {
+            recordException(entry.span, error);
+          }
+        }
+      } finally {
+        entry.close();
+      }
+    },
+    {
+      onStart(info) {
+        let close!: () => void;
+        const done = new Promise<void>((resolve) => {
+          close = resolve;
+        });
+        let entry: PendingSpan | undefined;
+        tracing.enterSpan(info.name, (span) => {
+          entry = { span, close };
+          return done;
+        });
+        return entry;
+      },
+    }
+  );
+});
+
+/** OTLP `IAnyValue` (from the shared describers) → Cloudflare attribute value. */
+function attributeValue(value: IAnyValue): string | number | boolean | undefined {
+  if (value.stringValue != null) return value.stringValue;
+  if (value.intValue != null) return value.intValue;
+  if (value.doubleValue != null) return value.doubleValue;
+  if (value.boolValue != null) return value.boolValue;
+}
+
+/**
+ * OTEL exception semconv, flattened onto span attributes — the Cloudflare API
+ * has no span events, and `setOutcome` is not available yet.
+ */
+function recordException(span: Span, error: unknown): void {
+  const err = error as Partial<Error> | undefined;
+  if (typeof err?.name === "string") {
+    span.setAttribute("exception.type", err.name);
+  }
+  span.setAttribute(
+    "exception.message",
+    typeof err?.message === "string" ? err.message : String(error)
+  );
+  if (typeof err?.stack === "string") {
+    span.setAttribute("exception.stacktrace", err.stack);
+  }
+}

src/runtime/internal/telemetry/subscribe.ts

@@ -3,16 +3,35 @@ import { TRACED_CHANNELS } from "./channels.ts";
 import { Span } from "./span.ts";
 
 /**
- * Called once per completed traced operation with the derived span info, the
- * operation start time (unix nanoseconds, as an OTLP `*UnixNano` string) and the
- * operation's error (`undefined` when it succeeded). A sink turns this into
- * whatever its platform consumes — an OTLP export, a log line, …
+ * Called once per completed traced operation with the span info (derived from
+ * the completed payload, falling back to the start-time payload for `onStart`
+ * subscriptions), the operation start time (unix nanoseconds, as an OTLP
+ * `*UnixNano` string), the operation's error (`undefined` when it succeeded)
+ * and the state returned by `onStart` (`undefined` without one). A sink turns
+ * this into whatever its platform consumes — an OTLP export, a log line, a
+ * platform span, …
  */
-export type SpanSink = (info: SpanInfo, startTimeUnixNano: string, error: unknown) => void;
+export type SpanSink<S = unknown> = (
+  info: SpanInfo,
+  startTimeUnixNano: string,
+  error: unknown,
+  state: S | undefined
+) => void;
+
+export interface SubscribeTracedChannelsOptions<S> {
+  /**
+   * Called synchronously when a traced operation starts, inside its execution
+   * context — where platform span APIs like Cloudflare's `enterSpan` must be
+   * called. Returned state is handed back to `onSpan` at completion.
+   */
+  onStart?: (info: SpanInfo) => S | undefined;
+}
 
 /**
  * Subscribes to the tracing channels declared in `TRACED_CHANNELS` (produced by
- * h3, srvx, unstorage, …) and invokes `onSpan` for each completed operation.
+ * h3, srvx, unstorage, …) and invokes `onSpan` once per traced operation:
+ * normally at `asyncEnd`, or at `end` when the traced function threw
+ * synchronously (`tracePromise` never publishes `asyncEnd` in that case).
  *
  * A `tracingChannel(<name>)` publishes to plain named channels
  * (`tracing:<name>:start`, `tracing:<name>:asyncEnd`, …). Subscribing to those
@@ -23,34 +42,74 @@ export type SpanSink = (info: SpanInfo, startTimeUnixNano: string, error: unknow
  *
  * A no-op when `node:diagnostics_channel` is unavailable (non-Node runtimes).
  */
-export function subscribeTracedChannels(onSpan: SpanSink): void {
+export function subscribeTracedChannels<S = unknown>(
+  onSpan: SpanSink<S>,
+  options?: SubscribeTracedChannelsOptions<S>
+): void {
   const diagnostics = globalThis.process?.getBuiltinModule?.("node:diagnostics_channel");
   if (!diagnostics?.subscribe) return;
 
-  // Carry the start time from `start` to `asyncEnd` without mutating the producer's context object.
-  const starts = new WeakMap<object, string>();
+  const onStart = options?.onStart;
+
+  // Carry the start time (and any start-time info / sink state) from `start`
+  // to completion without mutating the producer's context object.
+  interface Pending {
+    start: string;
+    info: SpanInfo | undefined;
+    state: S | undefined;
+  }
+  const pending = new WeakMap<object, Pending>();
 
   for (const name of Object.keys(TRACED_CHANNELS)) {
     const describe = TRACED_CHANNELS[name];
 
     diagnostics.subscribe(`tracing:${name}:start`, (message) => {
-      starts.set(message as object, Span.nowUnixNano());
+      const entry: Pending = { start: Span.nowUnixNano(), info: undefined, state: undefined };
+      if (onStart) {
+        try {
+          entry.info = describe(name, message);
+          entry.state = onStart(entry.info);
+        } catch {
+          // Malformed payload, or a sink failure (e.g. the platform refused to
+          // open a span) — no state; the completion callback still fires.
+        }
+      }
+      pending.set(message as object, entry);
     });
 
-    diagnostics.subscribe(`tracing:${name}:asyncEnd`, (message) => {
+    const complete = (message: unknown) => {
+      const entry = pending.get(message as object);
+      if (entry === undefined) return;
+      pending.delete(message as object);
+
+      // Derive span name, kind and semantic attributes from the completed
+      // operation. A describer only throws on a payload shape it doesn't
+      // recognise (a producer that changed shape); fall back to the start-time
+      // info (held for `onStart` subscriptions) so stateful sinks still get
+      // the completion and can release their span.
+      let info: SpanInfo | undefined;
       try {
-        const start = starts.get(message as object);
-        if (start === undefined) return;
-        starts.delete(message as object);
-
-        // Derive span name, kind and semantic attributes from the operation. A
-        // describer only throws on a payload shape it doesn't recognise (a
-        // producer that changed shape); drop that span via the catch below
-        // rather than emit a contentless one.
-        const info = describe(name, message);
-        onSpan(info, start, (message as { error?: unknown }).error);
+        info = describe(name, message);
+      } catch {}
+      info ??= entry.info;
+      if (info === undefined) return;
+
+      try {
+        onSpan(info, entry.start, (message as { error?: unknown }).error, entry.state);
       } catch {
-        // Malformed payload, or a sink failure — never break the traced operation.
+        // A sink failure must never break the traced operation.
+      }
+    };
+
+    diagnostics.subscribe(`tracing:${name}:asyncEnd`, complete);
+
+    // `tracePromise` never publishes `asyncEnd` when the traced function
+    // throws synchronously — only `end`, with `error` already set. In the
+    // normal async path `end` fires before the promise settles, while `error`
+    // is still unset, so the guard makes this a no-op there.
+    diagnostics.subscribe(`tracing:${name}:end`, (message) => {
+      if ((message as { error?: unknown }).error !== undefined) {
+        complete(message);
       }
     });
   }