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

solution: metrics for proxy requests

Browse Source
This commit is contained in:
Igor Artamonov
2021-02-10 21:25:39 -05:00
parent 81c5f2ef7b
commit bd6ce103f7
3 changed files with 87 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
src/main/kotlin/io/emeraldpay/dshackle/ChainValue.kt Normal file
View File

@@ -0,0 +1,55 @@
/**
* Copyright (c) 2021 EmeraldPay, Inc
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle
import io.emeraldpay.api.proto.Common
import io.emeraldpay.grpc.Chain
import java.util.*
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock
/**
* Keeps a lazily created value associated with a Chain
*/
class ChainValue<V>(
private val factory: (chain: Chain) -> V
) {
private val values = EnumMap<Chain, V>(Chain::class.java)
private val valuesInitLock = ReentrantLock()
fun get(chain: Common.ChainRef): V {
return get(Chain.byId(chain.number))
}
fun get(chain: Chain): V {
val existing = values[chain]
if (existing != null) {
return existing
}
return valuesInitLock.withLock {
// second check in case it was updated while getting the lock
val existing2 = values[chain]
if (existing2 != null) {
existing2
} else {
val created = factory(chain)
values[chain] = created
created
}
}
}
}

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

@@ -18,12 +18,17 @@ package io.emeraldpay.dshackle.proxy
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.ChainValue
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.TlsSetup
import io.emeraldpay.dshackle.config.ProxyConfig
import io.emeraldpay.dshackle.rpc.NativeCall
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.rpc.RpcException
import io.micrometer.core.instrument.Counter
import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Timer
import io.netty.buffer.ByteBuf
import io.netty.buffer.Unpooled
import io.netty.channel.ChannelHandler
@@ -39,6 +44,7 @@ import reactor.netty.http.server.HttpServer
import reactor.netty.http.server.HttpServerRequest
import reactor.netty.http.server.HttpServerResponse
import reactor.netty.http.server.HttpServerRoutes
import java.util.concurrent.TimeUnit
import java.util.function.BiFunction
/**
@@ -56,6 +62,8 @@ class ProxyServer(
private val log = LoggerFactory.getLogger(ProxyServer::class.java)
}
private val chainMetrics = ChainValue { chain -> RequestMetrics(chain) }
private val errorHandler: ChannelHandler = object : ChannelHandler {
override fun handlerAdded(p0: ChannelHandlerContext?) {
}
@@ -123,9 +131,16 @@ class ProxyServer(
}
fun processRequest(chain: Common.ChainRef, request: Mono<ByteArray>): Flux<ByteBuf> {
return request.map(readRpcJson)
val metrics = chainMetrics.get(chain)
val startTime = System.currentTimeMillis()
return request
.map(readRpcJson)
.flatMapMany { call -> execute(chain, call) }
.doOnNext {
metrics.callMetric.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS)
}
.onErrorResume(RpcException::class.java) { err ->
metrics.errorMetric.increment()
val id = err.details?.let {
if (it is JsonRpcResponse.Id) it else JsonRpcResponse.NumberId(-1)
} ?: JsonRpcResponse.NumberId(-1)
@@ -147,4 +162,13 @@ class ProxyServer(
.send(results)
}
}
class RequestMetrics(chain: Chain) {
val callMetric = Timer.builder("request.jsonrpc.call")
.tag("chain", chain.chainCode)
.register(Metrics.globalRegistry)
val errorMetric = Counter.builder("request.jsonrpc.err")
.tag("chain", chain.chainCode)
.register(Metrics.globalRegistry)
}
}

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

@@ -19,6 +19,7 @@ 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.ChainValue
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.grpc.Chain
import io.micrometer.core.instrument.Counter
@@ -54,6 +55,7 @@ class BlockchainRpc(
.register(Metrics.globalRegistry)
private val errorMetric = Counter.builder("request.grpc.err")
.register(Metrics.globalRegistry)
private val chainMetrics = ChainValue { chain -> RequestMetrics(chain) }
override fun nativeCall(request: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
var startTime = 0L
@@ -61,7 +63,7 @@ class BlockchainRpc(
return nativeCall.nativeCall(
request
.doOnNext {
metrics = getMetrics(it.chain)
metrics = chainMetrics.get(it.chain)
metrics!!.nativeCallMetric.increment()
startTime = System.currentTimeMillis()
}
@@ -73,14 +75,14 @@ class BlockchainRpc(
override fun subscribeHead(request: Mono<Common.Chain>): Flux<BlockchainOuterClass.ChainHead> {
return streamHead.add(
request
.doOnNext { getMetrics(it.type).subscribeHeadMetric.increment() }
.doOnNext { chainMetrics.get(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)
val metrics = chainMetrics.get(chain)
metrics.subscribeTxMetric.increment()
try {
trackTx.find { it.isSupported(chain) }?.let { track ->
@@ -99,7 +101,7 @@ 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)
val metrics = chainMetrics.get(chain)
metrics.subscribeBalanceMetric.increment()
val asset = request.asset.code.toLowerCase()
try {
@@ -122,7 +124,7 @@ 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)
val metrics = chainMetrics.get(chain)
metrics.getBalanceMetric.increment()
val asset = request.asset.code.toLowerCase()
val startTime = System.currentTimeMillis()
@@ -156,30 +158,6 @@ class BlockchainRpc(
.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")