Merge pull request #25 from p2p-org/handle_no_matching_upstreams

Handle no matching upstreams

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()}]")
 }

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?,

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<ParsedCallDetails>) =
+        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<Any> {
         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)

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<Chain, UpstreamsConfig.Options> {
         val defaultOptions = HashMap<Chain, UpstreamsConfig.Options>()
         config.defaultOptions.forEach { defaultsConfig ->

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<Upstream> {
 
@@ -26,4 +27,6 @@ interface ApiSource : Publisher<Upstream> {
      * Must be called before actual use, it spins off control flow of the API Source
      */
     fun request(tries: Int)
+
+    fun attempts(): AtomicInteger
 }

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<Upstream>
     private val standardWithFallback: List<Upstream>
 
+    private val counter: AtomicInteger = AtomicInteger(0)
+
     private var started = false
     private val control = Sinks.many().unicast().onBackpressureBuffer<Boolean>()
 
@@ -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"
     }

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"() {