From b96723f808b0270e125e333f10a702f6602a9138 Mon Sep 17 00:00:00 2001 From: Maxksim Fomenkov Date: Wed, 28 Sep 2022 10:54:23 +0300 Subject: [PATCH 1/4] handle no matching upstreams --- .../kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index 5f72e89b..d7a9a4eb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.grpc.Chain import io.micrometer.core.instrument.DistributionSummary @@ -156,6 +157,10 @@ class FilteredApis( } override fun subscribe(subscriber: Subscriber) { + if (allUpstreams.none { matcher.matches(it) }) { + Flux.empty().subscribe(subscriber) + return + } // initially try only standard upstreams val first = Flux.fromIterable(primaryUpstreams) val second = Flux.fromIterable(secondaryUpstreams) From f4ff5ab046ecdbd8f3f0f75f0d7df369dee7e900 Mon Sep 17 00:00:00 2001 From: Maxksim Fomenkov Date: Wed, 28 Sep 2022 11:38:34 +0300 Subject: [PATCH 2/4] fix code style --- src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt | 1 - 1 file changed, 1 deletion(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index d7a9a4eb..d7f3ede3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -17,7 +17,6 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.grpc.Chain import io.micrometer.core.instrument.DistributionSummary From 69f1f18a5afa637c008a2613dc40a98b015c4e14 Mon Sep 17 00:00:00 2001 From: Maxksim Fomenkov Date: Thu, 29 Sep 2022 02:20:26 +0300 Subject: [PATCH 3/4] better solution --- src/main/kotlin/io/emeraldpay/dshackle/SilentException.kt | 3 +++ src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt | 7 +++++++ .../io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt | 6 ++++++ .../io/emeraldpay/dshackle/upstream/FilteredApis.kt | 4 ---- .../io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy | 8 ++++++-- 5 files changed, 22 insertions(+), 6 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/SilentException.kt b/src/main/kotlin/io/emeraldpay/dshackle/SilentException.kt index 7b15b6c0..e92e67f1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/SilentException.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/SilentException.kt @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.grpc.Chain /** @@ -30,4 +31,6 @@ open class SilentException(message: String) : Exception(message) { } class DataUnavailable(val code: String) : SilentException("Data is unavailable: $code") + + class NoMatchingUpstream(matcher: Selector.LabelSelectorMatcher) : SilentException("No configured upstream matching selector [${matcher.describeInternal()}]") } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 6867a786..a6d99c3e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.dshackle.quorum.NotLaggingQuorum import io.emeraldpay.dshackle.quorum.QuorumReaderFactory +import io.emeraldpay.dshackle.startup.ConfiguredUpstreams import io.emeraldpay.dshackle.upstream.ApiSource import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.MultistreamHolder @@ -51,6 +52,7 @@ import java.util.EnumMap @Service open class NativeCall( @Autowired private val multistreamHolder: MultistreamHolder, + @Autowired private val configuredUpstreams: ConfiguredUpstreams, @Autowired private val signer: ResponseSigner ) { @@ -150,6 +152,11 @@ open class NativeCall( return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(request.chain.number))) } + val matcher = Selector.convertToMatcher(request.selector) + if (!configuredUpstreams.hasMatchingUpstream(chain, matcher)) { + return Flux.error(CallFailure(0, SilentException.NoMatchingUpstream(matcher))) + } + val upstream = multistreamHolder.getUpstream(chain) ?: return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(chain))) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index d59b3348..250361d9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -25,6 +25,7 @@ import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.HttpRpcFactory import io.emeraldpay.dshackle.upstream.MergedHead +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream @@ -107,6 +108,11 @@ open class ConfiguredUpstreams( } } + fun hasMatchingUpstream(chain: Chain, matcher: Selector.LabelSelectorMatcher): Boolean = + config.upstreams.any { up -> + up.chain?.equals(chain.chainName, ignoreCase = true) ?: true && matcher.matches(up.labels) + } + private fun buildDefaultOptions(config: UpstreamsConfig): HashMap { val defaultOptions = HashMap() config.defaultOptions.forEach { defaultsConfig -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index d7f3ede3..5f72e89b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -156,10 +156,6 @@ class FilteredApis( } override fun subscribe(subscriber: Subscriber) { - if (allUpstreams.none { matcher.matches(it) }) { - Flux.empty().subscribe(subscriber) - return - } // initially try only standard upstreams val first = Flux.fromIterable(primaryUpstreams) val second = Flux.fromIterable(secondaryUpstreams) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 2704fdbb..1d27097a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.startup.ConfiguredUpstreams import io.emeraldpay.dshackle.test.MultistreamHolderMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.quorum.AlwaysQuorum @@ -51,14 +52,17 @@ class NativeCallSpec extends Specification { ObjectMapper objectMapper = Global.objectMapper - def nativeCall(MultistreamHolder upstreams = null, ResponseSigner signer = null) { + def nativeCall(MultistreamHolder upstreams = null, ResponseSigner signer = null, ConfiguredUpstreams configuredUpstreams = null) { if (upstreams == null) { upstreams = Stub(MultistreamHolder) } if (signer == null) { signer = Stub(ResponseSigner) } - new NativeCall(upstreams, signer) + if (configuredUpstreams == null) { + configuredUpstreams = Stub(ConfiguredUpstreams) + } + new NativeCall(upstreams, configuredUpstreams, signer) } def "Tries router first"() { From 9af1b1f64b8f4dc447aa1031e13c3e6a7369a236 Mon Sep 17 00:00:00 2001 From: Maxksim Fomenkov Date: Tue, 4 Oct 2022 16:52:57 +0300 Subject: [PATCH 4/4] metrics and better error handling --- .../dshackle/quorum/QuorumRpcReader.kt | 4 + .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 84 +++++++++++++++++-- .../emeraldpay/dshackle/upstream/ApiSource.kt | 3 + .../dshackle/upstream/FilteredApis.kt | 14 +++- 4 files changed, 98 insertions(+), 7 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt index 4a8911b9..214b95bc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt @@ -31,6 +31,7 @@ import reactor.util.function.Tuple2 import reactor.util.function.Tuple3 import reactor.util.function.Tuples import java.util.Optional +import java.util.concurrent.atomic.AtomicInteger import java.util.function.BiFunction import java.util.function.Function @@ -190,6 +191,9 @@ class QuorumRpcReader( } } + fun getValidAttemptsCount(): AtomicInteger = + apiControl.attempts() + class Result( val value: ByteArray, val signature: ResponseSigner.Signature?, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index a6d99c3e..3f006391 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.dshackle.quorum.NotLaggingQuorum import io.emeraldpay.dshackle.quorum.QuorumReaderFactory +import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.startup.ConfiguredUpstreams import io.emeraldpay.dshackle.upstream.ApiSource import io.emeraldpay.dshackle.upstream.Multistream @@ -40,6 +41,7 @@ import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.etherjar.rpc.RpcResponseError import io.emeraldpay.grpc.BlockchainType import io.emeraldpay.grpc.Chain +import io.micrometer.core.instrument.Metrics import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired @@ -48,6 +50,7 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.kotlin.core.publisher.toMono import java.util.EnumMap +import java.util.concurrent.atomic.AtomicInteger @Service open class NativeCall( @@ -64,7 +67,9 @@ open class NativeCall( init { multistreamHolder.observeChains().subscribe { chain -> - if ((BlockchainType.from(chain) == BlockchainType.ETHEREUM_POS || BlockchainType.from(chain) == BlockchainType.ETHEREUM) && !ethereumCallSelectors.containsKey(chain) + if ((BlockchainType.from(chain) == BlockchainType.ETHEREUM_POS || BlockchainType.from(chain) == BlockchainType.ETHEREUM) && !ethereumCallSelectors.containsKey( + chain + ) ) { multistreamHolder.getUpstream(chain)?.let { up -> val reader = up.cast(EthereumPosMultiStream::class.java).getReader() @@ -118,7 +123,10 @@ open class NativeCall( return result.build() } - fun buildSignature(nonce: Long, signature: ResponseSigner.Signature): BlockchainOuterClass.NativeCallReplySignature { + fun buildSignature( + nonce: Long, + signature: ResponseSigner.Signature + ): BlockchainOuterClass.NativeCallReplySignature { val msg = BlockchainOuterClass.NativeCallReplySignature.newBuilder() msg.signature = ByteString.copyFrom(signature.value) msg.keyId = signature.keyId @@ -154,6 +162,11 @@ open class NativeCall( val matcher = Selector.convertToMatcher(request.selector) if (!configuredUpstreams.hasMatchingUpstream(chain, matcher)) { + if (Global.metricsExtended) { + Metrics.globalRegistry + .counter("no_matching_upstream", "chain", chain.chainCode, "matcher", matcher.describeInternal()) + .increment() + } return Flux.error(CallFailure(0, SilentException.NoMatchingUpstream(matcher))) } @@ -188,7 +201,11 @@ open class NativeCall( val errorMessage = "The method $method does not exist/is not available" return Mono.just( InvalidCallContext( - CallError(requestItem.id, errorMessage, JsonRpcError(RpcResponseError.CODE_METHOD_NOT_EXIST, errorMessage)) + CallError( + requestItem.id, + errorMessage, + JsonRpcError(RpcResponseError.CODE_METHOD_NOT_EXIST, errorMessage) + ) ) ) } @@ -216,7 +233,14 @@ open class NativeCall( matcher.withMatcher(heightMatcher) } val nonce = requestItem.nonce.let { if (it == 0L) null else it } - ValidCallContext(requestItem.id, nonce, upstream, matcher.build(), callQuorum, RawCallDetails(method, params)) + ValidCallContext( + requestItem.id, + nonce, + upstream, + matcher.build(), + callQuorum, + RawCallDetails(method, params) + ) } } @@ -250,6 +274,11 @@ open class NativeCall( return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method")) } val reader = quorumReaderFactory.create(ctx.getApis(), ctx.callQuorum, signer) + val counter = if (reader is QuorumRpcReader) { + reader.getValidAttemptsCount() + } else { + AtomicInteger(-1) + } return reader .read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce)) .map { @@ -264,10 +293,47 @@ open class NativeCall( Mono.just(failure) } .switchIfEmpty( - Mono.just(CallResult.fail(ctx.id, ctx.nonce, 1, "No response or no available upstream for ${ctx.payload.method}")) + Mono.fromSupplier { + counter.get().let { attempts -> + CallResult.fail( + ctx.id, + ctx.nonce, + 1, + errorMessage(attempts, ctx.payload.method) + ).also { + countFailure(attempts, ctx) + } + } + } ) } + private fun errorMessage(attempts: Int, method: String): String = + when (attempts) { + -1 -> "No response or no available upstream for $method" + 0 -> "No available upstream for $method" + else -> "No response for $method" + } + + private fun countFailure(counter: Int, ctx: ValidCallContext) = + when (counter) { + -1 -> "UNDEFINED" + 0 -> "NO_AVAIL_UPSTREAM" + else -> "NO_RESPONSE" + }.let { reason -> + if (Global.metricsExtended) { + Metrics.globalRegistry.counter( + "native_call_failure", + "upstream", + ctx.upstream.getId(), + "reason", + reason, + "chain", + ctx.upstream.chain.chainCode, + ).increment() + } + } + @Suppress("UNCHECKED_CAST") private fun extractParams(jsonParams: String): List { if (StringUtils.isEmpty(jsonParams)) { @@ -346,7 +412,13 @@ open class NativeCall( } } - open class CallResult(val id: Int, val nonce: Long?, val result: ByteArray?, val error: CallError?, val signature: ResponseSigner.Signature?) { + open class CallResult( + val id: Int, + val nonce: Long?, + val result: ByteArray?, + val error: CallError?, + val signature: ResponseSigner.Signature? + ) { companion object { fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?): CallResult { return CallResult(id, nonce, result, null, signature) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt index c9f747fe..66a977d4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream import org.reactivestreams.Publisher +import java.util.concurrent.atomic.AtomicInteger interface ApiSource : Publisher { @@ -26,4 +27,6 @@ interface ApiSource : Publisher { * Must be called before actual use, it spins off control flow of the API Source */ fun request(tries: Int) + + fun attempts(): AtomicInteger } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index 5f72e89b..36ee6615 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -28,6 +28,7 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Sinks import java.time.Duration import java.util.EnumMap +import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.locks.Lock import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.withLock @@ -89,6 +90,8 @@ class FilteredApis( private val secondaryUpstreams: List private val standardWithFallback: List + private val counter: AtomicInteger = AtomicInteger(0) + private var started = false private val control = Sinks.many().unicast().onBackpressureBuffer() @@ -177,7 +180,13 @@ class FilteredApis( .doFinally { metrics[chain]?.tried?.record(count.toDouble()) } } - result.filter { up -> up.isAvailable() && matcher.matches(up) } + result.filter { up -> + (up.isAvailable() && matcher.matches(up)).also { + if (it) { + counter.incrementAndGet() + } + } + } .zipWith(control.asFlux()) .map { it.t1 } .doOnSubscribe { @@ -201,6 +210,9 @@ class FilteredApis( } } + override fun attempts(): AtomicInteger = + counter + override fun toString(): String { return "Filter API: ${allUpstreams.size} upstreams with $matcher" }