perf(sdk-metrics): optionally capture active context for sync instruments

CHANGELOG.md

@@ -34,6 +34,7 @@ For notes on migrating to 2.x / 0.200.x see [the upgrade guide](doc/upgrade-to-2
 
 * perf(sdk-metrics): defer allocation of HrTime to accumulation creation [#6839](https://github.com/open-telemetry/opentelemetry-js/pull/6839) @legendecas
 * chore(\*): migrate use of sdk-trace-base and sdk-trace-node to sdk-trace [#6851](https://github.com/open-telemetry/opentelemetry-js/pull/6851) @trentm
+* perf(sdk-metrics): optionally capture active context for sync instruments [#6848](https://github.com/open-telemetry/opentelemetry-js/pull/6848) @legendecas
 
 ## 2.8.0
 

packages/sdk-metrics/src/Instruments.ts

@@ -16,7 +16,7 @@ import type {
   ObservableGauge,
   ObservableUpDownCounter,
 } from '@opentelemetry/api';
-import { context as contextApi, diag, ValueType } from '@opentelemetry/api';
+import { diag, ValueType } from '@opentelemetry/api';
 import type { InstrumentDescriptor } from './InstrumentDescriptor';
 import type { ObservableRegistry } from './state/ObservableRegistry';
 import type {
@@ -39,7 +39,7 @@ export class SyncInstrument {
   protected _record(
     value: number,
     attributes: Attributes = {},
-    context: Context = contextApi.active()
+    context?: Context
   ) {
     if (typeof value !== 'number') {
       diag.warn(

packages/sdk-metrics/src/state/AsyncMetricStorage.ts

@@ -28,12 +28,12 @@ export class AsyncMetricStorage<T extends Maybe<Accumulation>>
   private _aggregationCardinalityLimit?: number;
   private _deltaMetricStorage: DeltaMetricProcessor<T>;
   private _temporalMetricStorage: TemporalMetricProcessor<T>;
-  private _attributesProcessor: IAttributesProcessor;
+  private _attributesProcessor?: IAttributesProcessor;
 
   constructor(
     _instrumentDescriptor: InstrumentDescriptor,
     aggregator: Aggregator<T>,
-    attributesProcessor: IAttributesProcessor,
+    attributesProcessor: IAttributesProcessor | undefined,
     collectorHandles: MetricCollectorHandle[],
     aggregationCardinalityLimit?: number
   ) {
@@ -51,6 +51,10 @@ export class AsyncMetricStorage<T extends Maybe<Accumulation>>
   }
 
   record(measurements: AttributeHashMap<number>, observationTime: HrTime) {
+    if (this._attributesProcessor === undefined) {
+      this._deltaMetricStorage.batchCumulate(measurements, observationTime);
+      return;
+    }
     const processed = new AttributeHashMap<number>();
     for (const [attributes, value] of measurements.entries()) {
       processed.set(this._attributesProcessor.process(attributes), value);

packages/sdk-metrics/src/state/DeltaMetricProcessor.ts

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import type { Context, HrTime, Attributes } from '@opentelemetry/api';
+import type { HrTime, Attributes } from '@opentelemetry/api';
 import { millisToHrTime } from '@opentelemetry/core';
 import type { Maybe } from '../utils';
 import { hashAttributes } from '../utils';
@@ -33,12 +33,7 @@ export class DeltaMetricProcessor<T extends Maybe<Accumulation>> {
     this._overflowHashCode = hashAttributes(this._overflowAttributes);
   }
 
-  record(
-    value: number,
-    attributes: Attributes,
-    _context: Context,
-    collectionTime: number
-  ) {
+  record(value: number, attributes: Attributes, collectionTime: number) {
     let accumulation = this._activeCollectionStorage.get(attributes);
 
     if (!accumulation) {

packages/sdk-metrics/src/state/MeterSharedState.ts

@@ -20,7 +20,6 @@ import { ObservableRegistry } from './ObservableRegistry';
 import { SyncMetricStorage } from './SyncMetricStorage';
 import type { Accumulation, Aggregator } from '../aggregator/types';
 import type { IAttributesProcessor } from '../view/AttributesProcessor';
-import { createNoopAttributesProcessor } from '../view/AttributesProcessor';
 import type { MetricStorage } from './MetricStorage';
 
 /**
@@ -166,7 +165,7 @@ export class MeterSharedState {
           const storage = new MetricStorageType(
             descriptor,
             aggregator,
-            createNoopAttributesProcessor(),
+            undefined,
             [collector],
             cardinalityLimit
           ) as R;
@@ -190,7 +189,7 @@ interface MetricStorageConstructor {
   new (
     instrumentDescriptor: InstrumentDescriptor,
     aggregator: Aggregator<Maybe<Accumulation>>,
-    attributesProcessor: IAttributesProcessor,
+    attributesProcessor: IAttributesProcessor | undefined,
     collectors: MetricCollectorHandle[],
     aggregationCardinalityLimit?: number
   ): MetricStorage;

packages/sdk-metrics/src/state/MultiWritableMetricStorage.ts

@@ -4,23 +4,32 @@
  */
 
 import type { Context, Attributes } from '@opentelemetry/api';
+import { context as contextApi } from '@opentelemetry/api';
 import type { WritableMetricStorage } from './WritableMetricStorage';
 
 /**
  * Internal interface.
  */
 export class MultiMetricStorage implements WritableMetricStorage {
   private readonly _backingStorages: WritableMetricStorage[];
+  readonly hasAttributeProcessor: boolean;
+
   constructor(backingStorages: WritableMetricStorage[]) {
     this._backingStorages = backingStorages;
+    this.hasAttributeProcessor = backingStorages.some(
+      s => s.hasAttributeProcessor
+    );
   }
 
   record(
     value: number,
     attributes: Attributes,
-    context: Context,
+    context: Context | undefined,
     recordTime: number
   ) {
+    if (this.hasAttributeProcessor && context === undefined) {
+      context = contextApi.active();
+    }
     const storages = this._backingStorages;
     for (let i = 0; i < storages.length; i++) {
       storages[i].record(value, attributes, context, recordTime);

packages/sdk-metrics/src/state/SyncMetricStorage.ts

@@ -4,6 +4,7 @@
  */
 
 import type { Context, HrTime, Attributes } from '@opentelemetry/api';
+import { context as contextApi } from '@opentelemetry/api';
 import type { WritableMetricStorage } from './WritableMetricStorage';
 import type { Accumulation, Aggregator } from '../aggregator/types';
 import type { InstrumentDescriptor } from '../InstrumentDescriptor';
@@ -27,12 +28,12 @@ export class SyncMetricStorage<T extends Maybe<Accumulation>>
   private _aggregationCardinalityLimit?: number;
   private _deltaMetricStorage: DeltaMetricProcessor<T>;
   private _temporalMetricStorage: TemporalMetricProcessor<T>;
-  private _attributesProcessor: IAttributesProcessor;
+  private _attributesProcessor?: IAttributesProcessor;
 
   constructor(
     instrumentDescriptor: InstrumentDescriptor,
     aggregator: Aggregator<T>,
-    attributesProcessor: IAttributesProcessor,
+    attributesProcessor: IAttributesProcessor | undefined,
     collectorHandles: MetricCollectorHandle[],
     aggregationCardinalityLimit?: number
   ) {
@@ -47,16 +48,24 @@ export class SyncMetricStorage<T extends Maybe<Accumulation>>
       collectorHandles
     );
     this._attributesProcessor = attributesProcessor;
+    this.hasAttributeProcessor = attributesProcessor !== undefined;
   }
 
+  readonly hasAttributeProcessor: boolean;
+
   record(
     value: number,
     attributes: Attributes,
-    context: Context,
+    context: Context | undefined,
     recordTime: number
   ) {
-    attributes = this._attributesProcessor.process(attributes, context);
-    this._deltaMetricStorage.record(value, attributes, context, recordTime);
+    if (this._attributesProcessor !== undefined) {
+      attributes = this._attributesProcessor.process(
+        attributes,
+        context ?? contextApi.active()
+      );
+    }
+    this._deltaMetricStorage.record(value, attributes, recordTime);
   }
 
   /**

packages/sdk-metrics/src/state/WritableMetricStorage.ts

@@ -13,11 +13,14 @@ import type { AttributeHashMap } from './HashMap';
  * An interface representing SyncMetricStorage with type parameters removed.
  */
 export interface WritableMetricStorage {
+  /** Whether this storage has an attribute processor that needs context. */
+  hasAttributeProcessor: boolean;
+
   /** Records a measurement. */
   record(
     value: number,
     attributes: Attributes,
-    context: Context,
+    context: Context | undefined,
     recordTime: number
   ): void;
 }

packages/sdk-metrics/test/performance/benchmark/counter-add.js

@@ -4,12 +4,28 @@
  */
 
 const Benchmark = require('benchmark');
-const { MeterProvider } = require('../../../build/src');
+const {
+  MeterProvider,
+  createAllowListAttributesProcessor,
+} = require('../../../build/src');
 
 const provider = new MeterProvider();
 const meter = provider.getMeter('bench');
 const counter = meter.createCounter('bench.counter');
 
+const viewProvider = new MeterProvider({
+  views: [
+    {
+      instrumentName: 'bench.counter.view',
+      attributesProcessors: [
+        createAllowListAttributesProcessor(['service', 'route']),
+      ],
+    },
+  ],
+});
+const viewMeter = viewProvider.getMeter('bench');
+const viewCounter = viewMeter.createCounter('bench.counter.view');
+
 const fixedAttributes = {
   service: 'checkout',
   route: '/api/cart',
@@ -42,4 +58,8 @@ suite.add('Counter.add (varied attributes, 100 combos)', () => {
   counter.add(1, variedAttributes[variedIdx++ % 100]);
 });
 
+suite.add('Counter.add (with view attribute processor)', () => {
+  viewCounter.add(1, fixedAttributes);
+});
+
 suite.run();

packages/sdk-metrics/test/state/DeltaMetricProcessor.test.ts

@@ -3,7 +3,6 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import * as api from '@opentelemetry/api';
 import * as assert from 'assert';
 import { DropAggregator, SumAggregator } from '../../src/aggregator';
 import { DeltaMetricProcessor } from '../../src/state/DeltaMetricProcessor';
@@ -17,7 +16,7 @@ describe('DeltaMetricProcessor', () => {
 
       for (const value of commonValues) {
         for (const attributes of commonAttributes) {
-          metricProcessor.record(value, attributes, api.context.active(), 0);
+          metricProcessor.record(value, attributes, 0);
         }
       }
     });
@@ -27,7 +26,7 @@ describe('DeltaMetricProcessor', () => {
 
       for (const value of commonValues) {
         for (const attributes of commonAttributes) {
-          metricProcessor.record(value, attributes, api.context.active(), 0);
+          metricProcessor.record(value, attributes, 0);
         }
       }
     });
@@ -134,9 +133,9 @@ describe('DeltaMetricProcessor', () => {
     it('should export', () => {
       const metricProcessor = new DeltaMetricProcessor(new SumAggregator(true));
 
-      metricProcessor.record(1, { attribute: '1' }, api.ROOT_CONTEXT, 0);
-      metricProcessor.record(2, { attribute: '1' }, api.ROOT_CONTEXT, 1000);
-      metricProcessor.record(1, { attribute: '2' }, api.ROOT_CONTEXT, 2000);
+      metricProcessor.record(1, { attribute: '1' }, 0);
+      metricProcessor.record(2, { attribute: '1' }, 1000);
+      metricProcessor.record(1, { attribute: '2' }, 2000);
 
       let accumulations = metricProcessor.collect();
       assert.strictEqual(accumulations.size, 2);

packages/sdk-metrics/test/state/MultiWritableMetricStorage.test.ts

@@ -4,15 +4,21 @@
  */
 
 import * as api from '@opentelemetry/api';
-import type { Attributes } from '@opentelemetry/api';
+import type { Attributes, Context } from '@opentelemetry/api';
 import * as assert from 'assert';
+import { SumAggregator } from '../../src/aggregator';
+import { AggregationTemporality } from '../../src/export/AggregationTemporality';
+import type { MetricCollectorHandle } from '../../src/state/MetricCollector';
 import { MultiMetricStorage } from '../../src/state/MultiWritableMetricStorage';
+import { SyncMetricStorage } from '../../src/state/SyncMetricStorage';
 import type { WritableMetricStorage } from '../../src/state/WritableMetricStorage';
+import type { IAttributesProcessor } from '../../src/view/AttributesProcessor';
 import type { Measurement } from '../util';
 import {
   assertMeasurementEqual,
   commonAttributes,
   commonValues,
+  defaultInstrumentDescriptor,
 } from '../util';
 
 describe('MultiMetricStorage', () => {
@@ -22,13 +28,14 @@ describe('MultiMetricStorage', () => {
 
       for (const value of commonValues) {
         for (const attribute of commonAttributes) {
-          metricStorage.record(value, attribute, api.context.active(), 0);
+          metricStorage.record(value, attribute, undefined, 0);
         }
       }
     });
 
     it('record with multiple backing storages', () => {
       class TestWritableMetricStorage implements WritableMetricStorage {
+        hasAttributeProcessor = false;
         records: Measurement[] = [];
         record(
           value: number,
@@ -68,5 +75,69 @@ describe('MultiMetricStorage', () => {
         assertMeasurementEqual(backingStorage2.records[idx], expected);
       }
     });
+
+    it('should resolve active context for attribute processors', () => {
+      const deltaCollector: MetricCollectorHandle = {
+        selectAggregationTemporality: () => AggregationTemporality.DELTA,
+        selectCardinalityLimit: () => 2000,
+      };
+
+      const processor: IAttributesProcessor = {
+        process(incoming: Attributes, context?: Context) {
+          assert.strictEqual(context, api.context.active());
+          return incoming;
+        },
+      };
+
+      const storage1 = new SyncMetricStorage(
+        defaultInstrumentDescriptor,
+        new SumAggregator(true),
+        processor,
+        [deltaCollector]
+      );
+      const storage2 = new SyncMetricStorage(
+        defaultInstrumentDescriptor,
+        new SumAggregator(true),
+        processor,
+        [deltaCollector]
+      );
+      const multi = new MultiMetricStorage([storage1, storage2]);
+
+      multi.record(1, {}, undefined, 0);
+    });
+
+    it('should pass provided context to attribute processor', () => {
+      const deltaCollector: MetricCollectorHandle = {
+        selectAggregationTemporality: () => AggregationTemporality.DELTA,
+        selectCardinalityLimit: () => 2000,
+      };
+
+      const expectedContext = api.ROOT_CONTEXT.setValue(
+        api.createContextKey('test'),
+        'value'
+      );
+      const processor: IAttributesProcessor = {
+        process(incoming: Attributes, context?: Context) {
+          assert.strictEqual(context, expectedContext);
+          return incoming;
+        },
+      };
+
+      const storage1 = new SyncMetricStorage(
+        defaultInstrumentDescriptor,
+        new SumAggregator(true),
+        processor,
+        [deltaCollector]
+      );
+      const storage2 = new SyncMetricStorage(
+        defaultInstrumentDescriptor,
+        new SumAggregator(true),
+        processor,
+        [deltaCollector]
+      );
+      const multi = new MultiMetricStorage([storage1, storage2]);
+
+      multi.record(1, {}, expectedContext, 0);
+    });
   });
 });