fix(sdk-node): pass OTLP HTTP metric config

experimental/CHANGELOG.md

@@ -21,6 +21,7 @@ For notes on migrating to 2.x / 0.200.x see [the upgrade guide](doc/upgrade-to-2
 
 ### :bug: Bug Fixes
 
+* fix(sdk-node): pass OTLP HTTP metric exporter configuration from declarative config [#6665](https://github.com/open-telemetry/opentelemetry-js/issues/6665) @cyphercodes
 * fix(sdk-logs): default BatchLogRecordProcessor `scheduleDelayMillis` is 1000 [#6796](https://github.com/open-telemetry/opentelemetry-js/pull/6796) @trentm
 * fix(configuration): percent-decode keys and values in `resource.attributes_list` per spec [#6787](https://github.com/open-telemetry/opentelemetry-js/pull/6787) @MikeGoldsmith
 * fix(configuration): default `log_level` to `info` in env-based config initialization for consistency with file-based config [#6788](https://github.com/open-telemetry/opentelemetry-js/pull/6788) @MikeGoldsmith

experimental/packages/configuration/src/index.ts

@@ -27,6 +27,8 @@ export type {
   SimpleLogRecordProcessor as SimpleLogRecordProcessorConfigModel,
   OtlpHttpExporter as OtlpHttpExporterConfigModel,
   OtlpGrpcExporter as OtlpGrpcExporterConfigModel,
+  OtlpHttpMetricExporter as OtlpHttpMetricExporterConfigModel,
+  ExporterTemporalityPreference as ExporterTemporalityPreferenceConfigModel,
   LogRecordLimits as LogRecordLimitsConfigModel,
 } from './generated/types';
 export { createConfigFactory } from './ConfigFactory';

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

@@ -10,6 +10,7 @@ import {
   getNumberFromEnv,
   getStringFromEnv,
   getStringListFromEnv,
+  parseKeyPairsIntoRecord,
   W3CBaggagePropagator,
   W3CTraceContextPropagator,
 } from '@opentelemetry/core';
@@ -74,11 +75,14 @@ import type {
   LogRecordProcessorConfigModel,
   OtlpGrpcExporterConfigModel,
   OtlpHttpExporterConfigModel,
+  OtlpHttpMetricExporterConfigModel,
+  ExporterTemporalityPreferenceConfigModel,
   SimpleLogRecordProcessorConfigModel,
   LogRecordLimitsConfigModel,
 } from '@opentelemetry/configuration';
 import type {
   AggregationOption,
+  AggregationSelector,
   IAttributesProcessor,
   IMetricReader,
   PushMetricExporter,
@@ -93,7 +97,10 @@ import {
   PeriodicExportingMetricReader,
 } from '@opentelemetry/sdk-metrics';
 import { OTLPMetricExporter as OTLPGrpcMetricExporter } from '@opentelemetry/exporter-metrics-otlp-grpc';
-import { OTLPMetricExporter as OTLPHttpMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http';
+import {
+  AggregationTemporalityPreference,
+  OTLPMetricExporter as OTLPHttpMetricExporter,
+} from '@opentelemetry/exporter-metrics-otlp-http';
 import { OTLPMetricExporter as OTLPProtoMetricExporter } from '@opentelemetry/exporter-metrics-otlp-proto';
 import type {
   BatchLogRecordProcessorOptions,
@@ -531,6 +538,65 @@ export function getOtlpMetricExporterFromEnv(): PushMetricExporter {
   return new OTLPProtoMetricExporter();
 }
 
+function getMetricExporterCompression(
+  compression: string | null | undefined
+): CompressionAlgorithm {
+  return compression === 'gzip'
+    ? CompressionAlgorithm.GZIP
+    : CompressionAlgorithm.NONE;
+}
+
+function getMetricExporterTemporalityPreference(
+  temporalityPreference: ExporterTemporalityPreferenceConfigModel | undefined
+): AggregationTemporalityPreference | undefined {
+  switch (temporalityPreference) {
+    case 'delta':
+      return AggregationTemporalityPreference.DELTA;
+    case 'low_memory':
+      return AggregationTemporalityPreference.LOWMEMORY;
+    case 'cumulative':
+      return AggregationTemporalityPreference.CUMULATIVE;
+    default:
+      return undefined;
+  }
+}
+
+function getMetricExporterDefaultHistogramAggregation(
+  defaultHistogramAggregation: NonNullable<OtlpHttpMetricExporterConfigModel>['default_histogram_aggregation']
+): AggregationSelector | undefined {
+  if (defaultHistogramAggregation !== 'base2_exponential_bucket_histogram') {
+    return undefined;
+  }
+
+  return instrumentType => {
+    if (instrumentType === InstrumentType.HISTOGRAM) {
+      return { type: AggregationType.EXPONENTIAL_HISTOGRAM };
+    }
+    return { type: AggregationType.DEFAULT };
+  };
+}
+
+function getOtlpHttpMetricExporterConfig(
+  config: NonNullable<OtlpHttpMetricExporterConfigModel>
+) {
+  return {
+    compression: getMetricExporterCompression(config.compression),
+    url: config.endpoint ?? undefined,
+    headers: getHeadersFromConfiguration(
+      config.headers,
+      config.headers_list ?? undefined
+    ),
+    timeoutMillis: validateExporterTimeout(config.timeout),
+    httpAgentOptions: getHttpAgentOptionsFromTls(config.tls),
+    temporalityPreference: getMetricExporterTemporalityPreference(
+      config.temporality_preference
+    ),
+    aggregationPreference: getMetricExporterDefaultHistogramAggregation(
+      config.default_histogram_aggregation
+    ),
+  };
+}
+
 function getMetricProducersFromConfiguration(
   producers: MetricProducerConfigModel[] | undefined
 ): MetricProducer[] | undefined {
@@ -563,32 +629,22 @@ export function getPeriodicMetricReaderFromConfiguration(
 ): IMetricReader | undefined {
   if (periodic.exporter) {
     let exporter;
-    if (periodic.exporter.otlp_http !== undefined) {
-      const encoding = periodic.exporter.otlp_http?.encoding ?? 'protobuf';
+    if (periodic.exporter.otlp_http) {
+      const config = periodic.exporter.otlp_http;
+      const encoding = config.encoding ?? 'protobuf';
+      const exporterConfig = getOtlpHttpMetricExporterConfig(config);
       if (encoding === 'json') {
-        exporter = new OTLPHttpMetricExporter({
-          compression:
-            periodic.exporter.otlp_http?.compression === 'gzip'
-              ? CompressionAlgorithm.GZIP
-              : CompressionAlgorithm.NONE,
-        });
+        exporter = new OTLPHttpMetricExporter(exporterConfig);
       } else if (encoding === 'protobuf') {
-        exporter = new OTLPProtoMetricExporter({
-          compression:
-            periodic.exporter.otlp_http?.compression === 'gzip'
-              ? CompressionAlgorithm.GZIP
-              : CompressionAlgorithm.NONE,
-        });
+        exporter = new OTLPProtoMetricExporter(exporterConfig);
       } else {
         diag.warn(`Unsupported OTLP metrics encoding: ${encoding}.`);
       }
     }
-    if (periodic.exporter.otlp_grpc !== undefined) {
+    if (periodic.exporter.otlp_grpc) {
+      const config = periodic.exporter.otlp_grpc;
       exporter = new OTLPGrpcMetricExporter({
-        compression:
-          periodic.exporter.otlp_grpc?.compression === 'gzip'
-            ? CompressionAlgorithm.GZIP
-            : CompressionAlgorithm.NONE,
+        compression: getMetricExporterCompression(config.compression),
       });
     }
 
@@ -788,13 +844,14 @@ export function createLogRecordProcessorFromConfig(
 }
 
 export function getHeadersFromConfiguration(
-  headers: NameStringValuePairConfigModel[] | undefined
+  headers: NameStringValuePairConfigModel[] | undefined,
+  headersList?: string
 ): Record<string, string> | undefined {
-  if (!headers) {
+  if (!headers && headersList === undefined) {
     return undefined;
   }
-  const result: Record<string, string> = {};
-  headers.forEach(header => {
+  const result: Record<string, string> = parseKeyPairsIntoRecord(headersList);
+  headers?.forEach(header => {
     if (header.value !== null) {
       result[header.name] = header.value;
     }

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

@@ -33,7 +33,52 @@ import {
   serviceInstanceIdDetector,
 } from '@opentelemetry/resources';
 import type { LoggerProviderOptions } from '@opentelemetry/sdk-logs';
-import { AggregationType, InstrumentType } from '@opentelemetry/sdk-metrics';
+import type { AggregationOption } from '@opentelemetry/sdk-metrics';
+import {
+  AggregationTemporality,
+  AggregationType,
+  InstrumentType,
+} from '@opentelemetry/sdk-metrics';
+
+interface OtlpHttpTransportParameters {
+  url: string;
+  compression: string;
+  headers: () => Promise<Record<string, string>>;
+  agentFactory: (protocol: string) => Promise<{
+    options: {
+      ca?: Buffer;
+      cert?: Buffer;
+      key?: Buffer;
+    };
+  }>;
+}
+
+interface OtlpMetricExporterInternals {
+  _delegate: {
+    _timeout: number;
+    _transport: {
+      _transport: {
+        _parameters: OtlpHttpTransportParameters;
+      };
+    };
+  };
+  selectAggregation(instrumentType: InstrumentType): AggregationOption;
+  selectAggregationTemporality(
+    instrumentType: InstrumentType
+  ): AggregationTemporality;
+}
+
+function getMetricExporterInternals(
+  reader: unknown
+): OtlpMetricExporterInternals {
+  return (reader as any)._exporter;
+}
+
+function getHttpTransportParameters(
+  exporter: OtlpMetricExporterInternals
+): OtlpHttpTransportParameters {
+  return exporter._delegate._transport._transport._parameters;
+}
 
 describe('getPropagatorFromEnv', function () {
   afterEach(() => {
@@ -296,6 +341,95 @@ describe('getLoggerProviderConfigFromEnv', function () {
   });
 });
 
+describe('getPeriodicMetricReaderFromConfiguration', function () {
+  afterEach(function () {
+    sinon.restore();
+  });
+
+  it('should return warning message for invalid compression type for meter provider', function () {
+    const warnStub = sinon.stub(diag, 'warn');
+    getPeriodicMetricReaderFromConfiguration({
+      exporter: { otlp_http: { encoding: 'invalid' } },
+    } as any);
+    sinon.assert.calledWithExactly(
+      warnStub,
+      'Unsupported OTLP metrics encoding: invalid.'
+    );
+  });
+
+  it('passes OTLP HTTP metric exporter connection options from configuration', async function () {
+    const reader = getPeriodicMetricReaderFromConfiguration({
+      interval: 7000,
+      timeout: 5000,
+      exporter: {
+        otlp_http: {
+          endpoint: 'https://collector.example/v1/metrics',
+          tls: {
+            ca_file: 'test/fixtures/ca.pem',
+            key_file: 'test/fixtures/ca-key.pem',
+            cert_file: 'test/fixtures/cert.pem',
+          },
+          headers_list: 'x-list=list-value,x-overridden=list-value',
+          headers: [
+            { name: 'x-test-header', value: 'test-value' },
+            { name: 'x-overridden', value: 'header-value' },
+          ],
+          compression: 'gzip',
+          timeout: 1234,
+        },
+      },
+    });
+
+    assert.ok(reader);
+    const exporter = getMetricExporterInternals(reader);
+    const parameters = getHttpTransportParameters(exporter);
+
+    assert.strictEqual(parameters.url, 'https://collector.example/v1/metrics');
+    assert.strictEqual(parameters.compression, 'gzip');
+    assert.strictEqual(exporter._delegate._timeout, 1234);
+    assert.deepStrictEqual(await parameters.headers(), {
+      'x-list': 'list-value',
+      'x-overridden': 'header-value',
+      'x-test-header': 'test-value',
+      'Content-Type': 'application/x-protobuf',
+    });
+
+    const agent = await parameters.agentFactory('https:');
+    assert.ok(agent.options.ca);
+    assert.ok(agent.options.key);
+    assert.ok(agent.options.cert);
+  });
+
+  it('passes OTLP HTTP metric exporter aggregation options from configuration', function () {
+    const reader = getPeriodicMetricReaderFromConfiguration({
+      exporter: {
+        otlp_http: {
+          encoding: 'json',
+          temporality_preference: 'delta',
+          default_histogram_aggregation: 'base2_exponential_bucket_histogram',
+        },
+      },
+    });
+
+    assert.ok(reader);
+    const exporter = getMetricExporterInternals(reader);
+
+    assert.strictEqual(
+      exporter.selectAggregationTemporality(InstrumentType.COUNTER),
+      AggregationTemporality.DELTA
+    );
+    assert.deepStrictEqual(
+      exporter.selectAggregation(InstrumentType.HISTOGRAM),
+      {
+        type: AggregationType.EXPONENTIAL_HISTOGRAM,
+      }
+    );
+    assert.deepStrictEqual(exporter.selectAggregation(InstrumentType.COUNTER), {
+      type: AggregationType.DEFAULT,
+    });
+  });
+});
+
 describe('getBatchLogRecordProcessorConfigFromEnv', function () {
   afterEach(function () {
     delete process.env.OTEL_BLRP_MAX_QUEUE_SIZE;
@@ -420,17 +554,6 @@ describe('getBatchLogRecordProcessorConfigFromEnv', function () {
     sinon.assert.callCount(warnStub, 4);
   });
 
-  it('should return warning message for invalid compression type for meter provider', function () {
-    const warnStub = sinon.stub(diag, 'warn');
-    getPeriodicMetricReaderFromConfiguration({
-      exporter: { otlp_http: { encoding: 'invalid' } },
-    } as any);
-    sinon.assert.calledWithExactly(
-      warnStub,
-      'Unsupported OTLP metrics encoding: invalid.'
-    );
-  });
-
   it('should return values for getInstrumentType', function () {
     assert.deepStrictEqual(
       getInstrumentType('counter' as InstrumentTypeConfigModel),