fix: clear references to metrics on stop (#3154)

Otherwise the node can leak memory.

Author: achingbrain

packages/metrics-opentelemetry/package.json

@@ -47,6 +47,7 @@
     "it-stream-types": "^2.0.2"
   },
   "devDependencies": {
+    "@libp2p/logger": "^5.1.18",
     "aegir": "^47.0.14"
   },
   "browser": {

packages/metrics-opentelemetry/src/index.ts

@@ -48,14 +48,15 @@ import { OpenTelemetryMetric } from './metric.js'
 import { OpenTelemetrySummaryGroup } from './summary-group.js'
 import { OpenTelemetrySummary } from './summary.js'
 import { collectSystemMetrics } from './system-metrics.js'
-import type { MultiaddrConnection, Stream, Connection, Metric, MetricGroup, Metrics, CalculatedMetricOptions, MetricOptions, Counter, CounterGroup, Histogram, HistogramOptions, HistogramGroup, Summary, SummaryOptions, SummaryGroup, CalculatedHistogramOptions, CalculatedSummaryOptions, NodeInfo, TraceFunctionOptions, TraceGeneratorFunctionOptions, TraceAttributes } from '@libp2p/interface'
-import type { Span, Attributes } from '@opentelemetry/api'
+import type { MultiaddrConnection, Stream, Connection, Metric, MetricGroup, Metrics, CalculatedMetricOptions, MetricOptions, Counter, CounterGroup, Histogram, HistogramOptions, HistogramGroup, Summary, SummaryOptions, SummaryGroup, CalculatedHistogramOptions, CalculatedSummaryOptions, NodeInfo, TraceFunctionOptions, TraceGeneratorFunctionOptions, TraceAttributes, ComponentLogger, Logger } from '@libp2p/interface'
+import type { Span, Attributes, Meter, Observable } from '@opentelemetry/api'
 import type { Duplex } from 'it-stream-types'
 
 // see https://betterstack.com/community/guides/observability/opentelemetry-metrics-nodejs/#prerequisites
 
 export interface OpenTelemetryComponents {
   nodeInfo: NodeInfo
+  logger: ComponentLogger
 }
 
 export interface OpenTelemetryMetricsInit {
@@ -92,14 +93,20 @@ export interface OpenTelemetryMetricsInit {
 class OpenTelemetryMetrics implements Metrics {
   private transferStats: Map<string, number>
   private readonly tracer: ReturnType<typeof trace.getTracer>
-  private readonly meterName: string
+  private readonly meter: Meter
+  private readonly log: Logger
+  private metrics: Map<string, OpenTelemetryMetric | OpenTelemetryMetricGroup | OpenTelemetryCounter | OpenTelemetryCounterGroup | OpenTelemetryHistogram | OpenTelemetryHistogramGroup | OpenTelemetrySummary | OpenTelemetrySummaryGroup>
+  private observables: Map<string, Observable>
 
   constructor (components: OpenTelemetryComponents, init?: OpenTelemetryMetricsInit) {
+    this.log = components.logger.forComponent('libp2p:open-telemetry-metrics')
     this.tracer = trace.getTracer(init?.appName ?? components.nodeInfo.name, init?.appVersion ?? components.nodeInfo.version)
+    this.metrics = new Map()
+    this.observables = new Map()
 
     // holds global and per-protocol sent/received stats
     this.transferStats = new Map()
-    this.meterName = init?.meterName ?? components.nodeInfo.name
+    this.meter = metrics.getMeterProvider().getMeter(init?.meterName ?? components.nodeInfo.name)
 
     this.registerCounterGroup('libp2p_data_transfer_bytes_total', {
       label: 'protocol',
@@ -126,6 +133,16 @@ class OpenTelemetryMetrics implements Metrics {
     '@libp2p/metrics'
   ]
 
+  start (): void {
+
+  }
+
+  stop (): void {
+    this.transferStats.clear()
+    this.metrics.clear()
+    this.observables.clear()
+  }
+
   /**
    * Increment the transfer stat for the passed key, making sure
    * it exists first
@@ -177,23 +194,32 @@ class OpenTelemetryMetrics implements Metrics {
       throw new InvalidParametersError('Metric name is required')
     }
 
-    const meter = metrics.getMeterProvider().getMeter(this.meterName)
-
     if (isCalculatedMetricOptions<CalculatedMetricOptions>(opts)) {
-      const calculate = opts.calculate
-      const counter = meter.createObservableGauge(name, {
+      const gauge = this.observables.get(name) ?? this.meter.createObservableGauge(name, {
         description: opts?.help ?? name
       })
-      counter.addCallback(async (result) => {
+
+      const calculate = opts.calculate
+      gauge.addCallback(async (result) => {
         result.observe(await calculate())
       })
 
+      this.observables.set(name, gauge)
+
       return
     }
 
-    return new OpenTelemetryMetric(meter.createGauge(name, {
-      description: opts?.help ?? name
-    }))
+    let metric = this.metrics.get(name)
+
+    if (metric == null) {
+      metric = new OpenTelemetryMetric(this.meter.createGauge(name, {
+        description: opts?.help ?? name
+      }))
+
+      this.metrics.set(name, metric)
+    }
+
+    return metric
   }
 
   registerMetricGroup (name: string, opts: CalculatedMetricOptions<Record<string, number>>): void
@@ -203,14 +229,14 @@ class OpenTelemetryMetrics implements Metrics {
       throw new InvalidParametersError('Metric name is required')
     }
 
-    const meter = metrics.getMeterProvider().getMeter(this.meterName)
     const label = opts?.label ?? name
 
     if (isCalculatedMetricOptions<CalculatedMetricOptions<Record<string, number>>>(opts)) {
-      const calculate = opts.calculate
-      const gauge = meter.createObservableGauge(name, {
+      const gauge = this.observables.get(name) ?? this.meter.createObservableGauge(name, {
         description: opts?.help ?? name
       })
+
+      const calculate = opts.calculate
       gauge.addCallback(async (observable) => {
         const observed = await calculate()
 
@@ -221,12 +247,22 @@ class OpenTelemetryMetrics implements Metrics {
         }
       })
 
+      this.observables.set(name, gauge)
+
       return
     }
 
-    return new OpenTelemetryMetricGroup(label, meter.createGauge(name, {
-      description: opts?.help ?? name
-    }))
+    let metric = this.metrics.get(name)
+
+    if (metric == null) {
+      metric = new OpenTelemetryMetricGroup(label, this.meter.createGauge(name, {
+        description: opts?.help ?? name
+      }))
+
+      this.metrics.set(name, metric)
+    }
+
+    return metric
   }
 
   registerCounter (name: string, opts: CalculatedMetricOptions): void
@@ -236,23 +272,32 @@ class OpenTelemetryMetrics implements Metrics {
       throw new InvalidParametersError('Metric name is required')
     }
 
-    const meter = metrics.getMeterProvider().getMeter(this.meterName)
-
     if (isCalculatedMetricOptions<CalculatedMetricOptions>(opts)) {
-      const calculate = opts.calculate
-      const counter = meter.createObservableCounter(name, {
+      const counter = this.observables.get(name) ?? this.meter.createObservableCounter(name, {
         description: opts?.help ?? name
       })
+
+      const calculate = opts.calculate
       counter.addCallback(async (result) => {
         result.observe(await calculate())
       })
 
+      this.observables.set(name, counter)
+
       return
     }
 
-    return new OpenTelemetryCounter(meter.createCounter(name, {
-      description: opts?.help ?? name
-    }))
+    let metric = this.metrics.get(name)
+
+    if (metric == null) {
+      metric = new OpenTelemetryCounter(this.meter.createCounter(name, {
+        description: opts?.help ?? name
+      }))
+
+      this.metrics.set(name, metric)
+    }
+
+    return metric
   }
 
   registerCounterGroup (name: string, opts: CalculatedMetricOptions<Record<string, number>>): void
@@ -262,15 +307,15 @@ class OpenTelemetryMetrics implements Metrics {
       throw new InvalidParametersError('Metric name is required')
     }
 
-    const meter = metrics.getMeterProvider().getMeter(this.meterName)
     const label = opts?.label ?? name
 
     if (isCalculatedMetricOptions<CalculatedMetricOptions<Record<string, number>>>(opts)) {
-      const values: Record<string, number> = {}
-      const calculate = opts.calculate
-      const counter = meter.createObservableGauge(name, {
+      const counter = this.observables.get(name) ?? this.meter.createObservableCounter(name, {
         description: opts?.help ?? name
       })
+
+      const values: Record<string, number> = {}
+      const calculate = opts.calculate
       counter.addCallback(async (observable) => {
         const observed = await calculate()
 
@@ -290,9 +335,17 @@ class OpenTelemetryMetrics implements Metrics {
       return
     }
 
-    return new OpenTelemetryCounterGroup(label, meter.createCounter(name, {
-      description: opts?.help ?? name
-    }))
+    let metric = this.metrics.get(name)
+
+    if (metric == null) {
+      metric = new OpenTelemetryCounterGroup(label, this.meter.createCounter(name, {
+        description: opts?.help ?? name
+      }))
+
+      this.metrics.set(name, metric)
+    }
+
+    return metric
   }
 
   registerHistogram (name: string, opts: CalculatedHistogramOptions): void
@@ -302,18 +355,24 @@ class OpenTelemetryMetrics implements Metrics {
       throw new InvalidParametersError('Metric name is required')
     }
 
-    const meter = metrics.getMeterProvider().getMeter(this.meterName)
-
     if (isCalculatedMetricOptions<CalculatedHistogramOptions>(opts)) {
       return
     }
 
-    return new OpenTelemetryHistogram(meter.createHistogram(name, {
-      advice: {
-        explicitBucketBoundaries: opts.buckets
-      },
-      description: opts?.help ?? name
-    }))
+    let metric = this.metrics.get(name)
+
+    if (metric == null) {
+      metric = new OpenTelemetryHistogram(this.meter.createHistogram(name, {
+        advice: {
+          explicitBucketBoundaries: opts.buckets
+        },
+        description: opts?.help ?? name
+      }))
+
+      this.metrics.set(name, metric)
+    }
+
+    return metric
   }
 
   registerHistogramGroup (name: string, opts: CalculatedHistogramOptions<Record<string, number>>): void
@@ -323,19 +382,26 @@ class OpenTelemetryMetrics implements Metrics {
       throw new InvalidParametersError('Metric name is required')
     }
 
-    const meter = metrics.getMeterProvider().getMeter(this.meterName)
     const label = opts?.label ?? name
 
     if (isCalculatedMetricOptions<CalculatedHistogramOptions<Record<string, number>>>(opts)) {
       return
     }
 
-    return new OpenTelemetryHistogramGroup(label, meter.createHistogram(name, {
-      advice: {
-        explicitBucketBoundaries: opts.buckets
-      },
-      description: opts?.help ?? name
-    }))
+    let metric = this.metrics.get(name)
+
+    if (metric == null) {
+      metric = new OpenTelemetryHistogramGroup(label, this.meter.createHistogram(name, {
+        advice: {
+          explicitBucketBoundaries: opts.buckets
+        },
+        description: opts?.help ?? name
+      }))
+
+      this.metrics.set(name, metric)
+    }
+
+    return metric
   }
 
   registerSummary (name: string, opts: CalculatedSummaryOptions): void
@@ -345,15 +411,21 @@ class OpenTelemetryMetrics implements Metrics {
       throw new InvalidParametersError('Metric name is required')
     }
 
-    const meter = metrics.getMeterProvider().getMeter(this.meterName)
-
     if (isCalculatedMetricOptions<CalculatedHistogramOptions>(opts)) {
       return
     }
 
-    return new OpenTelemetrySummary(meter.createGauge(name, {
-      description: opts?.help ?? name
-    }))
+    let metric = this.metrics.get(name)
+
+    if (metric == null) {
+      metric = new OpenTelemetrySummary(this.meter.createGauge(name, {
+        description: opts?.help ?? name
+      }))
+
+      this.metrics.set(name, metric)
+    }
+
+    return metric
   }
 
   registerSummaryGroup (name: string, opts: CalculatedSummaryOptions<Record<string, number>>): void
@@ -363,16 +435,23 @@ class OpenTelemetryMetrics implements Metrics {
       throw new InvalidParametersError('Metric name is required')
     }
 
-    const meter = metrics.getMeterProvider().getMeter(this.meterName)
     const label = opts?.label ?? name
 
     if (isCalculatedMetricOptions<CalculatedSummaryOptions>(opts)) {
       return
     }
 
-    return new OpenTelemetrySummaryGroup(label, meter.createGauge(name, {
-      description: opts?.help ?? name
-    }))
+    let metric = this.metrics.get(name)
+
+    if (metric == null) {
+      metric = new OpenTelemetrySummaryGroup(label, this.meter.createGauge(name, {
+        description: opts?.help ?? name
+      }))
+
+      this.metrics.set(name, metric)
+    }
+
+    return metric
   }
 
   createTrace (): any {