solution: metrics for upstream calls

src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt

@@ -32,10 +32,15 @@ import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcHttpClient
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
 import io.emeraldpay.grpc.Chain
+import io.micrometer.core.instrument.Counter
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Repository
+import io.micrometer.core.instrument.Metrics
+import io.micrometer.core.instrument.Tag
+import io.micrometer.core.instrument.Timer
 import java.net.URI
 import java.util.*
 import java.util.concurrent.atomic.AtomicInteger
@@ -235,9 +240,27 @@ open class ConfiguredUpstreams(
                     fileResolver.resolve(ca).readBytes()
                 }
             }
+            val metricsTags = listOf(
+                    // "unknown" is not supposed to happen
+                    Tag.of("upstream", config.id ?: "unknown"),
+                    // UNSPECIFIED shouldn't happen too
+                    Tag.of("chain", (chainNames[config.chain ?: ""] ?: Chain.UNSPECIFIED ).chainCode)
+            )
+            val metrics = RpcMetrics(
+                    Timer.builder("upstream.rpc.conn")
+                            .description("Request time through a HTTP JSON RPC connection")
+                            .tags(metricsTags)
+                            .publishPercentileHistogram()
+                            .register(Metrics.globalRegistry),
+                    Counter.builder("upstream.rpc.err")
+                            .description("Errors received on request through HTTP JSON RPC connection")
+                            .tags(metricsTags)
+                            .register(Metrics.globalRegistry)
+            )
             urls.add(endpoint.url)
             JsonRpcHttpClient(
                     endpoint.url.toString(),
+                    metrics,
                     conn.rpc?.basicAuth,
                     tls
             )

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt

@@ -26,9 +26,14 @@ import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.startup.UpstreamChange
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
+import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
 import io.emeraldpay.grpc.Chain
 import io.grpc.ManagedChannelBuilder
 import io.grpc.netty.NettyChannelBuilder
+import io.micrometer.core.instrument.Counter
+import io.micrometer.core.instrument.Metrics
+import io.micrometer.core.instrument.Tag
+import io.micrometer.core.instrument.Timer
 import io.netty.handler.ssl.*
 import org.apache.commons.lang3.StringUtils
 import org.apache.commons.lang3.exception.ExceptionUtils
@@ -157,21 +162,40 @@ class GrpcUpstreams(
     }
 
     fun getOrCreate(chain: Chain): UpstreamChange {
+        val metricsTags = listOf(
+                // "unknown" is not supposed to happen
+                Tag.of("upstream", id ?: "unknown"),
+                // UNSPECIFIED shouldn't happen too
+                Tag.of("chain", chain.chainCode)
+        )
+
+        val metrics = RpcMetrics(
+                Timer.builder("upstream.grpc.conn")
+                        .description("Request time through a gRPC connection")
+                        .tags(metricsTags)
+                        .publishPercentileHistogram()
+                        .register(Metrics.globalRegistry),
+                Counter.builder("upstream.grpc.err")
+                        .description("Errors received on request through gRPC connection")
+                        .tags(metricsTags)
+                        .register(Metrics.globalRegistry)
+        )
+
         val blockchainType = BlockchainType.fromBlockchain(chain)
         if (blockchainType == BlockchainType.ETHEREUM) {
-            return getOrCreateEthereum(chain)
+            return getOrCreateEthereum(chain, metrics)
         } else if (blockchainType == BlockchainType.BITCOIN) {
-            return getOrCreateBitcoin(chain)
+            return getOrCreateBitcoin(chain, metrics)
         } else {
             throw IllegalArgumentException("Unsupported blockchain: $chain")
         }
     }
 
-    fun getOrCreateEthereum(chain: Chain): UpstreamChange {
+    fun getOrCreateEthereum(chain: Chain, metrics: RpcMetrics): UpstreamChange {
         lock.withLock {
             val current = known[chain]
             return if (current == null) {
-                val rpcClient = JsonRpcGrpcClient(client!!, chain)
+                val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
                 val created = EthereumGrpcUpstream(id, chain, client!!, rpcClient)
                 created.timeout = this.timeout
                 known[chain] = created
@@ -183,11 +207,11 @@ class GrpcUpstreams(
         }
     }
 
-    fun getOrCreateBitcoin(chain: Chain): UpstreamChange {
+    fun getOrCreateBitcoin(chain: Chain, metrics: RpcMetrics): UpstreamChange {
         lock.withLock {
             val current = known[chain]
             return if (current == null) {
-                val rpcClient = JsonRpcGrpcClient(client!!, chain)
+                val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
                 val created = BitcoinGrpcUpstream(id, chain, client!!, rpcClient)
                 created.timeout = this.timeout
                 known[chain] = created

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt

@@ -27,11 +27,14 @@ import io.grpc.Channel
 import io.infinitape.etherjar.rpc.RpcException
 import io.infinitape.etherjar.rpc.RpcResponseError
 import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import java.util.concurrent.TimeUnit
 
 class JsonRpcGrpcClient(
         private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
-        private val chain: Chain
+        private val chain: Chain,
+        private val metrics: RpcMetrics
 ) {
 
     companion object {
@@ -39,18 +42,18 @@ class JsonRpcGrpcClient(
     }
 
     fun forSelector(matcher: Selector.Matcher): Reader<JsonRpcRequest, JsonRpcResponse> {
-        return Executor(stub, chain, matcher)
+        return Executor(stub, chain, matcher, metrics)
     }
 
     class Executor(
             private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
             private val chain: Chain,
-            private val matcher: Selector.Matcher
+            private val matcher: Selector.Matcher,
+            private val metrics: RpcMetrics
     ) : Reader<JsonRpcRequest, JsonRpcResponse> {
 
-        private val parser = JsonRpcParser()
-
         override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
+            var startTime: Long = 0
             val req = BlockchainOuterClass.NativeCallRequest.newBuilder()
                     .setChainValue(chain.id)
 
@@ -68,14 +71,26 @@ class JsonRpcGrpcClient(
                         req.addItems(it)
                     }
 
-            return stub.nativeCall(req.build())
+            return Mono.just(key)
-                    .single()
+                    .doOnNext {
-                    .flatMap { resp ->
+                        startTime = System.nanoTime()
-                        if (resp.succeed) {
+                    }.flatMap {
-                            val bytes = resp.payload.toByteArray()
+                        stub.nativeCall(req.build())
-                            Mono.just(JsonRpcResponse(bytes, null))
+                                .single()
-                        } else {
+                                .flatMap { resp ->
-                            Mono.error(RpcException(RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, resp.errorMessage))
+                                    if (resp.succeed) {
+                                        val bytes = resp.payload.toByteArray()
+                                        Mono.just(JsonRpcResponse(bytes, null))
+                                    } else {
+                                        metrics.errors.increment()
+                                        Mono.error(RpcException(RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, resp.errorMessage))
+                                    }
+                                }
+                    }
+                    .doOnNext {
+                        if (startTime > 0) {
+                            val now = System.nanoTime()
+                            metrics.timer.record(now - startTime, TimeUnit.NANOSECONDS)
                         }
                     }
         }

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt

@@ -19,6 +19,8 @@ import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.reader.Reader
 import io.infinitape.etherjar.rpc.RpcException
 import io.infinitape.etherjar.rpc.RpcResponseError
+import io.micrometer.core.instrument.Counter
+import io.micrometer.core.instrument.Timer
 import io.netty.buffer.Unpooled
 import io.netty.handler.codec.http.HttpHeaderNames
 import io.netty.handler.codec.http.HttpHeaders
@@ -31,6 +33,7 @@ import java.security.KeyStore
 import java.security.cert.CertificateFactory
 import java.security.cert.X509Certificate
 import java.util.*
+import java.util.concurrent.TimeUnit
 import java.util.function.Consumer
 
 /**
@@ -38,6 +41,7 @@ import java.util.function.Consumer
  */
 class JsonRpcHttpClient(
         private val target: String,
+        private val metrics: RpcMetrics,
         basicAuth: AuthConfig.ClientBasicAuth? = null,
         tlsCAAuth: ByteArray? = null
 ) : Reader<JsonRpcRequest, JsonRpcResponse> {
@@ -98,9 +102,19 @@ class JsonRpcHttpClient(
     }
 
     override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
+        var startTime: Long = 0
         return Mono.just(key)
                 .map(JsonRpcRequest::toJson)
+                .doOnNext {
+                    startTime = System.nanoTime()
+                }
                 .flatMap(this@JsonRpcHttpClient::execute)
+                .doOnNext {
+                    if (startTime > 0) {
+                        val now = System.nanoTime()
+                        metrics.timer.record(now - startTime, TimeUnit.NANOSECONDS)
+                    }
+                }
                 .map(parser::parse)
                 .onErrorResume { t ->
                     val err = when (t) {
@@ -108,7 +122,9 @@ class JsonRpcHttpClient(
                         is JsonRpcException -> JsonRpcResponse.error(t.error, JsonRpcResponse.NumberId(1))
                         else -> JsonRpcResponse.error(1, t.message ?: t.javaClass.name)
                     }
+                    metrics.errors.increment()
                     Mono.just(err)
                 }
     }
+
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/RpcMetrics.kt

@@ -0,0 +1,24 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.rpcclient
+
+import io.micrometer.core.instrument.Counter
+import io.micrometer.core.instrument.Timer
+
+class RpcMetrics(
+        val timer: Timer,
+        val errors: Counter
+)

src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy

@@ -34,6 +34,8 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.rpc.json.BlockJson
+import io.micrometer.core.instrument.MeterRegistry
+import io.micrometer.core.instrument.logging.LoggingMeterRegistry
 import org.apache.commons.lang3.StringUtils
 
 import java.time.Duration
@@ -113,4 +115,6 @@ class TestingCommons {
         Instant.ofEpochSecond(1577876400)
                 .plusSeconds(x * stepSeconds)
     }
+
+    static MeterRegistry meterRegistry = new LoggingMeterRegistry()
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy

@@ -27,10 +27,13 @@ import io.emeraldpay.dshackle.test.MockGrpcServer
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
+import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
 import io.emeraldpay.grpc.Chain
 import io.grpc.stub.StreamObserver
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.rpc.json.BlockJson
+import io.micrometer.core.instrument.Counter
+import io.micrometer.core.instrument.Timer
 import spock.lang.Specification
 
 import java.time.Duration
@@ -41,6 +44,10 @@ class EthereumGrpcUpstreamSpec extends Specification {
 
     MockGrpcServer mockServer = new MockGrpcServer()
     ObjectMapper objectMapper = Global.objectMapper
+    RpcMetrics metrics = new RpcMetrics(
+            Timer.builder("test1").register(TestingCommons.meterRegistry),
+            Counter.builder("test2").register(TestingCommons.meterRegistry)
+    )
 
     def "Subscribe to head"() {
         setup:
@@ -73,7 +80,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 )
             }
         })
-        def upstream = new EthereumGrpcUpstream("test", chain, client, new JsonRpcGrpcClient(client, chain))
+        def upstream = new EthereumGrpcUpstream("test", chain, client, new JsonRpcGrpcClient(client, chain, metrics))
         upstream.setLag(0)
         upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
                 .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
@@ -131,7 +138,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 )
             }
         })
-        def upstream = new EthereumGrpcUpstream("test", Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM))
+        def upstream = new EthereumGrpcUpstream("test", Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics))
         upstream.setLag(0)
         upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
                 .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
@@ -193,7 +200,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 finished.complete(true)
             }
         })
-        def upstream = new EthereumGrpcUpstream("test", chain, client, new JsonRpcGrpcClient(client, chain))
+        def upstream = new EthereumGrpcUpstream("test", chain, client, new JsonRpcGrpcClient(client, chain, metrics))
         upstream.setLag(0)
         upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
                 .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))

src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy

@@ -19,6 +19,8 @@ import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.infinitape.etherjar.rpc.RpcException
 import io.infinitape.etherjar.rpc.RpcResponseError
+import io.micrometer.core.instrument.Counter
+import io.micrometer.core.instrument.Timer
 import org.mockserver.integration.ClientAndServer
 import org.mockserver.model.HttpRequest
 import org.mockserver.model.HttpResponse
@@ -33,6 +35,10 @@ class JsonRpcHttpClientSpec extends Specification {
 
     ClientAndServer mockServer
     int port = 19332
+    RpcMetrics metrics = new RpcMetrics(
+            Timer.builder("test1").register(TestingCommons.meterRegistry),
+            Counter.builder("test2").register(TestingCommons.meterRegistry)
+    )
 
     def setup() {
         port = SocketUtils.findAvailableTcpPort(19332)
@@ -45,7 +51,7 @@ class JsonRpcHttpClientSpec extends Specification {
 
     def "Make a request"() {
         setup:
-        JsonRpcHttpClient client = new JsonRpcHttpClient("localhost:${port}", null, null)
+        JsonRpcHttpClient client = new JsonRpcHttpClient("localhost:${port}", metrics,null, null)
         def resp = '{' +
                 '  "jsonrpc": "2.0",' +
                 '  "result": "0x98de45",' +
@@ -67,7 +73,7 @@ class JsonRpcHttpClientSpec extends Specification {
     def "Make request with basic auth"() {
         setup:
         def auth = new AuthConfig.ClientBasicAuth("user", "passwd")
-        def client = new JsonRpcHttpClient("localhost:${port}", auth, null)
+        def client = new JsonRpcHttpClient("localhost:${port}", metrics, auth, null)
 
         mockServer.when(
                 HttpRequest.request()
@@ -95,7 +101,7 @@ class JsonRpcHttpClientSpec extends Specification {
 
     def "Produces RPC Exception on error status code"() {
         setup:
-        def client = new JsonRpcHttpClient("localhost:${port}", null, null)
+        def client = new JsonRpcHttpClient("localhost:${port}", metrics, null, null)
 
         mockServer.when(
                 HttpRequest.request()