Add json logging and tracing (#131)

2023-02-17 16:39:35 +04:00
parent 2c33410d0a
commit 04d1334af6
14 changed files with 274 additions and 109 deletions
--- 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<io.grpc.BindableService>,
    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)
        }
--- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/TraceGrpcConfiguration.kt
+++ 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()
+}
--- 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<Flux<Upstream>, Mono<CallQuorum>> {
        val quorumReduce = BiFunction<CallQuorum, Tuple4<ByteArray, Optional<ResponseSigner.Signature>, Upstream, Optional<String>>, 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 <T> withErrorResume(api: Upstream, key: JsonRpcRequest): Function<Mono<T>, Mono<T>> {
        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<Result> {
        return Mono.just(quorum).flatMap { q ->
            if (q.isFailed()) {
-                Mono.error<Result>(
-                    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<Result>(err)
            } else {
                log.warn("Did not get any result from upstream. Method [${key.method}] using [$q]")
                Mono.empty<Result>()
--- 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<BlockchainOuterClass.NativeCallRequest>): Flux<CallResult> {
+        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<CallResult> {
+        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<RawCallDetails>): ValidCallContext<ParsedCallDetails> {
        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<CallContext> {
+        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<Any>
    }

-    interface CallContext {
-        fun isValid(): Boolean
-        fun <T> get(): ValidCallContext<T>
-        fun getError(): CallError
+    abstract class CallContext(
+        val requestId: String,
+        val requestCount: Int
+    ) {
+        abstract fun isValid(): Boolean
+        abstract fun <T> get(): ValidCallContext<T>
+        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 <X> withPayload(payload: X): ValidCallContext<X> {
-            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}")
--- 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
        }
--- 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)) {