feat(CLDSRV-884): Add OpenTelemetry tracing instrumentation

index.js

@@ -7,4 +7,6 @@ require('werelogs').stderrUtils.catchAndTimestampStderr(
     require('cluster').isPrimary ? 1 : null,
 );
 
+require('./lib/tracing').init();
+
 require('./lib/server.js')();

lib/server.js

@@ -1,11 +1,13 @@
 const http = require('http');
 const https = require('https');
 const cluster = require('cluster');
+const { promisify } = require('util');
 const { series } = require('async');
 const arsenal = require('arsenal');
 const { setServerHeader } = arsenal.s3routes.routesUtils;
 const { RedisClient, StatsClient } = arsenal.metrics;
 const monitoringClient = require('./utilities/monitoringHandler');
+const tracing = require('./tracing');
 
 const logger = require('./utilities/logger');
 const { internalHandlers } = require('./utilities/internalHandlers');
@@ -15,15 +17,11 @@
 const api = require('./api/api');
 const dataWrapper = require('./data/wrapper');
 const kms = require('./kms/wrapper');
-const locationStorageCheck =
-    require('./api/apiUtils/object/locationStorageCheck');
+const locationStorageCheck = require('./api/apiUtils/object/locationStorageCheck');
 const vault = require('./auth/vault');
 const metadata = require('./metadata/wrapper');
 const { initManagement } = require('./management');
-const {
-    initManagementClient,
-    isManagementAgentUsed,
-} = require('./management/agentClient');
+const { initManagementClient, isManagementAgentUsed } = require('./management/agentClient');
 const { startCleanupJob } = require('./api/apiUtils/rateLimit/cleanup');
 const { startRefillJob, stopRefillJob } = require('./api/apiUtils/rateLimit/refillJob');
 
@@ -46,8 +44,7 @@
 _config.on('location-constraints-update', () => {
     if (implName === 'multipleBackends') {
         const clients = parseLC(_config, vault);
-        client = new MultipleBackendGateway(
-            clients, metadata, locationStorageCheck);
+        client = new MultipleBackendGateway(clients, metadata, locationStorageCheck);
     }
 });
 
@@ -59,8 +56,7 @@
 // stats client
 const STATS_INTERVAL = 5; // 5 seconds
 const STATS_EXPIRY = 30; // 30 seconds
-const statsClient = new StatsClient(localCacheClient, STATS_INTERVAL,
-    STATS_EXPIRY);
+const statsClient = new StatsClient(localCacheClient, STATS_INTERVAL, STATS_EXPIRY);
 const enableRemoteManagement = true;
 
 class S3Server {
@@ -84,7 +80,7 @@
         process.on('SIGHUP', this.cleanUp.bind(this));
         process.on('SIGQUIT', this.cleanUp.bind(this));
         process.on('SIGTERM', this.cleanUp.bind(this));
-        process.on('SIGPIPE', () => { });
+        process.on('SIGPIPE', () => {});
         // This will pick up exceptions up the stack
         process.on('uncaughtException', err => {
             // If just send the error object results in empty
@@ -95,7 +91,7 @@
                 workerId: this.worker ? this.worker.id : undefined,
                 workerPid: this.worker ? this.worker.process.pid : undefined,
             });
-            this.caughtExceptionShutdown();
+            void this.caughtExceptionShutdown();
         });
         this.started = false;
     }
@@ -130,9 +126,10 @@
         const requestStartTime = process.hrtime.bigint();
 
         // Skip server access logs for heartbeat.
-        const isLoggingEnabled = _config.serverAccessLogs
-            && (_config.serverAccessLogs.mode === serverAccessLogsModes.LOG_ONLY
-                || _config.serverAccessLogs.mode === serverAccessLogsModes.ENABLED);
+        const isLoggingEnabled =
+            _config.serverAccessLogs &&
+            (_config.serverAccessLogs.mode === serverAccessLogsModes.LOG_ONLY ||
+                _config.serverAccessLogs.mode === serverAccessLogsModes.ENABLED);
         const isInternalRoute = req.url.startsWith('/_');
         const isBackbeatRoute = req.url.startsWith('/_/backbeat/');
         if (isLoggingEnabled && (!isInternalRoute || isBackbeatRoute)) {
@@ -176,9 +173,7 @@
                 labels.action = req.apiMethod;
             }
             monitoringClient.httpRequestsTotal.labels(labels).inc();
-            monitoringClient.httpRequestDurationSeconds
-                .labels(labels)
-                .observe(responseTimeInNs / 1e9);
+            monitoringClient.httpRequestDurationSeconds.labels(labels).observe(responseTimeInNs / 1e9);
             monitoringClient.httpActiveRequests.dec();
         };
         res.on('close', monitorEndOfRequest);
@@ -231,14 +226,13 @@
         };
 
         let reqUids = req.headers['x-scal-request-uids'];
-        if (reqUids !== undefined && !/*isValidReqUids*/(reqUids.length < 128)) {
+        if (reqUids !== undefined && !(/*isValidReqUids*/ (reqUids.length < 128))) {
             // simply ignore invalid id (any user can provide an
             // invalid request ID through a crafted header)
             reqUids = undefined;
         }
-        const log = (reqUids !== undefined ?
-            logger.newRequestLoggerFromSerializedUids(reqUids) :
-            logger.newRequestLogger());
+        const log =
+            reqUids !== undefined ? logger.newRequestLoggerFromSerializedUids(reqUids) : logger.newRequestLogger();
         log.end().addDefaultFields(clientInfo);
 
         log.debug('received admin request', clientInfo);
@@ -292,8 +286,7 @@
         server.requestTimeout = 0; // disabling request timeout
 
         server.on('connection', socket => {
-            socket.on('error', err => logger.info('request rejected',
-                { error: err }));
+            socket.on('error', err => logger.info('request rejected', { error: err }));
         });
 
         // https://nodejs.org/dist/latest-v6.x/
@@ -309,8 +302,11 @@
             };
             const { address } = addr;
             logger.info('server started', {
-                address, port,
-                pid: process.pid, serverIP: address, serverPort: port
+                address,
+                port,
+                pid: process.pid,
+                serverIP: address,
+                serverPort: port,
             });
         });
 
@@ -323,32 +319,47 @@
         this.servers.push(server);
     }
 
-    /*
-     * This exits the running process properly.
-     */
-    cleanUp() {
+    async cleanUp() {
         logger.info('server shutting down');
-        // Stop token refill job if running
         if (this.config.rateLimiting?.enabled) {
             stopRefillJob(logger);
         }
-        Promise.all(this.servers.map(server =>
-            new Promise(resolve => server.close(resolve))
-        )).then(() => process.exit(0));
+        try {
+            await Promise.all(this.servers.map(server => promisify(server.close.bind(server))()));
+        } finally {
+            // Nested finally so tracing.close() runs even if a server's
+            // close() errored — otherwise the BatchSpanProcessor would
+            // lose buffered spans on shutdown.
+            try {
+                await tracing.close();
+            } finally {
+                process.exit(0);
+            }
+        }
     }
 
-    caughtExceptionShutdown() {
+    async caughtExceptionShutdown() {
         if (!this.cluster) {
-            process.exit(1);
+            try {
+                await tracing.close();
+            } finally {
+                process.exit(1);
+            }
+            return;
         }
         logger.error('shutdown of worker due to exception', {
             workerId: this.worker ? this.worker.id : undefined,
             workerPid: this.worker ? this.worker.process.pid : undefined,
         });
-        // Will close all servers, cause disconnect event on primary and kill
-        // worker process with 'SIGTERM'.
+        // worker.kill() is graceful (closes servers, disconnects IPC) but
+        // does not fire our SIGTERM handler, so the BatchSpanProcessor
+        // would lose buffered spans without an explicit flush here.
         if (this.worker) {
-            this.worker.kill();
+            try {
+                await tracing.close();
+            } finally {
+                this.worker.kill();
+            }
         }
     }
 
@@ -363,10 +374,7 @@
     }
 
     initiateStartup(log) {
-        series([
-            next => metadata.setup(next),
-            next => clientCheck(true, log, next),
-        ], (err, results) => {
+        series([next => metadata.setup(next), next => clientCheck(true, log, next)], (err, results) => {
             if (err) {
                 log.warn('initial health check failed, delaying startup', {
                     error: err,
@@ -417,8 +425,10 @@
 
             try {
                 logger.info('ServerAccessLogger config', { config: _config.serverAccessLogs });
-                if (_config.serverAccessLogs.mode === serverAccessLogsModes.LOG_ONLY
-                    || _config.serverAccessLogs.mode === serverAccessLogsModes.ENABLED) {
+                if (
+                    _config.serverAccessLogs.mode === serverAccessLogsModes.LOG_ONLY ||
+                    _config.serverAccessLogs.mode === serverAccessLogsModes.ENABLED
+                ) {
                     var serverAccessLogger = new ServerAccessLogger(
                         _config.serverAccessLogs.outputFile,
                         _config.serverAccessLogs.highWaterMarkBytes,
@@ -434,7 +444,6 @@
                 logger.error('ServerAccessLogger creation error', error);
             }
 
-
             this.started = true;
         });
     }
@@ -490,8 +499,7 @@
         });
 
         const metricServer = new S3Server(_config);
-        metricServer.startServer(_config.metricsListenOn,
-            _config.metricsPort, metricServer.routeAdminRequest);
+        metricServer.startServer(_config.metricsListenOn, _config.metricsPort, metricServer.routeAdminRequest);
     }
     if (_config.isCluster && cluster.isWorker) {
         const server = new S3Server(_config, cluster.worker);

lib/tracing/healthPaths.js

@@ -0,0 +1,18 @@
+'use strict';
+
+// Probe + scrape paths that should never produce a span. Filtered at
+// ingest (not at the trace backend) because probe rate × pod count ×
+// always-on sampling overwhelms the exporter and storage with traffic
+// nobody queries.
+const HEALTH_PATHS = new Set(['/live', '/ready', '/_/healthcheck', '/_/healthcheck/deep', '/metrics']);
+
+function isHealthPath(url) {
+    if (typeof url !== 'string' || url.length === 0) {
+        return false;
+    }
+    const qIdx = url.indexOf('?');
+    const path = qIdx === -1 ? url : url.slice(0, qIdx);
+    return HEALTH_PATHS.has(path);
+}
+
+module.exports = { isHealthPath };

lib/tracing/index.js

@@ -0,0 +1,129 @@
+'use strict';
+
+const assert = require('assert');
+const cluster = require('cluster');
+
+const { loadTrustedHosts, makeRequestHook } = require('./trustedHosts');
+const { isHealthPath } = require('./healthPaths');
+
+let sdk = null;
+
+function isEnabled() {
+    return process.env.ENABLE_OTEL === 'true';
+}
+
+function init() {
+    if (!isEnabled() || sdk) {
+        return;
+    }
+
+    const endpoint = process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT;
+    assert(endpoint, 'ENABLE_OTEL=true but OTEL_EXPORTER_OTLP_TRACES_ENDPOINT is unset');
+
+    // Cluster primary doesn't serve HTTP; workers re-exec the entry and init() themselves.
+    const { config } = require('../Config');
+    if (config.isCluster && cluster.isPrimary) {
+        return;
+    }
+
+    const { diag, DiagConsoleLogger, DiagLogLevel } = require('@opentelemetry/api');
+    const { NodeSDK } = require('@opentelemetry/sdk-node');
+    const { resourceFromAttributes } = require('@opentelemetry/resources');
+    const { OTLPTraceExporter } = require('@opentelemetry/exporter-trace-otlp-http');
+    const { HttpInstrumentation } = require('@opentelemetry/instrumentation-http');
+    const { IORedisInstrumentation } = require('@opentelemetry/instrumentation-ioredis');
+    const { MongoDBInstrumentation } = require('@opentelemetry/instrumentation-mongodb');
+    const { ParentBasedSampler, TraceIdRatioBasedSampler } = require('@opentelemetry/sdk-trace-base');
+    const { version } = require('../../package.json');
+
+    diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.WARN);
+
+    let samplingRatio = 0.01;
+    if (process.env.OTEL_SAMPLING_RATIO !== undefined) {
+        const parsed = Number(process.env.OTEL_SAMPLING_RATIO);
+        assert(
+            Number.isFinite(parsed) && parsed >= 0 && parsed <= 1,
+            `OTEL_SAMPLING_RATIO must be a finite number in [0, 1], got: ${process.env.OTEL_SAMPLING_RATIO}`,
+        );
+        samplingRatio = parsed;
+    }
+
+    const trustedHosts = loadTrustedHosts();
+
+    const ignoreIncomingRequestHook = req => req.method === 'OPTIONS' || isHealthPath(req.url);
+
+    sdk = new NodeSDK({
+        resource: resourceFromAttributes({
+            'service.name': process.env.OTEL_SERVICE_NAME || 'cloudserver',
+            'service.version': process.env.OTEL_SERVICE_VERSION || version,
+            'service.namespace': process.env.OTEL_SERVICE_NAMESPACE || 'scality',
+        }),
+        traceExporter: new OTLPTraceExporter({ url: endpoint }),
+        logRecordProcessors: [],
+        metricReaders: [],
+        spanLimits: {
+            attributeValueLengthLimit: 4096,
+            attributeCountLimit: 128,
+            eventCountLimit: 128,
+            linkCountLimit: 128,
+        },
+        sampler: new ParentBasedSampler({
+            root: new TraceIdRatioBasedSampler(samplingRatio),
+        }),
+        instrumentations: [
+            new HttpInstrumentation({
+                ignoreIncomingRequestHook,
+                requestHook: makeRequestHook(trustedHosts),
+            }),
+            new IORedisInstrumentation({ requireParentSpan: true }),
+            // Mask leaf values in db.statement so query shape is captured
+            // without user data (object keys, filter values) flowing to
+            // the trace backend.
+            new MongoDBInstrumentation({ enhancedDatabaseReporting: false }),
+        ],
+    });
+
+    sdk.start();
+}
+
+// Cap the flush window. The BatchSpanProcessor's default 30s export
+// timeout would otherwise block process.exit, and Kubernetes' default
+// terminationGracePeriodSeconds is also 30s — we'd get SIGKILL'd
+// before the flush ever completed.
+const SHUTDOWN_DEADLINE_MS = 5000;
+
+async function close() {
+    // Capture + clear before awaiting so concurrent callers (SIGTERM
+    // during an uncaught-exception flow, for example) don't both call
+    // sdk.shutdown() — the SDK doesn't guarantee idempotent shutdown.
+    const local = sdk;
+    if (!local) {
+        return;
+    }
+    sdk = null;
+    // Attach the rejection handler before racing the timeout. If the
+    // timeout wins the race, sdk.shutdown() keeps running unattached;
+    // a late rejection on that orphaned promise would otherwise crash
+    // the process via Node's default unhandledRejection behavior.
+    const shutdown = (async () => {
+        try {
+            await local.shutdown();
+        } catch (err) {
+            // Loggers may already be torn down at this point in shutdown;
+            // log to stderr directly.
+            // eslint-disable-next-line no-console
+            console.error('tracing close failed', err);
+        }
+    })();
+    await Promise.race([
+        shutdown,
+        // .unref() so the timer doesn't pin the event loop open
+        // when sdk.shutdown() resolves first.
+        new Promise(resolve => {
+            setTimeout(resolve, SHUTDOWN_DEADLINE_MS).unref();
+        }),
+    ]);
+    require('./instrumentation').resetTracer();
+}
+
+module.exports = { init, close, isEnabled };