diff --git a/src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt b/src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt index 862b6b81..52ba194b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt @@ -9,6 +9,7 @@ const val SPAN_REQUEST_API_TYPE = "request.api.type" const val SPAN_REQUEST_UPSTREAM_ID = "request.upstreamId" const val SPAN_RESPONSE_UPSTREAM_ID = "response.upstreamId" const val SPAN_REQUEST_ID = "request.id" +const val SPAN_NO_RESPONSE_MESSAGE = "no-response.message" const val LOCAL_READER = "localReader" const val REMOTE_QUORUM_RPC_READER = "remoteQuorumRpcReader" diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/spans/ErrorSpanHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/spans/ProviderSpanHandler.kt similarity index 89% rename from src/main/kotlin/io/emeraldpay/dshackle/config/spans/ErrorSpanHandler.kt rename to src/main/kotlin/io/emeraldpay/dshackle/config/spans/ProviderSpanHandler.kt index f19173f7..58936392 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/spans/ErrorSpanHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/spans/ProviderSpanHandler.kt @@ -5,14 +5,14 @@ import brave.handler.SpanHandler import brave.propagation.TraceContext import com.fasterxml.jackson.databind.ObjectMapper import com.github.benmanes.caffeine.cache.Caffeine -import io.emeraldpay.dshackle.commons.SPAN_ERROR import org.springframework.beans.factory.annotation.Qualifier import org.springframework.cloud.sleuth.Span import java.time.Duration -class ErrorSpanHandler( +class ProviderSpanHandler( @Qualifier("spanMapper") private val spanMapper: ObjectMapper, + private val spanExportableList: List ) : SpanHandler() { private val spans = Caffeine .newBuilder() @@ -48,7 +48,7 @@ class ErrorSpanHandler( } } - return if (spansInfo.hasError) { + return if (spansInfo.exportable) { spanMapper.writeValueAsString(spansInfo.spans) } else { "" @@ -68,13 +68,13 @@ class ErrorSpanHandler( private fun processSpanInfo(span: MutableSpan, spansInfo: SpansInfo) { spansInfo.spans.add(span) - if (span.tags().containsKey(SPAN_ERROR)) { - spansInfo.hasError = true + if (spanExportableList.any { it.isExportable(span) }) { + spansInfo.exportable = true } } private data class SpansInfo( - var hasError: Boolean = false, + var exportable: Boolean = false, val spans: MutableList = mutableListOf() ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/spans/ServerSpansInterceptor.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/spans/ServerSpansInterceptor.kt index 37841bf6..f47fca39 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/spans/ServerSpansInterceptor.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/spans/ServerSpansInterceptor.kt @@ -9,7 +9,7 @@ import org.springframework.cloud.sleuth.Tracer class ServerSpansInterceptor( private val tracer: Tracer, - private val errorSpanHandler: ErrorSpanHandler + private val providerSpanHandler: ProviderSpanHandler ) : ServerInterceptor { override fun interceptCall( call: ServerCall, @@ -32,7 +32,7 @@ class ServerSpansInterceptor( tracer.currentSpan()?.let { val parentId = it.context().parentId() if (parentId != null) { - val spans = errorSpanHandler.getErrorSpans(it.context().spanId(), it) + val spans = providerSpanHandler.getErrorSpans(it.context().spanId(), it) if (spans.isNotBlank()) { headers.put(Metadata.Key.of(SPAN_HEADER, Metadata.ASCII_STRING_MARSHALLER), spans) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/spans/SpanConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/spans/SpanConfig.kt index 365c90f4..c24be4d5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/spans/SpanConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/spans/SpanConfig.kt @@ -56,13 +56,14 @@ open class SpanConfig { @Bean open fun errorSpanHandler( @Qualifier("spanMapper") - spanMapper: ObjectMapper - ): ErrorSpanHandler = ErrorSpanHandler(spanMapper) + spanMapper: ObjectMapper, + spanExportableList: List + ): ProviderSpanHandler = ProviderSpanHandler(spanMapper, spanExportableList) @Bean open fun serverSpansInterceptor( tracer: org.springframework.cloud.sleuth.Tracer, - errorSpanHandler: ErrorSpanHandler - ): ServerInterceptor = ServerSpansInterceptor(tracer, errorSpanHandler) + providerSpanHandler: ProviderSpanHandler + ): ServerInterceptor = ServerSpansInterceptor(tracer, providerSpanHandler) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/spans/export.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/spans/export.kt new file mode 100644 index 00000000..fc60f2c5 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/spans/export.kt @@ -0,0 +1,20 @@ +package io.emeraldpay.dshackle.config.spans + +import brave.handler.MutableSpan +import io.emeraldpay.dshackle.commons.SPAN_ERROR +import io.emeraldpay.dshackle.commons.SPAN_NO_RESPONSE_MESSAGE +import org.springframework.stereotype.Component + +interface SpanExportable { + fun isExportable(span: MutableSpan): Boolean +} + +@Component +class ErrorSpanExportable : SpanExportable { + override fun isExportable(span: MutableSpan): Boolean = span.tags().containsKey(SPAN_ERROR) +} + +@Component +class NoResponseSpanExportable : SpanExportable { + override fun isExportable(span: MutableSpan): Boolean = span.tags().containsKey(SPAN_NO_RESPONSE_MESSAGE) +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt index 930fbb27..d2821865 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt @@ -20,6 +20,11 @@ import io.emeraldpay.dshackle.upstream.ApiSource import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import org.springframework.cloud.sleuth.Tracer +import java.util.concurrent.atomic.AtomicInteger + +interface QuorumReader : Reader { + fun attempts(): AtomicInteger +} // creates instance of a Quorum based reader interface QuorumReaderFactory { @@ -30,10 +35,10 @@ interface QuorumReaderFactory { } } - fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): Reader + fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): QuorumReader class Default : QuorumReaderFactory { - override fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): Reader { + override fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): QuorumReader { return QuorumRpcReader(apis, quorum, signer, tracer) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt index 664a970f..546d9903 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt @@ -15,10 +15,11 @@ */ package io.emeraldpay.dshackle.quorum +import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.commons.API_READER +import io.emeraldpay.dshackle.commons.SPAN_NO_RESPONSE_MESSAGE import io.emeraldpay.dshackle.commons.SPAN_REQUEST_API_TYPE import io.emeraldpay.dshackle.commons.SPAN_REQUEST_UPSTREAM_ID -import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.SpannedReader import io.emeraldpay.dshackle.upstream.ApiSource import io.emeraldpay.dshackle.upstream.Upstream @@ -48,7 +49,7 @@ class QuorumRpcReader( private val quorum: CallQuorum, private val signer: ResponseSigner?, private val tracer: Tracer -) : Reader { +) : QuorumReader { companion object { private val log = LoggerFactory.getLogger(QuorumRpcReader::class.java) @@ -56,6 +57,8 @@ class QuorumRpcReader( constructor(apiControl: ApiSource, quorum: CallQuorum, tracer: Tracer) : this(apiControl, quorum, null, tracer) + override fun attempts(): AtomicInteger = apiControl.attempts() + override fun read(key: JsonRpcRequest): Mono { // needs at least one response, so start a request apiControl.request(1) @@ -95,7 +98,7 @@ class QuorumRpcReader( .transform(processResult(defaultResult)) } - fun execute(key: JsonRpcRequest, retrySpec: reactor.util.retry.Retry): Function, Mono> { + private fun execute(key: JsonRpcRequest, retrySpec: reactor.util.retry.Retry): Function, Mono> { val quorumReduce = BiFunction, Upstream, Optional>, CallQuorum> { res, a -> if (res.record(a.t1, a.t2.orElse(null), a.t3, a.t4.orElse(null))) { log.trace("Quorum is resolved for method ${key.method}") @@ -122,7 +125,7 @@ class QuorumRpcReader( } } - fun processResult(defaultResult: Mono): Function, Mono> { + private fun processResult(defaultResult: Mono): Function, Mono> { return Function { quorumResult -> quorumResult .filter { it.isResolved() } // return nothing if not resolved @@ -134,7 +137,7 @@ class QuorumRpcReader( } } - fun callApi(api: Upstream, key: JsonRpcRequest): Mono, Upstream, Optional>> { + private fun callApi(api: Upstream, key: JsonRpcRequest): Mono, Upstream, Optional>> { val apiReader = api.getIngressReader() val spanParams = mapOf( SPAN_REQUEST_API_TYPE to apiReader.javaClass.name, @@ -152,7 +155,7 @@ class QuorumRpcReader( .map { Tuples.of(it.t1, it.t2, api, it.t3) } } - fun withSignatureAndUpstream(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function, Mono, Optional>>> { + private fun withSignatureAndUpstream(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function, Mono, Optional>>> { return Function { src -> src.map { val signature = response.providedSignature @@ -166,7 +169,7 @@ class QuorumRpcReader( } } - fun withErrorResume(api: Upstream, key: JsonRpcRequest): Function, Mono> { + private fun withErrorResume(api: Upstream, key: JsonRpcRequest): Function, Mono> { return Function { src -> src.onErrorResume { err -> log.error("Error during call upstream ${api.getId()} with method ${key.method}", err) @@ -195,22 +198,33 @@ class QuorumRpcReader( } } - fun setupDefaultResult(key: JsonRpcRequest): Mono { + private fun setupDefaultResult(key: JsonRpcRequest): Mono { return Mono.just(quorum).flatMap { q -> if (q.isFailed()) { val err = q.getError()?.asException(JsonRpcResponse.NumberId(key.id)) ?: JsonRpcException(JsonRpcResponse.NumberId(key.id), JsonRpcError(-32603, "Unhandled Upstream error")) log.warn("Quorum is failed. Method ${key.method}, message ${err.message}") - Mono.error(err) + Mono.error(err) } else { log.warn("Did not get any result from upstream. Method [${key.method}] using [$q]") - Mono.empty() + noResponse(key.method, q) } } } - fun getValidAttemptsCount(): AtomicInteger = - apiControl.attempts() + private fun noResponse(method: String, q: CallQuorum): Mono { + return apiControl.upstreamsMatchesResponse()?.run { + tracer.currentSpan()?.tag(SPAN_NO_RESPONSE_MESSAGE, getFullCause()) + val cause = getCause(method) ?: return Mono.empty() + if (cause.shouldReturnNull) { + Mono.just( + Result(Global.nullValue, null, 1, emptyList(), null) + ) + } else { + Mono.error(RpcException(1, "No response for method $method. Cause - ${cause.cause}")) + } + } ?: Mono.error(RpcException(1, "Quorum [$q] is not resolved [isResolved - ${q.isResolved()}]")) + } class Result( val value: ByteArray, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index b00b5228..79d00444 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -66,7 +66,6 @@ import reactor.core.publisher.Mono import reactor.kotlin.core.publisher.toMono import reactor.util.context.Context import java.util.EnumMap -import java.util.concurrent.atomic.AtomicInteger @Service open class NativeCall( @@ -392,18 +391,16 @@ open class NativeCall( return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method")) } val reader = quorumReaderFactory.create(ctx.getApis(), ctx.callQuorum, signer, tracer) - val counter = if (reader is QuorumRpcReader) { - reader.getValidAttemptsCount() - } else { - AtomicInteger(-1) - } + val counter = reader.attempts() return SpannedReader(reader, tracer, REMOTE_QUORUM_RPC_READER) .read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector)) .map { val bytes = ctx.resultDecorator.processResult(it) validateResult(bytes, "remote", ctx) - CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, it.providedUpstreamId ?: it.resolvers.first().getId(), ctx) + val upId = it.providedUpstreamId + ?: if (it.resolvers.isEmpty()) ctx.upstream.getId() else it.resolvers.first().getId() + CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, upId, ctx) } .onErrorResume { t -> Mono.just(CallResult.fail(ctx.id, ctx.nonce, t, ctx)) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt index 9df5ed7d..afea9881 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt @@ -63,9 +63,11 @@ class TrackBitcoinAddress( /** * Criteria for a remote grpc upstream that can provide a balance */ - private val balanceUpstreamMatcher = Selector.LocalAndMatcher( - Selector.GrpcMatcher(), - Selector.CapabilityMatcher(Capability.BALANCE) + private val balanceUpstreamMatcher = Selector.MultiMatcher( + listOf( + Selector.GrpcMatcher(), + Selector.CapabilityMatcher(Capability.BALANCE) + ) ) @EventListener diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt index 66a977d4..b97b90b3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt @@ -29,4 +29,6 @@ interface ApiSource : Publisher { fun request(tries: Int) fun attempts(): AtomicInteger + + fun upstreamsMatchesResponse(): UpstreamsMatchesResponse? } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index 0e370c70..024f392d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -49,6 +49,7 @@ class FilteredApis( private val retryLimit: Long, jitter: Int ) : ApiSource { + private val internalMatcher: Selector.Matcher companion object { private val log = LoggerFactory.getLogger(FilteredApis::class.java) @@ -94,6 +95,7 @@ class FilteredApis( private var started = false private val control = Sinks.many().unicast().onBackpressureBuffer() + private var upstreamsMatchesResponse: UpstreamsMatchesResponse? = UpstreamsMatchesResponse() init { delay = if (jitter > 0) { @@ -129,6 +131,9 @@ class FilteredApis( monitoring.countFallback.record(fallbackUpstreams.size.toDouble()) } } + internalMatcher = Selector.MultiMatcher( + listOf(Selector.AvailabilityMatcher(), matcher) + ) } private fun getMetrics(chain: Chain): Monitoring { @@ -181,14 +186,16 @@ class FilteredApis( } result.filter { up -> - (up.isAvailable() && matcher.matches(up)).also { - if (it) { - counter.incrementAndGet() - } - } + val matchesResponse = internalMatcher.matchesWithCause(up) + processMatchesResponse(up.getId(), matchesResponse) + matchesResponse.matched() } .zipWith(control.asFlux()) - .map { it.t1 } + .map { + upstreamsMatchesResponse = null + counter.incrementAndGet() + it.t1 + } .doOnSubscribe { if (!started) { // in addition to subscription the FilteredAPI should use request() method to prepare the control flow @@ -198,6 +205,14 @@ class FilteredApis( .subscribe(subscriber) } + private fun processMatchesResponse(upstreamId: String, matchesResponse: MatchesResponse) { + upstreamsMatchesResponse?.run { + if (!matchesResponse.matched()) { + addUpstreamMatchesResponse(upstreamId, matchesResponse) + } + } + } + override fun resolve() { control.tryEmitComplete() } @@ -213,8 +228,10 @@ class FilteredApis( override fun attempts(): AtomicInteger = counter + override fun upstreamsMatchesResponse(): UpstreamsMatchesResponse? = upstreamsMatchesResponse + override fun toString(): String { - return "Filter API: ${allUpstreams.size} upstreams with $matcher" + return "Filter API: ${allUpstreams.size} upstreams with $internalMatcher" } class Monitoring(chain: Chain) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MatchesResponse.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MatchesResponse.kt new file mode 100644 index 00000000..0b5c56e4 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MatchesResponse.kt @@ -0,0 +1,83 @@ +package io.emeraldpay.dshackle.upstream + +sealed class MatchesResponse { + + fun matched(): Boolean { + return when (this) { + is Success -> true + else -> false + } + } + + fun getCause(): String? = + when (this) { + is LabelResponse -> "No label `${this.name}` with values ${this.values}" + AvailabilityResponse -> "Upstream is not available" + is CapabilityResponse -> "Upstream does not have capability ${this.capability}" + is ExistsResponse -> "Label ${this.name} does not exist" + GrpcResponse -> "Upstream is not grpc" + is HeightResponse -> "Upstream height ${this.currentHeight} is less than ${this.height}" + is MethodResponse -> "Method ${this.method} is not supported" + is MultiResponse -> + this.allResponses + .filter { it !is Success } + .joinToString("; ") { it.getCause()!! } + is NotMatchedResponse -> "Not matched - ${response.getCause()}" + is SameNodeResponse -> "Upstream does not have hash ${this.upstreamHash}" + else -> null + } + + object Success : MatchesResponse() + + data class LabelResponse( + val name: String, + val values: Collection + ) : MatchesResponse() + + data class NotMatchedResponse( + val response: MatchesResponse + ) : MatchesResponse() + + data class MultiResponse( + private val responses: Set + ) : MatchesResponse() { + val allResponses = mutableSetOf() + + init { + responses.forEach { + if (it is MultiResponse) { + it.allResponses.forEach { resp -> + allResponses.add(resp) + } + } else { + allResponses.add(it) + } + } + } + } + + data class MethodResponse( + val method: String + ) : MatchesResponse() + + data class ExistsResponse( + val name: String + ) : MatchesResponse() + + data class CapabilityResponse( + val capability: Capability + ) : MatchesResponse() + + object GrpcResponse : MatchesResponse() + + data class HeightResponse( + val height: Long, + val currentHeight: Long + ) : MatchesResponse() + + data class SameNodeResponse( + val upstreamHash: Byte + ) : MatchesResponse() + + object AvailabilityResponse : MatchesResponse() +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt index 7ce6172f..d972e3b4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt @@ -18,6 +18,14 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.upstream.MatchesResponse.AvailabilityResponse +import io.emeraldpay.dshackle.upstream.MatchesResponse.CapabilityResponse +import io.emeraldpay.dshackle.upstream.MatchesResponse.ExistsResponse +import io.emeraldpay.dshackle.upstream.MatchesResponse.GrpcResponse +import io.emeraldpay.dshackle.upstream.MatchesResponse.HeightResponse +import io.emeraldpay.dshackle.upstream.MatchesResponse.NotMatchedResponse +import io.emeraldpay.dshackle.upstream.MatchesResponse.SameNodeResponse +import io.emeraldpay.dshackle.upstream.MatchesResponse.Success import org.apache.commons.lang3.StringUtils import java.util.Collections @@ -121,17 +129,27 @@ class Selector { } } - interface Matcher { - fun matches(up: Upstream): Boolean + abstract class Matcher { + fun matches(up: Upstream): Boolean = matchesWithCause(up).matched() - fun describeInternal(): String + abstract fun matchesWithCause(up: Upstream): MatchesResponse + + abstract fun describeInternal(): String } data class MultiMatcher( private val matchers: Collection - ) : Matcher { - override fun matches(up: Upstream): Boolean { - return matchers.all { it.matches(up) } + ) : Matcher() { + + override fun matchesWithCause(up: Upstream): MatchesResponse { + val responses = matchers.map { it.matchesWithCause(up) } + return if (responses.all { it is Success }) { + Success + } else { + MatchesResponse.MultiResponse( + responses.filter { it !is Success }.toSet() + ) + } } fun getMatchers(): Collection { @@ -158,10 +176,14 @@ class Selector { data class MethodMatcher( val method: String - ) : Matcher { - override fun matches(up: Upstream): Boolean { - return up.getMethods().isCallable(method) - } + ) : Matcher() { + + override fun matchesWithCause(up: Upstream): MatchesResponse = + if (up.getMethods().isCallable(method)) { + Success + } else { + MatchesResponse.MethodResponse(method) + } override fun describeInternal(): String { return "allow method $method" @@ -172,19 +194,26 @@ class Selector { } } - abstract class LabelSelectorMatcher : Matcher { - override fun matches(up: Upstream): Boolean { - return up.getLabels().any(this::matches) + abstract class LabelSelectorMatcher : Matcher() { + + override fun matchesWithCause(up: Upstream): MatchesResponse { + val labelsResponses = up.getLabels().map { matchesWithCause(it) } + return if (labelsResponses.any { it is Success }) { + Success + } else { + MatchesResponse.MultiResponse( + labelsResponses.filter { it !is Success }.toSet() + ) + } } - abstract fun matches(labels: UpstreamsConfig.Labels): Boolean + abstract fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse abstract fun asProto(): BlockchainOuterClass.Selector? } - class EmptyMatcher : Matcher { - override fun matches(up: Upstream): Boolean { - return true - } + class EmptyMatcher : Matcher() { + + override fun matchesWithCause(up: Upstream): MatchesResponse = Success override fun describeInternal(): String { return "empty" @@ -197,18 +226,12 @@ class Selector { class AnyLabelMatcher : LabelSelectorMatcher() { - override fun matches(labels: UpstreamsConfig.Labels): Boolean { - return true - } + override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse = Success override fun asProto(): BlockchainOuterClass.Selector? { return null } - override fun matches(up: Upstream): Boolean { - return true - } - override fun describeInternal(): String { return "any label" } @@ -218,26 +241,22 @@ class Selector { } } - class LocalAndMatcher(vararg val matchers: Matcher) : Matcher { + class LabelMatcher( + val name: String, + val values: Collection + ) : LabelSelectorMatcher() { - override fun matches(up: Upstream): Boolean { - return matchers.all { it.matches(up) } - } - - override fun describeInternal(): String { - return "local upstream" - } - - override fun toString(): String { - return "Matcher: ${describeInternal()}" - } - } - - class LabelMatcher(val name: String, val values: Collection) : LabelSelectorMatcher() { - override fun matches(labels: UpstreamsConfig.Labels): Boolean { - return labels.get(name)?.let { labelValue -> + override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse { + val response = labels[name]?.let { labelValue -> values.any { it == labelValue } } ?: false + return if (response) { + Success + } else { + MatchesResponse.LabelResponse( + name, values + ) + } } override fun asProto(): BlockchainOuterClass.Selector { @@ -257,9 +276,19 @@ class Selector { } } - class OrMatcher(val matchers: Collection) : LabelSelectorMatcher() { - override fun matches(labels: UpstreamsConfig.Labels): Boolean { - return matchers.any { matcher -> matcher.matches(labels) } + class OrMatcher( + val matchers: Collection, + ) : LabelSelectorMatcher() { + + override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse { + val responses = matchers.map { it.matchesWithCause(labels) } + return if (responses.any { it is Success }) { + Success + } else { + MatchesResponse.MultiResponse( + responses.filter { it !is Success }.toSet() + ) + } } override fun asProto(): BlockchainOuterClass.Selector { @@ -279,9 +308,19 @@ class Selector { } } - class AndMatcher(val matchers: Collection) : LabelSelectorMatcher() { - override fun matches(labels: UpstreamsConfig.Labels): Boolean { - return matchers.all { matcher -> matcher.matches(labels) } + class AndMatcher( + val matchers: Collection + ) : LabelSelectorMatcher() { + + override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse { + val responses = matchers.map { it.matchesWithCause(labels) } + return if (responses.all { it is Success }) { + Success + } else { + MatchesResponse.MultiResponse( + responses.filter { it !is Success }.toSet() + ) + } } override fun asProto(): BlockchainOuterClass.Selector { @@ -301,9 +340,17 @@ class Selector { } } - class NotMatcher(val matcher: LabelSelectorMatcher) : LabelSelectorMatcher() { - override fun matches(labels: UpstreamsConfig.Labels): Boolean { - return !matcher.matches(labels) + class NotMatcher( + val matcher: LabelSelectorMatcher + ) : LabelSelectorMatcher() { + + override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse { + val response = matcher.matchesWithCause(labels) + return if (response !is Success) { + Success + } else { + NotMatchedResponse(response) + } } override fun asProto(): BlockchainOuterClass.Selector { @@ -323,10 +370,16 @@ class Selector { } } - class ExistsMatcher(val name: String) : LabelSelectorMatcher() { - override fun matches(labels: UpstreamsConfig.Labels): Boolean { - return labels.containsKey(name) - } + class ExistsMatcher( + val name: String + ) : LabelSelectorMatcher() { + + override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse = + if (labels.containsKey(name)) { + Success + } else { + ExistsResponse(name) + } override fun asProto(): BlockchainOuterClass.Selector { return BlockchainOuterClass.Selector.newBuilder().setExistsSelector( @@ -345,10 +398,14 @@ class Selector { } } - class CapabilityMatcher(val capability: Capability) : Matcher { - override fun matches(up: Upstream): Boolean { - return up.getCapabilities().contains(capability) - } + class CapabilityMatcher(val capability: Capability) : Matcher() { + + override fun matchesWithCause(up: Upstream): MatchesResponse = + if (up.getCapabilities().contains(capability)) { + Success + } else { + CapabilityResponse(capability) + } override fun describeInternal(): String { return "provides $capability API" @@ -359,10 +416,14 @@ class Selector { } } - class GrpcMatcher : Matcher { - override fun matches(up: Upstream): Boolean { - return up.isGrpc() - } + class GrpcMatcher : Matcher() { + + override fun matchesWithCause(up: Upstream): MatchesResponse = + if (up.isGrpc()) { + Success + } else { + GrpcResponse + } override fun describeInternal(): String { return "is gRPC" @@ -373,9 +434,15 @@ class Selector { } } - class HeightMatcher(val height: Long) : Matcher { - override fun matches(up: Upstream): Boolean { - return (up.getHead().getCurrentHeight() ?: 0) >= height + class HeightMatcher(val height: Long) : Matcher() { + + override fun matchesWithCause(up: Upstream): MatchesResponse { + val currentHeight = up.getHead().getCurrentHeight() ?: 0 + return if (currentHeight >= height) { + Success + } else { + HeightResponse(height, currentHeight) + } } override fun equals(other: Any?): Boolean { @@ -400,9 +467,14 @@ class Selector { } } - class SameNodeMatcher(private val upstreamHash: Byte) : Matcher { - override fun matches(up: Upstream): Boolean = - up.nodeId() == upstreamHash + class SameNodeMatcher(private val upstreamHash: Byte) : Matcher() { + + override fun matchesWithCause(up: Upstream): MatchesResponse = + if (up.nodeId() == upstreamHash) { + Success + } else { + SameNodeResponse(upstreamHash) + } override fun describeInternal(): String = "upstream node-id=${upstreamHash.toUByte()}" @@ -417,4 +489,17 @@ class Selector { return other.upstreamHash == upstreamHash } } + + class AvailabilityMatcher : Matcher() { + override fun matchesWithCause(up: Upstream): MatchesResponse = + if (up.isAvailable()) { + Success + } else { + AvailabilityResponse + } + + override fun describeInternal(): String = "availability" + + override fun toString(): String = "Matcher: ${describeInternal()}" + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamsMatchesResponse.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamsMatchesResponse.kt new file mode 100644 index 00000000..dcef474a --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamsMatchesResponse.kt @@ -0,0 +1,99 @@ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.upstream.MatchesResponse.HeightResponse +import io.emeraldpay.dshackle.upstream.MatchesResponse.MultiResponse +import kotlin.math.min + +class UpstreamsMatchesResponse { + + companion object { + private val possibleNullReturnMethods = listOf( + "eth_getTransactionByHash", + "eth_getTransactionReceipt", + "eth_getBlockByHash", + "eth_getBlockByNumber", + "eth_getTransactionByBlockHashAndIndex", + "eth_getTransactionByBlockNumberAndIndex", + "eth_getUncleByBlockHashAndIndex", + "eth_getUncleByBlockNumberAndIndex" + ) + } + + private val responses = LinkedHashSet() + + fun addUpstreamMatchesResponse(upstreamId: String, response: MatchesResponse) { + if (!response.matched()) { + responses.add(UpstreamNotMatchedResponse(upstreamId, response)) + } + } + + fun getFullCause(): String? = + if (responses.isEmpty()) { + null + } else { + responses + .joinToString("; ") { "${it.upstreamId} - ${it.matchesResponse.getCause()}" } + .run { + substring(0, min(200, this.length)) + } + } + + fun getCause(method: String): NotMatchesCause? { + if (responses.isEmpty()) { + return null + } + val commonMatchesResponse = hasCommonMatchesResponse() + return if (commonMatchesResponse is HeightResponse && possibleNullReturnMethods.contains(method)) { + NotMatchesCause(true) + } else if (commonMatchesResponse != null && commonMatchesResponse !is HeightResponse) { + NotMatchesCause(false, commonMatchesResponse.getCause()) + } else { + null + } + } + + private fun hasCommonMatchesResponse(): MatchesResponse? { + val matchesResponses = responses.map { it.matchesResponses } + val commonResponses = mutableListOf() + for (matchesResponse in matchesResponses[0]) { + if (matchesResponses.all { it.any { resp -> matchesResponse.javaClass == resp.javaClass } }) { + commonResponses.add(matchesResponse) + } + } + if (commonResponses.isEmpty()) { + return null + } + val heightResponse = commonResponses.find { it is HeightResponse } + return heightResponse ?: commonResponses[0] + } + + data class NotMatchesCause( + val shouldReturnNull: Boolean, + val cause: String? = null + ) + + private class UpstreamNotMatchedResponse( + val upstreamId: String, + val matchesResponse: MatchesResponse + ) { + val matchesResponses: Set = when (matchesResponse) { + is MultiResponse -> matchesResponse.allResponses + else -> setOf(matchesResponse) + } + + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (javaClass != other?.javaClass) return false + + other as UpstreamNotMatchedResponse + + if (upstreamId != other.upstreamId) return false + + return true + } + + override fun hashCode(): Int { + return upstreamId.hashCode() + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt index 60d20209..a96b3d6a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt @@ -17,9 +17,11 @@ class RemoteUnspentReader( private val log = LoggerFactory.getLogger(RemoteUnspentReader::class.java) } - private val selector = Selector.LocalAndMatcher( - Selector.GrpcMatcher(), - Selector.CapabilityMatcher(Capability.BALANCE) + private val selector = Selector.MultiMatcher( + listOf( + Selector.GrpcMatcher(), + Selector.CapabilityMatcher(Capability.BALANCE) + ) ) override fun read(key: Address): Mono> { diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy index 8b75a784..b75a57dd 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy @@ -289,4 +289,33 @@ class QuorumRpcReaderSpec extends Specification { .verify(Duration.ofSeconds(1)) } + def "Error if no upstreams"() { + setup: + def api = Stub(Reader) + def up = Mock(Upstream) { + _ * getId() >> "id1" + _ * isAvailable() >> false + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * getIngressReader() >> api + } + def apis = new FilteredApis( + Chain.ETHEREUM, + [up], Selector.empty + ) + def reader = new QuorumRpcReader(apis, new AlwaysQuorum(), Stub(Tracer)) + + when: + def act = reader.read(new JsonRpcRequest("eth_test", [])) + .map { + new String(it.value) + } + + then: + StepVerifier.create(act) + .expectErrorMatches { t -> + t instanceof RpcException && t.rpcMessage == "No response for method eth_test. Cause - Upstream is not available" && t.error.code == 1 + } + .verify(Duration.ofSeconds(4)) + } + } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index ac090b50..82508463 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -25,6 +25,7 @@ import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.CacheConfig import io.emeraldpay.dshackle.config.MainConfig import io.emeraldpay.dshackle.quorum.AlwaysQuorum +import io.emeraldpay.dshackle.quorum.QuorumReader import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.reader.Reader @@ -52,6 +53,7 @@ import spock.lang.Ignore import spock.lang.Specification import java.time.Duration +import java.util.concurrent.atomic.AtomicInteger class NativeCallSpec extends Specification { @@ -131,7 +133,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall() nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, null, 1, Collections.singletonList(ups), null)) } } @@ -152,7 +154,8 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall() nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { + 1 * attempts() >> new AtomicInteger(1) 1 * read(new JsonRpcRequest("eth_test", [], 10)) >> Mono.empty() } } @@ -176,7 +179,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall() nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_test", [], 10)) >> Mono.error( new JsonRpcException(JsonRpcResponse.Id.from(12), new JsonRpcError(-32123, "Foo Bar", "Foo Bar Baz"), true) ) @@ -613,7 +616,7 @@ class NativeCallSpec extends Specification { } def nativeCall = nativeCall(multistreamHolder) nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups), null)) } } @@ -648,7 +651,7 @@ class NativeCallSpec extends Specification { } def nativeCall = nativeCall(multistreamHolder) nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups), null)) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index a1e07b37..b38a4795 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -33,6 +33,8 @@ import spock.lang.Specification import java.time.Duration +import static java.util.List.of + class FilteredApisSpec extends Specification { def ethereumTargets = new DefaultEthereumMethods(Chain.ETHEREUM) @@ -194,6 +196,29 @@ class FilteredApisSpec extends Specification { .expectNext(ups[2], ups[3], ups[4], ups[5], ups[0], ups[1]) .expectComplete() .verify(Duration.ofSeconds(1)) + act.attempts().get() == 6 + } + + def "FilteredApis is requested 3 times"() { + setup: + def apis = (0..5).collect { + new EthereumApiStub(it) + } + def ups = apis.collect { + TestingCommons.upstream(it) + } + when: + def act = new FilteredApis(Chain.ETHEREUM, ups, Selector.empty, 2, 1, 0) + act.request(3) + then: + StepVerifier.create(act) + .expectNext(ups[2], ups[3], ups[4]) + .then { + act.resolve() + } + .expectComplete() + .verify(Duration.ofSeconds(1)) + act.attempts().get() == 3 } def "Start with offset - 5 items"() { @@ -330,4 +355,207 @@ class FilteredApisSpec extends Specification { .expectComplete() .verify(Duration.ofSeconds(1)) } + + def "No upstreams if they all are unavailable"() { + setup: + List ups = [ + Mock(Upstream) { + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * isAvailable() >> false + _ * getId() >> "id1" + _ * getStatus() >> UpstreamAvailability.SYNCING + }, + Mock(Upstream) { + _ * getId() >> "id2" + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * isAvailable() >> false + _ * getStatus() >> UpstreamAvailability.SYNCING + } + ] + when: + def act = new FilteredApis(Chain.ETHEREUM, ups, Selector.empty) + act.request(1) + then: + StepVerifier.create(act) + .expectNextCount(0) + .expectComplete() + .verify(Duration.ofSeconds(5)) + act.upstreamsMatchesResponse() != null + act.upstreamsMatchesResponse().getFullCause() == "id1 - Upstream is not available; id2 - Upstream is not available" + act.upstreamsMatchesResponse().getCause("").cause == "Upstream is not available" + } + + def "No upstreams if they all are not matched"() { + setup: + List ups = [ + Mock(Upstream) { + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * isAvailable() >> true + _ * getId() >> "id1" + _ * getStatus() >> UpstreamAvailability.OK + _ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "archive"))) + }, + Mock(Upstream) { + _ * getId() >> "id2" + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * isAvailable() >> true + _ * getStatus() >> UpstreamAvailability.OK + _ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "archive"))) + } + ] + when: + def act = new FilteredApis(Chain.ETHEREUM, ups, new Selector.LabelMatcher("node", of("test"))) + act.request(1) + then: + StepVerifier.create(act) + .expectNextCount(0) + .expectComplete() + .verify(Duration.ofSeconds(5)) + act.upstreamsMatchesResponse() != null + act.upstreamsMatchesResponse().getFullCause() == "id1 - No label `node` with values [test]; id2 - No label `node` with values [test]" + act.upstreamsMatchesResponse().getCause("").cause == "No label `node` with values [test]" + } + + def "No upstreams if they all are not matched by first matcher"() { + setup: + List ups = [ + Mock(Upstream) { + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * isAvailable() >> false + _ * getId() >> "id1" + _ * getStatus() >> UpstreamAvailability.OK + _ * getHead() >> Mock(Head) { + _ * getCurrentHeight() >> 100000 + } + _ * getLabels() >> of( + UpstreamsConfig.Labels.fromMap( + Map.of("node", "archive", "type", "super") + ) + ) + }, + Mock(Upstream) { + _ * getId() >> "id2" + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * isAvailable() >> false + _ * getHead() >> Mock(Head) { + _ * getCurrentHeight() >> 100000 + } + _ * getStatus() >> UpstreamAvailability.OK + _ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "archive"))) + } + ] + when: + def act = new FilteredApis( + Chain.ETHEREUM, ups, + new Selector.MultiMatcher( + of( + new Selector.HeightMatcher(100000000), + ) + ) + ) + act.request(1) + then: + StepVerifier.create(act) + .expectNextCount(0) + .expectComplete() + .verify(Duration.ofSeconds(5)) + act.upstreamsMatchesResponse() != null + act.upstreamsMatchesResponse().getFullCause() == "id1 - Upstream is not available; Upstream height 100000 is less than 100000000; id2 - Upstream is not available; Upstream height 100000 is less than 100000000" + act.upstreamsMatchesResponse().getCause("eth_getTransactionByHash").cause == null + act.upstreamsMatchesResponse().getCause("eth_getTransactionByHash").shouldReturnNull + } + + def "No upstreams if they all are not matched and return null cause"() { + setup: + List ups = [ + Mock(Upstream) { + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * isAvailable() >> false + _ * getId() >> "id1" + _ * getStatus() >> UpstreamAvailability.OK + _ * getHead() >> Mock(Head) { + _ * getCurrentHeight() >> 100000 + } + _ * getLabels() >> of( + UpstreamsConfig.Labels.fromMap( + Map.of("node", "archive", "type", "super") + ) + ) + }, + Mock(Upstream) { + _ * getId() >> "id2" + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * isAvailable() >> false + _ * getHead() >> Mock(Head) { + _ * getCurrentHeight() >> 100000 + } + _ * getStatus() >> UpstreamAvailability.OK + _ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "archive"))) + } + ] + when: + def act = new FilteredApis( + Chain.ETHEREUM, ups, + new Selector.MultiMatcher( + of( + new Selector.HeightMatcher(100000000), + ) + ) + ) + act.request(1) + then: + StepVerifier.create(act) + .expectNextCount(0) + .expectComplete() + .verify(Duration.ofSeconds(5)) + act.upstreamsMatchesResponse() != null + act.upstreamsMatchesResponse().getFullCause() == "id1 - Upstream is not available; Upstream height 100000 is less than 100000000; id2 - Upstream is not available; Upstream height 100000 is less than 100000000" + act.upstreamsMatchesResponse().getCause("other") == null + } + + def "Second upstream if first is not matched"() { + setup: + def up = Mock(Upstream) { + _ * getId() >> "id2" + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * isAvailable() >> true + _ * getHead() >> Mock(Head) { + _ * getCurrentHeight() >> 100000001 + } + _ * getStatus() >> UpstreamAvailability.OK + _ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "test"))) + } + List ups = [ + Mock(Upstream) { + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * isAvailable() >> true + _ * getId() >> "id1" + _ * getStatus() >> UpstreamAvailability.OK + _ * getHead() >> Mock(Head) { + _ * getCurrentHeight() >> 100000 + } + _ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "archive"))) + }, up + ] + when: + def act = new FilteredApis( + Chain.ETHEREUM, ups, + new Selector.MultiMatcher( + of( + new Selector.HeightMatcher(100000000), + new Selector.LabelMatcher("node", of("test")) + ) + ) + ) + act.request(1) + then: + StepVerifier.create(act) + .expectNext(up) + .then { + act.resolve() + } + .expectComplete() + .verify(Duration.ofSeconds(5)) + act.upstreamsMatchesResponse() == null + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/SelectorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/SelectorSpec.groovy index fd8de016..34516c3d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/SelectorSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/SelectorSpec.groovy @@ -178,9 +178,12 @@ class SelectorSpec extends Specification { def "LABEL matches single label"() { setup: def matcher = new Selector.LabelMatcher("test", ["foo"]) + def up = Mock(Upstream) { + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)] + } expect: - matcher.matches(UpstreamsConfig.Labels.fromMap(maps)) + matcher.matches(up) where: maps << [ @@ -193,9 +196,12 @@ class SelectorSpec extends Specification { def "LABEL matches one label two values"() { setup: def matcher = new Selector.LabelMatcher("test", ["foo", "bar"]) + def up = Mock(Upstream) { + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)] + } expect: - matcher.matches(UpstreamsConfig.Labels.fromMap(maps)) + matcher.matches(up) where: maps << [ @@ -212,9 +218,12 @@ class SelectorSpec extends Specification { new Selector.LabelMatcher("test", ["foo", "bar"]) ] ) + def up = Mock(Upstream) { + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)] + } expect: - matcher.matches(UpstreamsConfig.Labels.fromMap(maps)) + matcher.matches(up) where: maps << [ @@ -232,9 +241,12 @@ class SelectorSpec extends Specification { new Selector.LabelMatcher("test2", ["baz"]) ] ) + def up = Mock(Upstream) { + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)] + } expect: - matcher.matches(UpstreamsConfig.Labels.fromMap(maps)) + matcher.matches(up) where: maps << [ @@ -252,9 +264,12 @@ class SelectorSpec extends Specification { new Selector.LabelMatcher("test2", ["baz"]) ] ) + def up = Mock(Upstream) { + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)] + } expect: - matcher.matches(UpstreamsConfig.Labels.fromMap(maps)) + matcher.matches(up) where: maps << [ @@ -277,9 +292,12 @@ class SelectorSpec extends Specification { ) ] ) + def up = Mock(Upstream) { + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)] + } expect: - matcher.matches(UpstreamsConfig.Labels.fromMap(maps)) + matcher.matches(up) where: maps << [ diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy index 3f047a27..d81a6b26 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -5,9 +5,9 @@ import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CurrentBlockCache import io.emeraldpay.dshackle.data.DefaultContainer +import io.emeraldpay.dshackle.quorum.QuorumReader import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.quorum.QuorumRpcReader -import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.ApiSource import io.emeraldpay.dshackle.upstream.Head @@ -56,7 +56,7 @@ class EthereumDirectReaderSpec extends Specification { up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _,) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null) @@ -86,7 +86,7 @@ class EthereumDirectReaderSpec extends Specification { up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers, null @@ -124,7 +124,7 @@ class EthereumDirectReaderSpec extends Specification { up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null @@ -160,7 +160,7 @@ class EthereumDirectReaderSpec extends Specification { up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null @@ -196,7 +196,7 @@ class EthereumDirectReaderSpec extends Specification { up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList(), null @@ -233,7 +233,7 @@ class EthereumDirectReaderSpec extends Specification { up, caches, new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList(), null @@ -261,7 +261,7 @@ class EthereumDirectReaderSpec extends Specification { up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers, null @@ -292,7 +292,7 @@ class EthereumDirectReaderSpec extends Specification { up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers, null @@ -324,7 +324,7 @@ class EthereumDirectReaderSpec extends Specification { up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_getBalance", [address1, "0xa8c9bb"])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers, null @@ -365,11 +365,11 @@ class EthereumDirectReaderSpec extends Specification { up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) ethereumDirectReader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 2 * create(_, _, _, _) >> Mock(Reader) { + 2 * create(_, _, _, _) >> Mock(QuorumReader) { 2 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >>> [Mono.error(new RuntimeException()), Mono.error(new RuntimeException())] } - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> result } } @@ -408,11 +408,11 @@ class EthereumDirectReaderSpec extends Specification { up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) ethereumDirectReader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 2 * create(_, _, _, _) >> Mock(Reader) { + 2 * create(_, _, _, _) >> Mock(QuorumReader) { 2 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >>> [Mono.error(new RuntimeException()), Mono.error(new RuntimeException())] } - 1 * create(_, _, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(QuorumReader) { 1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> result } } @@ -442,7 +442,7 @@ class EthereumDirectReaderSpec extends Specification { up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 4 * create(_, _, _, _) >> Mock(Reader) { + 4 * create(_, _, _, _) >> Mock(QuorumReader) { 4 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >>> [Mono.error(new RuntimeException()), Mono.error(new RuntimeException()), Mono.error(new RuntimeException()), Mono.error(new RuntimeException())] diff --git a/src/test/kotlin/io/emeraldpay/dshackle/config/spans/CollectSpanConfigTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/config/spans/CollectSpanConfigTest.kt index 473021ea..f0fca008 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/config/spans/CollectSpanConfigTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/config/spans/CollectSpanConfigTest.kt @@ -37,7 +37,7 @@ class CollectSpanConfigTest { appCtx.getBean(SpanConfig::class.java) } assertThrows(NoSuchBeanDefinitionException::class.java) { - appCtx.getBean(ErrorSpanHandler::class.java) + appCtx.getBean(ProviderSpanHandler::class.java) } assertThrows(NoSuchBeanDefinitionException::class.java) { appCtx.getBean(ServerSpansInterceptor::class.java) @@ -72,7 +72,7 @@ class CollectSpanConfigTest { appCtx.getBean(SpanConfig::class.java) } assertDoesNotThrow { - appCtx.getBean(ErrorSpanHandler::class.java) + appCtx.getBean(ProviderSpanHandler::class.java) } assertDoesNotThrow { appCtx.getBean(ServerSpansInterceptor::class.java) @@ -113,7 +113,7 @@ class CollectSpanConfigTest { appCtx.getBean(SpanConfig::class.java) } assertThrows(NoSuchBeanDefinitionException::class.java) { - appCtx.getBean(ErrorSpanHandler::class.java) + appCtx.getBean(ProviderSpanHandler::class.java) } assertThrows(NoSuchBeanDefinitionException::class.java) { appCtx.getBean(ServerSpansInterceptor::class.java) diff --git a/src/test/kotlin/io/emeraldpay/dshackle/config/spans/ErrorSpanHandlerTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/config/spans/ProviderSpanHandlerTest.kt similarity index 73% rename from src/test/kotlin/io/emeraldpay/dshackle/config/spans/ErrorSpanHandlerTest.kt rename to src/test/kotlin/io/emeraldpay/dshackle/config/spans/ProviderSpanHandlerTest.kt index db18703b..3b9451aa 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/config/spans/ErrorSpanHandlerTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/config/spans/ProviderSpanHandlerTest.kt @@ -5,16 +5,20 @@ import brave.handler.SpanHandler import brave.propagation.TraceContext import com.fasterxml.jackson.module.kotlin.readValue import io.emeraldpay.dshackle.commons.SPAN_ERROR +import io.emeraldpay.dshackle.commons.SPAN_NO_RESPONSE_MESSAGE import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.MethodSource import org.mockito.Mockito import org.mockito.Mockito.`when` import org.springframework.cloud.sleuth.Span import org.springframework.cloud.sleuth.brave.bridge.BraveTraceContext -class ErrorSpanHandlerTest { +class ProviderSpanHandlerTest { private val mapper = SpanConfig().spanMapper() + private val spanExportableList = listOf(ErrorSpanExportable(), NoResponseSpanExportable()) private val ctx = TraceContext.newBuilder() .traceId(1223324) .spanId(234235) @@ -48,13 +52,11 @@ class ErrorSpanHandlerTest { assertEquals("", result) } - @Test - fun `span with length of traceId greater than 20 and with parentId is collected`() { - val spanId = "f7e83f2b69ec684d" + @ParameterizedTest + @MethodSource("spans") + fun `span with length of traceId greater than 20 and with parentId is collected`(span: MutableSpan) { val currentSpan = Mockito.mock(Span::class.java) val handler = spanHandler() - val span = span("6666632728347823749827349723985", spanId) - .apply { parentId("f7e83f2b69ec682d") } `when`(currentSpan.context()).thenReturn(BraveTraceContext(ctx)) @@ -92,5 +94,25 @@ class ErrorSpanHandlerTest { tag(SPAN_ERROR, "true") } - private fun spanHandler() = ErrorSpanHandler(mapper) + companion object { + @JvmStatic + fun spans() = listOf( + MutableSpan() + .apply { + traceId("6666632728347823749827349723985") + id("f7e83f2b69ec682d") + tag(SPAN_ERROR, "true") + parentId("f7e83f2b69ec682d") + }, + MutableSpan() + .apply { + traceId("6666632728347823749827349723985") + id("f7e83f2b69ec111d") + parentId("f7e83f2b69ec682d") + tag(SPAN_NO_RESPONSE_MESSAGE, "noResp") + } + ) + } + + private fun spanHandler() = ProviderSpanHandler(mapper, spanExportableList) }