Flush OTEL logs with Vercel waitUntil (#354)

Author: ben-fornefeld

bun.lock

@@ -49,7 +49,7 @@
     "@hookform/resolvers": "^5.2.2",
     "@marsidev/react-turnstile": "^1.4.1",
     "@next-safe-action/adapter-react-hook-form": "^2.0.0",
-    "@next/env": "^16.2.6",
+    "@next/env": "^16.2.7",
     "@opentelemetry/api": "^1.9.0",
     "@opentelemetry/auto-instrumentations-node": "^0.62.1",
     "@opentelemetry/exporter-logs-otlp-http": "^0.203.0",
@@ -110,7 +110,7 @@
     "micromatch": "^4.0.8",
     "motion": "^12.23.25",
     "nanoid": "^5.0.9",
-    "next": "^16.2.6",
+    "next": "^16.2.7",
     "next-safe-action": "^8.0.11",
     "next-themes": "^0.4.6",
     "nuqs": "^2.7.0",

package.json

@@ -0,0 +1,73 @@
+import { type Context, diag } from '@opentelemetry/api'
+import type { LogRecordProcessor, SdkLogRecord } from '@opentelemetry/sdk-logs'
+
+type WaitUntilInput = Promise<unknown> | (() => Promise<unknown>)
+
+interface VercelRequestContext {
+  waitUntil: (promiseOrFunc: WaitUntilInput) => void
+}
+
+interface VercelRequestContextReader {
+  get: () => VercelRequestContext | undefined
+}
+
+// This mirrors the private request-context hook used by @vercel/otel.
+const VERCEL_REQUEST_CONTEXT_SYMBOL = Symbol.for('@vercel/request-context')
+
+function getVercelRequestContext(): VercelRequestContext | undefined {
+  const reader = Reflect.get(globalThis, VERCEL_REQUEST_CONTEXT_SYMBOL) as
+    | VercelRequestContextReader
+    | undefined
+
+  return reader?.get()
+}
+
+export class VercelWaitUntilLogRecordProcessor implements LogRecordProcessor {
+  private readonly pendingFlushContexts = new WeakSet<VercelRequestContext>()
+
+  constructor(private readonly delegate: LogRecordProcessor) {}
+
+  onEmit(logRecord: SdkLogRecord, context?: Context): void {
+    this.delegate.onEmit(logRecord, context)
+    this.scheduleRequestFlush()
+  }
+
+  forceFlush(): Promise<void> {
+    return this.delegate.forceFlush()
+  }
+
+  shutdown(): Promise<void> {
+    return this.delegate.shutdown()
+  }
+
+  private scheduleRequestFlush(): void {
+    const requestContext = getVercelRequestContext()
+
+    if (!requestContext || this.pendingFlushContexts.has(requestContext)) {
+      return
+    }
+
+    this.pendingFlushContexts.add(requestContext)
+
+    try {
+      requestContext.waitUntil(async () => {
+        try {
+          await this.delegate.forceFlush()
+        } catch (error) {
+          diag.warn(
+            'Failed to flush OpenTelemetry logs in Vercel waitUntil',
+            error
+          )
+        } finally {
+          this.pendingFlushContexts.delete(requestContext)
+        }
+      })
+    } catch (error) {
+      this.pendingFlushContexts.delete(requestContext)
+      diag.warn(
+        'Failed to schedule OpenTelemetry log flush in Vercel waitUntil',
+        error
+      )
+    }
+  }
+}

src/core/server/observability/vercel-wait-until-log-record-processor.ts

@@ -15,6 +15,7 @@ import {
   ATTR_SERVICE_VERSION,
 } from '@opentelemetry/semantic-conventions'
 import { FetchInstrumentation } from '@vercel/otel'
+import { VercelWaitUntilLogRecordProcessor } from './core/server/observability/vercel-wait-until-log-record-processor'
 
 function parseResourceAttributes(
   resourceAttrs?: string
@@ -76,10 +77,12 @@ const sdk = new NodeSDK({
     }),
   }),
   logRecordProcessors: [
-    new BatchLogRecordProcessor(
-      new OTLPLogExporter({
-        url: `${OTEL_EXPORTER_OTLP_ENDPOINT}/v1/logs`,
-      })
+    new VercelWaitUntilLogRecordProcessor(
+      new BatchLogRecordProcessor(
+        new OTLPLogExporter({
+          url: `${OTEL_EXPORTER_OTLP_ENDPOINT}/v1/logs`,
+        })
+      )
     ),
   ],
   instrumentations: [

src/instrumentation.node.ts

@@ -0,0 +1,72 @@
+import type { Context } from '@opentelemetry/api'
+import type { LogRecordProcessor, SdkLogRecord } from '@opentelemetry/sdk-logs'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { VercelWaitUntilLogRecordProcessor } from '@/core/server/observability/vercel-wait-until-log-record-processor'
+
+type WaitUntilCallback = () => Promise<unknown>
+type WaitUntilInput = Promise<unknown> | WaitUntilCallback
+
+const requestContextSymbol = Symbol.for('@vercel/request-context')
+const logRecord = {} as SdkLogRecord
+
+class TestLogRecordProcessor implements LogRecordProcessor {
+  readonly onEmit = vi.fn()
+  readonly forceFlush = vi.fn(async () => {})
+  readonly shutdown = vi.fn(async () => {})
+}
+
+function installVercelRequestContext(
+  waitUntil: (input: WaitUntilInput) => void
+) {
+  const requestContext = {
+    waitUntil,
+  }
+
+  Reflect.set(globalThis, requestContextSymbol, {
+    get: () => requestContext,
+  })
+}
+
+afterEach(() => {
+  Reflect.deleteProperty(globalThis, requestContextSymbol)
+  vi.clearAllMocks()
+})
+
+describe('VercelWaitUntilLogRecordProcessor', () => {
+  it('delegates log records without a Vercel request context', () => {
+    const delegate = new TestLogRecordProcessor()
+    const processor = new VercelWaitUntilLogRecordProcessor(delegate)
+
+    processor.onEmit(logRecord)
+
+    expect(delegate.onEmit).toHaveBeenCalledWith(logRecord, undefined)
+    expect(delegate.forceFlush).not.toHaveBeenCalled()
+  })
+
+  it('schedules one waitUntil flush per request context', async () => {
+    const callbacks: WaitUntilCallback[] = []
+    const waitUntil = vi.fn((input: WaitUntilInput) => {
+      callbacks.push(
+        typeof input === 'function' ? input : async () => await input
+      )
+    })
+    installVercelRequestContext(waitUntil)
+
+    const delegate = new TestLogRecordProcessor()
+    const processor = new VercelWaitUntilLogRecordProcessor(delegate)
+
+    processor.onEmit(logRecord)
+    processor.onEmit(logRecord)
+
+    expect(delegate.onEmit).toHaveBeenCalledTimes(2)
+    expect(waitUntil).toHaveBeenCalledTimes(1)
+
+    await callbacks[0]?.()
+
+    expect(delegate.forceFlush).toHaveBeenCalledTimes(1)
+
+    processor.onEmit(logRecord, {} as Context)
+
+    expect(waitUntil).toHaveBeenCalledTimes(2)
+  })
+})