|
| 1 | +import type { Attributes, Context, TextMapGetter } from '@opentelemetry/api' |
| 2 | +import { diag, SpanKind } from '@opentelemetry/api' |
| 3 | +import type { |
| 4 | + ReadableSpan, |
| 5 | + Span, |
| 6 | + SpanProcessor, |
| 7 | +} from '@opentelemetry/sdk-trace-base' |
| 8 | + |
| 9 | +import { TraceFlags } from '@opentelemetry/api' |
| 10 | + |
| 11 | +type AttributesFromHeaderFunc = <Carrier = unknown>( |
| 12 | + headers: Carrier, |
| 13 | + getter: TextMapGetter<Carrier> |
| 14 | +) => Attributes | undefined |
| 15 | + |
| 16 | +type AttributesFromHeaders = Record<string, string> | AttributesFromHeaderFunc |
| 17 | + |
| 18 | +/** |
| 19 | + * Helper ──────────────────────────────────────────────────────────────────── |
| 20 | + * Try to obtain the callback that extends the lifetime of the request so that |
| 21 | + * our exporter has time to flush. |
| 22 | + * |
| 23 | + * Priority: |
| 24 | + * 1. `after(cb)` (Next.js 15) |
| 25 | + * 2. fallback (best-effort `setTimeout`) |
| 26 | + */ |
| 27 | +function scheduleAfterResponse(task: () => Promise<void>) { |
| 28 | + try { |
| 29 | + // avoid a hard dependency so this file can be imported outside a route. |
| 30 | + // `require` is evaluated lazily – if Next isn't around it will throw. |
| 31 | + // eslint-disable-next-line |
| 32 | + const mod = require('next/server') as { after?: (cb: () => void) => void } |
| 33 | + |
| 34 | + if (typeof mod.after === 'function') { |
| 35 | + mod.after(() => { |
| 36 | + // no await – Next treats sync or async the same, |
| 37 | + // we just fire the promise and let it resolve. |
| 38 | + void task() |
| 39 | + }) |
| 40 | + return |
| 41 | + } |
| 42 | + } catch { |
| 43 | + /* ignored – we're probably not inside a Next context */ |
| 44 | + } |
| 45 | + |
| 46 | + // 2. Node / local fallback – try our best and hope the |
| 47 | + // process stays alive long enough. |
| 48 | + setTimeout(() => { |
| 49 | + void task() |
| 50 | + }, 0) |
| 51 | +} |
| 52 | + |
| 53 | +function isSampled(traceFlags: number): boolean { |
| 54 | + // Use bitwise AND to inspect the sampled flag |
| 55 | + return (traceFlags & TraceFlags.SAMPLED) !== 0 |
| 56 | +} |
| 57 | + |
| 58 | +/** Custom CompositeSpanProcessor for Next.js */ |
| 59 | +export class NextCompositeSpanProcessor implements SpanProcessor { |
| 60 | + private readonly rootSpanIds = new Map< |
| 61 | + string, |
| 62 | + { rootSpanId: string; open: Span[] } |
| 63 | + >() |
| 64 | + private readonly waitSpanEnd = new Map<string, () => void>() |
| 65 | + /** makes concurrent `forceFlush()` invocations queue instead of collide */ |
| 66 | + private flushInFlight: Promise<void> | null = null |
| 67 | + |
| 68 | + constructor( |
| 69 | + private readonly processors: SpanProcessor[], |
| 70 | + private readonly attributesFromHeaders?: AttributesFromHeaders |
| 71 | + ) {} |
| 72 | + |
| 73 | + // ───────────────────────────── infrastructure ──────────────────────────── |
| 74 | + forceFlush(): Promise<void> { |
| 75 | + // Serialise: if a flush is already happening, share that promise. |
| 76 | + if (this.flushInFlight) return this.flushInFlight |
| 77 | + |
| 78 | + const flushPromise = Promise.all( |
| 79 | + this.processors.map((p) => |
| 80 | + p.forceFlush().catch((e) => { |
| 81 | + diag.error('forceFlush failed:', e) |
| 82 | + }) |
| 83 | + ) |
| 84 | + ) |
| 85 | + .then(() => undefined) // ensure Promise<void> |
| 86 | + .catch(() => undefined) // already logged |
| 87 | + .finally(() => { |
| 88 | + this.flushInFlight = null |
| 89 | + }) |
| 90 | + |
| 91 | + this.flushInFlight = flushPromise |
| 92 | + |
| 93 | + return this.flushInFlight as Promise<void> |
| 94 | + } |
| 95 | + |
| 96 | + async shutdown(): Promise<void> { |
| 97 | + return Promise.all( |
| 98 | + this.processors.map((p) => p.shutdown().catch(() => undefined)) |
| 99 | + ).then(() => undefined) |
| 100 | + } |
| 101 | + |
| 102 | + // ────────────────────────────────── onStart ────────────────────────────── |
| 103 | + onStart(span: Span, parentContext: Context): void { |
| 104 | + const { traceId, spanId, traceFlags } = span.spanContext() |
| 105 | + const isRoot = !this.rootSpanIds.has(traceId) |
| 106 | + |
| 107 | + if (isRoot) { |
| 108 | + this.rootSpanIds.set(traceId, { rootSpanId: spanId, open: [] }) |
| 109 | + } else { |
| 110 | + this.rootSpanIds.get(traceId)?.open.push(span) |
| 111 | + } |
| 112 | + |
| 113 | + // Attach request-specific attributes only on the root span |
| 114 | + if (isRoot && isSampled(traceFlags)) { |
| 115 | + // When the *response* (or prerender) is done, flush traces. |
| 116 | + scheduleAfterResponse(async () => { |
| 117 | + if (this.rootSpanIds.has(traceId)) { |
| 118 | + // Root hasn’t finished yet – wait via onEnd(). |
| 119 | + const waiter = new Promise<void>((resolve) => |
| 120 | + this.waitSpanEnd.set(traceId, resolve) |
| 121 | + ) |
| 122 | + let timer: NodeJS.Timeout | undefined |
| 123 | + |
| 124 | + await Promise.race([ |
| 125 | + waiter, |
| 126 | + new Promise((res) => { |
| 127 | + timer = setTimeout(() => { |
| 128 | + this.waitSpanEnd.delete(traceId) |
| 129 | + res(undefined) |
| 130 | + }, 50) // same 50 ms guard as Vercel’s impl |
| 131 | + }), |
| 132 | + ]) |
| 133 | + if (timer) clearTimeout(timer) |
| 134 | + } |
| 135 | + |
| 136 | + await this.forceFlush() |
| 137 | + }) |
| 138 | + } |
| 139 | + |
| 140 | + // Fan-out start to underlying processors |
| 141 | + for (const p of this.processors) p.onStart(span, parentContext) |
| 142 | + } |
| 143 | + |
| 144 | + // ─────────────────────────────────── onEnd ─────────────────────────────── |
| 145 | + onEnd(span: ReadableSpan): void { |
| 146 | + const { traceId, spanId, traceFlags } = span.spanContext() |
| 147 | + const root = this.rootSpanIds.get(traceId) |
| 148 | + const isRoot = root?.rootSpanId === spanId |
| 149 | + |
| 150 | + // Datadog-style resource/operation name enrichment |
| 151 | + if (isSampled(traceFlags)) { |
| 152 | + const resAttrs = getResourceAttributes(span) |
| 153 | + if (resAttrs) Object.assign(span.attributes, resAttrs) |
| 154 | + } |
| 155 | + |
| 156 | + // Maintain open-span book-keeping |
| 157 | + if (isRoot) { |
| 158 | + // Root finished: no need to force-end children; they will end naturally. |
| 159 | + this.rootSpanIds.delete(traceId) |
| 160 | + } else if (root) { |
| 161 | + root.open = root.open.filter((s) => s.spanContext().spanId !== spanId) |
| 162 | + } |
| 163 | + |
| 164 | + // Fan-out end |
| 165 | + for (const p of this.processors) p.onEnd(span) |
| 166 | + |
| 167 | + // Release waiter if anyone is waiting for the root span to finish |
| 168 | + if (isRoot) { |
| 169 | + const pending = this.waitSpanEnd.get(traceId) |
| 170 | + if (pending) { |
| 171 | + this.waitSpanEnd.delete(traceId) |
| 172 | + pending() |
| 173 | + } |
| 174 | + } |
| 175 | + } |
| 176 | +} |
| 177 | + |
| 178 | +/* ───────────────────────── Helpers copied from Vercel impl ─────────────── */ |
| 179 | +const SPAN_KIND_NAME: { [k in SpanKind]: string } = { |
| 180 | + [SpanKind.INTERNAL]: 'internal', |
| 181 | + [SpanKind.SERVER]: 'server', |
| 182 | + [SpanKind.CLIENT]: 'client', |
| 183 | + [SpanKind.PRODUCER]: 'producer', |
| 184 | + [SpanKind.CONSUMER]: 'consumer', |
| 185 | +} |
| 186 | + |
| 187 | +function getResourceAttributes(span: ReadableSpan): Attributes | undefined { |
| 188 | + const { kind, attributes } = span |
| 189 | + const { |
| 190 | + 'operation.name': opName, |
| 191 | + 'resource.name': resName, |
| 192 | + 'span.type': spanTypeAttr, |
| 193 | + 'next.span_type': nextSpanType, |
| 194 | + 'http.method': httpMethod, |
| 195 | + 'http.route': httpRoute, |
| 196 | + } = attributes |
| 197 | + if (opName) return undefined |
| 198 | + |
| 199 | + const resourceName = |
| 200 | + resName ?? |
| 201 | + (httpMethod && httpRoute ? `${httpMethod} ${httpRoute}` : httpRoute) |
| 202 | + |
| 203 | + if ( |
| 204 | + kind === SpanKind.SERVER && |
| 205 | + typeof httpMethod === 'string' && |
| 206 | + typeof httpRoute === 'string' |
| 207 | + ) { |
| 208 | + return { 'operation.name': 'web.request', 'resource.name': resourceName } |
| 209 | + } |
| 210 | + |
| 211 | + const spanType = nextSpanType ?? spanTypeAttr |
| 212 | + if (typeof spanType === 'string') { |
| 213 | + return httpRoute |
| 214 | + ? { 'operation.name': spanType, 'resource.name': resourceName } |
| 215 | + : { 'operation.name': spanType } |
| 216 | + } |
| 217 | + |
| 218 | + return { |
| 219 | + 'operation.name': |
| 220 | + kind === SpanKind.INTERNAL ? 'internal' : SPAN_KIND_NAME[kind], |
| 221 | + } |
| 222 | +} |
| 223 | + |
| 224 | +function toOperationName(lib: string, name: string) { |
| 225 | + if (!lib) return name |
| 226 | + let clean = lib.replace(/[ @./]/g, '_') |
| 227 | + if (clean.startsWith('_')) clean = clean.slice(1) |
| 228 | + return name ? `${clean}.${name}` : clean |
| 229 | +} |
0 commit comments