feat: add span metrics processor

packages/opentelemetry-node/lib/processors.js

@@ -0,0 +1,210 @@
+/*
+ * Copyright Elasticsearch B.V. and contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+// This replicates the SDKs logic of getting processors from env
+// and also adds a new one to collect span metrics like:
+// - metric.otel.sdk.span.live.count
+// - metric.otel.sdk.span.ended.count
+
+const {metrics} = require('@opentelemetry/api');
+const {getStringListFromEnv, getStringFromEnv} = require('@opentelemetry/core');
+const {
+    BatchSpanProcessor,
+    SimpleSpanProcessor,
+    ConsoleSpanExporter,
+} = require('@opentelemetry/sdk-trace-base');
+const {log} = require('./logging');
+
+/**
+ * @typedef {import('@opentelemetry/api').Meter} Meter
+ * @typedef {import('@opentelemetry/api').UpDownCounter} UpDownCounter
+ * @typedef {import('@opentelemetry/api').Counter} Counter
+ * @typedef {import('@opentelemetry/sdk-trace-base').SpanProcessor} SpanProcessor
+ * @typedef {import('@opentelemetry/sdk-trace-base').SpanExporter} SpanExporter
+ */
+
+// @ts-ignore - compiler options do not allow lookp outside `lib` folder
+const ELASTIC_PKG = require('../package.json');
+const ELASTIC_SDK_VERSION = ELASTIC_PKG.version;
+const ELASTIC_SDK_SCOPE = ELASTIC_PKG.name;
+
+const otlpPkgPreffix = '@opentelemetry/exporter-trace-otlp-';
+const otlpProtocol =
+    getStringFromEnv('OTEL_EXPORTER_OTLP_TRACES_PROTOCOL') ??
+    getStringFromEnv('OTEL_EXPORTER_OTLP_PROTOCOL') ??
+    'http/protobuf';
+
+// NOTE: assuming the meter provider is not going to be replaced once
+// the EDOT is started we can cache the meter and metrics in these vars
+/** @type {Meter} */
+let selfMetricsMeter;
+/** @type {UpDownCounter} */
+let liveSpans;
+/** @type {Counter} */
+let closedSpans;
+
+/**
+ * @returns {Meter}
+ */
+function getSpansMeter() {
+    if (selfMetricsMeter) {
+        return selfMetricsMeter;
+    }
+    selfMetricsMeter = metrics.getMeter(ELASTIC_SDK_SCOPE, ELASTIC_SDK_VERSION);
+    return selfMetricsMeter;
+}
+
+/**
+ * @returns {UpDownCounter}
+ */
+function getLiveSpansCounter() {
+    if (liveSpans) {
+        return liveSpans;
+    }
+    liveSpans = getSpansMeter().createUpDownCounter(
+        'otel.sdk.span.live.count',
+        {
+            description:
+                'Number of created spans for which the end operation has not been called yet',
+        }
+    );
+    return liveSpans;
+}
+
+/**
+ * @returns {Counter}
+ */
+function getClosedSpansCounter() {
+    if (closedSpans) {
+        return closedSpans;
+    }
+    closedSpans = getSpansMeter().createCounter('otel.sdk.span.closed.count', {
+        description:
+            'Number of created spans for which the end operation was called',
+    });
+    return closedSpans;
+}
+
+/** @type {SpanProcessor} */
+const spanMetricsPrcessor = {
+    forceFlush: function () {
+        return Promise.resolve();
+    },
+    onStart: function (span, parentContext) {
+        getLiveSpansCounter().add(1);
+    },
+    onEnd: function (span) {
+        getLiveSpansCounter().add(-1);
+        getClosedSpansCounter().add(1);
+    },
+    shutdown: function () {
+        return Promise.resolve();
+    },
+};
+
+/**
+ * @param {'otlp' | 'zipkin' | 'jaeger' | 'console'} type
+ * @returns {SpanExporter}
+ */
+function getSpanExporter(type) {
+    if (type === 'zipkin') {
+        const {ZipkinExporter} = require('@opentelemetry/exporter-zipkin');
+        return new ZipkinExporter();
+    } else if (type === 'jaeger') {
+        // TODO: this should be installed and there is a possible issue with bundlers. refs:
+        // - is a dev-dependency? https://github.com/open-telemetry/opentelemetry-js/blob/ec17ce48d0e5a99a122da5add612a20e2dd84ed5/experimental/packages/opentelemetry-sdk-node/package.json#L76
+        // - surreunded with try catch in https://github.com/open-telemetry/opentelemetry-js/blob/ec17ce48d0e5a99a122da5add612a20e2dd84ed5/experimental/packages/opentelemetry-sdk-node/src/utils.ts#L120
+        // const {JaegerExporter} = require('@opentelemetry/exporter-jaeger');
+        // result.push(new BatchSpanProcessor(new JaegerExporter()));
+    } else if (type === 'console') {
+        return new ConsoleSpanExporter();
+    }
+
+    let exporterPkgName = `${otlpPkgPreffix}`;
+    switch (otlpProtocol) {
+        case 'grpc':
+            exporterPkgName += 'grpc';
+            break;
+        case 'http/json':
+            exporterPkgName += 'http';
+            break;
+        case 'http/protobuf':
+            exporterPkgName += 'proto';
+            break;
+        default:
+            log.warn(
+                `Unsupported OTLP traces protocol: ${otlpProtocol}. Using http/protobuf.`
+            );
+            exporterPkgName += 'proto';
+    }
+    const {OTLPTraceExporter} = require(exporterPkgName);
+    return new OTLPTraceExporter();
+}
+
+/**
+ * @param {SpanProcessor[]} [processors]
+ */
+function getSpanProcessors(processors) {
+    const metricsExporters =
+        getStringListFromEnv('OTEL_METRICS_EXPORTER') || [];
+    const metricsEnabled = metricsExporters.every((e) => e !== 'none');
+
+    if (Array.isArray(processors)) {
+        if (metricsEnabled) {
+            processors.push(spanMetricsPrcessor);
+        }
+        return processors;
+    }
+
+    // Get from env
+    const exporters = getStringListFromEnv('OTEL_TRACES_EXPORTER') ?? [];
+    const result = metricsEnabled ? [spanMetricsPrcessor] : [];
+
+    if (exporters.some((exp) => exp === 'none')) {
+        log.warn(
+            'OTEL_TRACES_EXPORTER contains "none". No trace information or Spans will be exported.'
+        );
+        return [];
+    }
+
+    if (exporters.length === 0) {
+        log.trace(
+            'OTEL_TRACES_EXPORTER is empty. Using the default "otlp" exporter.'
+        );
+        exporters.push('otlp');
+    }
+
+    for (const name of exporters) {
+        log.trace(`Initializing "${name}" traces exporter.`);
+        switch (name) {
+            case 'otlp':
+                result.push(new BatchSpanProcessor(getSpanExporter('otlp')));
+                break;
+            case 'console':
+                result.push(
+                    new SimpleSpanProcessor(getSpanExporter('console'))
+                );
+                break;
+            case 'zipkin':
+                result.push(new BatchSpanProcessor(getSpanExporter('zipkin')));
+                break;
+            case 'jaeger':
+                // TODO: check comment in `getSpanExporter` function
+                // result.push(new BatchSpanProcessor(getSpanExporter('zipkin')));
+                log.warn(
+                    `OTEL_TRACES_EXPORTER value "${name}" not available yet.`
+                );
+                break;
+            default:
+                log.warn(`Unrecognized OTEL_TRACES_EXPORTER value: ${name}.`);
+        }
+    }
+
+    return result;
+}
+
+module.exports = {
+    getSpanProcessors,
+};

packages/opentelemetry-node/lib/sdk.js

@@ -33,6 +33,7 @@
 const {resolveDetectors} = require('./detectors');
 const {setupEnvironment, restoreEnvironment} = require('./environment');
 const {getInstrumentations} = require('./instrumentations');
+const {getSpanProcessors} = require('./processors');
 const {setupCentralConfig} = require('./central-config');
 const DISTRO_VERSION = require('../package.json').version;
 
@@ -55,7 +56,7 @@
         try {
             await shutdownFn();
         } catch (err) {
             console.warn('warning: error shutting down OTel SDK', err);
         }
         process.exit(128 + os.constants.signals.SIGTERM);
     });
@@ -65,7 +66,7 @@
         try {
             await shutdownFn();
         } catch (err) {
             console.warn('warning: error shutting down OTel SDK', err);
         }
     });
 }
@@ -124,8 +125,7 @@
     const defaultConfig = {
         resourceDetectors: resolveDetectors(cfg.resourceDetectors),
         instrumentations: cfg.instrumentations || getInstrumentations(),
-        // Avoid setting `spanProcessor` or `traceExporter` to have NodeSDK
-        // use its `TracerProviderWithEnvExporters` for tracing setup.
+        spanProcessors: getSpanProcessors(cfg.spanProcessors),
     };
 
     const exporterPkgNameFromEnvVar = {

packages/opentelemetry-node/package.json

@@ -81,6 +81,7 @@
     "@opentelemetry/exporter-metrics-otlp-grpc": "^0.202.0",
     "@opentelemetry/exporter-metrics-otlp-http": "^0.202.0",
     "@opentelemetry/exporter-metrics-otlp-proto": "^0.202.0",
+    "@opentelemetry/exporter-zipkin": "^2.0.1",
     "@opentelemetry/host-metrics": "^0.36.0",
     "@opentelemetry/instrumentation-amqplib": "^0.49.0",
     "@opentelemetry/instrumentation-aws-sdk": "^0.54.0",
@@ -129,6 +130,7 @@
     "@opentelemetry/sdk-logs": "^0.202.0",
     "@opentelemetry/sdk-metrics": "^2.0.0",
     "@opentelemetry/sdk-node": "^0.202.0",
+    "@opentelemetry/sdk-trace-base": "^2.0.1",
     "@opentelemetry/semantic-conventions": "^1.30.0",
     "@opentelemetry/winston-transport": "^0.13.0",
     "import-in-the-middle": "^1.12.0",

packages/opentelemetry-node/test/OTEL_METRICS_EXPORTER.test.js

@@ -34,6 +34,9 @@ const testFixtures = [
             t.ok(hasLog(`name: 'nodejs.eventloop.delay.min'`));
             t.ok(hasLog(`name: 'nodejs.eventloop.delay.max'`));
             t.ok(hasLog(`name: 'process.cpu.utilization'`));
+            t.ok(hasLog(`name: 'process.cpu.utilization'`));
+            t.ok(hasLog(`name: 'otel.sdk.span.live.count'`));
+            t.ok(hasLog(`name: 'otel.sdk.span.closed.count'`));
         },
         checkTelemetry: (t, col) => {
             t.ok(col.metrics.length > 0);

packages/opentelemetry-node/test/OTEL_TRACES_EXPORTER.test.js

@@ -0,0 +1,101 @@
+/*
+ * Copyright Elasticsearch B.V. and contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+const {test} = require('tape');
+const {runTestFixtures} = require('./testutils');
+
+/** @type {import('./testutils').TestFixture[]} */
+const testFixtures = [
+    {
+        name: 'basic scenario with no values in env',
+        // Using an existing fixture since we're only interested in startup logs
+        // from the distro.
+        args: ['./fixtures/use-exporter-protocol.js'],
+        cwd: __dirname,
+        env: {
+            NODE_OPTIONS: '--import=@elastic/opentelemetry-node',
+            OTEL_LOG_LEVEL: 'verbose',
+        },
+        // verbose: true,
+        checkResult: (t, err, stdout, stderr) => {
+            t.error(err);
+            const lines = stdout.split('\n');
+            const hasLog = (text) =>
+                getLogs(lines).some((log) => log.msg.includes(text));
+            t.ok(hasLog('Initializing "otlp" traces exporter.'));
+            t.ok(!hasLog('Initializing "zipkin" traces exporter.'));
+            t.ok(!hasLog('Initializing "jaeger" traces exporter.'));
+            t.ok(!hasLog('Initializing "console" traces exporter.'));
+        },
+    },
+    {
+        name: 'scenario with "none" in OTEL_TRACES_EXPORTER',
+        // Using an existing fixture since we're only interested in startup logs
+        // from the distro.
+        args: ['./fixtures/use-exporter-protocol.js'],
+        cwd: __dirname,
+        env: {
+            NODE_OPTIONS: '--import=@elastic/opentelemetry-node',
+            OTEL_LOG_LEVEL: 'verbose',
+            OTEL_TRACES_EXPORTER: 'console, zipkin, otlp, none',
+        },
+        // verbose: true,
+        checkResult: (t, err, stdout, stderr) => {
+            t.error(err);
+            const lines = stdout.split('\n');
+            const hasLog = (text) =>
+                getLogs(lines).some((log) => log.msg.includes(text));
+            t.ok(
+                hasLog(
+                    'OTEL_TRACES_EXPORTER contains "none". No trace information or Spans will be exported.'
+                )
+            );
+            t.ok(!hasLog('Initializing "otlp" traces exporter.'));
+            t.ok(!hasLog('Initializing "zipkin" traces exporter.'));
+            t.ok(!hasLog('Initializing "jaeger" traces exporter.'));
+            t.ok(!hasLog('Initializing "console" traces exporter.'));
+        },
+    },
+    {
+        name: 'scenario with values in OTEL_TRACES_EXPORTER',
+        // Using an existing fixture since we're only interested in startup logs
+        // from the distro.
+        args: ['./fixtures/use-exporter-protocol.js'],
+        cwd: __dirname,
+        env: {
+            NODE_OPTIONS: '--import=@elastic/opentelemetry-node',
+            OTEL_LOG_LEVEL: 'verbose',
+            OTEL_TRACES_EXPORTER: 'console, otlp',
+        },
+        // verbose: true,
+        checkResult: (t, err, stdout, stderr) => {
+            t.error(err);
+            const lines = stdout.split('\n');
+            const hasLog = (text) =>
+                getLogs(lines).some((log) => log.msg.includes(text));
+            t.ok(hasLog('Initializing "otlp" traces exporter.'));
+            t.ok(hasLog('Initializing "console" traces exporter.'));
+            t.ok(!hasLog('Initializing "zipkin" traces exporter.'));
+            t.ok(!hasLog('Initializing "jaeger" traces exporter.'));
+        },
+    },
+];
+
+// ----- helper functions -----
+
+/**
+ * @param {Array<string>} lines
+ * @returns {Array<{name: string, lelvel: number; msg: string}>}
+ */
+function getLogs(lines) {
+    return lines.filter((l) => l.startsWith('{')).map(JSON.parse);
+}
+
+// ----- main line -----
+
+test('OTEL_TRACES_EXPORTER', (suite) => {
+    runTestFixtures(suite, testFixtures);
+    suite.end();
+});

packages/opentelemetry-node/types/processors.d.ts

@@ -0,0 +1,9 @@
+export type Meter = import('@opentelemetry/api').Meter;
+export type UpDownCounter = import('@opentelemetry/api').UpDownCounter;
+export type Counter = import('@opentelemetry/api').Counter;
+export type SpanProcessor = import('@opentelemetry/sdk-trace-base').SpanProcessor;
+export type SpanExporter = import('@opentelemetry/sdk-trace-base').SpanExporter;
+/**
+ * @param {SpanProcessor[]} [processors]
+ */
+export function getSpanProcessors(processors?: SpanProcessor[]): import("@opentelemetry/sdk-trace-base").SpanProcessor[];