Skip to content
This repository was archived by the owner on Aug 29, 2023. It is now read-only.

Commit bc0bf79

Browse files
dappliondapplion
committed
Add metrics
1 parent c538cf0 commit bc0bf79

File tree

4 files changed

+86
-34
lines changed

4 files changed

+86
-34
lines changed

‎src/index.ts

+10-2
Original file line numberDiff line numberDiff line change
@@ -11,9 +11,12 @@ import { CreateListenerOptions, DialOptions, symbol, Transport } from '@libp2p/i
1111
import type { AbortOptions, Multiaddr } from '@multiformats/multiaddr'
1212
import type { Socket, IpcSocketConnectOpts, TcpSocketConnectOpts } from 'net'
1313
import type { Connection } from '@libp2p/interface-connection'
14+
import type { TcpMetrics } from './metrics.js'
1415

1516
const log = logger('libp2p:tcp')
1617

18+
export { TcpMetrics }
19+
1720
export interface TCPOptions {
1821
/**
1922
* An optional number in ms that is used as an inactivity timeout after which the socket will be closed
@@ -51,9 +54,11 @@ export interface TCPCreateListenerOptions extends CreateListenerOptions, TCPSock
5154

5255
export class TCP implements Transport {
5356
private readonly opts: TCPOptions
57+
private readonly metrics: TcpMetrics | null
5458

55-
constructor (options: TCPOptions = {}) {
59+
constructor (options: TCPOptions = {}, metrics?: TcpMetrics | null) {
5660
this.opts = options
61+
this.metrics = metrics ?? null
5762
}
5863

5964
get [symbol] (): true {
@@ -72,6 +77,7 @@ export class TCP implements Transport {
7277
// Avoid uncaught errors caused by unstable connections
7378
socket.on('error', err => {
7479
log('socket error', err)
80+
this.metrics?.socketEvents.inc({ event: 'error' })
7581
})
7682

7783
const maConn = toMultiaddrConnection(socket, {
@@ -107,6 +113,7 @@ export class TCP implements Transport {
107113

108114
const onTimeout = () => {
109115
log('connection timeout %s', cOptsStr)
116+
this.metrics?.listenerErrors.inc({ error: 'outbound_connection_timeout' })
110117

111118
const err = errCode(new Error(`connection timeout after ${Date.now() - start}ms`), 'ERR_CONNECT_TIMEOUT')
112119
// Note: this will result in onError() being called
@@ -159,7 +166,8 @@ export class TCP implements Transport {
159166
return new TcpListener({
160167
...options,
161168
socketInactivityTimeout: this.opts.inboundSocketInactivityTimeout,
162-
socketCloseTimeout: this.opts.socketCloseTimeout
169+
socketCloseTimeout: this.opts.socketCloseTimeout,
170+
metrics: this.metrics
163171
})
164172
}
165173

‎src/listener.ts

+35-30
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import type { MultiaddrConnection, Connection } from '@libp2p/interface-connecti
1111
import type { Upgrader, Listener } from '@libp2p/interface-transport'
1212
import type { Multiaddr } from '@multiformats/multiaddr'
1313
import type { TCPCreateListenerOptions } from './index.js'
14+
import { ServerStatusMetric, TcpMetrics } from './metrics.js'
1415

1516
const log = logger('libp2p:tcp:listener')
1617

@@ -30,6 +31,7 @@ interface Context extends TCPCreateListenerOptions {
3031
upgrader: Upgrader
3132
socketInactivityTimeout?: number
3233
socketCloseTimeout?: number
34+
metrics: TcpMetrics | null
3335
}
3436

3537
type Status = {started: false} | {started: true, listeningAddr: Multiaddr, peerId: string | null }
@@ -38,6 +40,7 @@ export class TcpListener extends EventEmitter<any> implements Listener {
3840
private readonly server: net.Server
3941
/** Keep track of open connections to destroy in case of timeout */
4042
private readonly connections = new Set<MultiaddrConnection>()
43+
private readonly metrics: TcpMetrics | null
4144

4245
private status: Status = { started: false }
4346

@@ -46,64 +49,66 @@ export class TcpListener extends EventEmitter<any> implements Listener {
4649

4750
context.keepAlive = context.keepAlive ?? true
4851

49-
this.server = net.createServer(context, this.onSocket.bind(this))
52+
this.server = net.createServer(context, socket => {
53+
this.onSocket(socket).catch(e => log('onSocket error', e))
54+
})
5055

5156
this.server
5257
.on('listening', () => this.dispatchEvent(new CustomEvent('listening')))
5358
.on('error', err => this.dispatchEvent(new CustomEvent<Error>('error', { detail: err })))
5459
.on('close', () => this.dispatchEvent(new CustomEvent('close')))
60+
61+
this.metrics = context.metrics
62+
this.metrics?.connections.addCollect(() => {
63+
this.metrics?.connections.set(this.connections.size)
64+
this.metrics?.serverStatus.set(this.status.started ? ServerStatusMetric.started : ServerStatusMetric.stopped)
65+
})
5566
}
5667

57-
private onSocket (socket: net.Socket) {
68+
private async onSocket (socket: net.Socket) {
5869
// Avoid uncaught errors caused by unstable connections
5970
socket.on('error', err => {
6071
log('socket error', err)
72+
this.metrics?.socketEvents.inc({ event: 'error' })
6173
})
6274

6375
let maConn: MultiaddrConnection
6476
try {
6577
maConn = toMultiaddrConnection(socket, {
6678
listeningAddr: this.status.started ? this.status.listeningAddr : undefined,
6779
socketInactivityTimeout: this.context.socketInactivityTimeout,
68-
socketCloseTimeout: this.context.socketCloseTimeout
80+
socketCloseTimeout: this.context.socketCloseTimeout,
81+
metrics: this.metrics
6982
})
7083
} catch (err) {
7184
log.error('inbound connection failed', err)
85+
this.metrics?.listenerErrors.inc({ error: 'inbound_to_connection' })
7286
return
7387
}
7488

7589
log('new inbound connection %s', maConn.remoteAddr)
7690
try {
77-
this.context.upgrader.upgradeInbound(maConn)
78-
.then((conn) => {
79-
log('inbound connection %s upgraded', maConn.remoteAddr)
80-
this.connections.add(maConn)
81-
82-
socket.once('close', () => {
83-
this.connections.delete(maConn)
84-
})
85-
86-
if (this.context.handler != null) {
87-
this.context.handler(conn)
88-
}
89-
90-
this.dispatchEvent(new CustomEvent<Connection>('connection', { detail: conn }))
91-
})
92-
.catch(async err => {
93-
log.error('inbound connection failed', err)
94-
95-
await attemptClose(maConn)
96-
})
97-
.catch(err => {
98-
log.error('closing inbound connection failed', err)
99-
})
91+
const conn = await this.context.upgrader.upgradeInbound(maConn)
92+
log('inbound connection %s upgraded', maConn.remoteAddr)
93+
this.connections.add(maConn)
94+
95+
socket.once('close', () => {
96+
this.connections.delete(maConn)
97+
})
98+
99+
if (this.context.handler != null) {
100+
this.context.handler(conn)
101+
}
102+
103+
this.dispatchEvent(new CustomEvent<Connection>('connection', { detail: conn }))
100104
} catch (err) {
101105
log.error('inbound connection failed', err)
106+
this.metrics?.listenerErrors.inc({ error: 'inbound_upgrade' })
102107

103-
attemptClose(maConn)
104-
.catch(err => {
105-
log.error('closing inbound connection failed', err)
106-
})
108+
attemptClose(maConn).catch(err => {
109+
log.error('closing inbound connection failed', err)
110+
this.metrics?.listenerErrors.inc({ error: 'inbound_closing_failed' })
111+
})
107112
}
108113
}
109114

‎src/metrics.ts

+34
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
/* eslint-disable etc/prefer-interface, @typescript-eslint/method-signature-style */
2+
3+
type LabelsGeneric = Record<string, string | undefined>
4+
type CollectFn<Labels extends LabelsGeneric> = (metric: Gauge<Labels>) => void
5+
6+
interface Gauge<Labels extends LabelsGeneric = never> {
7+
// Sorry for this mess, `prom-client` API choices are not great
8+
// If the function signature was `inc(value: number, labels?: Labels)`, this would be simpler
9+
inc(value?: number): void
10+
inc(labels: Labels, value?: number): void
11+
inc(arg1?: Labels | number, arg2?: number): void
12+
13+
dec(value?: number): void
14+
dec(labels: Labels, value?: number): void
15+
dec(arg1?: Labels | number, arg2?: number): void
16+
17+
set(value: number): void
18+
set(labels: Labels, value: number): void
19+
set(arg1?: Labels | number, arg2?: number): void
20+
21+
addCollect(collectFn: CollectFn<Labels>): void
22+
}
23+
24+
export enum ServerStatusMetric {
25+
stopped = 0,
26+
started = 1
27+
}
28+
29+
export interface TcpMetrics {
30+
serverStatus: Gauge
31+
connections: Gauge
32+
listenerErrors: Gauge<{error: string}>
33+
socketEvents: Gauge<{event: string}>
34+
}

‎src/socket-to-conn.ts

+7-2
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import errCode from 'err-code'
99
import type { Socket } from 'net'
1010
import type { Multiaddr } from '@multiformats/multiaddr'
1111
import type { MultiaddrConnection } from '@libp2p/interface-connection'
12+
import type { TcpMetrics } from './metrics.js'
1213

1314
const log = logger('libp2p:tcp:socket')
1415

@@ -19,14 +20,15 @@ interface ToConnectionOptions {
1920
signal?: AbortSignal
2021
socketInactivityTimeout?: number
2122
socketCloseTimeout?: number
23+
metrics?: TcpMetrics | null
2224
}
2325

2426
/**
2527
* Convert a socket into a MultiaddrConnection
2628
* https://github.com/libp2p/interface-transport#multiaddrconnection
2729
*/
28-
export const toMultiaddrConnection = (socket: Socket, options?: ToConnectionOptions) => {
29-
options = options ?? {}
30+
export const toMultiaddrConnection = (socket: Socket, options: ToConnectionOptions) => {
31+
const metrics = options.metrics
3032
const inactivityTimeout = options.socketInactivityTimeout ?? SOCKET_TIMEOUT
3133
const closeTimeout = options.socketCloseTimeout ?? CLOSE_TIMEOUT
3234

@@ -61,6 +63,7 @@ export const toMultiaddrConnection = (socket: Socket, options?: ToConnectionOpti
6163
// https://nodejs.org/dist/latest-v16.x/docs/api/net.html#socketsettimeouttimeout-callback
6264
socket.setTimeout(inactivityTimeout, () => {
6365
log('%s socket read timeout', lOptsStr)
66+
metrics?.socketEvents.inc({ event: 'timeout' })
6467

6568
// only destroy with an error if the remote has not sent the FIN message
6669
let err: Error | undefined
@@ -75,6 +78,7 @@ export const toMultiaddrConnection = (socket: Socket, options?: ToConnectionOpti
7578

7679
socket.once('close', () => {
7780
log('%s socket read timeout', lOptsStr)
81+
metrics?.socketEvents.inc({ event: 'close' })
7882

7983
// In instances where `close` was not explicitly called,
8084
// such as an iterable stream ending, ensure we have set the close
@@ -88,6 +92,7 @@ export const toMultiaddrConnection = (socket: Socket, options?: ToConnectionOpti
8892
// the remote sent a FIN packet which means no more data will be sent
8993
// https://nodejs.org/dist/latest-v16.x/docs/api/net.html#event-end
9094
log('socket ended', maConn.remoteAddr.toString())
95+
metrics?.socketEvents.inc({ event: 'end' })
9196
})
9297

9398
const maConn: MultiaddrConnection = {

0 commit comments

Comments
 (0)