From 22aad348d82fc7e971ae9782ae7575082c92a5df Mon Sep 17 00:00:00 2001 From: Maxksim Fomenkov Date: Mon, 5 Sep 2022 14:45:00 +0300 Subject: [PATCH] added metrics for native call inner requests --- .../emeraldpay/dshackle/proxy/ProxyServer.kt | 31 ++--------- .../emeraldpay/dshackle/rpc/BlockchainRpc.kt | 55 +++++++++++++------ 2 files changed, 44 insertions(+), 42 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt index 69a2b109..f6228137 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -32,9 +32,7 @@ import org.slf4j.LoggerFactory import reactor.netty.http.server.HttpServer import reactor.netty.http.server.HttpServerRoutes import java.util.EnumMap -import java.util.concurrent.locks.ReentrantReadWriteLock -import kotlin.concurrent.read -import kotlin.concurrent.write +import java.util.concurrent.ConcurrentHashMap /** * HTTP Proxy Server @@ -141,32 +139,13 @@ class ProxyServer( } } - /** - * Monitoring that has separate metrics per RPC method. - * Slightly slower to use than StandardRequestMetrics - */ class ExtendedRequestMetrics : RequestMetricsFactory { - private val current = HashMap() - private val lock = ReentrantReadWriteLock() + private val current = ConcurrentHashMap() - override fun get(chain: Chain, method: String): RequestMetrics { - val ref = Ref(chain, method) - lock.read { - val existing = current[ref] - if (existing != null) { - return existing - } + override fun get(chain: Chain, method: String): RequestMetrics = + current.computeIfAbsent(Ref(chain, method)) { + RequestMetricsWithMethod(it.chain, it.method) } - lock.write { - val existing = current[ref] - if (existing != null) { - return existing - } - val created = RequestMetricsWithMethod(chain, method) - current[ref] = created - return created - } - } data class Ref(val chain: Chain, val method: String) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index e87f4c7a..ff8ca66a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -32,6 +32,7 @@ import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.util.Locale +import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.TimeUnit @Service @@ -65,18 +66,28 @@ class BlockchainRpc( override fun nativeCall(request: Mono): Flux { var startTime = 0L var metrics: RequestMetrics? = null + val idsMap = mutableMapOf() return nativeCall.nativeCall( request - .doOnNext { - metrics = chainMetrics.get(it.chain) - metrics!!.nativeCallMetric.increment() + .doOnNext { req -> + metrics = chainMetrics.get(req.chain) + metrics?.let { m -> + m.nativeCallMetric.increment() + req.itemsList.forEach { item -> + idsMap[item.id] = item.method + m.getNativeItemMetrics(item.method).nativeItemRequest.increment() + } + } startTime = System.currentTimeMillis() } ).doOnNext { reply -> - metrics?.let { m -> - m.nativeCallRespMetric?.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS) + metrics?.getNativeItemMetrics(idsMap[reply.id] ?: "unknown")?.let { itemMetrics -> + itemMetrics.nativeItemResponse.record( + System.currentTimeMillis() - startTime, + TimeUnit.MILLISECONDS + ) if (!reply.succeed) { - m.nativeCallErrRespMetric.increment() + itemMetrics.nativeItemResponseErr.increment() } } }.doOnError { failMetric.increment() } @@ -204,20 +215,11 @@ class BlockchainRpc( .doOnError { failMetric.increment() } } - class RequestMetrics(chain: Chain) { + class RequestMetrics(val 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 nativeCallErrRespMetric = Counter.builder("request.grpc.response.err") - .tag("type", "nativeCall") - .tag("chain", chain.chainCode) - .register(Metrics.globalRegistry) val nativeSubscribeMetric = Counter.builder("request.grpc.request") .tag("type", "nativeSubscribe") .tag("chain", chain.chainCode) @@ -264,5 +266,26 @@ class BlockchainRpc( .tag("chain", chain.chainCode) .publishPercentileHistogram() .register(Metrics.globalRegistry) + + private val nativeItemMetrics = ConcurrentHashMap() + + fun getNativeItemMetrics(method: String) = + nativeItemMetrics.computeIfAbsent(method) { NativeRequestItemsMetrics(chain, it) } + } + + class NativeRequestItemsMetrics(chain: Chain, method: String) { + val nativeItemRequest = Counter.builder("request.grpc.native.request") + .tag("chain", chain.chainCode) + .tag("method", method) + .register(Metrics.globalRegistry) + val nativeItemResponse = Timer.builder("request.grpc.native.response") + .tag("chain", chain.chainCode) + .tag("method", method) + .publishPercentileHistogram() + .register(Metrics.globalRegistry) + val nativeItemResponseErr = Counter.builder("request.grpc.native.request") + .tag("chain", chain.chainCode) + .tag("method", method) + .register(Metrics.globalRegistry) } }