feat(sdk-logs): implement log 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-logs): implement log creation metrics  [#6433](https://github.com/open-telemetry/opentelemetry-js/pull/6433) @anuraaga
+* feat(sdk-logs): implement log processor metrics  [#6554](https://github.com/open-telemetry/opentelemetry-js/pull/6554) @anuraaga
 
 ### :bug: Bug Fixes
 

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

@@ -249,8 +249,6 @@ export class NodeSDK {
       diag.warn(
         "The 'logRecordProcessor' option is deprecated. Please use 'logRecordProcessors' instead."
       );
-    } else {
-      this.configureLoggerProviderFromEnv();
     }
 
     if (configuration.metricReaders) {
@@ -355,6 +353,12 @@ export class NodeSDK {
       trace.setGlobalTracerProvider(this._tracerProvider);
     }
 
+    if (!this._loggerProviderConfig) {
+      this.configureLoggerProviderFromEnv(
+        sdkMetricsEnabled ? this._meterProvider : undefined
+      );
+    }
+
     if (this._loggerProviderConfig) {
       const loggerProvider = new LoggerProvider({
         ...getLoggerProviderConfigFromEnv(),
@@ -388,7 +392,9 @@ export class NodeSDK {
     );
   }
 
-  private configureLoggerProviderFromEnv(): void {
+  private configureLoggerProviderFromEnv(
+    meterProvider: MeterProvider | undefined
+  ): void {
     const enabledExporters = Array.from(
       new Set(getStringListFromEnv('OTEL_LOGS_EXPORTER') ?? [])
     );
@@ -444,9 +450,9 @@ export class NodeSDK {
       this._loggerProviderConfig = {
         logRecordProcessors: exporters.map(exporter => {
           if (exporter instanceof ConsoleLogRecordExporter) {
-            return new SimpleLogRecordProcessor(exporter);
+            return new SimpleLogRecordProcessor(exporter, { meterProvider });
           } else {
-            return getBatchLogRecordProcessorFromEnv(exporter);
+            return getBatchLogRecordProcessorFromEnv(exporter, meterProvider);
           }
         }),
       };

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

@@ -56,6 +56,7 @@ import type {
 import type {
   AggregationOption,
   IMetricReader,
+  MeterProvider,
   PushMetricExporter,
   ViewOptions,
 } from '@opentelemetry/sdk-metrics';
@@ -578,12 +579,13 @@ export function getBatchLogRecordProcessorConfigFromEnv(): BufferConfig {
 }
 
 export function getBatchLogRecordProcessorFromEnv(
-  exporter: LogRecordExporter
+  exporter: LogRecordExporter,
+  meterProvider: MeterProvider | undefined
 ): BatchLogRecordProcessor {
-  return new BatchLogRecordProcessor(
-    exporter,
-    getBatchLogRecordProcessorConfigFromEnv()
-  );
+  return new BatchLogRecordProcessor(exporter, {
+    ...getBatchLogRecordProcessorConfigFromEnv(),
+    meterProvider,
+  });
 }
 
 export function getLogRecordExporter(

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

@@ -411,27 +411,14 @@ describe('Node SDK', () => {
 
     it('should configure components for SDK metrics if enabled', async () => {
       process.env.OTEL_NODE_EXPERIMENTAL_SDK_METRICS = 'true';
-      const exporter = new ConsoleMetricExporter();
-      const metricReader = new PeriodicExportingMetricReader({
-        exporter: exporter,
-        exportIntervalMillis: 100,
-        exportTimeoutMillis: 100,
-      });
+      process.env.OTEL_TRACES_EXPORTER = 'console';
+      process.env.OTEL_LOGS_EXPORTER = 'console';
+      process.env.OTEL_METRICS_EXPORTER = 'console';
 
-      const sdk = new NodeSDK({
-        metricReader: metricReader,
-        traceExporter: new ConsoleSpanExporter(),
-        logRecordProcessors: [
-          new SimpleLogRecordProcessor(new InMemoryLogRecordExporter()),
-        ],
-        autoDetectResources: false,
-      });
+      const sdk = new NodeSDK();
 
       sdk.start();
 
-      assertDefaultContextManagerRegistered();
-      assertDefaultPropagatorRegistered();
-
       assert.strictEqual(setGlobalTracerProviderSpy.callCount, 1);
       const tracerProvider = setGlobalTracerProviderSpy.lastCall.args[0];
       assert.ok(tracerProvider instanceof NodeTracerProvider);
@@ -444,13 +431,21 @@ describe('Node SDK', () => {
         (loggerProvider as any)['_sharedState'].loggerMetrics.createdLogs,
         NOOP_COUNTER_METRIC
       );
+      assert.notDeepEqual(
+        (loggerProvider as any)['_sharedState'].registeredLogRecordProcessors[0]
+          ._metrics.processedLogs,
+        NOOP_COUNTER_METRIC
+      );
 
       assert.ok(metrics.getMeterProvider() instanceof MeterProvider);
 
       await sdk.shutdown();
     });
 
-    it('should not configure components for SDK metrics if disabled', async () => {
+    it('should configure initialized components for SDK metrics if enabled', async () => {
+      process.env.OTEL_NODE_EXPERIMENTAL_SDK_METRICS = 'true';
+      process.env.OTEL_LOGS_EXPORTER = 'console';
+
       const exporter = new ConsoleMetricExporter();
       const metricReader = new PeriodicExportingMetricReader({
         exporter: exporter,
@@ -461,16 +456,42 @@ describe('Node SDK', () => {
       const sdk = new NodeSDK({
         metricReader: metricReader,
         traceExporter: new ConsoleSpanExporter(),
-        logRecordProcessors: [
-          new SimpleLogRecordProcessor(new InMemoryLogRecordExporter()),
-        ],
         autoDetectResources: false,
       });
 
       sdk.start();
 
-      assertDefaultContextManagerRegistered();
-      assertDefaultPropagatorRegistered();
+      assert.strictEqual(setGlobalTracerProviderSpy.callCount, 1);
+      const tracerProvider = setGlobalTracerProviderSpy.lastCall.args[0];
+      assert.ok(tracerProvider instanceof NodeTracerProvider);
+      assert.ok(
+        (tracerProvider as any)._config.meterProvider instanceof MeterProvider
+      );
+
+      const loggerProvider = setGlobalLoggerProviderSpy.lastCall.args[0];
+      assert.notDeepEqual(
+        (loggerProvider as any)['_sharedState'].loggerMetrics.createdLogs,
+        NOOP_COUNTER_METRIC
+      );
+      assert.notDeepEqual(
+        (loggerProvider as any)['_sharedState'].registeredLogRecordProcessors[0]
+          ._metrics.processedLogs,
+        NOOP_COUNTER_METRIC
+      );
+
+      assert.ok(metrics.getMeterProvider() instanceof MeterProvider);
+
+      await sdk.shutdown();
+    });
+
+    it('should not configure components for SDK metrics if disabled', async () => {
+      process.env.OTEL_TRACES_EXPORTER = 'console';
+      process.env.OTEL_LOGS_EXPORTER = 'console';
+      process.env.OTEL_METRICS_EXPORTER = 'console';
+
+      const sdk = new NodeSDK();
+
+      sdk.start();
 
       assert.strictEqual(setGlobalTracerProviderSpy.callCount, 1);
       const tracerProvider = setGlobalTracerProviderSpy.lastCall.args[0];
@@ -482,6 +503,11 @@ describe('Node SDK', () => {
         (loggerProvider as any)['_sharedState'].loggerMetrics.createdLogs,
         NOOP_COUNTER_METRIC
       );
+      assert.deepEqual(
+        (loggerProvider as any)['_sharedState'].registeredLogRecordProcessors[0]
+          ._metrics.processedLogs,
+        NOOP_COUNTER_METRIC
+      );
 
       assert.ok(metrics.getMeterProvider() instanceof MeterProvider);
 

experimental/packages/sdk-logs/src/export/BatchLogRecordProcessorBase.ts

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import { context, diag } from '@opentelemetry/api';
+import { context, createNoopMeter, diag } from '@opentelemetry/api';
 import {
   ExportResultCode,
   globalErrorHandler,
@@ -15,6 +15,9 @@ import type { BufferConfig } from '../types';
 import type { SdkLogRecord } from './SdkLogRecord';
 import type { LogRecordExporter } from './LogRecordExporter';
 import type { LogRecordProcessor } from '../LogRecordProcessor';
+import type { LogRecordProcessorConfig } from './LogRecordProcessorConfig';
+import { LogRecordProcessorMetrics } from './LogRecordProcessorMetrics';
+import { OTEL_COMPONENT_TYPE_VALUE_BATCHING_LOG_PROCESSOR } from '../semconv';
 
 /**
  * Waits for all pending async resources in the log records to be resolved.
@@ -42,12 +45,14 @@ async function waitForResources(logRecords: SdkLogRecord[]): Promise<void> {
 class ExportOperation {
   private readonly _exportCompleted: Promise<void>;
   private readonly _exportScheduledPromise: Promise<void>;
+  private readonly _metrics: LogRecordProcessorMetrics;
   private _exportScheduledResolve!: () => void;
 
   constructor(
     exporter: LogRecordExporter,
     logRecords: SdkLogRecord[],
-    exportTimeoutMillis: number
+    exportTimeoutMillis: number,
+    metrics: LogRecordProcessorMetrics
   ) {
     this._exportScheduledPromise = new Promise<void>(resolve => {
       this._exportScheduledResolve = resolve;
@@ -57,6 +62,7 @@ class ExportOperation {
       logRecords,
       exportTimeoutMillis
     );
+    this._metrics = metrics;
   }
 
   /** Get the promise that resolves when the export completes */
@@ -106,6 +112,7 @@ class ExportOperation {
 
       // Call exporter.export() and immediately resolve exportScheduled
       exporter.export(logRecords, result => {
+        this._metrics.finishLogs(logRecords.length, result.error);
         clearTimeout(timer);
         if (result.code === ExportResultCode.SUCCESS) {
           resolve();
@@ -131,14 +138,18 @@ export abstract class BatchLogRecordProcessorBase<T extends BufferConfig>
   private readonly _scheduledDelayMillis: number;
   private readonly _exportTimeoutMillis: number;
   private readonly _exporter: LogRecordExporter;
+  private readonly _metrics: LogRecordProcessorMetrics;
 
   private _currentExport: ExportOperation | null = null;
   private _finishedLogRecords: SdkLogRecord[] = [];
   private _timer: NodeJS.Timeout | number | undefined;
   private _shutdownOnce: BindOnceFuture<void>;
   private _flushing: boolean = false;
 
-  constructor(exporter: LogRecordExporter, config?: T) {
+  constructor(
+    exporter: LogRecordExporter,
+    config?: T & LogRecordProcessorConfig
+  ) {
     this._exporter = exporter;
     this._maxExportBatchSize = config?.maxExportBatchSize ?? 512;
     this._maxQueueSize = config?.maxQueueSize ?? 2048;
@@ -153,6 +164,19 @@ export abstract class BatchLogRecordProcessorBase<T extends BufferConfig>
       );
       this._maxExportBatchSize = this._maxQueueSize;
     }
+
+    const meter = config?.meterProvider
+      ? config.meterProvider.getMeter('@opentelemetry/sdk-logs')
+      : createNoopMeter();
+
+    this._metrics = new LogRecordProcessorMetrics(
+      OTEL_COMPONENT_TYPE_VALUE_BATCHING_LOG_PROCESSOR,
+      meter,
+      {
+        capacity: this._maxQueueSize,
+        getQueueSize: () => this._finishedLogRecords.length,
+      }
+    );
   }
 
   public onEmit(logRecord: SdkLogRecord): void {
@@ -172,6 +196,7 @@ export abstract class BatchLogRecordProcessorBase<T extends BufferConfig>
   /** Add a LogRecord in the buffer. */
   private _addToBuffer(logRecord: SdkLogRecord) {
     if (this._finishedLogRecords.length >= this._maxQueueSize) {
+      this._metrics.dropLogs(1);
       return;
     }
     this._finishedLogRecords.push(logRecord);
@@ -185,6 +210,7 @@ export abstract class BatchLogRecordProcessorBase<T extends BufferConfig>
   private async _shutdown(): Promise<void> {
     this.onShutdown();
     await this._flushAll();
+    this._metrics.shutdown();
     await this._exporter.shutdown();
   }
 
@@ -230,7 +256,8 @@ export abstract class BatchLogRecordProcessorBase<T extends BufferConfig>
       const exportOp = new ExportOperation(
         this._exporter,
         batch,
-        this._exportTimeoutMillis
+        this._exportTimeoutMillis,
+        this._metrics
       );
       this._currentExport = exportOp;
 
@@ -279,7 +306,8 @@ export abstract class BatchLogRecordProcessorBase<T extends BufferConfig>
     const exportOp = new ExportOperation(
       this._exporter,
       logRecords,
-      this._exportTimeoutMillis
+      this._exportTimeoutMillis,
+      this._metrics
     );
     this._currentExport = exportOp;
 

experimental/packages/sdk-logs/src/export/LogRecordProcessorConfig.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 log record processors.
+ */
+export interface LogRecordProcessorConfig {
+  meterProvider?: MeterProvider;
+}