From f93f350a29d43250a01392fe864246735bd43faa Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Tue, 26 May 2026 12:11:13 +0300 Subject: [PATCH] Add method label to upstream.rpc.conn metric (#863) --- .../dshackle/upstream/BasicHttpFactory.kt | 13 ++++++++----- .../io/emeraldpay/dshackle/upstream/HttpReader.kt | 2 +- .../dshackle/upstream/RequestMetrics.kt | 15 +++++++++++++-- .../upstream/ethereum/WsConnectionImpl.kt | 4 ++-- .../upstream/restclient/RestHttpReader.kt | 2 +- .../upstream/rpcclient/JsonRpcGrpcClient.kt | 2 +- .../upstream/rpcclient/JsonRpcHttpReader.kt | 2 +- 7 files changed, 27 insertions(+), 13 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt index fcb4c9f2..38abb0bb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt @@ -34,11 +34,14 @@ class BasicHttpFactory( Tag.of("chain", chain.chainCode), ) val metrics = RequestMetrics( - Timer.builder("upstream.rpc.conn") - .description("Request time through a HTTP JSON RPC connection") - .tags(metricsTags) - .publishPercentileHistogram() - .register(Metrics.globalRegistry), + { method -> + Timer.builder("upstream.rpc.conn") + .description("Request time through a HTTP JSON RPC connection") + .tags(metricsTags) + .tag("method", method ?: "unknown") + .publishPercentileHistogram() + .register(Metrics.globalRegistry) + }, Counter.builder("upstream.rpc.fail") .description("Number of failures of HTTP JSON RPC requests") .tags(metricsTags) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpReader.kt index b944ac3d..28e19d75 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpReader.kt @@ -101,7 +101,7 @@ abstract class HttpReader( open fun onStop() { if (metrics != null) { - Metrics.globalRegistry.remove(metrics.timer) + metrics.registeredTimers().forEach { Metrics.globalRegistry.remove(it) } Metrics.globalRegistry.remove(metrics.fails) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestMetrics.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestMetrics.kt index 75ef2f1f..7caca387 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestMetrics.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestMetrics.kt @@ -17,9 +17,20 @@ package io.emeraldpay.dshackle.upstream import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Timer +import java.util.concurrent.ConcurrentHashMap class RequestMetrics( - val timer: Timer, + private val timerFactory: (String?) -> Timer, val fails: Counter, val nettyMetricsEnabled: Boolean, -) +) { + private val timerCache = ConcurrentHashMap() + + constructor(timer: Timer, fails: Counter, nettyMetricsEnabled: Boolean) : + this({ _ -> timer }, fails, nettyMetricsEnabled) + + fun timer(method: String? = null): Timer = + timerCache.computeIfAbsent(method ?: "") { timerFactory(method) } + + fun registeredTimers(): Collection = timerCache.values +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt index f7bbfef7..a7ec18e9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt @@ -404,7 +404,7 @@ open class WsConnectionImpl( return Mono.from(onResponse.asMono()).or(failOnDisconnect) .doOnSubscribe { sendRpc(request) } .take(Defaults.timeout) - .doOnNext { requestMetrics?.timer?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS) } + .doOnNext { requestMetrics?.timer()?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS) } .doOnError { requestMetrics?.fails?.increment() } .map { it.copyWithId(ChainResponse.Id.from(originalId)) } .switchIfEmpty( @@ -424,7 +424,7 @@ open class WsConnectionImpl( it.close() Metrics.globalRegistry.remove(it) } - requestMetrics?.timer?.let { + requestMetrics?.registeredTimers()?.forEach { it.close() Metrics.globalRegistry.remove(it) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt index 0b4a9075..aa0b2da7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt @@ -59,7 +59,7 @@ class RestHttpReader( .flatMap(this::execute) .doOnNext { if (startTime.isStarted) { - metrics?.timer?.record(startTime.nanoTime, TimeUnit.NANOSECONDS) + metrics?.timer(key.method)?.record(startTime.nanoTime, TimeUnit.NANOSECONDS) } } .handle { it, sink -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt index 4d0528ec..641cf3dd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt @@ -114,7 +114,7 @@ class JsonRpcGrpcClient( } .doOnNext { if (timer.isStarted) { - metrics?.timer?.record(timer.getTime(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS) + metrics?.timer()?.record(timer.getTime(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt index 843a0be8..6fc78e52 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt @@ -92,7 +92,7 @@ class JsonRpcHttpReader( .flatMap(this@JsonRpcHttpReader::execute) .doOnNext { if (startTime.isStarted) { - metrics?.timer?.record(startTime.nanoTime, TimeUnit.NANOSECONDS) + metrics?.timer(key.method)?.record(startTime.nanoTime, TimeUnit.NANOSECONDS) } } .transform(asJsonRpcResponse(key))