Enable telemetry control from env vars

package.json

@@ -47,9 +47,13 @@
   },
   "license": "MIT",
   "dependencies": {
+    "@opentelemetry/api": "^1.9.0",
+    "@opentelemetry/host-metrics": "0.35.5",
+    "@opentelemetry/instrumentation": "0.57.2",
+    "@opentelemetry/instrumentation-koa": "0.47.1",
     "@types/koa": "^2.11.0",
     "@types/koa-compose": "^3.2.3",
-    "@vtex/diagnostics-nodejs": "0.1.0-io-beta.19",
+    "@vtex/diagnostics-nodejs": "0.1.0-io-beta.20",
     "@vtex/node-error-report": "^0.0.3",
     "@wry/equality": "^0.1.9",
     "agentkeepalive": "^4.0.2",

src/constants.ts

@@ -74,3 +74,9 @@ export const INSPECT_DEBUGGER_PORT = 5858
 export const cancellableMethods = new Set(['GET', 'OPTIONS', 'HEAD'])
 
 export const LOG_CLIENT_INIT_TIMEOUT_MS = 5000
+
+export const OTEL_EXPORTER_OTLP_ENDPOINT = process.env.OTEL_EXPORTER_OTLP_ENDPOINT as string;
+
+export const DK_APP_ID = process.env.NODE_VTEX_API_DK_APP_ID as string || "apps-team";
+
+export const DIAGNOSTICS_TELEMETRY_ENABLED = process.env.VTEX_DIAGNOSTICS_TELEMETRY_ENABLED === 'true';

src/service/index.ts

@@ -1,20 +1,23 @@
+import { initializeTelemetry } from './telemetry'
 import cluster from 'cluster'
 
 import { HTTP_SERVER_PORT } from '../constants'
 import { getServiceJSON } from './loaders'
 import { LogLevel, logOnceToDevConsole } from './logger'
-import { startMaster } from './master'
-import { startWorker } from './worker'
 
-export const startApp = () => {
+export const startApp = async () => {
+  await initializeTelemetry()
   const serviceJSON = getServiceJSON()
   try {
     // if it is a master process then call setting up worker process
     if(cluster.isMaster) {
+      const { startMaster } = await import('./master')
       startMaster(serviceJSON)
     } else {
       // to setup server configurations and share port address for incoming requests
-      startWorker(serviceJSON).listen(HTTP_SERVER_PORT)
+      const { startWorker } = await import('./worker')
+      const app = await startWorker(serviceJSON)
+      app.listen(HTTP_SERVER_PORT)
     }
   } catch (err: any) {
     logOnceToDevConsole(err.stack || err.message, LogLevel.Error)

src/service/logger/logger.ts

@@ -32,7 +32,7 @@ export class Logger {
       }
     }
 
-    // this.initLogClient();
+    this.initLogClient();
   }
 
   private initLogClient(): Promise<Types.LogClient | undefined> {

src/service/metrics/client.ts

@@ -0,0 +1,46 @@
+import { Types } from "@vtex/diagnostics-nodejs";
+import { initializeTelemetry } from '../telemetry';
+
+class MetricClientSingleton {
+  private static instance: MetricClientSingleton | undefined;
+  private client: Types.MetricClient | undefined;
+  private initPromise: Promise<Types.MetricClient> | undefined;
+
+  private constructor() {}
+
+  public static getInstance(): MetricClientSingleton {
+    if (!MetricClientSingleton.instance) {
+      MetricClientSingleton.instance = new MetricClientSingleton();
+    }
+    return MetricClientSingleton.instance;
+  }
+
+  public async getClient(): Promise<Types.MetricClient> {
+    if (this.client) {
+      return this.client;
+    }
+
+    if (this.initPromise) {
+      return this.initPromise;
+    }
+
+    this.initPromise = this.initializeClient();
+
+    return this.initPromise;
+  }
+
+  private async initializeClient(): Promise<Types.MetricClient> {
+    try {
+      const { metricsClient } = await initializeTelemetry();
+      this.client = metricsClient;
+      this.initPromise = undefined;
+      return metricsClient;
+    } catch (error) {
+      console.error('Failed to initialize metrics client:', error);
+      this.initPromise = undefined;
+      throw error;
+    }
+  }
+}
+
+export const getMetricClient = () => MetricClientSingleton.getInstance().getClient();

src/service/metrics/instruments/hostMetrics.ts

@@ -0,0 +1,41 @@
+import { InstrumentationBase, InstrumentationConfig } from "@opentelemetry/instrumentation";
+import { MeterProvider } from '@opentelemetry/api';
+import { HostMetrics } from "@opentelemetry/host-metrics";
+
+interface HostMetricsInstrumentationConfig extends InstrumentationConfig {
+  name?: string;
+  meterProvider?: MeterProvider;
+}
+
+export class HostMetricsInstrumentation extends InstrumentationBase<HostMetricsInstrumentationConfig> {
+  private hostMetrics?: HostMetrics;
+
+  constructor(config: HostMetricsInstrumentationConfig = {}) {
+    const instrumentation_name = config.name || 'host-metrics-instrumentation';
+    const instrumentation_version = '1.0.0';
+    super(instrumentation_name, instrumentation_version, config);
+  }
+
+  init(): void {}
+
+  enable(): void {
+    if (!this._config.meterProvider) {
+      throw new Error('MeterProvider is required for HostMetricsInstrumentation');
+    }
+
+    this.hostMetrics = new HostMetrics({
+      meterProvider: this._config.meterProvider,
+      name: this._config.name || 'host-metrics',
+    });
+
+    this.hostMetrics.start();
+    console.debug('HostMetricsInstrumentation enabled');
+  }
+
+  disable(): void {
+    if (this.hostMetrics) {
+      this.hostMetrics = undefined;
+      console.debug('HostMetricsInstrumentation disabled');
+    }
+  }
+}

src/service/metrics/metrics.ts

@@ -0,0 +1,119 @@
+import { Types } from '@vtex/diagnostics-nodejs'
+import { getMetricClient } from './client'
+
+export const enum RequestsMetricLabels {
+  STATUS_CODE = 'status_code',
+  REQUEST_HANDLER = 'handler',
+}
+
+export interface OtelRequestInstruments {
+  concurrentRequests: Types.Gauge
+  requestTimings: Types.Histogram
+  totalRequests: Types.Counter
+  responseSizes: Types.Histogram
+  abortedRequests: Types.Counter
+}
+
+const createOtelConcurrentRequestsInstrument = async (): Promise<Types.Gauge> => {
+  const metricsClient = await getMetricClient()
+  return metricsClient.createGauge('io_http_requests_current', {
+    description: 'The current number of requests in course.',
+    unit: '1'
+  })
+}
+
+const createOtelRequestsTimingsInstrument = async (): Promise<Types.Histogram> => {
+  const metricsClient = await getMetricClient()
+  return metricsClient.createHistogram('runtime_http_requests_duration_milliseconds', {
+    description: 'The incoming http requests total duration.',
+    unit: 'ms'
+  })
+}
+
+const createOtelTotalRequestsInstrument = async (): Promise<Types.Counter> => {
+  const metricsClient = await getMetricClient()
+  return metricsClient.createCounter('runtime_http_requests_total', {
+    description: 'The total number of HTTP requests.',
+    unit: '1'
+  })
+}
+
+const createOtelRequestsResponseSizesInstrument = async (): Promise<Types.Histogram> => {
+  const metricsClient = await getMetricClient()
+  return metricsClient.createHistogram('runtime_http_response_size_bytes', {
+    description: 'The outgoing response sizes (only applicable when the response isn\'t a stream).',
+    unit: 'bytes'
+  })
+}
+
+const createOtelTotalAbortedRequestsInstrument = async (): Promise<Types.Counter> => {
+  const metricsClient = await getMetricClient()
+  return metricsClient.createCounter('runtime_http_aborted_requests_total', {
+    description: 'The total number of HTTP requests aborted.',
+    unit: '1'
+  })
+}
+
+class OtelInstrumentsSingleton {
+  private static instance: OtelInstrumentsSingleton | undefined;
+  private instruments: OtelRequestInstruments | undefined;
+  private initializingPromise: Promise<OtelRequestInstruments> | undefined;
+
+  private constructor() {}
+
+  public static getInstance(): OtelInstrumentsSingleton {
+    if (!OtelInstrumentsSingleton.instance) {
+      OtelInstrumentsSingleton.instance = new OtelInstrumentsSingleton();
+    }
+    return OtelInstrumentsSingleton.instance;
+  }
+
+  public async getInstruments(): Promise<OtelRequestInstruments> {
+    if (this.instruments) {
+      return this.instruments;
+    }
+
+    if (this.initializingPromise) {
+      return this.initializingPromise;
+    }
+
+    this.initializingPromise = this.initializeInstruments();
+
+    try {
+      this.instruments = await this.initializingPromise;
+      return this.instruments;
+    } catch (error) {
+      console.error('Failed to initialize OTel instruments:', error);
+      this.initializingPromise = undefined;
+      throw error;
+    } finally {
+      this.initializingPromise = undefined;
+    }
+  }
+
+  private async initializeInstruments(): Promise<OtelRequestInstruments> {
+    const [
+      concurrentRequests,
+      requestTimings,
+      totalRequests,
+      responseSizes,
+      abortedRequests
+    ] = await Promise.all([
+      createOtelConcurrentRequestsInstrument(),
+      createOtelRequestsTimingsInstrument(),
+      createOtelTotalRequestsInstrument(),
+      createOtelRequestsResponseSizesInstrument(),
+      createOtelTotalAbortedRequestsInstrument()
+    ])
+
+    return {
+      concurrentRequests,
+      requestTimings,
+      totalRequests,
+      responseSizes,
+      abortedRequests
+    }
+  }
+}
+
+export const getOtelInstruments = () => OtelInstrumentsSingleton.getInstance().getInstruments();

src/service/metrics/otelRequestMetricsMiddleware.ts

@@ -0,0 +1,89 @@
+import { finished as onStreamFinished } from 'stream'
+import { hrToMillisFloat } from '../../utils'
+import { getOtelInstruments, RequestsMetricLabels, OtelRequestInstruments } from './metrics'
+import { ServiceContext } from '../worker/runtime/typings'
+
+const INSTRUMENTS_INITIALIZATION_TIMEOUT = 500
+
+export const addOtelRequestMetricsMiddleware = () => {
+  let instruments: OtelRequestInstruments | undefined
+
+  const tryGetInstruments = async (ctx: ServiceContext): Promise<OtelRequestInstruments | undefined> => {
+    try {
+      return await Promise.race([
+        getOtelInstruments(),
+        new Promise<never>((_, reject) =>
+          setTimeout(
+            () => reject(new Error('Timeout waiting for OpenTelemetry instruments initialization')),
+            INSTRUMENTS_INITIALIZATION_TIMEOUT
+          )
+        )
+      ])
+    } catch (error) {
+      const errorMessage = error instanceof Error ? error.message : String(error)
+      console.warn(`OpenTelemetry instruments not ready for request ${ctx.requestHandlerName}: ${errorMessage}`)
+      return undefined
+    }
+  }
+
+  return async function addOtelRequestMetrics(ctx: ServiceContext, next: () => Promise<void>) {
+    instruments = instruments ? instruments : await tryGetInstruments(ctx)
+    if (!instruments) {
+      await next()
+      return
+    }
+
+    const start = process.hrtime()
+    instruments.concurrentRequests.add(1)
+
+    ctx.req.once('aborted', () => {
+      if (instruments) {
+        instruments.abortedRequests.add(1, { [RequestsMetricLabels.REQUEST_HANDLER]: ctx.requestHandlerName })
+      }
+    })
+
+    let responseClosed = false
+    ctx.res.once('close', () => (responseClosed = true))
+
+    try {
+      await next()
+    } finally {
+      const responseLength = ctx.response.length
+      if (responseLength && instruments) {
+        instruments.responseSizes.record(
+          responseLength,
+          { [RequestsMetricLabels.REQUEST_HANDLER]: ctx.requestHandlerName }
+        )
+      }
+
+      if (instruments) {
+        instruments.totalRequests.add(
+          1,
+          {
+            [RequestsMetricLabels.REQUEST_HANDLER]: ctx.requestHandlerName,
+            [RequestsMetricLabels.STATUS_CODE]: ctx.response.status,
+          }
+        )
+      }
+
+      const onResFinished = () => {
+        if (instruments) {
+          instruments.requestTimings.record(
+            hrToMillisFloat(process.hrtime(start)),
+            {
+              [RequestsMetricLabels.REQUEST_HANDLER]: ctx.requestHandlerName,
+            }
+          )
+
+          instruments.concurrentRequests.subtract(1)
+        }
+      }
+
+      if (responseClosed) {
+        onResFinished()
+      } else {
+        onStreamFinished(ctx.res, onResFinished)
+      }
+    }
+  }
+}