From 67ca1b85ea30a0289b7b7870982245ec832e5ac8 Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Fri, 24 Mar 2023 18:49:46 +0400 Subject: [PATCH] added protection for grpc head and grpc received blocks metric (#184) --- .../dshackle/upstream/grpc/GrpcHead.kt | 11 ++++ .../grpc/EthereumGrpcUpstreamSpec.groovy | 31 +++++----- .../upstream/grpc/GrpcHeadSpec.groovy | 62 ++++++++++--------- 3 files changed, 62 insertions(+), 42 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt index 6a16a7d6..d096d9b1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt @@ -25,6 +25,8 @@ import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.Metrics import org.reactivestreams.Publisher import reactor.core.Disposable import reactor.core.publisher.Flux @@ -75,6 +77,10 @@ class GrpcHead( .delayElements(Duration.ofSeconds(1)) ) .flatMap { it.subscribeHead(chainRef) } + .doOnNext { + headsCounter.increment() + } + .sample(Duration.ofMillis(1)) // protect from too many heads .doFinally { parent.setStatus(UpstreamAvailability.UNAVAILABLE) log.warn("Head subscription finished: $it") @@ -116,4 +122,9 @@ class GrpcHead( super.stop() headSubscription?.dispose() } + + val headsCounter = Counter.builder("grpc_head_received") + .tag("upstream", id) + .tag("chain", chain.chainCode) + .register(Metrics.globalRegistry) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy index d140dc45..8ab3f420 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy @@ -134,20 +134,23 @@ class EthereumGrpcUpstreamSpec extends Specification { @Override void subscribeHead(Common.Chain request, StreamObserver responseObserver) { - responseObserver.onNext( - BlockchainOuterClass.ChainHead.newBuilder() - .setBlockId(block1.hash.toHex().substring(2)) - .setHeight(block1.number) - .setWeight(ByteString.copyFrom(block1.totalDifficulty.toByteArray())) - .build() - ) - responseObserver.onNext( - BlockchainOuterClass.ChainHead.newBuilder() - .setBlockId(block2.hash.toHex().substring(2)) - .setHeight(block2.number) - .setWeight(ByteString.copyFrom(block2.totalDifficulty.toByteArray())) - .build() - ) + new Thread({ + responseObserver.onNext( + BlockchainOuterClass.ChainHead.newBuilder() + .setBlockId(block1.hash.toHex().substring(2)) + .setHeight(block1.number) + .setWeight(ByteString.copyFrom(block1.totalDifficulty.toByteArray())) + .build() + ) + Thread.sleep(100) + responseObserver.onNext( + BlockchainOuterClass.ChainHead.newBuilder() + .setBlockId(block2.hash.toHex().substring(2)) + .setHeight(block2.number) + .setWeight(ByteString.copyFrom(block2.totalDifficulty.toByteArray())) + .build() + ) + }).start() } }) def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null, ChainsConfig.ChainConfig.default()) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy index edcdbf94..2ebec4c3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy @@ -42,16 +42,20 @@ class GrpcHeadSpec extends Specification { responseObserver.onError(new IllegalStateException("Unsupported chain")) return } - [10, 11, 12, 14].forEach { height -> + new Thread({ [10, 11, 12, 14].forEach { height -> + println("next: $height") responseObserver.onNext( BlockchainOuterClass.ChainHead.newBuilder() .setChain(request.type) + .setBlockId(height.toString()) .setHeight(height) .build() ) - Thread.sleep(100) + Thread.sleep(500) } - responseObserver.onCompleted() + responseObserver.onCompleted() + }).start() + } }) def convert = { BlockchainOuterClass.ChainHead head -> @@ -88,32 +92,34 @@ class GrpcHeadSpec extends Specification { responseObserver.onError(new IllegalStateException("Unsupported chain")) return } - phase++ - if (phase == 1) { - responseObserver.onError(new RuntimeException("Phase 1 error")) - } else if (phase == 2) { - [10, 11].forEach { height -> - responseObserver.onNext( - BlockchainOuterClass.ChainHead.newBuilder() - .setChain(request.type) - .setHeight(height) - .build() - ) - Thread.sleep(100) + new Thread({ + phase++ + if (phase == 1) { + responseObserver.onError(new RuntimeException("Phase 1 error")) + } else if (phase == 2) { + [10, 11].forEach { height -> + responseObserver.onNext( + BlockchainOuterClass.ChainHead.newBuilder() + .setChain(request.type) + .setHeight(height) + .build() + ) + Thread.sleep(100) + } + responseObserver.onError(new RuntimeException("Phase 2 error")) + } else if (phase == 3) { + [11, 12, 13].forEach { height -> + responseObserver.onNext( + BlockchainOuterClass.ChainHead.newBuilder() + .setChain(request.type) + .setHeight(height) + .build() + ) + Thread.sleep(100) + } + responseObserver.onCompleted() } - responseObserver.onError(new RuntimeException("Phase 2 error")) - } else if (phase == 3) { - [11, 12, 13].forEach { height -> - responseObserver.onNext( - BlockchainOuterClass.ChainHead.newBuilder() - .setChain(request.type) - .setHeight(height) - .build() - ) - Thread.sleep(100) - } - responseObserver.onCompleted() - } + }).start() } }) def convert = { BlockchainOuterClass.ChainHead head ->