From 04d1334af65b361633a3a877f2a02911c0911288 Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Fri, 17 Feb 2023 16:39:35 +0400 Subject: [PATCH] Add json logging and tracing (#131) --- build.gradle | 17 ++- emerald-grpc | 2 +- gradle/libs.versions.toml | 15 +- .../io/emeraldpay/dshackle/GrpcServer.kt | 5 +- .../config/context/TraceGrpcConfiguration.kt | 18 +++ .../dshackle/quorum/QuorumRpcReader.kt | 15 +- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 137 +++++++++++++++--- .../dshackle/startup/ConfiguredUpstreams.kt | 7 +- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 5 +- src/main/resources/application.yml | 17 +++ src/main/resources/log4j2.xml | 44 ------ src/main/resources/logback-spring.xml | 52 +++++++ .../dshackle/rpc/NativeCallSpec.groovy | 31 ++-- .../startup/ConfiguredUpstreamsSpec.groovy | 18 ++- 14 files changed, 274 insertions(+), 109 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/context/TraceGrpcConfiguration.kt create mode 100644 src/main/resources/application.yml delete mode 100644 src/main/resources/log4j2.xml create mode 100644 src/main/resources/logback-spring.xml diff --git a/build.gradle b/build.gradle index c7819dd8..337157f5 100644 --- a/build.gradle +++ b/build.gradle @@ -67,9 +67,7 @@ dependencies { implementation libs.bundles.grpc implementation libs.bundles.netty implementation libs.zeromq - implementation(libs.bundles.spring.framework) { - exclude module: 'spring-boot-starter-logging' - } + implementation(libs.bundles.spring.framework) implementation libs.bundles.reactor @@ -79,8 +77,19 @@ dependencies { implementation libs.micrometer.registry.prometheus implementation libs.lettuce.core + implementation(libs.spring.cloud.starter.sleuth) { + exclude module: 'spring-security-rsa' + } + implementation libs.brave.instrumentation.grpc + implementation libs.bundles.etherjar + implementation libs.logstash.encoder + + implementation libs.janino + + implementation libs.spring.cloud.sleuth.zipkin + implementation libs.bitcoinj implementation libs.snake.yaml @@ -92,8 +101,6 @@ dependencies { implementation libs.caffeine implementation libs.javax.annotations - implementation libs.bundles.slf4j - testImplementation libs.cglib.nodep testImplementation libs.spockframework.core testImplementation libs.grpc.testing diff --git a/emerald-grpc b/emerald-grpc index 8e814567..8e74176f 160000 --- a/emerald-grpc +++ b/emerald-grpc @@ -1 +1 @@ -Subproject commit 8e814567e1a3820f22a78329ec24cb90d2fb2322 +Subproject commit 8e74176fd289c67d9f878733a855485f0ddcd0ad diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 78ef6e89..cc35c3b5 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -3,7 +3,6 @@ detekt = "1.22.0" etherjar = "0.11.1" groovy = "3.0.9" protoc = "3.21.7" -slf4j = "1.7.32" jackson = "2.11.0" grpc = "1.49.2" reactive-grpc = "1.2.0" @@ -21,6 +20,10 @@ apache-commons-collections4 = "org.apache.commons:commons-collections4:4.3" bitcoinj = "org.bitcoinj:bitcoinj-core:0.15.8" +logstash-encoder = "net.logstash.logback:logstash-logback-encoder:7.2" + +janino = "org.codehaus.janino:janino:3.1.9" + bouncycastle = "org.bouncycastle:bcprov-jdk15on:1.61" caffeine = "com.github.ben-manes.caffeine:caffeine:2.8.5" @@ -98,11 +101,6 @@ reactor-test = { module = "io.projectreactor:reactor-test", version.ref = "react reactor-grpc-stub = "com.salesforce.servicelibs:reactor-grpc-stub:1.2.0" -slf4j-api = { module = "org.slf4j:slf4j-api", version.ref = "slf4j" } -slf4j-jul = { module = "org.slf4j:jul-to-slf4j", version.ref = "slf4j" } -slf4j-jcl = { module = "org.slf4j:jcl-over-slf4j", version.ref = "slf4j" } -log4j-slf4j = "org.apache.logging.log4j:log4j-slf4j-impl:2.17.0" - snake-yaml = "org.yaml:snakeyaml:1.24" spockframework-core = "org.spockframework:spock-core:2.0-groovy-3.0" @@ -120,6 +118,10 @@ testcontainers-ganache = "io.github.ganchix:testcontainers-java-module-ganache:0 junit-jupiter = "org.junit.jupiter:junit-jupiter:5.9.1" assertj = "org.assertj:assertj-core:3.23.1" +spring-cloud-starter-sleuth = "org.springframework.cloud:spring-cloud-starter-sleuth:3.1.6" +spring-cloud-sleuth-zipkin = "org.springframework.cloud:spring-cloud-sleuth-zipkin:3.1.6" +brave-instrumentation-grpc = "io.zipkin.brave:brave-instrumentation-grpc:5.15.0" + [bundles] apache-commons = ["commons-io", "apache-commons-lang3", "apache-commons-collections4"] etherjar = ["etherjar-domain", "etherjar-hex", "etherjar-rpc-api", "etherjar-rpc-http", "etherjar-rpc-ws", "etherjar-tx", "etherjar-contract", "etherjar-erc20"] @@ -129,7 +131,6 @@ jackson = ["jackson-core", "jackson-databind", "jackson-datatype-jdk8", "jackson kotlin = ["kotlin-stdlib-jdk8", "kotlin-reflect"] netty = ["netty-common", "netty-transport", "netty-handler-core", "netty-handler-proxy", "netty-resolver-core", "netty-resolver-dns", "netty-codec-core", "netty-codec-http", "netty-codec-http2", "netty-buffer", "netty-tcnative-core", "netty-tcnative-boringssl"] reactor = ["reactor-core", "reactor-netty", "reactor-extra", "reactor-kotlin"] -slf4j = ["slf4j-api", "slf4j-jul", "slf4j-jcl", "log4j-slf4j"] spring-framework = ["spring-boot-starter", "spring-security-core", "spring-security-web", "spring-security-config"] testcontainers = ["testcontainers", "testcontainers-ganache"] junit = ["junit-jupiter", "assertj"] diff --git a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt index bd09d1eb..3b0b85cc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt @@ -39,7 +39,8 @@ open class GrpcServer( private val rpcs: List, private val mainConfig: MainConfig, private val tlsSetup: TlsSetup, - private val accessHandler: AccessHandlerGrpc + private val accessHandler: AccessHandlerGrpc, + private val grpcServerBraveInterceptor: ServerInterceptor ) { private val log = LoggerFactory.getLogger(GrpcServer::class.java) @@ -76,6 +77,8 @@ open class GrpcServer( it } + serverBuilder.intercept(grpcServerBraveInterceptor) + tlsSetup.setupServer("Native gRPC", mainConfig.tls, true)?.let { serverBuilder.sslContext(it) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/TraceGrpcConfiguration.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/TraceGrpcConfiguration.kt new file mode 100644 index 00000000..2aff6ed7 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/TraceGrpcConfiguration.kt @@ -0,0 +1,18 @@ +package io.emeraldpay.dshackle.config.context + +import brave.grpc.GrpcTracing +import brave.rpc.RpcTracing +import io.grpc.ServerInterceptor +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration + +@Configuration +open class TraceGrpcConfiguration { + + @Bean + open fun grpcTracing(rpcTracing: RpcTracing): GrpcTracing = GrpcTracing.create(rpcTracing) + + @Bean + open fun grpcServerBraveInterceptor(grpcTracing: GrpcTracing): ServerInterceptor = + grpcTracing.newServerInterceptor() +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt index 79d47a91..d3df3e8a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt @@ -92,8 +92,10 @@ class QuorumRpcReader( 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.debug("Quorum is resolved for method ${key.method}") apiControl.resolve() } else { + log.debug("Quorum needs more responses for method ${key.method}") // quorum needs more responses, so ask api controller to make another apiControl.request(1) } @@ -105,6 +107,7 @@ class QuorumRpcReader( quorum.isFailed() || quorum.isResolved() } .flatMap { api -> + log.debug("Calling upstream ${api.getId()} with method ${key.method}") callApi(api, key) } .retryWhen(retrySpec) @@ -129,6 +132,7 @@ class QuorumRpcReader( return api.getIngressReader() .read(key) .flatMap { response -> + log.debug("Received response from upstream ${api.getId()} for method ${key.method}") response.requireResult() .transform(withSignatureAndUpstream(api, key, response)) } @@ -154,6 +158,7 @@ class QuorumRpcReader( 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) // when the call failed with an error we want to notify the quorum because // it may use the error message or other details // @@ -168,8 +173,10 @@ class QuorumRpcReader( quorum.record(cleanErr, null, api,) // if it's failed after that, then we don't need more calls, stop api source if (quorum.isFailed()) { + log.debug("Quorum is failed, stop api source. Upstream ${api.getId()}, method ${key.method}") apiControl.resolve() } else { + log.debug("Received an error, trying to request next upstream") apiControl.request(1) } Mono.empty() @@ -180,10 +187,10 @@ class QuorumRpcReader( fun setupDefaultResult(key: JsonRpcRequest): Mono { return Mono.just(quorum).flatMap { q -> if (q.isFailed()) { - Mono.error( - q.getError()?.asException(JsonRpcResponse.NumberId(key.id)) - ?: JsonRpcException(JsonRpcResponse.NumberId(key.id), JsonRpcError(-32603, "Unhandled Upstream error")) - ) + 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) } else { log.warn("Did not get any result from upstream. Method [${key.method}] using [$q]") Mono.empty() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 0c82cdd8..07298e96 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -48,11 +48,15 @@ import io.emeraldpay.etherjar.rpc.RpcResponseError import io.micrometer.core.instrument.Metrics import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory +import org.springframework.cloud.sleuth.Span +import org.springframework.cloud.sleuth.Tracer +import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth import org.springframework.context.event.EventListener import org.springframework.stereotype.Service import reactor.core.publisher.Flux 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 @@ -60,7 +64,8 @@ import java.util.concurrent.atomic.AtomicInteger open class NativeCall( private val multistreamHolder: MultistreamHolder, private val signer: ResponseSigner, - config: MainConfig + config: MainConfig, + private val tracer: Tracer ) { private val log = LoggerFactory.getLogger(NativeCall::class.java) @@ -101,22 +106,79 @@ open class NativeCall( } open fun nativeCallResult(requestMono: Mono): Flux { + val requestSpan = tracer.currentSpan() return requestMono.flatMapMany(this::prepareCall) .flatMap { - if (it.isValid()) { - val parsed = parseParams(it.get()) - this.fetch(parsed) - .doOnError { e -> log.warn("Error during native call: ${e.message}") } - } else { - val error = it.getError() + val requestId = it.requestId + val requestCount = it.requestCount + val id = it.getContextId() + val result = processCallContext(it, requestSpan) - Mono.just( - CallResult(error.id, 0, null, error, null, null, null) - ) - } + return@flatMap result + .onErrorResume { err -> + Mono.just( + CallResult.fail(id, 0, err, null) + ) + } + .doOnNext { callRes -> completeSpan(callRes, requestCount) } + .contextWrite { ctx -> createTracingReactorContext(ctx, requestCount, requestId, requestSpan) } } } + private fun completeSpan(callResult: CallResult, requestCount: Int) { + if (requestCount > 1) { + val span = tracer.currentSpan() + if (callResult.isError()) { + span?.error( + RuntimeException(callResult.error?.message ?: "Internal error") + ) + } + span?.end() + } + } + + private fun createTracingReactorContext( + ctx: Context, + requestCount: Int, + requestId: String, + requestSpan: Span? + ): Context { + if (requestCount > 1) { + val span = tracer.nextSpan(requestSpan) + .name(requestId) + .tag("request.id", requestId) + .start() + return ReactorSleuth.putSpanInScope(tracer, ctx, span) + } + return ctx + } + + private fun processCallContext( + callContext: CallContext, + requestSpan: Span? + ): Mono { + return if (callContext.isValid()) { + run { + val parsed = try { + parseParams(callContext.get()) + } catch (e: Exception) { + return@run Mono.error(e) + } + if (callContext.requestCount == 1 && callContext.requestId.isNotBlank()) { + requestSpan?.tag("request.id", callContext.requestId) + } + this.fetch(parsed) + .doOnError { e -> log.warn("Error during native call: ${e.message}") } + } + } else { + val error = callContext.getError() + + Mono.just( + CallResult(error.id, 0, null, error, null, null, null) + ) + } + } + fun parseParams(it: ValidCallContext): ValidCallContext { val rawParams = extractParams(it.payload.params) val params = it.requestDecorator.processRequest(rawParams) @@ -160,6 +222,7 @@ open class NativeCall( log.error("Lost context for a native call", it) 0 } + tracer.currentSpan()?.error(it) return BlockchainOuterClass.NativeCallReplyItem.newBuilder() .setSucceed(false) .setErrorMessage(it?.message ?: "Internal error") @@ -212,6 +275,8 @@ open class NativeCall( requestItem: BlockchainOuterClass.NativeCallItem, upstream: Multistream ): Mono { + val requestId = requestItem.requestId + val requestCount = request.itemsCount val method = requestItem.method val params = requestItem.payload.toStringUtf8() val availableMethods = upstream.getMethods() @@ -224,7 +289,9 @@ open class NativeCall( requestItem.id, errorMessage, JsonRpcError(RpcResponseError.CODE_METHOD_NOT_EXIST, errorMessage) - ) + ), + requestId, + requestCount ) ) } @@ -266,7 +333,9 @@ open class NativeCall( RawCallDetails(method, params), requestDecorator, resultDecorator, - selector + selector, + requestId, + requestCount ) } } @@ -378,10 +447,14 @@ open class NativeCall( return req as List } - interface CallContext { - fun isValid(): Boolean - fun get(): ValidCallContext - fun getError(): CallError + abstract class CallContext( + val requestId: String, + val requestCount: Int + ) { + abstract fun isValid(): Boolean + abstract fun get(): ValidCallContext + abstract fun getError(): CallError + abstract fun getContextId(): Int } interface ResultDecorator { @@ -437,8 +510,10 @@ open class NativeCall( val payload: T, val requestDecorator: RequestDecorator, val resultDecorator: ResultDecorator, - val forwardedSelector: BlockchainOuterClass.Selector? - ) : CallContext { + val forwardedSelector: BlockchainOuterClass.Selector?, + requestId: String, + requestCount: Int + ) : CallContext(requestId, requestCount) { constructor( id: Int, @@ -446,8 +521,13 @@ open class NativeCall( upstream: Multistream, matcher: Selector.Matcher, callQuorum: CallQuorum, - payload: T - ) : this(id, nonce, upstream, matcher, callQuorum, payload, NoneRequestDecorator(), NoneResultDecorator(), null) + payload: T, + requestId: String, + requestCount: Int + ) : this( + id, nonce, upstream, matcher, callQuorum, payload, + NoneRequestDecorator(), NoneResultDecorator(), null, requestId, requestCount + ) override fun isValid(): Boolean { return true @@ -461,8 +541,13 @@ open class NativeCall( throw IllegalStateException("Invalid context $id") } + override fun getContextId(): Int = id + fun withPayload(payload: X): ValidCallContext { - return ValidCallContext(id, nonce, upstream, matcher, callQuorum, payload, requestDecorator, resultDecorator, forwardedSelector) + return ValidCallContext( + id, nonce, upstream, matcher, callQuorum, payload, + requestDecorator, resultDecorator, forwardedSelector, requestId, requestCount + ) } fun getApis(): ApiSource { @@ -474,8 +559,10 @@ open class NativeCall( * Call context when it's known in advance that the call is invalid and should return an error */ open class InvalidCallContext( - private val error: CallError - ) : CallContext { + private val error: CallError, + requestId: String, + requestCount: Int + ) : CallContext(requestId, requestCount) { override fun isValid(): Boolean { return false } @@ -487,6 +574,8 @@ open class NativeCall( override fun getError(): CallError { return error } + + override fun getContextId(): Int = error.id } open class CallFailure(val id: Int, val reason: Throwable) : Exception("Failed to call $id: ${reason.message}") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index d55b74cb..1886b3be 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -16,6 +16,7 @@ */ package io.emeraldpay.dshackle.startup +import brave.grpc.GrpcTracing import com.google.common.annotations.VisibleForTesting import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.Chain @@ -73,7 +74,8 @@ open class ConfiguredUpstreams( private val eventPublisher: ApplicationEventPublisher, @Qualifier("grpcChannelExecutor") private val channelExecutor: Executor, - private val chainsConfig: ChainsConfig + private val chainsConfig: ChainsConfig, + private val grpcTracing: GrpcTracing ) : ApplicationRunner { private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java) @@ -335,7 +337,8 @@ open class ConfiguredUpstreams( config.labels, grpcUpstreamsScheduler, channelExecutor, - chainsConfig + chainsConfig, + grpcTracing ).apply { timeout = options.timeout } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index 64f0de5b..6b8f3abc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -16,6 +16,7 @@ */ package io.emeraldpay.dshackle.upstream.grpc +import brave.grpc.GrpcTracing import io.emeraldpay.api.proto.BlockchainOuterClass.DescribeRequest import io.emeraldpay.api.proto.BlockchainOuterClass.DescribeResponse import io.emeraldpay.api.proto.BlockchainOuterClass.StatusRequest @@ -70,7 +71,8 @@ class GrpcUpstreams( private val labels: UpstreamsConfig.Labels, private val chainStatusScheduler: Scheduler, private val grpcExecutor: Executor, - private val chainsConfig: ChainsConfig + private val chainsConfig: ChainsConfig, + private val grpcTracing: GrpcTracing ) { private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java) @@ -85,6 +87,7 @@ class GrpcUpstreams( // some messages are very large. many of them in megabytes, some even in gigabytes (ex. ETH Traces) .maxInboundMessageSize(Defaults.maxMessageSize) .enableRetry() + .intercept(grpcTracing.newClientInterceptor()) .executor(grpcExecutor) .maxRetryAttempts(3) if (auth != null && StringUtils.isNotEmpty(auth.ca)) { diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml new file mode 100644 index 00000000..cbadad15 --- /dev/null +++ b/src/main/resources/application.yml @@ -0,0 +1,17 @@ +spring: + application: + name: ${DRPC_APP_NAME:dshackle} + zipkin: + enabled: ${ZIPKIN_ENABLE:false} + base-url: ${ZIPKIN_URL:http://localhost:9411/} + sleuth: + span-filter: + additional-span-name-patterns-to-ignore: + - "^grpcChannelExecutor$" + - ".+NativeSubscribe" + - ".+SubscribeHead" + - ".+SubscribeBalance" + - ".+SubscribeTxStatus" + - ".+SubscribeStatus" + - ".+SubscribeNodeStatus" + - ".+Describe" \ No newline at end of file diff --git a/src/main/resources/log4j2.xml b/src/main/resources/log4j2.xml deleted file mode 100644 index 807a731a..00000000 --- a/src/main/resources/log4j2.xml +++ /dev/null @@ -1,44 +0,0 @@ - - - - - %d{yyyy-dd-MM HH:mm:ss.SSS} | %-5level | %20.20c{1} | %msg%n - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/src/main/resources/logback-spring.xml b/src/main/resources/logback-spring.xml new file mode 100644 index 00000000..7bd93661 --- /dev/null +++ b/src/main/resources/logback-spring.xml @@ -0,0 +1,52 @@ + + + + + + + + + + + + TRACE + + + ${CONSOLE_LOG_PATTERN} + utf8 + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 328ae161..51894f9a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -44,6 +44,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.etherjar.rpc.RpcResponseError +import org.springframework.cloud.sleuth.Tracer import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.test.StepVerifier @@ -71,7 +72,7 @@ class NativeCallSpec extends Specification { config.cache = cacheConfig config.passthrough = passthrough - new NativeCall(upstreams, signer, config) + new NativeCall(upstreams, signer, config, Stub(Tracer)) } def "Tries router first"() { @@ -85,7 +86,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall() def ctx = new NativeCall.ValidCallContext( 1, null, upstream, Selector.empty, new AlwaysQuorum(), - new NativeCall.ParsedCallDetails("eth_test", []) + new NativeCall.ParsedCallDetails("eth_test", []), "reqId", 1 ) when: @@ -106,7 +107,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall() def ctx = new NativeCall.ValidCallContext( 15, null, upstream, Selector.empty, new AlwaysQuorum(), - new NativeCall.ParsedCallDetails("eth_test", []) + new NativeCall.ParsedCallDetails("eth_test", []), "reqId", 1 ) when: @@ -135,7 +136,7 @@ class NativeCallSpec extends Specification { } } def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum, - new NativeCall.ParsedCallDetails("eth_test", [])) + new NativeCall.ParsedCallDetails("eth_test", []), "reqId", 1) when: def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1)) @@ -156,7 +157,7 @@ class NativeCallSpec extends Specification { } } def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum, - new NativeCall.ParsedCallDetails("eth_test", [])) + new NativeCall.ParsedCallDetails("eth_test", []), "reqId", 1) when: def resp = nativeCall.executeOnRemote(call) @@ -182,7 +183,7 @@ class NativeCallSpec extends Specification { } } def call = new NativeCall.ValidCallContext(12, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum, - new NativeCall.ParsedCallDetails("eth_test", [])) + new NativeCall.ParsedCallDetails("eth_test", []), "reqId", 1) when: def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1)) @@ -531,7 +532,7 @@ class NativeCallSpec extends Specification { setup: def nativeCall = nativeCall() def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(), - new NativeCall.RawCallDetails("eth_test", "[]")) + new NativeCall.RawCallDetails("eth_test", "[]"), "reqId", 1) when: def act = nativeCall.parseParams(ctx) then: @@ -544,7 +545,7 @@ class NativeCallSpec extends Specification { setup: def nativeCall = nativeCall() def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(), - new NativeCall.RawCallDetails("eth_test", "")) + new NativeCall.RawCallDetails("eth_test", ""), "reqId", 1) when: def act = nativeCall.parseParams(ctx) then: @@ -557,7 +558,7 @@ class NativeCallSpec extends Specification { setup: def nativeCall = nativeCall() def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(), - new NativeCall.RawCallDetails("eth_test", "[false]")) + new NativeCall.RawCallDetails("eth_test", "[false]"), "reqId", 1) when: def act = nativeCall.parseParams(ctx) then: @@ -570,7 +571,7 @@ class NativeCallSpec extends Specification { setup: def nativeCall = nativeCall() def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(), - new NativeCall.RawCallDetails("eth_test", "[false, 123]")) + new NativeCall.RawCallDetails("eth_test", "[false, 123]"), "reqId", 1) when: def act = nativeCall.parseParams(ctx) then: @@ -584,7 +585,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall() def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(), new NativeCall.RawCallDetails("eth_getFilterUpdates", '["0xabcd"]'), - new NativeCall.WithFilterIdDecorator(), new NativeCall.NoneResultDecorator(), null) + new NativeCall.WithFilterIdDecorator(), new NativeCall.NoneResultDecorator(), null, "reqId", 1) when: def act = nativeCall.parseParams(ctx) then: @@ -618,7 +619,7 @@ class NativeCallSpec extends Specification { } def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum, new NativeCall.ParsedCallDetails("eth_getFilterChanges", []), - new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null) + new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null, "reqId", 1) when: def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1)) @@ -653,7 +654,7 @@ class NativeCallSpec extends Specification { } def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum, new NativeCall.ParsedCallDetails("eth_getFilterChanges", []), - new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null) + new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null, "reqId", 1) when: def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1)) @@ -674,7 +675,7 @@ class NativeCallSpec extends Specification { def ctx = new NativeCall.ValidCallContext(10, null, upstream, Selector.empty, new AlwaysQuorum(), - new NativeCall.ParsedCallDetails("eth_test", [])) + new NativeCall.ParsedCallDetails("eth_test", []), "reqId", 1) when: nativeCall.fetch(ctx) then: @@ -691,7 +692,7 @@ class NativeCallSpec extends Specification { def ctx = new NativeCall.ValidCallContext(10, null, upstream, Selector.empty, new AlwaysQuorum(), - new NativeCall.ParsedCallDetails("eth_test", [])) + new NativeCall.ParsedCallDetails("eth_test", []), "reqId", 1) when: def act = nativeCall.fetch(ctx) then: diff --git a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy index aad6f7f4..270030db 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy @@ -1,5 +1,8 @@ package io.emeraldpay.dshackle.startup +import io.emeraldpay.dshackle.Chain +import brave.Tracing +import brave.grpc.GrpcTracing import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.config.ChainsConfig @@ -25,7 +28,8 @@ class ConfiguredUpstreamsSpec extends Specification { callTargetsHolder, Mock(ApplicationEventPublisher), Executors.newFixedThreadPool(1), - ChainsConfig.default() + ChainsConfig.default(), + GrpcTracing.create(Tracing.newBuilder().build()) ) def methods = new UpstreamsConfig.Methods( [ @@ -53,7 +57,8 @@ class ConfiguredUpstreamsSpec extends Specification { callTargetsHolder, Mock(ApplicationEventPublisher), Executors.newFixedThreadPool(1), - ChainsConfig.default() + ChainsConfig.default(), + GrpcTracing.create(Tracing.newBuilder().build()) ) def methods = new UpstreamsConfig.Methods( [ @@ -80,7 +85,8 @@ class ConfiguredUpstreamsSpec extends Specification { callTargetsHolder, Mock(ApplicationEventPublisher), Executors.newFixedThreadPool(1), - ChainsConfig.default() + ChainsConfig.default(), + GrpcTracing.create(Tracing.newBuilder().build()) ) expect: configurer.getHash(node, src) == expected @@ -102,7 +108,8 @@ class ConfiguredUpstreamsSpec extends Specification { callTargetsHolder, Mock(ApplicationEventPublisher), Executors.newFixedThreadPool(1), - ChainsConfig.default() + ChainsConfig.default(), + GrpcTracing.create(Tracing.newBuilder().build()) ) when: def h1 = configurer.getHash(null, "hohoho") @@ -129,7 +136,8 @@ class ConfiguredUpstreamsSpec extends Specification { callTargetsHolder, Mock(ApplicationEventPublisher), Executors.newFixedThreadPool(1), - ChainsConfig.default() + ChainsConfig.default(), + GrpcTracing.create(Tracing.newBuilder().build()) ) def methodsGroup = new UpstreamsConfig.MethodGroups( ["filter"] as Set,