feat(sdk-trace): implement span processor metrics

CHANGELOG.md

@@ -17,7 +17,8 @@ For notes on migrating to 2.x / 0.200.x see [the upgrade guide](doc/upgrade-to-2
 
 ### :rocket: Features
 
-* feat(sdk-logs): implement log creation metrics [#6433](https://github.com/open-telemetry/opentelemetry-js/pull/6433) @anuraaga
+* feat(sdk-trace): implement span processor metrics [#6504](https://github.com/open-telemetry/opentelemetry-js/pull/6504) @anuraaga
+* feat(sdk-logs): implement log creation metrics  [#6433](https://github.com/open-telemetry/opentelemetry-js/pull/6433) @anuraaga
 
 ### :bug: Bug Fixes
 

experimental/packages/opentelemetry-sdk-node/src/sdk.ts

@@ -39,7 +39,10 @@ import {
   ConsoleMetricExporter,
   PeriodicExportingMetricReader,
 } from '@opentelemetry/sdk-metrics';
-import type { SpanProcessor } from '@opentelemetry/sdk-trace-base';
+import type {
+  SpanExporter,
+  SpanProcessor,
+} from '@opentelemetry/sdk-trace-base';
 import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-base';
 import type { NodeTracerConfig } from '@opentelemetry/sdk-trace-node';
 import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
@@ -65,7 +68,8 @@ import {
 
 type TracerProviderConfig = {
   tracerConfig: NodeTracerConfig;
-  spanProcessors: SpanProcessor[];
+  spanProcessors: SpanProcessor[] | undefined;
+  traceExporter: SpanExporter | undefined;
 };
 
 export type MeterProviderConfig = {
@@ -225,16 +229,16 @@ export class NodeSDK {
         );
       }
 
-      const spanProcessor =
-        configuration.spanProcessor ??
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        new BatchSpanProcessor(configuration.traceExporter!);
-
-      const spanProcessors = configuration.spanProcessors ?? [spanProcessor];
+      const spanProcessors =
+        configuration.spanProcessors ??
+        (configuration.spanProcessor
+          ? [configuration.spanProcessor]
+          : undefined);
 
       this._tracerProviderConfig = {
         tracerConfig: tracerProviderConfig,
         spanProcessors,
+        traceExporter: configuration.traceExporter,
       };
     }
 
@@ -340,9 +344,23 @@ export class NodeSDK {
       }
     }
 
-    const spanProcessors = this._tracerProviderConfig
-      ? this._tracerProviderConfig.spanProcessors
-      : getSpanProcessorsFromEnv();
+    let spanProcessors: SpanProcessor[];
+    if (this._tracerProviderConfig) {
+      // If tracerProviderConfig is set, either spanProcessors or traceExporter is set.
+      if (this._tracerProviderConfig.spanProcessors) {
+        spanProcessors = this._tracerProviderConfig.spanProcessors;
+      } else {
+        spanProcessors = [
+          new BatchSpanProcessor(this._tracerProviderConfig.traceExporter!, {
+            meterProvider: sdkMetricsEnabled ? this._meterProvider : undefined,
+          }),
+        ];
+      }
+    } else {
+      spanProcessors = getSpanProcessorsFromEnv(
+        sdkMetricsEnabled ? this._meterProvider : undefined
+      );
+    }
 
     // Only register if there is a span processor
     if (spanProcessors.length > 0) {

experimental/packages/opentelemetry-sdk-node/src/utils.ts

@@ -3,7 +3,11 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import type { ContextManager, TextMapPropagator } from '@opentelemetry/api';
+import type {
+  ContextManager,
+  MeterProvider,
+  TextMapPropagator,
+} from '@opentelemetry/api';
 import { context, diag, propagation } from '@opentelemetry/api';
 import {
   CompositePropagator,
@@ -177,7 +181,9 @@ function getOtlpExporterFromEnv(): SpanExporter {
   }
 }
 
-export function getSpanProcessorsFromEnv(): SpanProcessor[] {
+export function getSpanProcessorsFromEnv(
+  meterProvider: MeterProvider | undefined
+): SpanProcessor[] {
   const exportersMap = new Map<string, () => SpanExporter>([
     ['otlp', () => getOtlpExporterFromEnv()],
     ['zipkin', () => new ZipkinExporter()],
@@ -220,9 +226,9 @@ export function getSpanProcessorsFromEnv(): SpanProcessor[] {
 
   for (const exp of exporters) {
     if (exp instanceof ConsoleSpanExporter) {
-      processors.push(new SimpleSpanProcessor(exp));
+      processors.push(new SimpleSpanProcessor(exp, { meterProvider }));
     } else {
-      processors.push(new BatchSpanProcessor(exp));
+      processors.push(new BatchSpanProcessor(exp, { meterProvider }));
     }
   }
 

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

@@ -419,7 +419,7 @@ describe('Node SDK', () => {
       });
 
       const sdk = new NodeSDK({
-        metricReader: metricReader,
+        metricReaders: [metricReader],
         traceExporter: new ConsoleSpanExporter(),
         logRecordProcessors: [
           new SimpleLogRecordProcessor(new InMemoryLogRecordExporter()),
@@ -438,6 +438,11 @@ describe('Node SDK', () => {
       assert.ok(
         (tracerProvider as any)._config.meterProvider instanceof MeterProvider
       );
+      assert.notDeepEqual(
+        (tracerProvider as any)._activeSpanProcessor._spanProcessors[0]._metrics
+          .processedSpans,
+        NOOP_COUNTER_METRIC
+      );
 
       const loggerProvider = setGlobalLoggerProviderSpy.lastCall.args[0];
       assert.notDeepEqual(
@@ -476,6 +481,11 @@ describe('Node SDK', () => {
       const tracerProvider = setGlobalTracerProviderSpy.lastCall.args[0];
       assert.ok(tracerProvider instanceof NodeTracerProvider);
       assert.equal((tracerProvider as any)._config.meterProvider, undefined);
+      assert.deepEqual(
+        (tracerProvider as any)._activeSpanProcessor._spanProcessors[0]._metrics
+          .processedSpans,
+        NOOP_COUNTER_METRIC
+      );
 
       const loggerProvider = setGlobalLoggerProviderSpy.lastCall.args[0];
       assert.deepEqual(

packages/opentelemetry-sdk-trace-base/src/export/BatchSpanProcessorBase.ts

@@ -4,7 +4,7 @@
  */
 
 import type { Context } from '@opentelemetry/api';
-import { context, diag, TraceFlags } from '@opentelemetry/api';
+import { context, createNoopMeter, diag, TraceFlags } from '@opentelemetry/api';
 import {
   BindOnceFuture,
   ExportResultCode,
@@ -14,9 +14,12 @@ import {
 } from '@opentelemetry/core';
 import type { Span } from '../Span';
 import type { SpanProcessor } from '../SpanProcessor';
+import type { SpanProcessorConfig } from './SpanProcessorConfig';
 import type { BufferConfig } from '../types';
 import type { ReadableSpan } from './ReadableSpan';
 import type { SpanExporter } from './SpanExporter';
+import { SpanProcessorMetrics } from './SpanProcessorMetrics';
+import { OTEL_COMPONENT_TYPE_VALUE_BATCHING_SPAN_PROCESSOR } from '../semconv';
 
 /**
  * Implementation of the {@link SpanProcessor} that batches spans exported by
@@ -30,14 +33,15 @@ export abstract class BatchSpanProcessorBase<T extends BufferConfig>
   private readonly _scheduledDelayMillis: number;
   private readonly _exportTimeoutMillis: number;
   private readonly _exporter: SpanExporter;
+  private readonly _metrics: SpanProcessorMetrics;
 
   private _isExporting = false;
   private _finishedSpans: ReadableSpan[] = [];
   private _timer: NodeJS.Timeout | number | undefined;
   private _shutdownOnce: BindOnceFuture<void>;
   private _droppedSpansCount: number = 0;
 
-  constructor(exporter: SpanExporter, config?: T) {
+  constructor(exporter: SpanExporter, config?: T & SpanProcessorConfig) {
     this._exporter = exporter;
     this._maxExportBatchSize =
       typeof config?.maxExportBatchSize === 'number'
@@ -64,6 +68,18 @@ export abstract class BatchSpanProcessorBase<T extends BufferConfig>
       );
       this._maxExportBatchSize = this._maxQueueSize;
     }
+
+    const meter = config?.meterProvider
+      ? config.meterProvider.getMeter('@opentelemetry/sdk-trace')
+      : createNoopMeter();
+    this._metrics = new SpanProcessorMetrics(
+      OTEL_COMPONENT_TYPE_VALUE_BATCHING_SPAN_PROCESSOR,
+      meter,
+      {
+        capacity: this._maxQueueSize,
+        getQueueSize: () => this._finishedSpans.length,
+      }
+    );
   }
 
   forceFlush(): Promise<void> {
@@ -101,6 +117,7 @@ export abstract class BatchSpanProcessorBase<T extends BufferConfig>
         return this._flushAll();
       })
       .then(() => {
+        this._metrics.shutdown();
         return this._exporter.shutdown();
       });
   }
@@ -114,6 +131,7 @@ export abstract class BatchSpanProcessorBase<T extends BufferConfig>
         diag.debug('maxQueueSize reached, dropping spans');
       }
       this._droppedSpansCount++;
+      this._metrics.dropSpans(1);
 
       return;
     }
@@ -179,6 +197,7 @@ export abstract class BatchSpanProcessorBase<T extends BufferConfig>
         const doExport = () =>
           this._exporter.export(spans, result => {
             clearTimeout(timer);
+            this._metrics.finishSpans(spans.length, result.error);
             if (result.code === ExportResultCode.SUCCESS) {
               resolve();
             } else {

packages/opentelemetry-sdk-trace-base/src/export/SimpleSpanProcessor.ts

@@ -4,7 +4,7 @@
  */
 
 import type { Context } from '@opentelemetry/api';
-import { TraceFlags } from '@opentelemetry/api';
+import { createNoopMeter, TraceFlags } from '@opentelemetry/api';
 import {
   internal,
   ExportResultCode,
@@ -13,8 +13,11 @@ import {
 } from '@opentelemetry/core';
 import type { Span } from '../Span';
 import type { SpanProcessor } from '../SpanProcessor';
+import type { SpanProcessorConfig } from './SpanProcessorConfig';
 import type { ReadableSpan } from './ReadableSpan';
 import type { SpanExporter } from './SpanExporter';
+import { SpanProcessorMetrics } from './SpanProcessorMetrics';
+import { OTEL_COMPONENT_TYPE_VALUE_SIMPLE_SPAN_PROCESSOR } from '../semconv';
 
 /**
  * An implementation of the {@link SpanProcessor} that converts the {@link Span}
@@ -26,13 +29,22 @@ import type { SpanExporter } from './SpanExporter';
  */
 export class SimpleSpanProcessor implements SpanProcessor {
   private readonly _exporter: SpanExporter;
+  private readonly _metrics: SpanProcessorMetrics;
   private _shutdownOnce: BindOnceFuture<void>;
   private _pendingExports: Set<Promise<void>>;
 
-  constructor(exporter: SpanExporter) {
+  constructor(exporter: SpanExporter, config?: SpanProcessorConfig) {
     this._exporter = exporter;
     this._shutdownOnce = new BindOnceFuture(this._shutdown, this);
     this._pendingExports = new Set<Promise<void>>();
+
+    const meter = config?.meterProvider
+      ? config.meterProvider.getMeter('@opentelemetry/sdk-trace')
+      : createNoopMeter();
+    this._metrics = new SpanProcessorMetrics(
+      OTEL_COMPONENT_TYPE_VALUE_SIMPLE_SPAN_PROCESSOR,
+      meter
+    );
   }
 
   async forceFlush(): Promise<void> {
@@ -70,6 +82,7 @@ export class SimpleSpanProcessor implements SpanProcessor {
     }
 
     const result = await internal._export(this._exporter, [span]);
+    this._metrics.finishSpans(1, result.error);
     if (result.code !== ExportResultCode.SUCCESS) {
       throw (
         result.error ??
@@ -83,6 +96,7 @@ export class SimpleSpanProcessor implements SpanProcessor {
   }
 
   private _shutdown(): Promise<void> {
+    this._metrics.shutdown();
     return this._exporter.shutdown();
   }
 }

packages/opentelemetry-sdk-trace-base/src/export/SpanProcessorConfig.ts

@@ -0,0 +1,13 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import type { MeterProvider } from '@opentelemetry/api';
+
+/**
+ * Common options for SDK span processors.
+ */
+export interface SpanProcessorConfig {
+  meterProvider?: MeterProvider;
+}