From 2b8603df980a7a8707352178c2daf3371a1d3805 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 10 Feb 2022 23:21:10 -0500 Subject: [PATCH] problem: confusing metric name and no metric to monitor RPC error responses rel: #150 --- .../emeraldpay/dshackle/proxy/BaseHandler.kt | 5 ++- .../emeraldpay/dshackle/proxy/ProxyServer.kt | 15 ++++++++ .../emeraldpay/dshackle/rpc/BlockchainRpc.kt | 38 ++++++++++++------- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 1 + .../dshackle/startup/ConfiguredUpstreams.kt | 4 +- .../upstream/ethereum/EthereumWsUpstream.kt | 4 +- .../upstream/ethereum/WsConnection.kt | 2 +- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 6 +-- .../upstream/rpcclient/JsonRpcGrpcClient.kt | 2 +- .../upstream/rpcclient/JsonRpcHttpClient.kt | 2 +- .../dshackle/upstream/rpcclient/RpcMetrics.kt | 2 +- 11 files changed, 55 insertions(+), 26 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt index 8ca2fdbd..a37b0648 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt @@ -73,13 +73,16 @@ abstract class BaseHandler( metricById(it.id)?.let { metrics -> metrics.requestMetric.increment() metrics.callMetric.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS) + if (it.isError()) { + metrics.errorMetric.increment() + } } handler.onResponse(it) } .doOnError { // when error happened the whole flux is stopped and no result is produced, so we should mark all the requests as failed items.forEach { item -> - requestMetrics.get(chain, item.method).errorMetric.increment() + requestMetrics.get(chain, item.method).failMetric.increment() } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt index f0df599d..96198e58 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -173,29 +173,44 @@ class ProxyServer( interface RequestMetrics { val callMetric: Timer val errorMetric: Counter + val failMetric: Counter val requestMetric: Counter } class RequestMetricsBasic(chain: Chain) : RequestMetrics { override val callMetric = Timer.builder("request.jsonrpc.call") + .description("Time to process a call") .tags("chain", chain.chainCode) .register(Metrics.globalRegistry) override val errorMetric = Counter.builder("request.jsonrpc.err") + .description("Number of requests ended with an error response") + .tags("chain", chain.chainCode) + .register(Metrics.globalRegistry) + override val failMetric = Counter.builder("request.jsonrpc.fail") + .description("Number of requests failed to process") .tags("chain", chain.chainCode) .register(Metrics.globalRegistry) override val requestMetric = Counter.builder("request.jsonrpc.request.total") + .description("Number of requests") .tags("chain", chain.chainCode) .register(Metrics.globalRegistry) } class RequestMetricsWithMethod(chain: Chain, method: String) : RequestMetrics { override val callMetric = Timer.builder("request.jsonrpc.call") + .description("Time to process a call") .tags("chain", chain.chainCode, "method", method) .register(Metrics.globalRegistry) override val errorMetric = Counter.builder("request.jsonrpc.err") + .description("Number of requests ended with an error response") + .tags("chain", chain.chainCode, "method", method) + .register(Metrics.globalRegistry) + override val failMetric = Counter.builder("request.jsonrpc.fail") + .description("Number of requests failed to process") .tags("chain", chain.chainCode, "method", method) .register(Metrics.globalRegistry) override val requestMetric = Counter.builder("request.jsonrpc.request.total") + .description("Number of requests") .tags("chain", chain.chainCode, "method", method) .register(Metrics.globalRegistry) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 69dad367..7772dac0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -57,7 +57,8 @@ class BlockchainRpc( .tag("type", "subscribeStatus") .tag("chain", "NA") .register(Metrics.globalRegistry) - private val errorMetric = Counter.builder("request.grpc.err") + private val failMetric = Counter.builder("request.grpc.fail") + .description("Number of requests failed to process") .register(Metrics.globalRegistry) private val chainMetrics = ChainValue { chain -> RequestMetrics(chain) } @@ -71,9 +72,14 @@ class BlockchainRpc( metrics!!.nativeCallMetric.increment() startTime = System.currentTimeMillis() } - ).doOnNext { - metrics?.nativeCallRespMetric?.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS) - }.doOnError { errorMetric.increment() } + ).doOnNext { reply -> + metrics?.let { m -> + m.nativeCallRespMetric?.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS) + if (!reply.succeed) { + m.nativeCallErrRespMetric.increment() + } + } + }.doOnError { failMetric.increment() } } override fun nativeSubscribe(request: Mono): Flux { @@ -86,14 +92,14 @@ class BlockchainRpc( } ).doOnNext { metrics?.nativeSubscribeRespMetric?.increment() - }.doOnError { errorMetric.increment() } + }.doOnError { failMetric.increment() } } override fun subscribeHead(request: Mono): Flux { return streamHead.add( request .doOnNext { chainMetrics.get(it.type).subscribeHeadMetric.increment() } - ).doOnError { errorMetric.increment() } + ).doOnError { failMetric.increment() } } override fun subscribeTxStatus(requestMono: Mono): Flux { @@ -105,11 +111,11 @@ class BlockchainRpc( trackTx.find { it.isSupported(chain) }?.let { track -> track.subscribe(request) .doOnNext { metrics.subscribeHeadRespMetric.increment() } - .doOnError { errorMetric.increment() } + .doOnError { failMetric.increment() } } ?: Flux.error(SilentException.UnsupportedBlockchain(chain)) } catch (t: Throwable) { log.error("Internal error during Tx Subscription", t) - errorMetric.increment() + failMetric.increment() Flux.error(IllegalStateException("Internal Error")) } } @@ -125,14 +131,14 @@ class BlockchainRpc( trackAddress.find { it.isSupported(chain, asset) }?.let { track -> track.subscribe(request) .doOnNext { metrics.subscribeBalanceRespMetric.increment() } - .doOnError { errorMetric.increment() } + .doOnError { failMetric.increment() } } ?: Flux.error(SilentException.UnsupportedBlockchain(chain)) .doOnSubscribe { log.error("Balance for $chain:$asset is not supported") } } catch (t: Throwable) { log.error("Internal error during Balance Subscription", t) - errorMetric.increment() + failMetric.increment() Flux.error(IllegalStateException("Internal Error")) } } @@ -160,7 +166,7 @@ class BlockchainRpc( } } catch (t: Throwable) { log.error("Internal error during Balance Request", t) - errorMetric.increment() + failMetric.increment() Flux.error(IllegalStateException("Internal Error")) } } @@ -182,20 +188,20 @@ class BlockchainRpc( } .doOnError { t -> log.error("Internal error during Fee Estimation", t) - errorMetric.increment() + failMetric.increment() } } override fun describe(request: Mono): Mono { describeMetric.increment() return describe.describe(request) - .doOnError { errorMetric.increment() } + .doOnError { failMetric.increment() } } override fun subscribeStatus(request: Mono): Flux { subscribeStatusMetric.increment() return subscribeStatus.subscribeStatus(request) - .doOnError { errorMetric.increment() } + .doOnError { failMetric.increment() } } class RequestMetrics(chain: Chain) { @@ -208,6 +214,10 @@ class BlockchainRpc( .tag("chain", chain.chainCode) .publishPercentileHistogram() .register(Metrics.globalRegistry) + val nativeCallErrRespMetric = Counter.builder("request.grpc.response.err") + .tag("type", "nativeCall") + .tag("chain", chain.chainCode) + .register(Metrics.globalRegistry) val nativeSubscribeMetric = Counter.builder("request.grpc.request") .tag("type", "nativeSubscribe") .tag("chain", chain.chainCode) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 9bde8347..b1b26284 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -222,6 +222,7 @@ open class NativeCall( } fun executeOnRemote(ctx: ValidCallContext): Mono { + // check if method is allowed to be executed at all if (!ctx.upstream.getMethods().isCallable(ctx.payload.method)) { return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method")) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 7e7ec7fa..c02f2340 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -271,8 +271,8 @@ open class ConfiguredUpstreams( .tags(metricsTags) .publishPercentileHistogram() .register(Metrics.globalRegistry), - Counter.builder("upstream.rpc.err") - .description("Errors received on request through HTTP JSON RPC connection") + Counter.builder("upstream.rpc.fail") + .description("Number of failures of HTTP JSON RPC requests") .tags(metricsTags) .register(Metrics.globalRegistry) ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt index f03c14e0..2a4a24d4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt @@ -67,8 +67,8 @@ class EthereumWsUpstream( .tags(metricsTags) .publishPercentileHistogram() .register(Metrics.globalRegistry), - Counter.builder("upstream.ws.err") - .description("Errors received on request through WebSocket JSON RPC connection") + Counter.builder("upstream.ws.fail") + .description("Number of failures of WebSocket JSON RPC requests") .tags(metricsTags) .register(Metrics.globalRegistry) ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt index 6631ee90..6b8ebc32 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt @@ -387,7 +387,7 @@ class WsConnection( .take(1) .singleOrEmpty() .doOnNext { rpcMetrics?.timer?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS) } - .doOnError { rpcMetrics?.errors?.increment() } + .doOnError { rpcMetrics?.fails?.increment() } .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) } .defaultIfEmpty(failResponse) } 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 a2943e65..f9806f3e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -175,12 +175,12 @@ class GrpcUpstreams( val metrics = RpcMetrics( Timer.builder("upstream.grpc.conn") - .description("Request time through a gRPC connection") + .description("Request time through a Dshackle/gRPC connection") .tags(metricsTags) .publishPercentileHistogram() .register(Metrics.globalRegistry), - Counter.builder("upstream.grpc.err") - .description("Errors received on request through gRPC connection") + Counter.builder("upstream.grpc.fail") + .description("Number of failures of Dshackle/gRPC requests") .tags(metricsTags) .register(Metrics.globalRegistry) ) 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 f154b673..961eab02 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt @@ -79,7 +79,7 @@ class JsonRpcGrpcClient( val bytes = resp.payload.toByteArray() Mono.just(JsonRpcResponse(bytes, null)) } else { - metrics.errors.increment() + metrics.fails.increment() Mono.error( RpcException( RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, 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 07bc9601..8d3e7b76 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt @@ -126,7 +126,7 @@ class JsonRpcHttpClient( is JsonRpcException -> JsonRpcResponse.error(t.error, JsonRpcResponse.NumberId(1)) else -> JsonRpcResponse.error(1, t.message ?: t.javaClass.name) } - metrics.errors.increment() + metrics.fails.increment() Mono.just(err) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/RpcMetrics.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/RpcMetrics.kt index 67e8156b..b9c8fda7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/RpcMetrics.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/RpcMetrics.kt @@ -20,5 +20,5 @@ import io.micrometer.core.instrument.Timer class RpcMetrics( val timer: Timer, - val errors: Counter + val fails: Counter )