feat: configuration of metric export interval via environment variables

Plus: integration test for metrics.

Author: basti1302

test/integration/ChildProcessWrapper.ts

@@ -172,10 +172,17 @@ export function defaultAppConfiguration(appPort: number): ChildProcessWrapperOpt
     env: {
       ...process.env,
       PORT: appPort.toString(),
+
       // have the Node.js SDK send spans every 20 ms instead of every 5 seconds to speed up tests
       OTEL_BSP_SCHEDULE_DELAY: '20',
+
       // have the Node.js SDK send logs every 20 ms instead of every 5 seconds to speed up tests
       OTEL_BLRP_SCHEDULE_DELAY: '20',
+
+      // have the Node.js SDK send metrics every 100 ms instead of every 60 seconds to speed up tests
+      OTEL_METRIC_EXPORT_INTERVAL: '100',
+      OTEL_METRIC_EXPORT_TIMEOUT: '90',
+
       DASH0_OTEL_COLLECTOR_BASE_URL: 'http://localhost:4318',
     },
   };

test/integration/test.ts

@@ -9,8 +9,14 @@ import semver from 'semver';
 
 import { SeverityNumber } from '../collector/types/opentelemetry/proto/logs/v1/logs';
 import delay from '../util/delay';
-import { expectLogRecordAttribute, expectResourceAttribute, expectSpanAttribute } from '../util/expectAttribute';
+import {
+  expectLogRecordAttribute,
+  expectMetricDataPointAttribute,
+  expectResourceAttribute,
+  expectSpanAttribute,
+} from '../util/expectAttribute';
 import { expectMatchingLogRecord } from '../util/expectMatchingLogRecord';
+import { expectMatchingMetric } from '../util/expectMatchingMetric';
 import { expectMatchingSpan, expectMatchingSpanInFileDump } from '../util/expectMatchingSpan';
 import runCommand from '../util/runCommand';
 import waitUntil from '../util/waitUntil';
@@ -71,6 +77,33 @@ describe('attach', () => {
       });
     });
 
+    it('should attach via --require and capture metrics', async () => {
+      await waitUntil(async () => {
+        const metrics = await sendRequestAndWaitForMetrics();
+        expectMatchingMetric(
+          metrics,
+          [
+            resource => expectResourceAttribute(resource, 'telemetry.sdk.name', 'opentelemetry'),
+            resource => expectResourceAttribute(resource, 'telemetry.sdk.language', 'nodejs'),
+            resource => expectResourceAttribute(resource, 'telemetry.distro.name', 'dash0-nodejs'),
+            resource => expectResourceAttribute(resource, 'telemetry.distro.version', expectedDistroVersion),
+          ],
+          [
+            metric => expect(metric.name).to.equal('http.server.duration'),
+            metric => {
+              const dataPoints = metric.histogram?.data_points;
+              expect(dataPoints).to.exist;
+              expect(dataPoints).to.not.be.empty;
+              dataPoints?.forEach(dataPoint => {
+                expectMetricDataPointAttribute(dataPoint, 'http.method', 'GET');
+                expectMetricDataPointAttribute(dataPoint, 'http.route', '/ohai');
+              });
+            },
+          ],
+        );
+      });
+    });
+
     it('should attach via --require and capture logs', async () => {
       await waitUntil(async () => {
         const logs = await sendRequestAndWaitForLogRecords();
@@ -377,6 +410,11 @@ describe('attach', () => {
     return waitForTraceData();
   }
 
+  async function sendRequestAndWaitForMetrics() {
+    await sendRequestAndVerifyResponse();
+    return waitForMetrics();
+  }
+
   async function sendRequestAndWaitForLogRecords() {
     await sendRequestAndVerifyResponse();
     return waitForLogRecords();
@@ -396,6 +434,13 @@ describe('attach', () => {
     return (await collector().fetchTelemetry()).traces;
   }
 
+  async function waitForMetrics() {
+    if (!(await collector().hasMetrics())) {
+      throw new Error('The collector never received any metrics.');
+    }
+    return (await collector().fetchTelemetry()).metrics;
+  }
+
   async function waitForLogRecords() {
     if (!(await collector().hasLogs())) {
       throw new Error('The collector never received any log records.');

test/util/expectAttribute.ts

@@ -6,6 +6,7 @@ import { KeyValue } from '../collector/types/opentelemetry/proto/common/v1/commo
 import { Resource } from '../collector/types/opentelemetry/proto/resource/v1/resource';
 import { Span } from '../collector/types/opentelemetry/proto/trace/v1/trace';
 import { LogRecord } from '../collector/types/opentelemetry/proto/logs/v1/logs';
+import { HistogramDataPoint } from '../collector/types/opentelemetry/proto/metrics/v1/metrics';
 
 const { fail } = expect;
 
@@ -42,6 +43,10 @@ export function expectSpanAttribute(span: Span, key: string, expectedValue: any)
   expectAttribute(span, key, expectedValue, 'span');
 }
 
+export function expectMetricDataPointAttribute(dataPoint: HistogramDataPoint, key: string, expectedValue: any) {
+  expectAttribute(dataPoint, key, expectedValue, 'log record');
+}
+
 export function expectLogRecordAttribute(logRecord: LogRecord, key: string, expectedValue: any) {
   expectAttribute(logRecord, key, expectedValue, 'log record');
 }

test/util/expectMatchingMetric.ts

@@ -0,0 +1,46 @@
+// SPDX-FileCopyrightText: Copyright 2024 Dash0 Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+import { ExportMetricsServiceRequest } from '../collector/types/opentelemetry/proto/collector/metrics/v1/metrics_service';
+import { ResourceMetrics, ScopeMetrics, Metric } from '../collector/types/opentelemetry/proto/metrics/v1/metrics';
+import { Resource } from '../collector/types/opentelemetry/proto/resource/v1/resource';
+import {
+  Expectation,
+  findMatchingItemsInServiceRequest,
+  processFindItemsResult,
+  ServiceRequestMapper,
+} from './findMatchingItems';
+
+class MetricsServiceRequestMapper
+  implements ServiceRequestMapper<ExportMetricsServiceRequest, ResourceMetrics, ScopeMetrics, Metric>
+{
+  getResourceItems(serviceRequest: ExportMetricsServiceRequest): ResourceMetrics[] {
+    return serviceRequest.resource_metrics;
+  }
+
+  getResource(resourceMetrics: ResourceMetrics): Resource | undefined {
+    return resourceMetrics.resource;
+  }
+
+  getScopeItems(resourceMetrics: ResourceMetrics): ScopeMetrics[] {
+    return resourceMetrics.scope_metrics;
+  }
+
+  getItems(scopeMetrics: ScopeMetrics): Metric[] {
+    return scopeMetrics.metrics;
+  }
+}
+
+export function expectMatchingMetric(
+  metrics: ExportMetricsServiceRequest[],
+  resourceExpectations: Expectation<Resource>[],
+  metricExpectations: Expectation<Metric>[],
+): Metric {
+  const matchResult = findMatchingItemsInServiceRequest(
+    metrics,
+    new MetricsServiceRequestMapper(),
+    resourceExpectations,
+    metricExpectations,
+  );
+  return processFindItemsResult(matchResult, 'metric');
+}

test/util/waitUntil.ts

@@ -6,21 +6,23 @@ import isCi from 'is-ci';
 import delay from './delay';
 
 export interface RetryOptions {
-  attempts: number;
   maxAttempts: number;
   waitBetweenRetries: number;
 }
 
+export interface RetryInProgress {
+  attempts: number;
+  options: RetryOptions;
+}
+
 export function defaultRetryOptions(): RetryOptions {
   if (isCi) {
     return {
-      attempts: 0,
       maxAttempts: 30,
       waitBetweenRetries: 300,
     };
   } else {
     return {
-      attempts: 0,
       maxAttempts: 15,
       waitBetweenRetries: 200,
     };
@@ -33,16 +35,35 @@ export function defaultRetryOptions(): RetryOptions {
  * @param fn the function to retry
  * @param retryOptions the options for retrying
  */
-export default async function waitUntil(fn: () => any, options?: RetryOptions) {
-  options = options ?? defaultRetryOptions();
+export default async function waitUntil(fn: () => any, opts?: Partial<RetryOptions>) {
+  let retryInProgress: RetryInProgress;
+  const defaults = defaultRetryOptions();
+  if (!opts) {
+    retryInProgress = {
+      attempts: 0,
+      options: defaults,
+    };
+  } else {
+    retryInProgress = {
+      attempts: 0,
+      options: {
+        maxAttempts: opts.maxAttempts ?? defaults.maxAttempts,
+        waitBetweenRetries: opts.waitBetweenRetries ?? defaults.waitBetweenRetries,
+      },
+    };
+  }
+  return _waitUntil(fn, retryInProgress);
+}
+
+async function _waitUntil(fn: () => any, retryInProgress: RetryInProgress) {
   try {
     return await fn();
   } catch (e) {
-    await delay(options.waitBetweenRetries);
-    options.attempts += 1;
-    if (options.attempts > options.maxAttempts) {
+    await delay(retryInProgress.options.waitBetweenRetries);
+    retryInProgress.attempts += 1;
+    if (retryInProgress.attempts > retryInProgress.options.maxAttempts) {
       throw e;
     }
-    return waitUntil(fn, options);
+    return _waitUntil(fn, retryInProgress);
   }
 }