feat(node): Simplify SentrySpanProcessor (#11273)

Instead of extending `BatchSpanProcessor`, we handle this ourselves now.

There may be additional improvements we can do here, but this should be
OK for now!

I added two tests to cover what we really want to achieve/show here:

1. Ensure that if a span & child span are ended after each other, but in
the same tick, they are correctly sent.
2. Ensure that if a child span is ended later, it is skipped.

By skipping the batched span processor, we can simplify our flushing a
bit and also get rid of some of the delays.

For now I kept the exporter as a separate class (it is not exported
anyhow, so purely internal), we can eventually look into merging this or
not, but this is just a moving-parts-around exercise then. This way,
tests continued to work mostly, which is good I'd say.

Author: mydea

dev-packages/node-integration-tests/suites/tracing-experimental/hapi/test.ts

@@ -45,6 +45,11 @@ describe('hapi auto-instrumentation', () => {
 
   test('CJS - should handle returned plain errors in routes.', done => {
     createRunner(__dirname, 'scenario.js')
+      .expect({
+        transaction: {
+          transaction: 'GET /error',
+        },
+      })
       .expect({ event: EXPECTED_ERROR_EVENT })
       .expectError()
       .start(done)
@@ -53,6 +58,11 @@ describe('hapi auto-instrumentation', () => {
 
   test('CJS - should handle returned Boom errors in routes.', done => {
     createRunner(__dirname, 'scenario.js')
+      .expect({
+        transaction: {
+          transaction: 'GET /boom-error',
+        },
+      })
       .expect({ event: EXPECTED_ERROR_EVENT })
       .expectError()
       .start(done)
@@ -61,6 +71,11 @@ describe('hapi auto-instrumentation', () => {
 
   test('CJS - should handle promise rejections in routes.', done => {
     createRunner(__dirname, 'scenario.js')
+      .expect({
+        transaction: {
+          transaction: 'GET /promise-error',
+        },
+      })
       .expect({ event: EXPECTED_ERROR_EVENT })
       .expectError()
       .start(done)

dev-packages/node-integration-tests/suites/tracing-experimental/mysql/scenario-withoutCallback.js

@@ -24,7 +24,7 @@ connection.connect(function (err) {
   }
 });
 
-Sentry.startSpan(
+Sentry.startSpanManual(
   {
     op: 'transaction',
     name: 'Test Transaction',

packages/node-experimental/test/helpers/mockSdkInit.ts

@@ -1,6 +1,7 @@
 import { ProxyTracerProvider, context, propagation, trace } from '@opentelemetry/api';
 import { BasicTracerProvider } from '@opentelemetry/sdk-trace-base';
-import { getCurrentScope, getGlobalScope, getIsolationScope } from '@sentry/core';
+import { getClient, getCurrentScope, getGlobalScope, getIsolationScope } from '@sentry/core';
+import type { NodeClient } from '../../src';
 
 import { init } from '../../src/sdk/init';
 import type { NodeClientOptions } from '../../src/types';
@@ -36,7 +37,7 @@ export function cleanupOtel(_provider?: BasicTracerProvider): void {
 }
 
 export function getProvider(_provider?: BasicTracerProvider): BasicTracerProvider | undefined {
-  let provider = _provider || trace.getTracerProvider();
+  let provider = _provider || getClient<NodeClient>()?.traceProvider || trace.getTracerProvider();
 
   if (provider instanceof ProxyTracerProvider) {
     provider = provider.getDelegate();

packages/node-experimental/test/integration/transactions.test.ts

@@ -614,7 +614,6 @@ describe('Integration | Transactions', () => {
 
     mockSdkInit({ enableTracing: true, beforeSendTransaction });
 
-    const client = Sentry.getClient()!;
     const provider = getProvider();
     const multiSpanProcessor = provider?.activeSpanProcessor as
       | (SpanProcessor & { _spanProcessors?: SpanProcessor[] })
@@ -645,24 +644,17 @@ describe('Integration | Transactions', () => {
       await new Promise(resolve => setTimeout(resolve, 10 * 60 * 1000));
     });
 
-    // Nothing added to exporter yet
-    expect(exporter['_finishedSpans'].length).toBe(0);
-
-    void client.flush(5_000);
-    jest.advanceTimersByTime(5_000);
-
-    // Now the child-spans have been added to the exporter, but they are pending since they are waiting for their parant
+    // Child-spans have been added to the exporter, but they are pending since they are waiting for their parant
     expect(exporter['_finishedSpans'].length).toBe(2);
     expect(beforeSendTransaction).toHaveBeenCalledTimes(0);
 
     // Now wait for 5 mins
-    jest.advanceTimersByTime(5 * 60 * 1_000);
+    jest.advanceTimersByTime(5 * 60 * 1_000 + 1);
 
     // Adding another span will trigger the cleanup
     Sentry.startSpan({ name: 'other span' }, () => {});
 
-    void client.flush(5_000);
-    jest.advanceTimersByTime(5_000);
+    jest.advanceTimersByTime(1);
 
     // Old spans have been cleared away
     expect(exporter['_finishedSpans'].length).toBe(0);
@@ -672,7 +664,8 @@ describe('Integration | Transactions', () => {
 
     expect(logs).toEqual(
       expect.arrayContaining([
-        'SpanExporter exported 0 spans, 2 unsent spans remaining',
+        'SpanExporter has 1 unsent spans remaining',
+        'SpanExporter has 2 unsent spans remaining',
         'SpanExporter exported 1 spans, 2 unsent spans remaining',
         `SpanExporter dropping span inner span 1 (${innerSpan1Id}) because it is pending for more than 5 minutes.`,
         `SpanExporter dropping span inner span 2 (${innerSpan2Id}) because it is pending for more than 5 minutes.`,

packages/node-experimental/test/sdk/api.test.ts

@@ -1,3 +1,4 @@
+import type { Event } from '@sentry/types';
 import { getActiveSpan, getClient, startInactiveSpan, startSpan, withActiveSpan } from '../../src';
 import { cleanupOtel, mockSdkInit } from '../helpers/mockSdkInit';
 
@@ -64,7 +65,11 @@ describe('withActiveSpan()', () => {
   });
 
   it('when `null` is passed, should start a new trace for new spans', async () => {
-    const beforeSendTransaction = jest.fn(() => null);
+    const transactions: Event[] = [];
+    const beforeSendTransaction = jest.fn((event: Event) => {
+      transactions.push(event);
+      return null;
+    });
     mockSdkInit({ enableTracing: true, beforeSendTransaction });
     const client = getClient();
 

packages/opentelemetry/src/spanExporter.ts

@@ -1,8 +1,6 @@
 import type { Span } from '@opentelemetry/api';
 import { SpanKind } from '@opentelemetry/api';
-import type { ExportResult } from '@opentelemetry/core';
-import { ExportResultCode } from '@opentelemetry/core';
-import type { ReadableSpan, SpanExporter } from '@opentelemetry/sdk-trace-base';
+import type { ReadableSpan } from '@opentelemetry/sdk-trace-base';
 import { SemanticAttributes } from '@opentelemetry/semantic-conventions';
 import { captureEvent, getMetricSummaryJsonForSpan } from '@sentry/core';
 import {
@@ -21,6 +19,7 @@ import { convertOtelTimeToSeconds } from './utils/convertOtelTimeToSeconds';
 import { getDynamicSamplingContextFromSpan } from './utils/dynamicSamplingContext';
 import { getRequestSpanData } from './utils/getRequestSpanData';
 import type { SpanNode } from './utils/groupSpansWithParents';
+import { getLocalParentId } from './utils/groupSpansWithParents';
 import { groupSpansWithParents } from './utils/groupSpansWithParents';
 import { mapStatus } from './utils/mapStatus';
 import { parseSpanDescription } from './utils/parseSpanDescription';
@@ -31,38 +30,72 @@ type SpanNodeCompleted = SpanNode & { span: ReadableSpan };
 /**
  * A Sentry-specific exporter that converts OpenTelemetry Spans to Sentry Spans & Transactions.
  */
-export class SentrySpanExporter implements SpanExporter {
+export class SentrySpanExporter {
+  private _flushTimeout: ReturnType<typeof setTimeout> | undefined;
   private _finishedSpans: ReadableSpan[];
-  private _stopped: boolean;
 
   public constructor() {
-    this._stopped = false;
     this._finishedSpans = [];
   }
 
-  /** @inheritDoc */
-  public export(spans: ReadableSpan[], resultCallback: (result: ExportResult) => void): void {
-    if (this._stopped) {
-      return resultCallback({
-        code: ExportResultCode.FAILED,
-        error: new Error('Exporter has been stopped'),
-      });
+  /** Export a single span. */
+  public export(span: ReadableSpan): void {
+    this._finishedSpans.push(span);
+
+    // If the span has a local parent ID, we don't need to export anything just yet
+    if (getLocalParentId(span)) {
+      const openSpanCount = this._finishedSpans.length;
+      DEBUG_BUILD && logger.log(`SpanExporter has ${openSpanCount} unsent spans remaining`);
+      this._cleanupOldSpans();
+      return;
     }
 
-    const openSpanCount = this._finishedSpans.length;
-    const newSpanCount = spans.length;
+    this._clearTimeout();
+
+    // If we got a parent span, we try to send the span tree
+    // Wait a tick for this, to ensure we avoid race conditions
+    this._flushTimeout = setTimeout(() => {
+      this.flush();
+    }, 1);
+  }
+
+  /** Try to flush any pending spans immediately. */
+  public flush(): void {
+    this._clearTimeout();
 
-    this._finishedSpans.push(...spans);
+    const openSpanCount = this._finishedSpans.length;
 
     const remainingSpans = maybeSend(this._finishedSpans);
 
     const remainingOpenSpanCount = remainingSpans.length;
-    const sentSpanCount = openSpanCount + newSpanCount - remainingOpenSpanCount;
+    const sentSpanCount = openSpanCount - remainingOpenSpanCount;
 
     DEBUG_BUILD &&
       logger.log(`SpanExporter exported ${sentSpanCount} spans, ${remainingOpenSpanCount} unsent spans remaining`);
 
-    this._finishedSpans = remainingSpans.filter(span => {
+    this._cleanupOldSpans(remainingSpans);
+  }
+
+  /** Clear the exporter. */
+  public clear(): void {
+    this._finishedSpans = [];
+    this._clearTimeout();
+  }
+
+  /** Clear the flush timeout. */
+  private _clearTimeout(): void {
+    if (this._flushTimeout) {
+      clearTimeout(this._flushTimeout);
+      this._flushTimeout = undefined;
+    }
+  }
+
+  /**
+   * Remove any span that is older than 5min.
+   * We do this to avoid leaking memory.
+   */
+  private _cleanupOldSpans(spans = this._finishedSpans): void {
+    this._finishedSpans = spans.filter(span => {
       const shouldDrop = shouldCleanupSpan(span, 5 * 60);
       DEBUG_BUILD &&
         shouldDrop &&
@@ -73,23 +106,6 @@ export class SentrySpanExporter implements SpanExporter {
         );
       return !shouldDrop;
     });
-
-    resultCallback({ code: ExportResultCode.SUCCESS });
-  }
-
-  /** @inheritDoc */
-  public shutdown(): Promise<void> {
-    const forceFlush = this.forceFlush();
-    this._stopped = true;
-    this._finishedSpans = [];
-    return forceFlush;
-  }
-
-  /** @inheritDoc */
-  public forceFlush(): Promise<void> {
-    return new Promise(resolve => {
-      this.export(this._finishedSpans, () => resolve());
-    });
   }
 }
 
@@ -132,8 +148,12 @@ function maybeSend(spans: ReadableSpan[]): ReadableSpan[] {
     .filter((span): span is ReadableSpan => !!span);
 }
 
+function nodeIsCompletedRootNode(node: SpanNode): node is SpanNodeCompleted {
+  return !!node.span && !node.parentNode;
+}
+
 function getCompletedRootNodes(nodes: SpanNode[]): SpanNodeCompleted[] {
-  return nodes.filter((node): node is SpanNodeCompleted => !!node.span && !node.parentNode);
+  return nodes.filter(nodeIsCompletedRootNode);
 }
 
 function shouldCleanupSpan(span: ReadableSpan, maxStartTimeOffsetSeconds: number): boolean {

packages/opentelemetry/src/spanProcessor.ts

@@ -1,7 +1,6 @@
 import type { Context } from '@opentelemetry/api';
-import { ROOT_CONTEXT, trace } from '@opentelemetry/api';
-import type { Span, SpanProcessor as SpanProcessorInterface } from '@opentelemetry/sdk-trace-base';
-import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-base';
+import { ROOT_CONTEXT, TraceFlags, trace } from '@opentelemetry/api';
+import type { ReadableSpan, Span, SpanProcessor as SpanProcessorInterface } from '@opentelemetry/sdk-trace-base';
 import { addChildSpanToSpan, getClient, getDefaultCurrentScope, getDefaultIsolationScope } from '@sentry/core';
 import { logger } from '@sentry/utils';
 
@@ -46,20 +45,35 @@ function onSpanStart(span: Span, parentContext: Context): void {
   client?.emit('spanStart', span);
 }
 
-function onSpanEnd(span: Span): void {
+function onSpanEnd(span: ReadableSpan): void {
   const client = getClient();
-  client?.emit('spanEnd', span);
+  client?.emit('spanEnd', span as Span);
 }
 
 /**
  * Converts OpenTelemetry Spans to Sentry Spans and sends them to Sentry via
  * the Sentry SDK.
  */
-export class SentrySpanProcessor extends BatchSpanProcessor implements SpanProcessorInterface {
-  public constructor() {
-    super(new SentrySpanExporter());
+export class SentrySpanProcessor implements SpanProcessorInterface {
+  private _exporter: SentrySpanExporter;
 
+  public constructor() {
     setIsSetup('SentrySpanProcessor');
+    this._exporter = new SentrySpanExporter();
+  }
+
+  /**
+   * @inheritDoc
+   */
+  public async forceFlush(): Promise<void> {
+    this._exporter.flush();
+  }
+
+  /**
+   * @inheritDoc
+   */
+  public async shutdown(): Promise<void> {
+    this._exporter.clear();
   }
 
   /**
@@ -72,29 +86,19 @@ export class SentrySpanProcessor extends BatchSpanProcessor implements SpanProce
     // once we decoupled opentelemetry from SentrySpan
 
     DEBUG_BUILD && logger.log(`[Tracing] Starting span "${span.name}" (${span.spanContext().spanId})`);
-
-    return super.onStart(span, parentContext);
   }
 
   /** @inheritDoc */
-  public onEnd(span: Span): void {
-    DEBUG_BUILD && logger.log(`[Tracing] Finishing span "${span.name}" (${span.spanContext().spanId})`);
-
-    if (!this._shouldSendSpanToSentry(span)) {
-      // Prevent this being called to super.onEnd(), which would pass this to the span exporter
+  public onEnd(span: ReadableSpan): void {
+    if (span.spanContext().traceFlags !== TraceFlags.SAMPLED) {
+      DEBUG_BUILD && logger.log(`[Tracing] Finishing unsampled span "${span.name}" (${span.spanContext().spanId})`);
       return;
     }
 
-    onSpanEnd(span);
+    DEBUG_BUILD && logger.log(`[Tracing] Finishing span "${span.name}" (${span.spanContext().spanId})`);
 
-    return super.onEnd(span);
-  }
+    onSpanEnd(span);
 
-  /**
-   * You can overwrite this in a sub class to implement custom behavior for dropping spans.
-   * If you return `false` here, the span will not be passed to the exporter and thus not be sent.
-   */
-  protected _shouldSendSpanToSentry(_span: Span): boolean {
-    return true;
+    this._exporter.export(span);
   }
 }

packages/opentelemetry/src/utils/groupSpansWithParents.ts

@@ -26,14 +26,19 @@ export function groupSpansWithParents(spans: ReadableSpan[]): SpanNode[] {
   });
 }
 
-function createOrUpdateSpanNodeAndRefs(nodeMap: SpanMap, span: ReadableSpan): void {
+/**
+ * This returns the _local_ parent ID - `parentId` on the span may point to a remote span.
+ */
+export function getLocalParentId(span: ReadableSpan): string | undefined {
   const parentIsRemote = span.attributes[SEMANTIC_ATTRIBUTE_SENTRY_PARENT_IS_REMOTE] === true;
-
-  const id = span.spanContext().spanId;
-
   // If the parentId is the trace parent ID, we pretend it's undefined
   // As this means the parent exists somewhere else
-  const parentId = !parentIsRemote ? span.parentSpanId : undefined;
+  return !parentIsRemote ? span.parentSpanId : undefined;
+}
+
+function createOrUpdateSpanNodeAndRefs(nodeMap: SpanMap, span: ReadableSpan): void {
+  const id = span.spanContext().spanId;
+  const parentId = getLocalParentId(span);
 
   if (!parentId) {
     createOrUpdateNode(nodeMap, { id, span, children: [] });