fix: provide unique metric identity for workers

Signed-off-by: ferhat elmas <elmas.ferhat@gmail.com>

Author: ferhatelmas

monitoring/otel/config/otel-collector-config.yml

@@ -15,14 +15,18 @@ processors:
   batch/metrics:
     send_batch_size: 10000
     timeout: 10s
-  # Copy region and instance from Resource attributes to metric data point attributes
-  # This ensures all metrics have the region and instance labels
+  # Copy Resource attributes to metric data point attributes.
+  # This ensures all metrics have stable routing labels and a unique writer label.
   transform/add_resource_attributes:
     metric_statements:
       - context: datapoint
         statements:
           - set(attributes["region"], resource.attributes["region"]) where resource.attributes["region"] != nil
           - set(attributes["instance"], resource.attributes["instance"]) where resource.attributes["instance"] != nil
+          - set(attributes["service.name"], resource.attributes["service.name"]) where resource.attributes["service.name"] != nil
+          - set(attributes["service_instance_id"], resource.attributes["service.instance.id"]) where resource.attributes["service.instance.id"] != nil
+          - set(attributes["worker_id"], resource.attributes["worker.id"]) where resource.attributes["worker.id"] != nil
+          - set(attributes["platformatic_application_id"], resource.attributes["platformatic.application.id"]) where resource.attributes["platformatic.application.id"] != nil
   # Add storage_api_otel_ prefix to all metrics
   # Note: storage_api_ transform must be FIRST to avoid double prefix
   metricstransform/host:

src/internal/monitoring/otel-metrics.test.ts

@@ -2,6 +2,7 @@ interface OTelGlobalState {
   __otelMetricsShutdown?: () => Promise<void>
 }
 
+import fs from 'node:fs'
 import { vi } from 'vitest'
 
 const mockedMetricsModules = [
@@ -16,6 +17,8 @@ const mockedMetricsModules = [
   '@opentelemetry/instrumentation-runtime-node',
   '@opentelemetry/resources',
   '@opentelemetry/sdk-metrics',
+  '@platformatic/globals',
+  'os',
 ] as const
 
 async function importOtelMetricsModule() {
@@ -262,4 +265,142 @@ describe('otel metrics', () => {
       })
     )
   })
+
+  test('uses Watt worker identity as the OTel service instance id', async () => {
+    delete process.env.OTEL_EXPORTER_OTLP_ENDPOINT
+    delete process.env.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT
+
+    const registerInstrumentations = vi.fn(() => vi.fn())
+    const resourceFromAttributes = vi.fn((attributes) => attributes)
+    const HostMetrics = vi.fn(function () {
+      return {
+        start: vi.fn(),
+      }
+    })
+    const MeterProvider = vi.fn(function () {
+      return {
+        shutdown: vi.fn().mockResolvedValue(undefined),
+        getMeter: vi.fn(() => ({})),
+      }
+    })
+    const prometheusExporterOptions: Array<{ withResourceConstantLabels: RegExp }> = []
+    const PrometheusExporter = vi.fn(function (options: { withResourceConstantLabels: RegExp }) {
+      prometheusExporterOptions.push(options)
+      return {
+        getMetricsRequestHandler: vi.fn(),
+      }
+    })
+    const RuntimeNodeInstrumentation = vi.fn(function () {
+      return {}
+    })
+    const StorageNodeInstrumentation = vi.fn(function () {
+      return {}
+    })
+
+    vi.doMock('../../config', () => ({
+      getConfig: vi.fn(() => ({
+        version: 'test-version',
+        otelMetricsExportIntervalMs: 1000,
+        otelMetricsEnabled: true,
+        otelMetricsTemporality: 'CUMULATIVE',
+        prometheusMetricsEnabled: true,
+        region: 'local',
+        serviceName: 'storage-api',
+      })),
+    }))
+    vi.doMock('@platformatic/globals', () => ({
+      getGlobal: vi.fn(() => ({
+        applicationId: 'storage-api:tenant',
+        workerId: '3',
+      })),
+    }))
+    vi.doMock('os', () => ({
+      hostname: vi.fn(() => 'storage-host-a'),
+    }))
+    vi.doMock('@internal/monitoring/logger', () => ({
+      logger: { info: vi.fn() },
+      logSchema: { error: vi.fn(), info: vi.fn() },
+    }))
+    vi.doMock('@internal/monitoring/system', () => ({
+      StorageNodeInstrumentation,
+    }))
+    vi.doMock('@opentelemetry/api', () => ({
+      metrics: {
+        setGlobalMeterProvider: vi.fn(),
+      },
+    }))
+    vi.doMock('@opentelemetry/exporter-metrics-otlp-grpc', () => ({
+      OTLPMetricExporter: vi.fn(function () {
+        return {}
+      }),
+    }))
+    vi.doMock('@opentelemetry/exporter-prometheus', () => ({
+      PrometheusExporter,
+    }))
+    vi.doMock('@opentelemetry/host-metrics', () => ({
+      HostMetrics,
+    }))
+    vi.doMock('@opentelemetry/instrumentation', () => ({
+      registerInstrumentations,
+    }))
+    vi.doMock('@opentelemetry/instrumentation-runtime-node', () => ({
+      RuntimeNodeInstrumentation,
+    }))
+    vi.doMock('@opentelemetry/resources', () => ({
+      resourceFromAttributes,
+    }))
+    vi.doMock('@opentelemetry/sdk-metrics', () => ({
+      AggregationTemporality: {
+        CUMULATIVE: 'CUMULATIVE',
+        DELTA: 'DELTA',
+      },
+      AggregationType: {
+        DROP: 'DROP',
+        EXPLICIT_BUCKET_HISTOGRAM: 'EXPLICIT_BUCKET_HISTOGRAM',
+      },
+      MeterProvider,
+      PeriodicExportingMetricReader: vi.fn(),
+    }))
+
+    await importOtelMetricsModule()
+
+    expect(resourceFromAttributes).toHaveBeenCalledWith(
+      expect.objectContaining({
+        instance: 'storage-host-a',
+        'service.instance.id': 'storage-host-a:storage-api:tenant:worker:3',
+        'platformatic.application.id': 'storage-api:tenant',
+        'worker.id': '3',
+        'process.pid': process.pid,
+      })
+    )
+    expect(PrometheusExporter).toHaveBeenCalledWith(
+      expect.objectContaining({
+        withResourceConstantLabels: expect.objectContaining({
+          test: expect.any(Function),
+        }),
+      })
+    )
+
+    const prometheusLabelFilter = prometheusExporterOptions[0].withResourceConstantLabels
+    expect(prometheusLabelFilter.test('service.instance.id')).toBe(true)
+    expect(prometheusLabelFilter.test('worker.id')).toBe(true)
+    expect(prometheusLabelFilter.test('platformatic.application.id')).toBe(true)
+  })
+
+  test('collector promotes service instance identity to metric labels', () => {
+    const collectorConfig = fs.readFileSync(
+      'monitoring/otel/config/otel-collector-config.yml',
+      'utf8'
+    )
+
+    expect(collectorConfig).toContain(
+      'set(attributes["service_instance_id"], resource.attributes["service.instance.id"])'
+    )
+    expect(collectorConfig).toContain(
+      'set(attributes["worker_id"], resource.attributes["worker.id"])'
+    )
+    expect(collectorConfig).toContain(
+      'set(attributes["platformatic_application_id"], resource.attributes["platformatic.application.id"])'
+    )
+  })
 })

src/internal/monitoring/otel-metrics.ts

@@ -16,6 +16,7 @@ import {
   PeriodicExportingMetricReader,
 } from '@opentelemetry/sdk-metrics'
 import { ATTR_SERVICE_NAME, ATTR_SERVICE_VERSION } from '@opentelemetry/semantic-conventions'
+import { getGlobal } from '@platformatic/globals'
 import { FastifyReply, FastifyRequest } from 'fastify'
 import * as os from 'os'
 import { getConfig } from '../../config'
@@ -39,6 +40,39 @@ interface OTelMetricsGlobalState {
   __otelMetricsShutdown?: () => Promise<void>
 }
 
+const SERVICE_INSTANCE_ID_ATTRIBUTE = 'service.instance.id'
+const PROCESS_PID_ATTRIBUTE = 'process.pid'
+const WORKER_ID_ATTRIBUTE = 'worker.id'
+const PLATFORMATIC_APPLICATION_ID_ATTRIBUTE = 'platformatic.application.id'
+
+function normalizeMetricIdentityPart(value: unknown): string | undefined {
+  if (typeof value === 'number' && Number.isFinite(value)) {
+    return `${value}`
+  }
+
+  if (typeof value === 'string') {
+    const trimmed = value.trim()
+    return trimmed === '' ? undefined : trimmed
+  }
+
+  return undefined
+}
+
+function resolveMetricIdentity() {
+  const hostname = os.hostname()
+  const platformatic = getGlobal()
+  const applicationId = normalizeMetricIdentityPart(platformatic?.applicationId)
+  const workerId = normalizeMetricIdentityPart(platformatic?.workerId)
+  const runtimeId = workerId === undefined ? `pid:${process.pid}` : `worker:${workerId}`
+
+  return {
+    instance: hostname,
+    serviceInstanceId: [hostname, applicationId, runtimeId].filter(Boolean).join(':'),
+    applicationId,
+    workerId,
+  }
+}
+
 function unregisterMetricInstrumentation(unregister: (() => void) | undefined) {
   if (!unregister) {
     return
@@ -57,7 +91,8 @@ function unregisterMetricInstrumentation(unregister: (() => void) | undefined) {
 // =============================================================================
 // Shared config
 // =============================================================================
-const instance = os.hostname()
+const metricIdentity = resolveMetricIdentity()
+const instance = metricIdentity.instance
 const headersEnv = process.env.OTEL_EXPORTER_OTLP_METRICS_HEADERS || ''
 const otlpEndpoint =
   process.env.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT || process.env.OTEL_EXPORTER_OTLP_ENDPOINT
@@ -85,6 +120,12 @@ const resource = resourceFromAttributes({
   'metric.version': '1',
   region,
   instance,
+  [SERVICE_INSTANCE_ID_ATTRIBUTE]: metricIdentity.serviceInstanceId,
+  [PROCESS_PID_ATTRIBUTE]: process.pid,
+  ...(metricIdentity.workerId ? { [WORKER_ID_ATTRIBUTE]: metricIdentity.workerId } : {}),
+  ...(metricIdentity.applicationId
+    ? { [PLATFORMATIC_APPLICATION_ID_ATTRIBUTE]: metricIdentity.applicationId }
+    : {}),
 })
 
 // Bucket boundaries for duration histograms (in seconds)
@@ -256,7 +297,8 @@ if (otelMetricsEnabled) {
     prometheusExporter = new PrometheusExporter({
       prefix: serviceName,
       preventServerStart: true,
-      withResourceConstantLabels: /^(region|instance|metric\.version|service\.name)$/,
+      withResourceConstantLabels:
+        /^(region|instance|metric\.version|service\.name|service\.instance\.id|worker\.id|platformatic\.application\.id)$/,
     })
     readers.push(prometheusExporter)
   }