Add logs and tracer span for logsoracle (#392)

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) }
     }
 

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<String>,
         topics: List<List<String>>,
     ): Mono<String> {
+        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)

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt

@@ -183,7 +183,7 @@ class EthereumLocalReader(
 
         val req = params[0] as LinkedHashMap<String, Any?>
 
-        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<List<String>> = try {
-            val tpcs = req.get("topics")?.let { it as List<Any> } ?: listOf<Any>()
+            val tpcs = req.get("topics")?.let { it as List<Any?> } ?: listOf<Any?>()
             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 }
     }
 

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)

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<GenericUpstream> upstreams) {

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

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

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<Tracer>(),
         )
     }