From cf43f4128cf5333ecd671a7bf94aac28d2f8cb9e Mon Sep 17 00:00:00 2001 From: Nikolay G Date: Fri, 12 Jan 2024 15:53:53 +0000 Subject: [PATCH] Add logs and tracer span for logsoracle (#392) --- .../dshackle/config/context/MultistreamsConfig.kt | 1 + .../io/emeraldpay/dshackle/upstream/LogsOracle.kt | 11 +++++++++-- .../upstream/ethereum/EthereumLocalReader.kt | 6 +++--- .../dshackle/upstream/generic/GenericMultistream.kt | 4 +++- .../dshackle/test/MultistreamHolderMock.groovy | 6 ++++-- .../emeraldpay/dshackle/test/TestingCommons.groovy | 9 ++++++--- .../dshackle/upstream/MultistreamSpec.groovy | 12 ++++++------ .../dshackle/config/reload/ReloadConfigTest.kt | 1 + 8 files changed, 33 insertions(+), 17 deletions(-) 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 dc0a299d..85494515 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt @@ -74,6 +74,7 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) cs.subscriptionBuilder(headScheduler), logsOracleConfig, logsOracleScheduler, + tracer, ).also { register(it, name) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LogsOracle.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/LogsOracle.kt index 87774a13..13b4e564 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LogsOracle.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/LogsOracle.kt @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.config.IndexConfig import org.slf4j.LoggerFactory +import org.springframework.cloud.sleuth.Tracer import reactor.core.Disposable import reactor.core.publisher.Mono import reactor.core.scheduler.Scheduler @@ -10,6 +11,7 @@ class LogsOracle( private val config: IndexConfig.Index, private val upstream: Multistream, private val scheduler: Scheduler, + private val tracer: Tracer, ) { private val log = LoggerFactory.getLogger(LogsOracle::class.java) @@ -18,7 +20,7 @@ class LogsOracle( private var conn: org.drpc.logsoracle.LogsOracle? = null fun start() { - log.info("liboracle starting") + log.info("liboracle starting: store=${config.store}, ram=${config.ram_limit}") conn = org.drpc.logsoracle.LogsOracle(config.store, config.ram_limit ?: 0L) subscription = upstream.getHead().getFlux() @@ -46,8 +48,11 @@ class LogsOracle( address: List, topics: List>, ): Mono { + val requestSpan = tracer.currentSpan() + return Mono.fromCallable { - log.info("query: from=$fromBlock, to=$toBlock") + val span = tracer.nextSpan(requestSpan).name("emerald.blockchain/logsoracle").start() + log.info("query: from=$fromBlock, to=$toBlock, limit=$limit") try { val estimate = conn?.Query(limit, fromBlock, toBlock, address, topics) @@ -58,6 +63,8 @@ class LogsOracle( } else { throw e } + } finally { + span.end() } } .publishOn(scheduler) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt index e4fdde35..296cffa6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt @@ -183,7 +183,7 @@ class EthereumLocalReader( val req = params[0] as LinkedHashMap - val limit = try { req.get("limit") as Integer? } catch (_: IllegalArgumentException) { + val limit = try { req.get("limit") as Int? } catch (_: IllegalArgumentException) { throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Invalid 'limit' parameter") } @@ -205,7 +205,7 @@ class EthereumLocalReader( throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Invalid 'address' parameter") } val topics: List> = try { - val tpcs = req.get("topics")?.let { it as List } ?: listOf() + val tpcs = req.get("topics")?.let { it as List } ?: listOf() if (tpcs.size > 4) { throw IllegalArgumentException() } @@ -223,7 +223,7 @@ class EthereumLocalReader( throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Invalid 'topics' parameter") } - return logsOracle.estimate(limit?.toLong() ?: null, fromBlock, toBlock, address, topics) + return logsOracle.estimate(limit?.toLong(), fromBlock, toBlock, address, topics) .map { it.toByteArray() to null } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt index 8f97e601..3bf1742c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt @@ -39,6 +39,7 @@ import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.calls.CallSelector import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream +import org.springframework.cloud.sleuth.Tracer import org.springframework.util.ConcurrentReferenceHashMap import org.springframework.util.ConcurrentReferenceHashMap.ReferenceType.WEAK import reactor.core.publisher.Flux @@ -57,6 +58,7 @@ open class GenericMultistream( private val subscriptionBuilder: SubscriptionBuilder, logsOracleConfig: IndexConfig.Index? = null, private val logsOracleScheduler: Scheduler, + private val tracer: Tracer, ) : Multistream(chain, caches, callSelector, multistreamEventsScheduler) { private val cachingReader = cachingReaderBuilder(this, caches, getMethodsFactory()) @@ -76,7 +78,7 @@ open class GenericMultistream( ) private val logsOracle: LogsOracle? = logsOracleConfig?.let { - LogsOracle(logsOracleConfig, this, logsOracleScheduler) + LogsOracle(logsOracleConfig, this, logsOracleScheduler, tracer) } private var subscription: EgressSubscription = subscriptionBuilder(this) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index d88a8403..933b4ba4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -52,7 +52,8 @@ class MultistreamHolderMock implements MultistreamHolder { EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), - null, Schedulers.immediate() + null, Schedulers.immediate(), + TestingCommons.tracerMock() ) upstreams[chain].addUpstream(up) } else { @@ -105,7 +106,8 @@ class MultistreamHolderMock implements MultistreamHolder { super(chain, Schedulers.immediate(), null, upstreams, caches, Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(new BraveTracer(null, null, null)), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate(), + 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 288cddf2..a9a10a1e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -99,7 +99,8 @@ class TestingCommons { EthereumChainSpecific.INSTANCE.&localReaderBuilder, EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, - Schedulers.immediate() + Schedulers.immediate(), + tracerMock() ).tap { it.processUpstreamsEvents( new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up, UpstreamChangeEvent.ChangeType.ADDED) @@ -128,7 +129,8 @@ class TestingCommons { EthereumChainSpecific.INSTANCE.&localReaderBuilder, EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, - Schedulers.immediate()) + Schedulers.immediate(), + tracerMock()) } static Multistream multistreamClassicWithoutUpstreams(Chain chain) { @@ -137,7 +139,8 @@ class TestingCommons { EthereumChainSpecific.INSTANCE.&localReaderBuilder, EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, - Schedulers.immediate()) + Schedulers.immediate(), + tracerMock()) } static FileResolver fileResolver() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index 594b00c7..133d6413 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -59,7 +59,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate(), TestingCommons.tracerMock()) when: aggr.onUpstreamsUpdated() def act = aggr.getMethods() @@ -194,7 +194,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate(), TestingCommons.tracerMock()) expect: multistream.getHead(new Selector.LabelMatcher("provider", ["internal"])).is(up1.ethereumHeadMock) @@ -267,7 +267,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate(), TestingCommons.tracerMock()) when: ms.processUpstreamsEvents( new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED) @@ -298,7 +298,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate(), TestingCommons.tracerMock()) def head1 = createBlock(250, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448401b") def head2 = createBlock(270, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448402b") def head3 = createBlock(100, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448412b") @@ -333,7 +333,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate(), TestingCommons.tracerMock()) multistream.processUpstreamsEvents( new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED) ) @@ -371,7 +371,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - StarknetChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) + StarknetChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate(), TestingCommons.tracerMock()) } @NotNull diff --git a/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt index 13af79d6..4675a5a4 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt @@ -181,6 +181,7 @@ class ReloadConfigTest { cs.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.fromExecutor(Executors.newFixedThreadPool(6)), + mock(), ) }