diff --git a/src/main/kotlin/io/emeraldpay/dshackle/ChainValue.kt b/src/main/kotlin/io/emeraldpay/dshackle/ChainValue.kt new file mode 100644 index 00000000..15d318e8 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/ChainValue.kt @@ -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( + private val factory: (chain: Chain) -> V +) { + + private val values = EnumMap(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 + } + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt index ebda0f8b..6de95683 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -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): Flux { - 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) + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 813e6f2a..1b83b5b9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -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): Flux { 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): Flux { 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): Flux { 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): Flux { 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): Flux { 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::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")