fix(observability): flush logs with Next after (#365)

Author: ben-fornefeld

bun.lock

@@ -92,7 +92,7 @@
     "@types/micromatch": "^4.0.9",
     "@vercel/analytics": "^1.5.0",
     "@vercel/kv": "^3.0.0",
-    "@vercel/otel": "^1.13.0",
+    "@vercel/otel": "^2.1.2",
     "@vercel/speed-insights": "^1.2.0",
     "@xterm/xterm": "^6.0.0",
     "cheerio": "^1.0.0",

package.json

@@ -0,0 +1,55 @@
+import { type Context, diag } from '@opentelemetry/api'
+import type { LogRecordProcessor, SdkLogRecord } from '@opentelemetry/sdk-logs'
+import { after } from 'next/server'
+
+const NEXT_AFTER_OUTSIDE_REQUEST_ERROR_CODE = 'E468'
+
+function isAfterOutsideRequestScopeError(error: unknown): boolean {
+  return (
+    error instanceof Error &&
+    (Reflect.get(error, '__NEXT_ERROR_CODE') ===
+      NEXT_AFTER_OUTSIDE_REQUEST_ERROR_CODE ||
+      error.message.includes('`after` was called outside a request scope'))
+  )
+}
+
+export class NextAfterLogRecordProcessor implements LogRecordProcessor {
+  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 {
+    try {
+      after(async () => {
+        try {
+          await this.delegate.forceFlush()
+        } catch (error) {
+          diag.warn(
+            'Failed to flush OpenTelemetry logs in Next.js after()',
+            error
+          )
+        }
+      })
+    } catch (error) {
+      if (isAfterOutsideRequestScopeError(error)) {
+        return
+      }
+
+      diag.warn(
+        'Failed to schedule OpenTelemetry log flush in Next.js after()',
+        error
+      )
+    }
+  }
+}

src/core/server/observability/next-after-log-record-processor.ts

@@ -15,7 +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'
+import { NextAfterLogRecordProcessor } from './core/server/observability/next-after-log-record-processor'
 
 function parseResourceAttributes(
   resourceAttrs?: string
@@ -77,7 +77,7 @@ const sdk = new NodeSDK({
     }),
   }),
   logRecordProcessors: [
-    new VercelWaitUntilLogRecordProcessor(
+    new NextAfterLogRecordProcessor(
       new BatchLogRecordProcessor(
         new OTLPLogExporter({
           url: `${OTEL_EXPORTER_OTLP_ENDPOINT}/v1/logs`,

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

@@ -0,0 +1,65 @@
+import type { LogRecordProcessor, SdkLogRecord } from '@opentelemetry/sdk-logs'
+import { after } from 'next/server'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { NextAfterLogRecordProcessor } from '@/core/server/observability/next-after-log-record-processor'
+
+vi.mock('next/server', () => ({
+  after: vi.fn(),
+}))
+
+const logRecord = {} as SdkLogRecord
+const mockedAfter = vi.mocked(after)
+type AfterTask = Parameters<typeof after>[0]
+
+class TestLogRecordProcessor implements LogRecordProcessor {
+  readonly onEmit = vi.fn()
+  readonly forceFlush = vi.fn(async () => {})
+  readonly shutdown = vi.fn(async () => {})
+}
+
+function createOutsideRequestScopeError(): Error {
+  const error = new Error('`after` was called outside a request scope')
+  Reflect.set(error, '__NEXT_ERROR_CODE', 'E468')
+  return error
+}
+
+afterEach(() => {
+  mockedAfter.mockReset()
+  vi.clearAllMocks()
+})
+
+describe('NextAfterLogRecordProcessor', () => {
+  it('delegates log records outside a request scope', () => {
+    mockedAfter.mockImplementation(() => {
+      throw createOutsideRequestScopeError()
+    })
+
+    const delegate = new TestLogRecordProcessor()
+    const processor = new NextAfterLogRecordProcessor(delegate)
+
+    processor.onEmit(logRecord)
+
+    expect(delegate.onEmit).toHaveBeenCalledWith(logRecord, undefined)
+    expect(delegate.forceFlush).not.toHaveBeenCalled()
+  })
+
+  it('schedules an after() flush for each emitted log record', async () => {
+    const callbacks: Array<() => unknown> = []
+    mockedAfter.mockImplementation((task: AfterTask) => {
+      callbacks.push(typeof task === 'function' ? task : async () => await task)
+    })
+
+    const delegate = new TestLogRecordProcessor()
+    const processor = new NextAfterLogRecordProcessor(delegate)
+
+    processor.onEmit(logRecord)
+    processor.onEmit(logRecord)
+
+    expect(delegate.onEmit).toHaveBeenCalledTimes(2)
+    expect(mockedAfter).toHaveBeenCalledTimes(2)
+
+    await Promise.all(callbacks.map((callback) => callback()))
+
+    expect(delegate.forceFlush).toHaveBeenCalledTimes(2)
+  })
+})