CLDSRV-884: Backport OpenTelemetry tracing instrumentation

index.js

@@ -7,4 +7,25 @@ require('werelogs').stderrUtils.catchAndTimestampStderr(
     require('cluster').isPrimary ? 1 : null,
 );
 
+const tracing = require('arsenal/build/lib/tracing');
+
+// Gated on isEnabled() so the OTEL-off path doesn't load Config early.
+if (tracing.isEnabled() && !(require('./lib/Config').config.isCluster && require('cluster').isPrimary)) {
+    tracing.init({
+        serviceName: 'cloudserver',
+        serviceVersion: require('./package.json').version,
+        instrumentations: () => {
+            const { HttpInstrumentation } = require('@opentelemetry/instrumentation-http');
+            const { IORedisInstrumentation } = require('@opentelemetry/instrumentation-ioredis');
+            const { MongoDBInstrumentation } = require('@opentelemetry/instrumentation-mongodb');
+            const healthPaths = ['/live', '/ready', '/_/healthcheck', '/_/healthcheck/deep', '/metrics'];
+            return [
+                new HttpInstrumentation(tracing.makeHttpInstrumentationConfig({ healthPaths })),
+                new IORedisInstrumentation({ requireParentSpan: true }),
+                new MongoDBInstrumentation({ enhancedDatabaseReporting: false }),
+            ];
+        },
+    });
+}
+
 require('./lib/server.js')();

lib/api/api.js

@@ -79,6 +79,7 @@ const parseCopySource = require('./apiUtils/object/parseCopySource');
 const { tagConditionKeyAuth } = require('./apiUtils/authorization/tagConditionKeys');
 const { isRequesterASessionUser } = require('./apiUtils/authorization/permissionChecks');
 const checkHttpHeadersSize = require('./apiUtils/object/checkHttpHeadersSize');
+const { instrumentApiMethod } = require('arsenal/build/lib/tracing');
 const constants = require('../../constants');
 const { config } = require('../Config.js');
 const metadata = require('../metadata/wrapper');
@@ -609,4 +610,11 @@ const api = {
     handleAuthorizationResults,
 };
 
+const NON_INSTRUMENTED_KEYS = new Set(['callApiMethod', 'checkAuthResults', 'handleAuthorizationResults']);
+for (const [name, handler] of Object.entries(api)) {
+    if (typeof handler === 'function' && !NON_INSTRUMENTED_KEYS.has(name)) {
+        api[name] = instrumentApiMethod(handler, name);
+    }
+}
+
 module.exports = api;

lib/server.js

@@ -6,6 +6,7 @@
 const { setServerHeader } = arsenal.s3routes.routesUtils;
 const { RedisClient, StatsClient } = arsenal.metrics;
 const monitoringClient = require('./utilities/monitoringHandler');
+const tracing = require('arsenal/build/lib/tracing');
 
 const logger = require('./utilities/logger');
 const { internalHandlers } = require('./utilities/internalHandlers');
@@ -15,15 +16,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 +43,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 +55,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 +79,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
@@ -130,9 +125,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 +172,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 +225,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 +285,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 +301,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,
             });
         });
 
@@ -332,14 +327,14 @@
         if (this.config.rateLimiting?.enabled) {
             stopRefillJob(logger);
         }
-        Promise.all(this.servers.map(server =>
-            new Promise(resolve => server.close(resolve))
-        )).then(() => process.exit(0));
+        return Promise.all(this.servers.map(server => new Promise(resolve => server.close(resolve))))
+            .finally(() => tracing.close())
+            .finally(() => process.exit(0));
     }
 
     caughtExceptionShutdown() {
         if (!this.cluster) {
-            process.exit(1);
+            return tracing.close().finally(() => process.exit(1));
         }
         logger.error('shutdown of worker due to exception', {
             workerId: this.worker ? this.worker.id : undefined,
@@ -348,8 +343,9 @@
         // Will close all servers, cause disconnect event on primary and kill
         // worker process with 'SIGTERM'.
         if (this.worker) {
-            this.worker.kill();
+            return tracing.close().finally(() => this.worker.kill());
         }
+        return undefined;
     }
 
     startServer(listenOn, port, routeRequest) {
@@ -363,10 +359,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 +410,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 +429,6 @@
                 logger.error('ServerAccessLogger creation error', error);
             }
 
-
             this.started = true;
         });
     }
@@ -490,8 +484,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);

package.json

@@ -29,6 +29,10 @@
         "@aws-sdk/signature-v4": "^3.374.0",
         "@azure/storage-blob": "^12.28.0",
         "@hapi/joi": "^17.1.1",
+        "@opentelemetry/api": "^1.9.0",
+        "@opentelemetry/instrumentation-http": "~0.218.0",
+        "@opentelemetry/instrumentation-ioredis": "~0.64.0",
+        "@opentelemetry/instrumentation-mongodb": "~0.69.0",
         "@smithy/node-http-handler": "^3.0.0",
         "arsenal": "git+https://github.com/scality/Arsenal#8.4.24",
         "async": "2.6.4",