From 01ea756e33419bd1c9773ff84d480384ebd161fd Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Tue, 7 Mar 2023 13:42:34 +0400 Subject: [PATCH] Add SpanReader wrapper for creating additional spans (#155) --- .../emeraldpay/dshackle/commons/constatnts.kt | 20 ++++++ .../config/context/MultistreamsConfig.kt | 20 ++++-- .../dshackle/quorum/QuorumReaderFactory.kt | 7 +- .../dshackle/quorum/QuorumRpcReader.kt | 15 ++++- .../dshackle/reader/SpannedReader.kt | 57 +++++++++++++++++ .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 21 ++++-- .../ethereum/EthereumCachingReader.kt | 36 +++++++---- .../upstream/ethereum/EthereumDirectReader.kt | 7 +- .../upstream/ethereum/EthereumMultistream.kt | 6 +- .../ethereum_pos/EthereumPosMultiStream.kt | 6 +- .../quorum/QuorumRpcReaderSpec.groovy | 19 +++--- .../dshackle/rpc/NativeCallSpec.groovy | 10 +-- .../test/MultistreamHolderMock.groovy | 15 +++-- .../dshackle/test/TestingCommons.groovy | 10 ++- .../dshackle/test/TracerMock.groovy | 64 +++++++++++++++++++ .../dshackle/upstream/MultistreamSpec.groovy | 6 +- .../ethereum/EthereumCachingReaderSpec.groovy | 49 +++++++------- .../ethereum/EthereumLocalReaderSpec.groovy | 6 +- 18 files changed, 283 insertions(+), 91 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/SpannedReader.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/test/TracerMock.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt b/src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt new file mode 100644 index 00000000..06520a2c --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt @@ -0,0 +1,20 @@ +package io.emeraldpay.dshackle.commons + +const val SPAN_READER_NAME = "reader.name" +const val SPAN_REQUEST_INFO = "request.info" +const val SPAN_ERROR = "error" +const val SPAN_STATUS_MESSAGE = "status.message" +const val SPAN_READER_RESULT = "reader.result" +const val SPAN_REQUEST_API_TYPE = "request.api.type" +const val SPAN_REQUEST_UPSTREAM_ID = "request.upstreamId" +const val SPAN_REQUEST_ID = "request.id" + +const val LOCAL_READER = "localReader" +const val REMOTE_QUORUM_RPC_READER = "remoteQuorumRpcReader" +const val API_READER = "apiReader" +const val CACHE_BLOCK_BY_HASH_READER = "cacheBlockByHashReader" +const val DIRECT_QUORUM_RPC_READER = "directQuorumRpcReader" +const val CACHE_HEIGHT_BY_HASH_READER = "cacheHeightByHashReader" +const val CACHE_BLOCK_BY_HEIGHT_READER = "cacheBlockByHeightReader" +const val CACHE_TX_BY_HASH_READER = "cacheTxByHashReader" +const val CACHE_RECEIPTS_READER = "cacheReceiptsReader" diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt index 63a43b18..59e0665c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt @@ -10,6 +10,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream import org.springframework.beans.factory.annotation.Qualifier import org.springframework.beans.factory.config.ConfigurableListableBeanFactory +import org.springframework.cloud.sleuth.Tracer import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import reactor.core.scheduler.Scheduler @@ -21,14 +22,15 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) cachesFactory: CachesFactory, callTargetsHolder: CallTargetsHolder, @Qualifier("headMergedScheduler") - headScheduler: Scheduler + headScheduler: Scheduler, + tracer: Tracer ): List { return Chain.values() .filterNot { it == Chain.UNSPECIFIED } .mapNotNull { chain -> when (BlockchainType.from(chain)) { - BlockchainType.EVM_POS -> ethereumPosMultistream(chain, cachesFactory, headScheduler) - BlockchainType.EVM_POW -> ethereumMultistream(chain, cachesFactory, headScheduler) + BlockchainType.EVM_POS -> ethereumPosMultistream(chain, cachesFactory, headScheduler, tracer) + BlockchainType.EVM_POW -> ethereumMultistream(chain, cachesFactory, headScheduler, tracer) BlockchainType.BITCOIN -> bitcoinMultistream(chain, cachesFactory) else -> null } @@ -38,7 +40,8 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) private fun ethereumMultistream( chain: Chain, cachesFactory: CachesFactory, - headScheduler: Scheduler + headScheduler: Scheduler, + tracer: Tracer ): EthereumMultistream { val name = "multi-ethereum-$chain" @@ -46,14 +49,16 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) chain, ArrayList(), cachesFactory.getCaches(chain), - headScheduler + headScheduler, + tracer ).also { register(it, name) } } open fun ethereumPosMultistream( chain: Chain, cachesFactory: CachesFactory, - headScheduler: Scheduler + headScheduler: Scheduler, + tracer: Tracer ): EthereumPosMultiStream { val name = "multi-ethereum-pos-$chain" @@ -61,7 +66,8 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) chain, ArrayList(), cachesFactory.getCaches(chain), - headScheduler + headScheduler, + tracer ).also { register(it, name) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt index 5a18d0ef..930fbb27 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt @@ -19,6 +19,7 @@ import io.emeraldpay.dshackle.reader.Reader 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 // creates instance of a Quorum based reader interface QuorumReaderFactory { @@ -29,11 +30,11 @@ interface QuorumReaderFactory { } } - fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?): Reader + fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): Reader class Default : QuorumReaderFactory { - override fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?): Reader { - return QuorumRpcReader(apis, quorum, signer) + override fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): Reader { + 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 d3df3e8a..1b1e7c9d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt @@ -15,7 +15,11 @@ */ package io.emeraldpay.dshackle.quorum +import io.emeraldpay.dshackle.commons.API_READER +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 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError @@ -25,6 +29,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import io.emeraldpay.etherjar.rpc.RpcException import org.slf4j.LoggerFactory +import org.springframework.cloud.sleuth.Tracer import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.util.function.Tuple3 @@ -42,13 +47,14 @@ class QuorumRpcReader( private val apiControl: ApiSource, private val quorum: CallQuorum, private val signer: ResponseSigner?, + private val tracer: Tracer ) : Reader { companion object { private val log = LoggerFactory.getLogger(QuorumRpcReader::class.java) } - constructor(apiControl: ApiSource, quorum: CallQuorum) : this(apiControl, quorum, null) + constructor(apiControl: ApiSource, quorum: CallQuorum, tracer: Tracer) : this(apiControl, quorum, null, tracer) override fun read(key: JsonRpcRequest): Mono { // needs at least one response, so start a request @@ -129,7 +135,12 @@ class QuorumRpcReader( } fun callApi(api: Upstream, key: JsonRpcRequest): Mono, Upstream, Optional>> { - return api.getIngressReader() + val apiReader = api.getIngressReader() + val spanParams = mapOf( + SPAN_REQUEST_API_TYPE to apiReader.javaClass.name, + SPAN_REQUEST_UPSTREAM_ID to api.getId() + ) + return SpannedReader(apiReader, tracer, API_READER, spanParams) .read(key) .flatMap { response -> log.debug("Received response from upstream ${api.getId()} for method ${key.method}") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/SpannedReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/SpannedReader.kt new file mode 100644 index 00000000..71ec09ed --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/SpannedReader.kt @@ -0,0 +1,57 @@ +package io.emeraldpay.dshackle.reader + +import io.emeraldpay.dshackle.commons.SPAN_ERROR +import io.emeraldpay.dshackle.commons.SPAN_READER_NAME +import io.emeraldpay.dshackle.commons.SPAN_READER_RESULT +import io.emeraldpay.dshackle.commons.SPAN_REQUEST_INFO +import io.emeraldpay.dshackle.commons.SPAN_STATUS_MESSAGE +import io.emeraldpay.dshackle.data.HashId +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import org.springframework.cloud.sleuth.Tracer +import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth +import reactor.core.publisher.Mono +import reactor.kotlin.core.publisher.switchIfEmpty + +class SpannedReader( + private val reader: Reader, + private val tracer: Tracer, + private val name: String, + private val additionalParams: Map = emptyMap() +) : Reader { + + override fun read(key: K): Mono { + val newSpan = tracer.nextSpan(tracer.currentSpan()) + .name(reader.javaClass.name) + .tag(SPAN_READER_NAME, name) + .start() + + extractInfoFromKey(key)?.let { + newSpan.tag(SPAN_REQUEST_INFO, it) + } + additionalParams.forEach { newSpan.tag(it.key, it.value) } + + return reader.read(key) + .contextWrite { ReactorSleuth.putSpanInScope(tracer, it, newSpan) } + .doOnError { + newSpan.apply { + tag(SPAN_ERROR, "true") + tag(SPAN_STATUS_MESSAGE, it.message) + end() + } + } + .doOnNext { newSpan.end() } + .switchIfEmpty { + newSpan.tag(SPAN_READER_RESULT, "empty result") + newSpan.end() + Mono.empty() + } + } + + private fun extractInfoFromKey(key: K): String? { + return when (key) { + is JsonRpcRequest -> "method: ${key.method}" + is HashId, Long -> "params: $key" + else -> null + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 8dfa38d6..91ca1744 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -23,11 +23,17 @@ import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.SilentException +import io.emeraldpay.dshackle.commons.LOCAL_READER +import io.emeraldpay.dshackle.commons.REMOTE_QUORUM_RPC_READER +import io.emeraldpay.dshackle.commons.SPAN_ERROR +import io.emeraldpay.dshackle.commons.SPAN_REQUEST_ID +import io.emeraldpay.dshackle.commons.SPAN_STATUS_MESSAGE import io.emeraldpay.dshackle.config.MainConfig 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.reader.SpannedReader import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.ApiSource import io.emeraldpay.dshackle.upstream.Multistream @@ -137,8 +143,8 @@ open class NativeCall( private fun errorSpan(span: Span?, message: String) { span?.apply { - tag("error", "true") - tag("status.message", message) + tag(SPAN_ERROR, "true") + tag(SPAN_STATUS_MESSAGE, message) } } @@ -151,7 +157,7 @@ open class NativeCall( if (requestCount > 1) { val span = tracer.nextSpan(requestSpan) .name(requestId) - .tag("request.id", requestId) + .tag(SPAN_REQUEST_ID, requestId) .start() return ReactorSleuth.putSpanInScope(tracer, ctx, span) } @@ -170,7 +176,7 @@ open class NativeCall( return@run Mono.error(e) } if (callContext.requestCount == 1 && callContext.requestId.isNotBlank()) { - requestSpan?.tag("request.id", callContext.requestId) + requestSpan?.tag(SPAN_REQUEST_ID, callContext.requestId) } this.fetch(parsed) .doOnError { e -> log.warn("Error during native call: ${e.message}") } @@ -358,7 +364,8 @@ open class NativeCall( fun fetch(ctx: ValidCallContext): Mono { return ctx.upstream.getLocalReader(localRouterEnabled) .flatMap { api -> - api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector)) + SpannedReader(api, tracer, LOCAL_READER) + .read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector)) .flatMap(JsonRpcResponse::requireResult) .map { validateResult(it, "local", ctx) @@ -381,14 +388,14 @@ open class NativeCall( if (!ctx.upstream.getMethods().isCallable(ctx.payload.method)) { return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method")) } - val reader = quorumReaderFactory.create(ctx.getApis(), ctx.callQuorum, signer) + val reader = quorumReaderFactory.create(ctx.getApis(), ctx.callQuorum, signer, tracer) val counter = if (reader is QuorumRpcReader) { reader.getValidAttemptsCount() } else { AtomicInteger(-1) } - return reader + 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) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt index ac77713a..57dff0b3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt @@ -20,6 +20,12 @@ import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CurrentBlockCache import io.emeraldpay.dshackle.cache.HeightByHashAdding +import io.emeraldpay.dshackle.commons.CACHE_BLOCK_BY_HASH_READER +import io.emeraldpay.dshackle.commons.CACHE_BLOCK_BY_HEIGHT_READER +import io.emeraldpay.dshackle.commons.CACHE_HEIGHT_BY_HASH_READER +import io.emeraldpay.dshackle.commons.CACHE_RECEIPTS_READER +import io.emeraldpay.dshackle.commons.CACHE_TX_BY_HASH_READER +import io.emeraldpay.dshackle.commons.DIRECT_QUORUM_RPC_READER import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.SourceContainer @@ -28,6 +34,7 @@ import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.reader.CompoundReader import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.RekeyingReader +import io.emeraldpay.dshackle.reader.SpannedReader import io.emeraldpay.dshackle.reader.TransformingReader import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Multistream @@ -41,6 +48,7 @@ import io.emeraldpay.etherjar.rpc.json.TransactionJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.apache.commons.collections4.Factory import org.slf4j.LoggerFactory +import org.springframework.cloud.sleuth.Tracer import java.util.function.Function /** @@ -49,7 +57,8 @@ import java.util.function.Function open class EthereumCachingReader( private val up: Multistream, private val caches: Caches, - private val callMethodsFactory: Factory + private val callMethodsFactory: Factory, + private val tracer: Tracer ) : Lifecycle { companion object { @@ -58,7 +67,7 @@ open class EthereumCachingReader( private val objectMapper: ObjectMapper = Global.objectMapper private val balanceCache = CurrentBlockCache() - private val directReader = EthereumDirectReader(up, caches, balanceCache, callMethodsFactory) + private val directReader = EthereumDirectReader(up, caches, balanceCache, callMethodsFactory, tracer) val extractBlock = Function> { block -> val existing = block.getParsed(BlockJson::class.java) @@ -86,16 +95,17 @@ open class EthereumCachingReader( private val idToTxHash = Function { id -> TransactionId.from(id.value) } private val blocksByIdAsCont = CompoundReader( - caches.getBlocksByHash(), - RekeyingReader(idToBlockHash, directReader.blockReader) + SpannedReader(caches.getBlocksByHash(), tracer, CACHE_BLOCK_BY_HASH_READER), + SpannedReader(RekeyingReader(idToBlockHash, directReader.blockReader), tracer, DIRECT_QUORUM_RPC_READER) ) - private val heightByHash = HeightByHashAdding(caches, blocksByIdAsCont) + private val heightByHash = + SpannedReader(HeightByHashAdding(caches, blocksByIdAsCont), tracer, CACHE_HEIGHT_BY_HASH_READER) fun blocksByHashAsCont(): Reader { return CompoundReader( - RekeyingReader(blockHashToId, caches.getBlocksByHash()), - directReader.blockReader + SpannedReader(RekeyingReader(blockHashToId, caches.getBlocksByHash()), tracer, CACHE_BLOCK_BY_HASH_READER), + SpannedReader(directReader.blockReader, tracer, DIRECT_QUORUM_RPC_READER) ) } @@ -119,8 +129,8 @@ open class EthereumCachingReader( open fun blocksByHeightAsCont(): Reader { return CompoundReader( - caches.getBlocksByHeight(), - directReader.blockByHeightReader + SpannedReader(caches.getBlocksByHeight(), tracer, CACHE_BLOCK_BY_HEIGHT_READER), + SpannedReader(directReader.blockByHeightReader, tracer, DIRECT_QUORUM_RPC_READER) ) } @@ -143,8 +153,8 @@ open class EthereumCachingReader( open fun txByHashAsCont(): Reader { return CompoundReader( - caches.getTxByHash(), - RekeyingReader(idToTxHash, directReader.txReader) + SpannedReader(caches.getTxByHash(), tracer, CACHE_TX_BY_HASH_READER), + SpannedReader(RekeyingReader(idToTxHash, directReader.txReader), tracer, DIRECT_QUORUM_RPC_READER) ) } @@ -161,8 +171,8 @@ open class EthereumCachingReader( directReader.receiptReader ) return CompoundReader( - caches.getReceipts(), - requested + SpannedReader(caches.getReceipts(), tracer, CACHE_RECEIPTS_READER), + SpannedReader(requested, tracer, DIRECT_QUORUM_RPC_READER) ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt index 39fac1bc..bc8c10a8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -29,6 +29,7 @@ import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.apache.commons.collections4.Factory import org.slf4j.LoggerFactory +import org.springframework.cloud.sleuth.Tracer import reactor.core.publisher.Mono import reactor.util.retry.Retry import java.time.Duration @@ -41,7 +42,8 @@ class EthereumDirectReader( private val up: Multistream, private val caches: Caches, private val balanceCache: CurrentBlockCache, - private val callMethodsFactory: Factory + private val callMethodsFactory: Factory, + private val tracer: Tracer ) { companion object { @@ -175,7 +177,8 @@ class EthereumDirectReader( up.getApiSource(matcher), callMethodsFactory.create().createQuorumFor(request.method), // we do not use Signer for internal requests because it doesn't make much sense - null + null, + tracer ) }.flatMap { it.read(request) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index f36d728b..89ee1663 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -39,6 +39,7 @@ import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream import org.slf4j.LoggerFactory +import org.springframework.cloud.sleuth.Tracer import org.springframework.util.ConcurrentReferenceHashMap import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -49,7 +50,8 @@ open class EthereumMultistream( chain: Chain, val upstreams: MutableList, caches: Caches, - headScheduler: Scheduler + headScheduler: Scheduler, + tracer: Tracer ) : Multistream(chain, upstreams as MutableList, caches), EthereumLikeMultistream { companion object { @@ -65,7 +67,7 @@ open class EthereumMultistream( private val filteredHeads: MutableMap = ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK) - private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory()) + private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory(), tracer) private var subscribe = EthereumEgressSubscription(this, NoPendingTxes()) private val supportsEIP1559 = when (chain) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index 99121b10..cd754ebd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -37,6 +37,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream import org.slf4j.LoggerFactory +import org.springframework.cloud.sleuth.Tracer import org.springframework.util.ConcurrentReferenceHashMap import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -47,7 +48,8 @@ open class EthereumPosMultiStream( chain: Chain, val upstreams: MutableList, caches: Caches, - headScheduler: Scheduler + headScheduler: Scheduler, + tracer: Tracer ) : Multistream(chain, upstreams as MutableList, caches), EthereumLikeMultistream { companion object { @@ -60,7 +62,7 @@ open class EthereumPosMultiStream( headScheduler ) - private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory()) + private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory(), tracer) private var subscribe = EthereumEgressSubscription(this, NoPendingTxes()) private val feeEstimation = EthereumPriorityFees(this, reader, 256) private val filteredHeads: MutableMap = diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy index 9e98aec4..6f8463e8 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy @@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.etherjar.rpc.RpcResponseError +import org.springframework.cloud.sleuth.Tracer import reactor.core.publisher.Mono import reactor.test.StepVerifier import spock.lang.Specification @@ -47,7 +48,7 @@ class QuorumRpcReaderSpec extends Specification { Chain.ETHEREUM, [up], Selector.empty ) - def reader = new QuorumRpcReader(apis, new AlwaysQuorum()) + def reader = new QuorumRpcReader(apis, new AlwaysQuorum(), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_test", [])) @@ -78,7 +79,7 @@ class QuorumRpcReaderSpec extends Specification { Chain.ETHEREUM, [up], Selector.empty ) - def reader = new QuorumRpcReader(apis, new AlwaysQuorum()) + def reader = new QuorumRpcReader(apis, new AlwaysQuorum(), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_test", [])) @@ -115,7 +116,7 @@ class QuorumRpcReaderSpec extends Specification { Chain.ETHEREUM, [up], Selector.empty ) - def reader = new QuorumRpcReader(apis, new AlwaysQuorum()) + def reader = new QuorumRpcReader(apis, new AlwaysQuorum(), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_test", [])) @@ -147,7 +148,7 @@ class QuorumRpcReaderSpec extends Specification { Chain.ETHEREUM, [up], Selector.empty ) - def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3)) + def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_test", [])) @@ -179,7 +180,7 @@ class QuorumRpcReaderSpec extends Specification { Chain.ETHEREUM, [up], Selector.empty ) - def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3)) + def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_test", [])) @@ -211,7 +212,7 @@ class QuorumRpcReaderSpec extends Specification { Chain.ETHEREUM, [up], Selector.empty ) - def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3)) + def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_test", [])) @@ -244,7 +245,7 @@ class QuorumRpcReaderSpec extends Specification { Chain.ETHEREUM, [up], Selector.empty ) - def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3)) + def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_test", [])) @@ -274,7 +275,7 @@ class QuorumRpcReaderSpec extends Specification { Chain.ETHEREUM, [up], Selector.empty ) - def reader = new QuorumRpcReader(apis, new AlwaysQuorum()) + def reader = new QuorumRpcReader(apis, new AlwaysQuorum(), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_test", [])) @@ -307,7 +308,7 @@ class QuorumRpcReaderSpec extends Specification { Chain.ETHEREUM, [up], Selector.empty ) - def reader = new QuorumRpcReader(apis, new NotLaggingQuorum(1)) + def reader = new QuorumRpcReader(apis, new NotLaggingQuorum(1), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_test", [])) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 51894f9a..ac090b50 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -131,7 +131,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall() nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, null, 1, Collections.singletonList(ups), null)) } } @@ -152,7 +152,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall() nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_test", [], 10)) >> Mono.empty() } } @@ -176,7 +176,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall() nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 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 +613,7 @@ class NativeCallSpec extends Specification { } def nativeCall = nativeCall(multistreamHolder) nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups), null)) } } @@ -648,7 +648,7 @@ class NativeCallSpec extends Specification { } def nativeCall = nativeCall(multistreamHolder) nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups), null)) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index a6ad2e56..797c5299 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -19,11 +19,7 @@ package io.emeraldpay.dshackle.test import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Multistream -import io.emeraldpay.dshackle.upstream.MultistreamHolder -import io.emeraldpay.dshackle.upstream.Selector -import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream import io.emeraldpay.dshackle.upstream.calls.CallMethods @@ -32,6 +28,8 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumCachingReader import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream import org.jetbrains.annotations.NotNull +import org.springframework.cloud.sleuth.Tracer +import org.springframework.cloud.sleuth.brave.bridge.BraveTracer import reactor.core.scheduler.Schedulers class MultistreamHolderMock implements MultistreamHolder { @@ -49,7 +47,10 @@ class MultistreamHolderMock implements MultistreamHolder { if (up instanceof EthereumPosMultiStream) { upstreams[chain] = up } else if (up instanceof EthereumPosRpcUpstream) { - upstreams[chain] = new EthereumPosMultiStream(chain, [up as EthereumPosRpcUpstream], Caches.default(), Schedulers.boundedElastic()) + upstreams[chain] = new EthereumPosMultiStream( + chain, [up as EthereumPosRpcUpstream], Caches.default(), + Schedulers.boundedElastic(), TestingCommons.tracerMock() + ) } else { throw new IllegalArgumentException("Unsupported upstream type ${up.class}") } @@ -97,7 +98,7 @@ class MultistreamHolderMock implements MultistreamHolder { Head customHead = null EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { - super(chain, upstreams, caches, Schedulers.boundedElastic()) + super(chain, upstreams, caches, Schedulers.boundedElastic(), new BraveTracer(null, null, null)) } EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index f84ad683..d9d509cb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -47,6 +47,10 @@ class TestingCommons { return new ApiReaderMock() } + static TracerMock tracerMock() { + return new TracerMock(null, null, null) + } + static EthereumPosRpcUpstreamMock upstream() { return new EthereumPosRpcUpstreamMock(Chain.ETHEREUM, api()) } @@ -84,7 +88,7 @@ class TestingCommons { } static Multistream multistream(EthereumPosRpcUpstreamMock up) { - return new EthereumPosMultiStream(Chain.ETHEREUM, [up], Caches.default(), Schedulers.boundedElastic()).tap { + return new EthereumPosMultiStream(Chain.ETHEREUM, [up], Caches.default(), Schedulers.boundedElastic(), tracerMock()).tap { start() } } @@ -105,11 +109,11 @@ class TestingCommons { } static Multistream multistreamWithoutUpstreams(Chain chain) { - return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic()) + return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), tracerMock()) } static Multistream multistreamClassicWithoutUpstreams(Chain chain) { - return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic()) + return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), tracerMock()) } static FileResolver fileResolver() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TracerMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TracerMock.groovy new file mode 100644 index 00000000..7d0efbbe --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TracerMock.groovy @@ -0,0 +1,64 @@ +package io.emeraldpay.dshackle.test + +import brave.Tracer +import org.springframework.cloud.sleuth.CurrentTraceContext +import org.springframework.cloud.sleuth.Span +import org.springframework.cloud.sleuth.TraceContext +import org.springframework.cloud.sleuth.brave.bridge.BraveBaggageManager +import org.springframework.cloud.sleuth.brave.bridge.BraveSpan +import org.springframework.cloud.sleuth.brave.bridge.BraveTracer +import spock.mock.DetachedMockFactory + +class TracerMock extends BraveTracer { + private final spanMock = new SpanMock(null) + private static final mockFactory = new DetachedMockFactory() + + TracerMock(Tracer tracer, CurrentTraceContext context, BraveBaggageManager braveBaggageManager) { + super(tracer, context, braveBaggageManager) + } + + @Override + Span nextSpan(Span parent) { + return spanMock + } + + @Override + Span currentSpan() { + return spanMock + } + + @Override + SpanInScope withSpan(Span span) { + return mockFactory.Stub(SpanInScope) + } + + private static class SpanMock extends BraveSpan { + SpanMock(brave.Span delegate) { + super(delegate) + } + + @Override + Span start() { + return this + } + + @Override + Span name(String name) { + return this + } + + @Override + Span tag(String key, String value) { + return this + } + + @Override + void end() { + } + + @Override + TraceContext context() { + return mockFactory.Stub(TraceContext) + } + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index bf35c197..2d9680f6 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -51,7 +51,7 @@ class MultistreamSpec extends Specification { setup: def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) def up2 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"])) - def aggr = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2], Caches.default(), Schedulers.boundedElastic()) + def aggr = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2], Caches.default(), Schedulers.boundedElastic(), TestingCommons.tracerMock()) when: aggr.onUpstreamsUpdated() def act = aggr.getMethods() @@ -187,7 +187,7 @@ class MultistreamSpec extends Specification { def up1 = TestingCommons.upstream("test-1", "internal") def up2 = TestingCommons.upstream("test-2", "external") def up3 = TestingCommons.upstream("test-3", "external") - def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default(), Schedulers.boundedElastic()) + def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default(), Schedulers.boundedElastic(), TestingCommons.tracerMock()) expect: multistream.getHead(new Selector.LabelMatcher("provider", ["internal"])).is(up1.ethereumHeadMock) @@ -255,7 +255,7 @@ class MultistreamSpec extends Specification { class TestEthereumPosMultistream extends EthereumPosMultiStream { TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { - super(chain, upstreams, caches, Schedulers.boundedElastic()) + super(chain, upstreams, caches, Schedulers.boundedElastic(), TestingCommons.tracerMock()) } @NotNull 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 2ba676d3..c76fce21 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -8,6 +8,7 @@ import io.emeraldpay.dshackle.data.DefaultContainer 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 import io.emeraldpay.dshackle.upstream.Multistream @@ -51,10 +52,10 @@ class EthereumDirectReaderSpec extends Specification { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls + up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null) @@ -81,10 +82,10 @@ class EthereumDirectReaderSpec extends Specification { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls + up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers, null @@ -116,10 +117,10 @@ class EthereumDirectReaderSpec extends Specification { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls + up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null @@ -152,10 +153,10 @@ class EthereumDirectReaderSpec extends Specification { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls + up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null @@ -188,10 +189,10 @@ class EthereumDirectReaderSpec extends Specification { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls + up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList(), null @@ -225,10 +226,10 @@ class EthereumDirectReaderSpec extends Specification { 1 * cacheReceipt(Caches.Tag.REQUESTED, { DefaultContainer data -> data.txId.toHex() == hash1.substring(2) && data.height == 100 }) } EthereumDirectReader reader = new EthereumDirectReader( - up, caches, new CurrentBlockCache(), calls + up, caches, new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList(), null @@ -253,10 +254,10 @@ class EthereumDirectReaderSpec extends Specification { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls + up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers, null @@ -284,10 +285,10 @@ class EthereumDirectReaderSpec extends Specification { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls + up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers, null @@ -316,10 +317,10 @@ class EthereumDirectReaderSpec extends Specification { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls + up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getBalance", [address1, "0xa8c9bb"])) >> Mono.just( new QuorumRpcReader.Result( Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers, null @@ -356,14 +357,14 @@ class EthereumDirectReaderSpec extends Specification { Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null) ) EthereumDirectReader ethereumDirectReader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls + up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) ethereumDirectReader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 2 * create(_, _, _) >> Mock(Reader) { + 2 * create(_, _, _, _) >> Mock(Reader) { 2 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >>> [Mono.error(new RuntimeException()), Mono.error(new RuntimeException())] } - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> result } } @@ -398,14 +399,14 @@ class EthereumDirectReaderSpec extends Specification { Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null) ) EthereumDirectReader ethereumDirectReader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls + up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) ethereumDirectReader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 2 * create(_, _, _) >> Mock(Reader) { + 2 * create(_, _, _, _) >> Mock(Reader) { 2 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >>> [Mono.error(new RuntimeException()), Mono.error(new RuntimeException())] } - 1 * create(_, _, _) >> Mock(Reader) { + 1 * create(_, _, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> result } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy index 952b7863..a3e19dac 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy @@ -26,7 +26,8 @@ class EthereumLocalReaderSpec extends Specification { new EthereumCachingReader( TestingCommons.multistream(TestingCommons.api()), Caches.default(), - ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM)) + ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM)), + TestingCommons.tracerMock() ), methods, new EmptyHead(), @@ -45,7 +46,8 @@ class EthereumLocalReaderSpec extends Specification { new EthereumCachingReader( TestingCommons.multistream(TestingCommons.api()), Caches.default(), - ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM)) + ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM)), + TestingCommons.tracerMock() ), methods, new EmptyHead(),