Skip to content

Commit 086c5c5

Browse files
achingbrainachingbrain
committed
feat: add traceFunction call to metrics
Allow tracing method calls using a metrics implementation. ```js const libp2p = await createLibp2p() for await (const foo of libp2p.contentRouting.findProviders(cid, { context: libp2p.metrics?.createTraceContext() signal: AbortSignal.timeout(20_000) }) { //... } ``` Adds tracing support to `libp2p.contentRouting.*` and `libp2p.peerRouting.*` to start with, other methods can have it added when necessary.
1 parent 7caee9f commit 086c5c5

File tree

16 files changed

+286
-13
lines changed

16 files changed

+286
-13
lines changed

‎packages/interface/src/content-routing/index.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import type { AbortOptions, RoutingOptions } from '../index.js'
1+
import type { RoutingOptions } from '../index.js'
22
import type { PeerInfo } from '../peer-info/index.js'
33
import type { CID } from 'multiformats/cid'
44

@@ -50,7 +50,7 @@ export interface ContentRouting {
5050
* provide content corresponding to the passed CID, call this function to no
5151
* longer remind them.
5252
*/
53-
cancelReprovide (key: CID, options?: AbortOptions): Promise<void>
53+
cancelReprovide (key: CID, options?: RoutingOptions): Promise<void>
5454

5555
/**
5656
* Find the providers of the passed CID.

‎packages/interface/src/index.ts‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -753,13 +753,22 @@ export interface LoggerOptions {
753753
log: Logger
754754
}
755755

756+
/**
757+
* An object that includes a context object that is passed onwards.
758+
*
759+
* This is used by metrics method tracing to link function calls together.
760+
*/
761+
export interface ContextOptions {
762+
context?: any
763+
}
764+
756765
/**
757766
* When a routing operation involves reading values, these options allow
758767
* controlling where the values are read from. By default libp2p will check
759768
* local caches but may not use the network if a valid local value is found,
760769
* these options allow tuning that behaviour.
761770
*/
762-
export interface RoutingOptions extends AbortOptions, ProgressOptions {
771+
export interface RoutingOptions extends AbortOptions, ProgressOptions, ContextOptions {
763772
/**
764773
* Pass `false` to not use the network
765774
*

‎packages/interface/src/metrics/index.ts‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -488,4 +488,61 @@ export interface Metrics {
488488
* method on the returned summary group object
489489
*/
490490
registerSummaryGroup: ((name: string, options?: SummaryOptions) => SummaryGroup) & ((name: string, options: CalculatedSummaryOptions<Record<string, number>>) => void)
491+
492+
/**
493+
* Wrap a function for tracing purposes.
494+
*
495+
* All functions wrapped like this should accept a final optional options arg.
496+
*
497+
* In order to pass an execution context along to create a multi-layered
498+
* trace, the index of the options arg must be specified.
499+
*/
500+
traceFunction <F extends (...args: any[]) => AsyncIterator<any>> (name: string, fn: F, options?: TraceGeneratorFunctionOptions<Parameters<F>, ReturnType<F>, YieldType<ReturnType<F>>>): F
501+
traceFunction <F extends (...args: any[]) => Iterator<any>> (name: string, fn: F, options?: TraceGeneratorFunctionOptions<Parameters<F>, ReturnType<F>, YieldType<ReturnType<F>>>): F
502+
traceFunction <F extends (...args: any[]) => any = (...args: any[]) => any> (name: string, fn: F, options?: TraceFunctionOptions<Parameters<F>, ReturnType<F>>): F
503+
504+
/**
505+
* Creates a tracing context that can be used to trace a method call
506+
*/
507+
createTraceContext(): any
508+
}
509+
510+
/**
511+
* Infer the yielded type of an (async)iterable
512+
*/
513+
type YieldType<T extends AsyncIterator<any> | Iterator<any>> = T extends AsyncIterator<infer Y> ? Y : T extends Iterator<infer Y, any, any> ? Y : never
514+
515+
export type TraceAttributes = Record<string, number | string | boolean | number[] | string[] | boolean[]>
516+
517+
export interface TraceFunctionOptions<A, B> {
518+
/**
519+
* To construct a trace that spans multiple method invocations, it's necessary
520+
* to pass the trace context onwards as part of the options object.
521+
*
522+
* Specify the index of the options object in the args array here.
523+
*
524+
* @default 0
525+
*/
526+
optionsIndex?: number
527+
528+
/**
529+
* Set attributes on the trace by modifying the passed attributes object.
530+
*/
531+
getAttributesFromArgs?(args: A, attributes: TraceAttributes): TraceAttributes
532+
533+
/**
534+
* Set attributes on the trace by modifying the passed attributes object. The
535+
* object will have previously been passed to `appendAttributesFromArgs`
536+
* and/or `appendAttributesFromYieldedValue` (if defined)
537+
*/
538+
getAttributesFromReturnValue?(value: B, attributes: TraceAttributes): TraceAttributes
539+
}
540+
541+
export interface TraceGeneratorFunctionOptions<A, B, C = any> extends TraceFunctionOptions<A, B> {
542+
/**
543+
* Set attributes on the trace by modifying the passed attributes object. The
544+
* object will have previously been passed to `appendAttributesFromArgs` (if
545+
* defined)
546+
*/
547+
getAttributesFromYieldedValue? (value: C, attributes: TraceAttributes, index: number): TraceAttributes
491548
}

‎packages/kad-dht/src/content-fetching/index.ts‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,13 @@ export class ContentFetching {
5555
this.peerRouting = peerRouting
5656
this.queryManager = queryManager
5757
this.network = network
58+
59+
this.get = components.metrics?.traceFunction('libp2p.kadDHT.get', this.get.bind(this), {
60+
optionsIndex: 1
61+
}) ?? this.get
62+
this.put = components.metrics?.traceFunction('libp2p.kadDHT.put', this.put.bind(this), {
63+
optionsIndex: 2
64+
}) ?? this.put
5865
}
5966

6067
/**
@@ -145,7 +152,10 @@ export class ContentFetching {
145152

146153
// put record to the closest peers
147154
yield * pipe(
148-
this.peerRouting.getClosestPeers(key, { signal: options.signal }),
155+
this.peerRouting.getClosestPeers(key, {
156+
...options,
157+
signal: options.signal
158+
}),
149159
(source) => map(source, (event) => {
150160
return async () => {
151161
if (event.name !== 'FINAL_PEER') {
@@ -252,7 +262,10 @@ export class ContentFetching {
252262
const self = this // eslint-disable-line @typescript-eslint/no-this-alias
253263

254264
const getValueQuery: QueryFunc = async function * ({ peer, signal }) {
255-
for await (const event of self.peerRouting.getValueOrPeers(peer, key, { signal })) {
265+
for await (const event of self.peerRouting.getValueOrPeers(peer, key, {
266+
...options,
267+
signal
268+
})) {
256269
yield event
257270

258271
if (event.name === 'PEER_RESPONSE' && (event.record != null)) {

‎packages/kad-dht/src/content-routing/index.ts‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,29 @@ export class ContentRouting {
5050
this.queryManager = queryManager
5151
this.routingTable = routingTable
5252
this.providers = providers
53+
54+
this.findProviders = components.metrics?.traceFunction('libp2p.kadDHT.findProviders', this.findProviders.bind(this), {
55+
optionsIndex: 1,
56+
getAttributesFromYieldedValue: (event, attrs: { providers?: string[] }) => {
57+
if (event.name === 'PROVIDER') {
58+
attrs.providers ??= []
59+
attrs.providers.push(...event.providers.map(info => info.id.toString()))
60+
}
61+
62+
return attrs
63+
}
64+
}) ?? this.findProviders
65+
this.provide = components.metrics?.traceFunction('libp2p.kadDHT.provide', this.provide.bind(this), {
66+
optionsIndex: 1,
67+
getAttributesFromYieldedValue: (event, attrs: { providers?: string[] }) => {
68+
if (event.name === 'PEER_RESPONSE' && event.messageName === 'ADD_PROVIDER') {
69+
attrs.providers ??= []
70+
attrs.providers.push(event.from.toString())
71+
}
72+
73+
return attrs
74+
}
75+
}) ?? this.provide
5376
}
5477

5578
/**

‎packages/kad-dht/src/network.ts‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,61 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
5757
operations: components.metrics?.registerCounterGroup(`${init.metricsPrefix}_outbound_rpc_requests_total`),
5858
errors: components.metrics?.registerCounterGroup(`${init.metricsPrefix}_outbound_rpc_errors_total`)
5959
}
60+
61+
this.sendRequest = components.metrics?.traceFunction('libp2p.kadDHT.sendRequest', this.sendRequest.bind(this), {
62+
optionsIndex: 2,
63+
getAttributesFromArgs ([to, message], attrs) {
64+
return {
65+
...attrs,
66+
to: to.toString(),
67+
'message type': `${message.type}`
68+
}
69+
},
70+
getAttributesFromYieldedValue: (event, attrs) => {
71+
if (event.name === 'PEER_RESPONSE') {
72+
if (event.providers.length > 0) {
73+
event.providers.forEach((value, index) => {
74+
attrs[`providers-${index}`] = value.id.toString()
75+
})
76+
}
77+
78+
if (event.closer.length > 0) {
79+
event.closer.forEach((value, index) => {
80+
attrs[`closer-${index}`] = value.id.toString()
81+
})
82+
}
83+
}
84+
85+
return attrs
86+
}
87+
}) ?? this.sendRequest
88+
this.sendMessage = components.metrics?.traceFunction('libp2p.kadDHT.sendMessage', this.sendMessage.bind(this), {
89+
optionsIndex: 2,
90+
getAttributesFromArgs ([to, message], attrs) {
91+
return {
92+
...attrs,
93+
to: to.toString(),
94+
'message type': `${message.type}`
95+
}
96+
},
97+
getAttributesFromYieldedValue: (event, attrs) => {
98+
if (event.name === 'PEER_RESPONSE') {
99+
if (event.providers.length > 0) {
100+
event.providers.forEach((value, index) => {
101+
attrs[`providers-${index}`] = value.id.toString()
102+
})
103+
}
104+
105+
if (event.closer.length > 0) {
106+
event.closer.forEach((value, index) => {
107+
attrs[`closer-${index}`] = value.id.toString()
108+
})
109+
}
110+
}
111+
112+
return attrs
113+
}
114+
}) ?? this.sendMessage
60115
}
61116

62117
/**

‎packages/kad-dht/src/peer-routing/index.ts‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,12 +22,13 @@ import type { Network } from '../network.js'
2222
import type { QueryManager, QueryOptions } from '../query/manager.js'
2323
import type { QueryFunc } from '../query/types.js'
2424
import type { RoutingTable } from '../routing-table/index.js'
25-
import type { ComponentLogger, Logger, PeerId, PeerInfo, PeerStore, RoutingOptions } from '@libp2p/interface'
25+
import type { ComponentLogger, Logger, Metrics, PeerId, PeerInfo, PeerStore, RoutingOptions } from '@libp2p/interface'
2626

2727
export interface PeerRoutingComponents {
2828
peerId: PeerId
2929
peerStore: PeerStore
3030
logger: ComponentLogger
31+
metrics?: Metrics
3132
}
3233

3334
export interface PeerRoutingInit {
@@ -55,6 +56,13 @@ export class PeerRouting {
5556
this.peerStore = components.peerStore
5657
this.peerId = components.peerId
5758
this.log = components.logger.forComponent(`${init.logPrefix}:peer-routing`)
59+
60+
this.findPeer = components.metrics?.traceFunction('libp2p.kadDHT.findPeer', this.findPeer.bind(this), {
61+
optionsIndex: 1
62+
}) ?? this.findPeer
63+
this.getClosestPeers = components.metrics?.traceFunction('libp2p.kadDHT.getClosestPeers', this.getClosestPeers.bind(this), {
64+
optionsIndex: 1
65+
}) ?? this.getClosestPeers
5866
}
5967

6068
/**

‎packages/kad-dht/src/query-self.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,9 +10,8 @@ import { timeOperationMethod } from './utils.js'
1010
import type { OperationMetrics } from './kad-dht.js'
1111
import type { PeerRouting } from './peer-routing/index.js'
1212
import type { RoutingTable } from './routing-table/index.js'
13-
import type { ComponentLogger, Logger, PeerId, Startable } from '@libp2p/interface'
13+
import type { ComponentLogger, Logger, Metrics, PeerId, Startable } from '@libp2p/interface'
1414
import type { DeferredPromise } from 'p-defer'
15-
1615
export interface QuerySelfInit {
1716
logPrefix: string
1817
peerRouting: PeerRouting
@@ -28,6 +27,7 @@ export interface QuerySelfInit {
2827
export interface QuerySelfComponents {
2928
peerId: PeerId
3029
logger: ComponentLogger
30+
metrics?: Metrics
3131
}
3232

3333
/**

‎packages/kad-dht/src/query/manager.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -171,6 +171,7 @@ export class QueryManager implements Startable {
171171
// Create query paths from the starting peers
172172
const paths = peersToQuery.map((peer, index) => {
173173
return queryPath({
174+
...options,
174175
key,
175176
startingPeer: peer,
176177
ourPeerId: this.peerId,

‎packages/kad-dht/src/query/query-path.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,7 @@ export async function * queryPath (options: QueryPathOptions): AsyncGenerator<Qu
121121

122122
try {
123123
for await (const event of query({
124+
...options,
124125
key,
125126
peer,
126127
signal: compoundSignal,

0 commit comments

Comments
 (0)