diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 516b44ad..813e6f2a 100644 --- 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): Flux { - 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): Flux { - return streamHead.add(request) + return streamHead.add( + request + .doOnNext { getMetrics(it.type).subscribeHeadMetric.increment() } + ).doOnError { errorMetric.increment() } } override fun subscribeTxStatus(request: Mono): Flux { 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(IllegalStateException("Internal Error")) } } @@ -64,15 +99,21 @@ class BlockchainRpc( override fun subscribeBalance(requestMono: Mono): Flux { 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(SilentException.UnsupportedBlockchain(chain)) + trackAddress.find { it.isSupported(chain, asset) }?.let { track -> + track.subscribe(request) + .doOnNext { metrics.subscribeBalanceRespMetric.increment() } + .doOnError { errorMetric.increment() } + } ?: Flux.error(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(IllegalStateException("Internal Error")) } } @@ -81,25 +122,102 @@ class BlockchainRpc( override fun getBalance(requestMono: Mono): Flux { 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(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(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(IllegalStateException("Internal Error")) } } } override fun describe(request: Mono): Mono { + describeMetric.increment() return describe.describe(request) + .doOnError { errorMetric.increment() } } override fun subscribeStatus(request: Mono): Flux { + subscribeStatusMetric.increment() return subscribeStatus.subscribeStatus(request) + .doOnError { errorMetric.increment() } + } + + private val metrics = EnumMap(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) } } \ No newline at end of file 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 6b67d70c..fc08e9db 100644 --- 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) )