From 981a5dc589e8a43925b3ffcd48ed5de50de97310 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 10 Feb 2021 14:36:44 -0500 Subject: [PATCH] solution: metrics for upstream calls --- .../dshackle/startup/ConfiguredUpstreams.kt | 23 +++++++++++ .../dshackle/upstream/grpc/GrpcUpstreams.kt | 36 +++++++++++++--- .../upstream/rpcclient/JsonRpcGrpcClient.kt | 41 +++++++++++++------ .../upstream/rpcclient/JsonRpcHttpClient.kt | 16 ++++++++ .../dshackle/upstream/rpcclient/RpcMetrics.kt | 24 +++++++++++ .../dshackle/test/TestingCommons.groovy | 4 ++ .../grpc/EthereumGrpcUpstreamSpec.groovy | 13 ++++-- .../rpcclient/JsonRpcHttpClientSpec.groovy | 12 ++++-- 8 files changed, 144 insertions(+), 25 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/RpcMetrics.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 3802150a..95fcb1cd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/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 ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index a57da6c0..6b67d70c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/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 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 a2fc023e..3383e854 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt +++ b/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 { - 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 { - private val parser = JsonRpcParser() - override fun read(key: JsonRpcRequest): Mono { + 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()) - .single() - .flatMap { resp -> - if (resp.succeed) { - val bytes = resp.payload.toByteArray() - Mono.just(JsonRpcResponse(bytes, null)) - } else { - Mono.error(RpcException(RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, resp.errorMessage)) + return Mono.just(key) + .doOnNext { + startTime = System.nanoTime() + }.flatMap { + stub.nativeCall(req.build()) + .single() + .flatMap { resp -> + 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) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt index 020672d4..d0e88fdb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt +++ b/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 { @@ -98,9 +102,19 @@ class JsonRpcHttpClient( } override fun read(key: JsonRpcRequest): Mono { + 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) } } + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/RpcMetrics.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/RpcMetrics.kt new file mode 100644 index 00000000..a21e40d1 --- /dev/null +++ b/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 +) \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index b4fa9663..d9aaad25 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/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() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy index aa26b997..99896483 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy +++ b/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)) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy index fd9c4c91..5cefeda7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy +++ b/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()