StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #8 from p2p-org/native_call_metrics

Browse Source 
added metrics for native call inner requests
This commit is contained in:
Vyacheslav Shebanov
2022-09-05 15:05:12 +03:00
committed by GitHub
parent c05c1ad0de 22aad348d8
commit 29657a64c3
2 changed files with 44 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt
View File

@@ -33,9 +33,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
@@ -143,32 +141,13 @@ class ProxyServer(
}
}
/**
* Monitoring that has separate metrics per RPC method.
* Slightly slower to use than StandardRequestMetrics
*/
class ExtendedRequestMetrics : RequestMetricsFactory {
private val current = HashMap<Ref, RequestMetrics>()
private val lock = ReentrantReadWriteLock()
private val current = ConcurrentHashMap<Ref, RequestMetrics>()
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)
}

55
src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
View File

@@ -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<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
var startTime = 0L
var metrics: RequestMetrics? = null
val idsMap = mutableMapOf<Int, String>()
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<String, NativeRequestItemsMetrics>()
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)
}
}