solution: metrics for grpc requests

2021-02-10 17:53:07 -05:00
parent 981a5dc589
commit 81c5f2ef7b
2 changed files with 128 additions and 12 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
@@ -19,14 +19,20 @@ package io.emeraldpay.dshackle.rpc
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
-import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.grpc.Chain
+import io.micrometer.core.instrument.Counter
+import io.micrometer.core.instrument.Metrics
+import io.micrometer.core.instrument.Timer
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import java.util.*
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock

@Service
 class BlockchainRpc(
@@ -40,22 +46,51 @@ class BlockchainRpc(

    private val log = LoggerFactory.getLogger(BlockchainRpc::class.java)

+    private val describeMetric = Counter.builder("request.grpc.request")
+            .tag("type", "describe")
+            .register(Metrics.globalRegistry)
+    private val subscribeStatusMetric = Counter.builder("request.grpc.request")
+            .tag("type", "subscribeStatus")
+            .register(Metrics.globalRegistry)
+    private val errorMetric = Counter.builder("request.grpc.err")
+            .register(Metrics.globalRegistry)
+
    override fun nativeCall(request: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
-        return nativeCall.nativeCall(request)
+        var startTime = 0L
+        var metrics: RequestMetrics? = null
+        return nativeCall.nativeCall(
+                request
+                        .doOnNext {
+                            metrics = getMetrics(it.chain)
+                            metrics!!.nativeCallMetric.increment()
+                            startTime = System.currentTimeMillis()
+                        }
+        ).doOnNext {
+            metrics?.nativeCallRespMetric?.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS)
+        }.doOnError { errorMetric.increment() }
    }

    override fun subscribeHead(request: Mono<Common.Chain>): Flux<BlockchainOuterClass.ChainHead> {
-        return streamHead.add(request)
+        return streamHead.add(
+                request
+                        .doOnNext { getMetrics(it.type).subscribeHeadMetric.increment() }
+        ).doOnError { errorMetric.increment() }
    }

    override fun subscribeTxStatus(request: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
        return request.flatMapMany { request ->
            val chain = Chain.byId(request.chainValue)
+            val metrics = getMetrics(chain)
+            metrics.subscribeTxMetric.increment()
            try {
-                trackTx.find { it.isSupported(chain) }?.subscribe(request)
-                        ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
+                trackTx.find { it.isSupported(chain) }?.let { track ->
+                    track.subscribe(request)
+                            .doOnNext { metrics.subscribeHeadRespMetric.increment() }
+                            .doOnError { errorMetric.increment() }
+                } ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
            } catch (t: Throwable) {
                log.error("Internal error during Tx Subscription", t)
+                errorMetric.increment()
                Flux.error<BlockchainOuterClass.TxStatus>(IllegalStateException("Internal Error"))
            }
        }
@@ -64,15 +99,21 @@ class BlockchainRpc(
    override fun subscribeBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
        return requestMono.flatMapMany { request ->
            val chain = Chain.byId(request.asset.chainValue)
+            val metrics = getMetrics(chain)
+            metrics.subscribeBalanceMetric.increment()
            val asset = request.asset.code.toLowerCase()
            try {
-                trackAddress.find { it.isSupported(chain, asset) }?.subscribe(request)
-                        ?: Flux.error<BlockchainOuterClass.AddressBalance>(SilentException.UnsupportedBlockchain(chain))
+                trackAddress.find { it.isSupported(chain, asset) }?.let { track ->
+                    track.subscribe(request)
+                            .doOnNext { metrics.subscribeBalanceRespMetric.increment() }
+                            .doOnError { errorMetric.increment() }
+                } ?: Flux.error<BlockchainOuterClass.AddressBalance>(SilentException.UnsupportedBlockchain(chain))
                                .doOnSubscribe {
                                    log.error("Balance for $chain:$asset is not supported")
                                }
            } catch (t: Throwable) {
                log.error("Internal error during Balance Subscription", t)
+                errorMetric.increment()
                Flux.error<BlockchainOuterClass.AddressBalance>(IllegalStateException("Internal Error"))
            }
        }
@@ -81,25 +122,102 @@ class BlockchainRpc(
    override fun getBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
        return requestMono.flatMapMany { request ->
            val chain = Chain.byId(request.asset.chainValue)
+            val metrics = getMetrics(chain)
+            metrics.getBalanceMetric.increment()
            val asset = request.asset.code.toLowerCase()
+            val startTime = System.currentTimeMillis()
            try {
-                trackAddress.find { it.isSupported(chain, asset) }?.getBalance(request)
-                        ?: Flux.error<BlockchainOuterClass.AddressBalance>(SilentException.UnsupportedBlockchain(chain))
+                trackAddress.find { it.isSupported(chain, asset) }?.let { track ->
+                    track.getBalance(request)
+                            .doOnNext {
+                                metrics.getBalanceRespMetric.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS)
+                            }
+                } ?: Flux.error<BlockchainOuterClass.AddressBalance>(SilentException.UnsupportedBlockchain(chain))
                                .doOnSubscribe {
                                    log.error("Balance for $chain:$asset is not supported")
                                }
            } catch (t: Throwable) {
                log.error("Internal error during Balance Request", t)
+                errorMetric.increment()
                Flux.error<BlockchainOuterClass.AddressBalance>(IllegalStateException("Internal Error"))
            }
        }
    }

    override fun describe(request: Mono<BlockchainOuterClass.DescribeRequest>): Mono<BlockchainOuterClass.DescribeResponse> {
+        describeMetric.increment()
        return describe.describe(request)
+                .doOnError { errorMetric.increment() }
    }

    override fun subscribeStatus(request: Mono<BlockchainOuterClass.StatusRequest>): Flux<BlockchainOuterClass.ChainStatus> {
+        subscribeStatusMetric.increment()
        return subscribeStatus.subscribeStatus(request)
+                .doOnError { errorMetric.increment() }
+    }
+
+    private val metrics = EnumMap<Chain, RequestMetrics>(Chain::class.java)
+    private val metricsSetupLock = ReentrantLock()
+
+    private fun getMetrics(chain: Common.ChainRef): RequestMetrics {
+        return getMetrics(Chain.byId(chain.number))
+    }
+    private fun getMetrics(chain: Chain): RequestMetrics {
+        val existing = metrics[chain]
+        if (existing != null) {
+            return existing
+        }
+        return metricsSetupLock.withLock {
+            // second check in case it was updated while getting the lock
+            val existing2 = metrics[chain]
+            if (existing2 != null) {
+                existing2
+            } else {
+                val created = RequestMetrics(chain)
+                metrics[chain] = created
+                created
+            }
+        }
+    }
+
+    class RequestMetrics(chain: Chain) {
+        val nativeCallMetric = Counter.builder("request.grpc.request")
+                .tag("type", "nativeCall")
+                .tag("chain", chain.chainCode)
+                .register(Metrics.globalRegistry)
+        val nativeCallRespMetric = Timer.builder("request.grpc.response")
+                .tag("type", "nativeCall")
+                .tag("chain", chain.chainCode)
+                .publishPercentileHistogram()
+                .register(Metrics.globalRegistry)
+        val subscribeHeadMetric = Counter.builder("request.grpc.request")
+                .tag("type", "subscribeHead")
+                .tag("chain", chain.chainCode)
+                .register(Metrics.globalRegistry)
+        val subscribeHeadRespMetric = Counter.builder("request.grpc.reply")
+                .tag("type", "subscribeHead")
+                .tag("chain", chain.chainCode)
+                .register(Metrics.globalRegistry)
+        val subscribeTxMetric = Counter.builder("request.grpc.request")
+                .tag("type", "subscribeTx")
+                .tag("chain", chain.chainCode)
+                .register(Metrics.globalRegistry)
+        val subscribeBalanceMetric = Counter.builder("request.grpc.request")
+                .tag("type", "subscribeBalance")
+                .tag("chain", chain.chainCode)
+                .register(Metrics.globalRegistry)
+        val subscribeBalanceRespMetric = Counter.builder("request.grpc.reply")
+                .tag("type", "subscribeBalance")
+                .tag("chain", chain.chainCode)
+                .register(Metrics.globalRegistry)
+        val getBalanceMetric = Counter.builder("request.grpc.request")
+                .tag("type", "getBalance")
+                .tag("chain", chain.chainCode)
+                .register(Metrics.globalRegistry)
+        val getBalanceRespMetric = Timer.builder("request.grpc.response")
+                .tag("type", "getBalance")
+                .tag("chain", chain.chainCode)
+                .publishPercentileHistogram()
+                .register(Metrics.globalRegistry)
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
@@ -163,9 +163,7 @@ class GrpcUpstreams(

    fun getOrCreate(chain: Chain): UpstreamChange {
        val metricsTags = listOf(
-                // "unknown" is not supposed to happen
-                Tag.of("upstream", id ?: "unknown"),
-                // UNSPECIFIED shouldn't happen too
+                Tag.of("upstream", id),
                Tag.of("chain", chain.chainCode)
        )