initial StreamingSpanExporter implementation for otel

Author: Lms24

packages/core/src/spans/captureSpan.ts

@@ -25,7 +25,7 @@ import { safeSetSpanAttributes } from './spanFirstUtils';
 /**
  * Captures a span and returns it to the caller, to be enqueued for sending.
  */
-export function captureSpan(span: Span, client = getClient()): void {
+export function captureSpan(span: Span, client = getClient()): Span | void {
   if (!client) {
     DEBUG_BUILD && debug.warn('No client available to capture span.');
     return;
@@ -59,6 +59,8 @@ export function captureSpan(span: Span, client = getClient()): void {
   // or construct a fully new span object. The latter is risky because users (or we) could hold
   // references to the original span instance.
   client.emit('enqueueSpan', span);
+
+  return span;
 }
 
 function applyScopeToSegmentSpan(

packages/node/src/sdk/initOtel.ts

@@ -107,6 +107,7 @@ export function setupOtel(
     spanProcessors: [
       new SentrySpanProcessor({
         timeout: _clampSpanProcessorTimeout(client.getOptions().maxSpanWaitDuration),
+        client,
       }),
       ...(options.spanProcessors || []),
     ],

packages/opentelemetry/src/spanExporter.ts

@@ -1,7 +1,7 @@
 /* eslint-disable max-lines */
 import type { Span } from '@opentelemetry/api';
 import { SpanKind } from '@opentelemetry/api';
-import type { ReadableSpan } from '@opentelemetry/sdk-trace-base';
+import type { ReadableSpan, SpanExporter } from '@opentelemetry/sdk-trace-base';
 import { ATTR_HTTP_RESPONSE_STATUS_CODE, SEMATTRS_HTTP_STATUS_CODE } from '@opentelemetry/semantic-conventions';
 import type {
   SpanAttributes,
@@ -46,10 +46,16 @@ interface FinishedSpanBucket {
   spans: Set<ReadableSpan>;
 }
 
+export interface ISentrySpanExporter {
+  export(span: ReadableSpan): void;
+  flush(): void;
+  clear(): void;
+}
+
 /**
  * A Sentry-specific exporter that converts OpenTelemetry Spans to Sentry Spans & Transactions.
  */
-export class SentrySpanExporter {
+export class SentrySpanExporter implements ISentrySpanExporter {
   /*
    * A quick explanation on the buckets: We do bucketing of finished spans for efficiency. This span exporter is
    * accumulating spans until a root span is encountered and then it flushes all the spans that are descendants of that

packages/opentelemetry/src/spanProcessor.ts

@@ -1,8 +1,10 @@
 import type { Context } from '@opentelemetry/api';
 import { ROOT_CONTEXT, trace } from '@opentelemetry/api';
 import type { ReadableSpan, Span, SpanProcessor as SpanProcessorInterface } from '@opentelemetry/sdk-trace-base';
+import type { Client } from '@sentry/core';
 import {
   addChildSpanToSpan,
+  captureSpan,
   getClient,
   getDefaultCurrentScope,
   getDefaultIsolationScope,
@@ -11,7 +13,9 @@ import {
   setCapturedScopesOnSpan,
 } from '@sentry/core';
 import { SEMANTIC_ATTRIBUTE_SENTRY_PARENT_IS_REMOTE } from './semanticAttributes';
+import type { ISentrySpanExporter } from './spanExporter';
 import { SentrySpanExporter } from './spanExporter';
+import { StreamingSpanExporter } from './streamedSpanExporter';
 import { getScopesFromContext } from './utils/contextData';
 import { setIsSetup } from './utils/setupCheck';
 
@@ -51,24 +55,22 @@ function onSpanStart(span: Span, parentContext: Context): void {
   client?.emit('spanStart', span);
 }
 
-function onSpanEnd(span: Span): void {
-  logSpanEnd(span);
-
-  const client = getClient();
-  client?.emit('spanEnd', span);
-  client?.emit('afterSpanEnd', span);
-}
-
 /**
  * Converts OpenTelemetry Spans to Sentry Spans and sends them to Sentry via
  * the Sentry SDK.
  */
 export class SentrySpanProcessor implements SpanProcessorInterface {
-  private _exporter: SentrySpanExporter;
+  private _exporter: ISentrySpanExporter;
+  private _client: Client | undefined;
 
-  public constructor(options?: { timeout?: number }) {
+  public constructor(options?: { timeout?: number; client?: Client }) {
     setIsSetup('SentrySpanProcessor');
-    this._exporter = new SentrySpanExporter(options);
+    this._client = options?.client ?? getClient();
+    if (this._client?.getOptions().traceLifecycle === 'stream') {
+      this._exporter = new StreamingSpanExporter(this._client, { flushInterval: options?.timeout });
+    } else {
+      this._exporter = new SentrySpanExporter(options);
+    }
   }
 
   /**
@@ -94,8 +96,16 @@ export class SentrySpanProcessor implements SpanProcessorInterface {
 
   /** @inheritDoc */
   public onEnd(span: Span & ReadableSpan): void {
-    onSpanEnd(span);
+    logSpanEnd(span);
+
+    this._client?.emit('spanEnd', span);
 
-    this._exporter.export(span);
+    if (this._client?.getOptions().traceLifecycle === 'stream') {
+      // we probably don't need to emit afterSpanEnd here but can call captureSpan directly.
+      // might need to revisit but let's see.
+      captureSpan(span, this._client);
+    } else {
+      this._exporter.export(span);
+    }
   }
 }

packages/vercel-edge/src/sdk.ts

@@ -170,6 +170,7 @@ export function setupOtel(client: VercelEdgeClient): void {
     spanProcessors: [
       new SentrySpanProcessor({
         timeout: client.getOptions().maxSpanWaitDuration,
+        client,
       }),
     ],
   });