Merge pull request #260 from Luchistack/Develop-comprehensive-unit-and-integration-test-coverage-for-the-pull-payment-cron-scheduler

Add OpenTelemetry integration and tracing support

Author: elizabetheonoja-art

.env.example

@@ -1,4 +1,11 @@
 PORT=3000
+OTEL_DISABLED=false
+OTEL_SERVICE_NAME=substream-protocol-backend
+OTEL_SERVICE_VERSION=1.0.0
+OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318/v1/traces
+OTEL_EXPORTER_OTLP_HEADERS=
+OTEL_IGNORE_PATHS=/health,/metrics
+OTEL_DIAG_LEVEL=error
 
 # Creator auth configuration
 CREATOR_AUTH_SECRET=replace-with-a-long-random-secret

index.js

@@ -1,11 +1,12 @@
-const express = require('express');
-const cors = require('cors');
 const dotenv = require('dotenv');
-const promClient = require('prom-client');
+const { initTracing } = require('./src/utils/opentelemetry');
 
 dotenv.config();
-require('dotenv').config();
-dotenv.config();
+initTracing({ serviceName: 'substream-protocol-backend', serviceVersion: '1.0.0' });
+
+const express = require('express');
+const cors = require('cors');
+const promClient = require('prom-client');
 
 // Initialize Prometheus metrics
 const register = new promClient.Registry();
@@ -179,6 +180,9 @@ async function createApp(dependencies = {}) {
   app.use(express.json({ limit: '10mb' }));
   app.use(express.urlencoded({ extended: true }));
 
+  // Request tracing middleware must be registered early for trace propagation
+  app.use(requestTracingMiddleware);
+
   // Add request start time for accurate monitoring
   app.use(addRequestStartTime);
 

package.json

@@ -95,7 +95,12 @@
     "uuid": "^14.0.0",
     "web3.storage": "^4.5.4",
     "winston": "^3.19.0",
-    "xlsx": "^0.18.5"
+    "xlsx": "^0.18.5",
+    "@opentelemetry/api": "^1.18.0",
+    "@opentelemetry/sdk-node": "^0.66.0",
+    "@opentelemetry/auto-instrumentations-node": "^0.66.0",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.66.0",
+    "@opentelemetry/sdk-trace-base": "^0.66.0"
   },
   "devDependencies": {
     "@nestjs/cli": "^11.0.21",

src/utils/logger.js

@@ -10,6 +10,7 @@
 
 const winston = require('winston');
 const { v4: uuidv4 } = require('uuid');
+const { trace, context } = require('@opentelemetry/api');
 
 // Custom format for structured logging
 const { combine, timestamp, json, errors, splat } = winston.format;
@@ -75,32 +76,41 @@ function createLoggerContext(traceId, additionalContext = {}) {
  * Middleware to add trace ID to all requests
  */
 function requestTracingMiddleware(req, res, next) {
-  // Get or create trace ID
-  const traceId = req.headers['x-trace-id'] || uuidv4();
-  
-  // Add trace ID to response headers
+  const activeSpan = trace.getSpan(context.active());
+  const spanContext = activeSpan?.spanContext();
+  const traceId = req.headers['x-trace-id'] || spanContext?.traceId || uuidv4();
+  const spanId = spanContext?.spanId;
+
+  // Add trace headers for downstream correlation
   res.setHeader('x-trace-id', traceId);
-  
-  // Attach logger to request with context
+  if (spanId) {
+    res.setHeader('x-span-id', spanId);
+  }
+
   req.logger = logger.child({
     traceId,
+    spanId,
     method: req.method,
     path: req.path,
     ip: req.ip,
     userAgent: req.headers['user-agent'],
   });
-  
+
+  // Preserve trace fields for downstream error handling
+  req.logger.fields = { traceId, spanId };
+
   // Track request timing
   const startTime = Date.now();
-  
+
   res.on('finish', () => {
     const duration = Date.now() - startTime;
     const context = {
       traceId,
+      spanId,
       statusCode: res.statusCode,
       duration: `${duration}ms`,
     };
-    
+
     if (res.statusCode >= 500) {
       logger.error('Request failed', context);
     } else if (res.statusCode >= 400) {
@@ -109,7 +119,7 @@ function requestTracingMiddleware(req, res, next) {
       logger.info('Request completed', context);
     }
   });
-  
+
   next();
 }
 

src/utils/opentelemetry.js

@@ -0,0 +1,101 @@
+const { NodeSDK } = require('@opentelemetry/sdk-node');
+const { getNodeAutoInstrumentations } = require('@opentelemetry/auto-instrumentations-node');
+const { Resource } = require('@opentelemetry/resources');
+const { SemanticResourceAttributes } = require('@opentelemetry/semantic-conventions');
+const { OTLPTraceExporter } = require('@opentelemetry/exporter-trace-otlp-http');
+const { ConsoleSpanExporter } = require('@opentelemetry/sdk-trace-base');
+const { diag, DiagConsoleLogger, DiagLogLevel } = require('@opentelemetry/api');
+
+function parseOtlpHeaders(headerValue) {
+  if (!headerValue || typeof headerValue !== 'string') {
+    return undefined;
+  }
+
+  return Object.fromEntries(
+    headerValue
+      .split(',')
+      .map((pair) => pair.trim())
+      .filter(Boolean)
+      .map((pair) => pair.split('=').map((value) => value.trim()))
+      .filter((tuple) => tuple.length === 2)
+  );
+}
+
+function createTraceExporter() {
+  const endpoint = process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT || process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
+  const headers = parseOtlpHeaders(process.env.OTEL_EXPORTER_OTLP_HEADERS);
+
+  if (endpoint) {
+    return new OTLPTraceExporter({ url: endpoint, headers });
+  }
+
+  return new ConsoleSpanExporter();
+}
+
+function getDiagLevel() {
+  const level = (process.env.OTEL_DIAG_LEVEL || 'error').toUpperCase();
+  return DiagLogLevel[level] || DiagLogLevel.ERROR;
+}
+
+function initTracing(options = {}) {
+  if (process.env.OTEL_DISABLED === 'true' || global.__substreamOtelInitialized) {
+    return;
+  }
+
+  diag.setLogger(new DiagConsoleLogger(), getDiagLevel());
+
+  const serviceName = process.env.OTEL_SERVICE_NAME || options.serviceName || 'substream-protocol-backend';
+  const serviceVersion = process.env.OTEL_SERVICE_VERSION || options.serviceVersion || '1.0.0';
+
+  const sdk = new NodeSDK({
+    resource: new Resource({
+      [SemanticResourceAttributes.SERVICE_NAME]: serviceName,
+      [SemanticResourceAttributes.SERVICE_VERSION]: serviceVersion,
+      [SemanticResourceAttributes.DEPLOYMENT_ENVIRONMENT]: process.env.NODE_ENV || 'development',
+    }),
+    traceExporter: createTraceExporter(),
+    instrumentations: [
+      getNodeAutoInstrumentations({
+        '@opentelemetry/instrumentation-http': {
+          ignoreIncomingPaths: process.env.OTEL_IGNORE_PATHS ? process.env.OTEL_IGNORE_PATHS.split(',') : ['/health', '/metrics'],
+        },
+        '@opentelemetry/instrumentation-express': {
+          requestHook: (span, info) => {
+            if (span && info) {
+              span.setAttribute('http.route', info.route || info.request?.url || '/');
+            }
+          },
+        },
+      }),
+    ],
+  });
+
+  sdk.start()
+    .then(() => {
+      console.log('[Tracing] OpenTelemetry initialized', {
+        serviceName,
+        serviceVersion,
+      });
+    })
+    .catch((error) => {
+      console.error('[Tracing] Failed to initialize OpenTelemetry', error);
+    });
+
+  const shutdown = async () => {
+    try {
+      await sdk.shutdown();
+      console.log('[Tracing] OpenTelemetry shutdown complete');
+    } catch (error) {
+      console.error('[Tracing] OpenTelemetry shutdown failed', error);
+    }
+  };
+
+  process.on('SIGTERM', shutdown);
+  process.on('SIGINT', shutdown);
+
+  global.__substreamOtelInitialized = true;
+}
+
+module.exports = {
+  initTracing,
+};

worker.js

@@ -1,5 +1,11 @@
 #!/usr/bin/env node
 
+const dotenv = require('dotenv');
+const { initTracing } = require('./src/utils/opentelemetry');
+
+dotenv.config();
+initTracing({ serviceName: 'substream-protocol-backend-worker', serviceVersion: '1.0.0' });
+
 const { loadConfig } = require('./src/config');
 const { BackgroundWorkerService } = require('./src/services/backgroundWorkerService');
 const { SorobanIndexerWorker } = require('./src/services/sorobanIndexerWorker');