Add method label to upstream.rpc.conn metric (#863)

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")
+            { method ->
-                .description("Request time through a HTTP JSON RPC connection")
+                Timer.builder("upstream.rpc.conn")
-                .tags(metricsTags)
+                    .description("Request time through a HTTP JSON RPC connection")
-                .publishPercentileHistogram()
+                    .tags(metricsTags)
-                .register(Metrics.globalRegistry),
+                    .tag("method", method ?: "unknown")
+                    .publishPercentileHistogram()
+                    .register(Metrics.globalRegistry)
+            },
             Counter.builder("upstream.rpc.fail")
                 .description("Number of failures of HTTP JSON RPC requests")
                 .tags(metricsTags)

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)
         }
     }

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<String, Timer>()
+
+    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<Timer> = timerCache.values
+}

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)
         }

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 ->

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)
                     }
                 }
         }

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))