feat: add opened/closed counters to protocol stream metrics

packages/libp2p/test/connection/index.spec.ts

@@ -12,7 +12,7 @@ import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
 import { createConnection } from '../../src/connection.js'
 import { UnhandledProtocolError } from '../../src/errors.ts'
 import type { ConnectionComponents, ConnectionInit } from '../../src/connection.js'
-import type { MultiaddrConnection, PeerStore, Stream, StreamMuxer } from '@libp2p/interface'
+import type { MultiaddrConnection, PeerStore, Stream, StreamMuxer, Metrics } from '@libp2p/interface'
 import type { Registrar } from '@libp2p/interface-internal'
 import type { StubbedInstance } from 'sinon-ts'
 
@@ -469,4 +469,64 @@ describe('connection', () => {
     expect(middleware2.called).to.be.false()
     expect(incomingStream).to.have.nested.property('abort.called', true)
   })
+
+  it('should call trackProtocolStream when a new outbound stream is opened', async () => {
+    const metrics = stubInterface<Metrics>()
+
+    const connection = createConnection({
+      ...components,
+      metrics
+    }, init)
+
+    await connection.newStream([ECHO_PROTOCOL])
+
+    // trackProtocolStream must be called exactly once with the negotiated stream
+    expect(metrics.trackProtocolStream.callCount).to.equal(1)
+    expect(metrics.trackProtocolStream.firstCall.args[0]).to.have.property('protocol', ECHO_PROTOCOL)
+    expect(metrics.trackProtocolStream.firstCall.args[0]).to.have.property('direction', 'outbound')
+  })
+
+  it('should call trackProtocolStream when an inbound stream is opened', async () => {
+    const streamProtocol = '/test/protocol'
+    const metrics = stubInterface<Metrics>()
+
+    registrar.getHandler.withArgs(streamProtocol).returns({
+      handler: () => {},
+      options: {}
+    })
+    registrar.getMiddleware.withArgs(streamProtocol).returns([])
+    registrar.getProtocols.returns([streamProtocol])
+
+    const stubbedMuxer = stubInterface<StreamMuxer>({ streams: [] })
+
+    createConnection({
+      ...components,
+      metrics
+    }, {
+      ...init,
+      muxer: stubbedMuxer
+    })
+
+    // grab the inbound stream listener registered on the muxer
+    const onIncomingStream = stubbedMuxer.addEventListener.getCall(0).args[1]
+
+    if (onIncomingStream == null || typeof onIncomingStream !== 'function') {
+      throw new Error('No incoming stream handler registered')
+    }
+
+    const incomingStream = stubInterface<Stream>({
+      log: defaultLogger().forComponent('stream'),
+      protocol: streamProtocol,
+      direction: 'inbound'
+    })
+
+    onIncomingStream(new CustomEvent('stream', { detail: incomingStream }))
+
+    // inbound stream handling is async
+    await delay(100)
+
+    expect(metrics.trackProtocolStream.callCount).to.equal(1)
+    expect(metrics.trackProtocolStream.firstCall.args[0]).to.have.property('protocol', streamProtocol)
+    expect(metrics.trackProtocolStream.firstCall.args[0]).to.have.property('direction', 'inbound')
+  })
 })

packages/metrics-opentelemetry/package.json

@@ -47,6 +47,7 @@
   },
   "devDependencies": {
     "@libp2p/logger": "^6.2.7",
+    "@opentelemetry/sdk-metrics": "^1.30.1",
     "aegir": "^47.0.22"
   },
   "browser": {

packages/metrics-opentelemetry/src/index.ts

@@ -36,16 +36,16 @@
 import { InvalidParametersError, serviceCapabilities } from '@libp2p/interface'
 import { isAsyncGenerator, isGenerator, isPromise } from '@libp2p/utils'
 import { trace, metrics, context, SpanStatusCode } from '@opentelemetry/api'
-import { OpenTelemetryCounterGroup } from './counter-group.ts'
-import { OpenTelemetryCounter } from './counter.ts'
-import { OpenTelemetryHistogramGroup } from './histogram-group.ts'
-import { OpenTelemetryHistogram } from './histogram.ts'
-import { OpenTelemetryMetricGroup } from './metric-group.ts'
-import { OpenTelemetryMetric } from './metric.ts'
-import { OpenTelemetrySummaryGroup } from './summary-group.ts'
-import { OpenTelemetrySummary } from './summary.ts'
-import { collectSystemMetrics } from './system-metrics.ts'
-import type { MultiaddrConnection, Stream, Metric, MetricGroup, Metrics, CalculatedMetricOptions, MetricOptions, Counter, CounterGroup, Histogram, HistogramOptions, HistogramGroup, Summary, SummaryOptions, SummaryGroup, CalculatedHistogramOptions, CalculatedSummaryOptions, NodeInfo, TraceFunctionOptions, TraceGeneratorFunctionOptions, TraceAttributes, ComponentLogger, Logger, MessageStream } from '@libp2p/interface'
+import { OpenTelemetryCounterGroup } from './counter-group.js'
+import { OpenTelemetryCounter } from './counter.js'
+import { OpenTelemetryHistogramGroup } from './histogram-group.js'
+import { OpenTelemetryHistogram } from './histogram.js'
+import { OpenTelemetryMetricGroup } from './metric-group.js'
+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, StreamCloseEvent, Metric, MetricGroup, Metrics, CalculatedMetricOptions, MetricOptions, Counter, CounterGroup, Histogram, HistogramOptions, HistogramGroup, Summary, SummaryOptions, SummaryGroup, CalculatedHistogramOptions, CalculatedSummaryOptions, NodeInfo, TraceFunctionOptions, TraceGeneratorFunctionOptions, TraceAttributes, ComponentLogger, Logger, MessageStream } from '@libp2p/interface'
 import type { Span, Attributes, Meter, Observable } from '@opentelemetry/api'
 
 // see https://betterstack.com/community/guides/observability/opentelemetry-metrics-nodejs/#prerequisites
@@ -93,6 +93,9 @@ class OpenTelemetryMetrics implements Metrics {
   private readonly log: Logger
   private metrics: Map<string, OpenTelemetryMetric | OpenTelemetryMetricGroup | OpenTelemetryCounter | OpenTelemetryCounterGroup | OpenTelemetryHistogram | OpenTelemetryHistogramGroup | OpenTelemetrySummary | OpenTelemetrySummaryGroup>
   private observables: Map<string, Observable>
+  private readonly streamsOpened: CounterGroup
+  private readonly streamsClosed: CounterGroup
+  private readonly streamsCloseErrors: CounterGroup
 
   constructor (components: OpenTelemetryComponents, init?: OpenTelemetryMetricsInit) {
     this.log = components.logger.forComponent('libp2p:open-telemetry-metrics')
@@ -120,6 +123,19 @@ class OpenTelemetryMetrics implements Metrics {
       }
     })
 
+    this.streamsOpened = this.registerCounterGroup('libp2p_protocol_streams_opened_total', {
+      label: 'protocol',
+      help: 'Total number of protocol streams opened, by direction and protocol'
+    })
+    this.streamsClosed = this.registerCounterGroup('libp2p_protocol_streams_closed_total', {
+      label: 'protocol',
+      help: 'Total number of protocol streams closed, by direction and protocol'
+    })
+    this.streamsCloseErrors = this.registerCounterGroup('libp2p_protocol_streams_close_errors_total', {
+      label: 'protocol',
+      help: 'Total number of protocol streams that ended with an error (abort, reset, etc.), by direction and protocol'
+    })
+
     collectSystemMetrics(this, init)
   }
 
@@ -188,6 +204,19 @@ class OpenTelemetryMetrics implements Metrics {
     }
 
     this._track(stream, stream.protocol)
+
+    const label = `${stream.direction} ${stream.protocol}`
+
+    this.streamsOpened.increment({ [label]: 1 })
+
+    stream.addEventListener('close', (evt: Event) => {
+      const e = evt as StreamCloseEvent
+      if (e.error != null) {
+        this.streamsCloseErrors.increment({ [label]: 1 })
+      } else {
+        this.streamsClosed.increment({ [label]: 1 })
+      }
+    }, { once: true })
   }
 
   registerMetric (name: string, opts: CalculatedMetricOptions): void

packages/metrics-opentelemetry/test/track-protocol-stream.spec.ts

@@ -0,0 +1,129 @@
+import { StreamAbortEvent, TypedEventEmitter, StreamCloseEvent } from '@libp2p/interface'
+import { defaultLogger } from '@libp2p/logger'
+import { metrics as otelApi } from '@opentelemetry/api'
+import {
+  AggregationTemporality,
+  DataPointType,
+  InMemoryMetricExporter,
+  MeterProvider,
+  PeriodicExportingMetricReader
+} from '@opentelemetry/sdk-metrics'
+import { expect } from 'aegir/chai'
+import { openTelemetryMetrics } from '../src/index.js'
+import type { MessageStreamEvents, Stream } from '@libp2p/interface'
+import type { ResourceMetrics, ScopeMetrics } from '@opentelemetry/sdk-metrics'
+
+function sumScopeMetrics (sm: ScopeMetrics, metricName: string, protocolValue: string): number {
+  let sub = 0
+  for (const metric of sm.metrics) {
+    if (metric.descriptor.name !== metricName || metric.dataPointType !== DataPointType.SUM) {
+      continue
+    }
+    for (const dp of metric.dataPoints) {
+      if (dp.attributes.protocol === protocolValue) {
+        sub += dp.value as number
+      }
+    }
+  }
+  return sub
+}
+
+function sumForProtocol (batches: ResourceMetrics[], metricName: string, protocolValue: string): number {
+  let total = 0
+  for (const batch of batches) {
+    for (const sm of batch.scopeMetrics) {
+      total += sumScopeMetrics(sm, metricName, protocolValue)
+    }
+  }
+  return total
+}
+
+describe('opentelemetry protocol stream counters', () => {
+  let previousProvider: ReturnType<typeof otelApi.getMeterProvider>
+  let reader: PeriodicExportingMetricReader
+  let exporter: InMemoryMetricExporter
+  let provider: MeterProvider
+  let meterId: number
+
+  before(() => {
+    previousProvider = otelApi.getMeterProvider()
+    exporter = new InMemoryMetricExporter(AggregationTemporality.CUMULATIVE)
+    reader = new PeriodicExportingMetricReader({
+      exporter,
+      exportIntervalMillis: 3_600_000
+    })
+    provider = new MeterProvider({ readers: [reader] })
+    otelApi.setGlobalMeterProvider(provider)
+  })
+
+  after(async () => {
+    await provider.shutdown()
+    otelApi.setGlobalMeterProvider(previousProvider)
+  })
+
+  beforeEach(() => {
+    meterId = Date.now() + Math.floor(Math.random() * 1e6)
+    exporter.reset()
+  })
+
+  function makeStream (direction: 'inbound' | 'outbound', protocol: string): Stream {
+    const target = new TypedEventEmitter<MessageStreamEvents>()
+    return {
+      direction,
+      protocol,
+      log: defaultLogger().forComponent('stream'),
+      addEventListener: target.addEventListener.bind(target),
+      removeEventListener: target.removeEventListener.bind(target),
+      dispatchEvent: target.dispatchEvent.bind(target),
+      send: () => true
+    } as unknown as Stream
+  }
+
+  it('increments opened and clean closed counters', async () => {
+    const metrics = openTelemetryMetrics()({
+      nodeInfo: {
+        name: `otel-ps-test-${meterId}`,
+        version: '1.0.0',
+        userAgent: 'test/1.0.0'
+      },
+      logger: defaultLogger()
+    })
+
+    const stream = makeStream('outbound', '/identify/1.0.0')
+    const label = `${stream.direction} ${stream.protocol}`
+
+    metrics.trackProtocolStream(stream)
+    stream.dispatchEvent(new StreamCloseEvent())
+
+    await reader.forceFlush()
+    const batches = exporter.getMetrics()
+
+    expect(sumForProtocol(batches, 'libp2p_protocol_streams_opened_total', label)).to.equal(1)
+    expect(sumForProtocol(batches, 'libp2p_protocol_streams_closed_total', label)).to.equal(1)
+    expect(sumForProtocol(batches, 'libp2p_protocol_streams_close_errors_total', label)).to.equal(0)
+  })
+
+  it('increments close-errors counter on StreamAbortEvent', async () => {
+    const metrics = openTelemetryMetrics()({
+      nodeInfo: {
+        name: `otel-ps-test-${meterId}`,
+        version: '1.0.0',
+        userAgent: 'test/1.0.0'
+      },
+      logger: defaultLogger()
+    })
+
+    const stream = makeStream('inbound', '/ping/1.0.0')
+    const label = `${stream.direction} ${stream.protocol}`
+
+    metrics.trackProtocolStream(stream)
+    stream.dispatchEvent(new StreamAbortEvent(new Error('aborted')))
+
+    await reader.forceFlush()
+    const batches = exporter.getMetrics()
+
+    expect(sumForProtocol(batches, 'libp2p_protocol_streams_opened_total', label)).to.equal(1)
+    expect(sumForProtocol(batches, 'libp2p_protocol_streams_closed_total', label)).to.equal(0)
+    expect(sumForProtocol(batches, 'libp2p_protocol_streams_close_errors_total', label)).to.equal(1)
+  })
+})

packages/metrics-prometheus/src/index.ts

@@ -70,15 +70,15 @@ import { statfs } from 'node:fs/promises'
 import { totalmem } from 'node:os'
 import { serviceCapabilities } from '@libp2p/interface'
 import { collectDefaultMetrics, register } from 'prom-client'
-import { PrometheusCounterGroup } from './counter-group.ts'
-import { PrometheusCounter } from './counter.ts'
-import { PrometheusHistogramGroup } from './histogram-group.ts'
-import { PrometheusHistogram } from './histogram.ts'
-import { PrometheusMetricGroup } from './metric-group.ts'
-import { PrometheusMetric } from './metric.ts'
-import { PrometheusSummaryGroup } from './summary-group.ts'
-import { PrometheusSummary } from './summary.ts'
-import type { ComponentLogger, Logger, MultiaddrConnection, Stream, CalculatedMetricOptions, Counter, CounterGroup, Metric, MetricGroup, MetricOptions, Metrics, CalculatedHistogramOptions, CalculatedSummaryOptions, HistogramOptions, Histogram, HistogramGroup, SummaryOptions, Summary, SummaryGroup, MessageStream } from '@libp2p/interface'
+import { PrometheusCounterGroup } from './counter-group.js'
+import { PrometheusCounter } from './counter.js'
+import { PrometheusHistogramGroup } from './histogram-group.js'
+import { PrometheusHistogram } from './histogram.js'
+import { PrometheusMetricGroup } from './metric-group.js'
+import { PrometheusMetric } from './metric.js'
+import { PrometheusSummaryGroup } from './summary-group.js'
+import { PrometheusSummary } from './summary.js'
+import type { ComponentLogger, Logger, MultiaddrConnection, Stream, StreamCloseEvent, CalculatedMetricOptions, Counter, CounterGroup, Metric, MetricGroup, MetricOptions, Metrics, CalculatedHistogramOptions, CalculatedSummaryOptions, HistogramOptions, Histogram, HistogramGroup, SummaryOptions, Summary, SummaryGroup, MessageStream } from '@libp2p/interface'
 import type { DefaultMetricsCollectorConfiguration, Registry, RegistryContentType } from 'prom-client'
 
 // export helper functions for creating buckets
@@ -141,6 +141,9 @@ class PrometheusMetrics implements Metrics {
   private readonly log: Logger
   private transferStats: Map<string, number>
   private readonly registry?: Registry
+  private readonly streamsOpened: CounterGroup
+  private readonly streamsClosed: CounterGroup
+  private readonly streamsCloseErrors: CounterGroup
 
   constructor (components: PrometheusMetricsComponents, init?: Partial<PrometheusMetricsInit>) {
     this.log = components.logger.forComponent('libp2p:prometheus-metrics')
@@ -205,6 +208,20 @@ class PrometheusMetrics implements Metrics {
         }
       }
     })
+
+    this.log('Collecting protocol stream open/close metrics')
+    this.streamsOpened = this.registerCounterGroup('libp2p_protocol_streams_opened_total', {
+      label: 'protocol',
+      help: 'Total number of protocol streams opened, by direction and protocol'
+    })
+    this.streamsClosed = this.registerCounterGroup('libp2p_protocol_streams_closed_total', {
+      label: 'protocol',
+      help: 'Total number of protocol streams closed, by direction and protocol'
+    })
+    this.streamsCloseErrors = this.registerCounterGroup('libp2p_protocol_streams_close_errors_total', {
+      label: 'protocol',
+      help: 'Total number of protocol streams that ended with an error (abort, reset, etc.), by direction and protocol'
+    })
   }
 
   readonly [Symbol.toStringTag] = '@libp2p/metrics-prometheus'
@@ -274,6 +291,22 @@ class PrometheusMetrics implements Metrics {
     }
 
     this._track(stream, stream.protocol)
+
+    // Label format: "direction protocol"  e.g. "inbound /identify/1.0.0"
+    // Matches the format of the existing libp2p_protocol_streams_total gauge.
+    const label = `${stream.direction} ${stream.protocol}`
+
+    // Stream is now open — increment the opened counter immediately.
+    this.streamsOpened.increment({ [label]: 1 })
+
+    stream.addEventListener('close', (evt: Event) => {
+      const e = evt as StreamCloseEvent
+      if (e.error != null) {
+        this.streamsCloseErrors.increment({ [label]: 1 })
+      } else {
+        this.streamsClosed.increment({ [label]: 1 })
+      }
+    }, { once: true })
   }
 
   registerMetric (name: string, opts: PrometheusCalculatedMetricOptions): void