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/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt index 67991ebf..3b9c9694 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 a2e5db89..d492db9c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -24,6 +24,8 @@ 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 import io.emeraldpay.dshackle.upstream.MultistreamHolder @@ -39,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 @@ -47,10 +50,12 @@ 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( @Autowired private val multistreamHolder: MultistreamHolder, + @Autowired private val configuredUpstreams: ConfiguredUpstreams, @Autowired private val signer: ResponseSigner ) { @@ -62,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() @@ -116,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 @@ -150,6 +160,16 @@ open class NativeCall( return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(request.chain.number))) } + 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))) + } + val upstream = multistreamHolder.getUpstream(chain) ?: return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(chain))) @@ -181,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) + ) ) ) } @@ -209,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) + ) } } @@ -243,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 { @@ -257,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)) { @@ -339,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/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/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" } 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"() {