From ad5d23cb328f5582281c53a224ede65a210c3604 Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Fri, 27 Jan 2023 20:44:49 +0400 Subject: [PATCH] Merge pull request #11 from p2p-org/node-status-sub-id id for subscription for node statuses --- build.gradle | 1 + emerald-grpc | 2 +- gradle/libs.versions.toml | 3 +- .../emeraldpay/dshackle/rpc/BlockchainRpc.kt | 18 ++- .../dshackle/rpc/SubscribeNodeStatus.kt | 121 +++++++++--------- .../dshackle/upstream/AbstractHead.kt | 4 +- .../dshackle/upstream/DefaultUpstream.kt | 2 +- .../NoChoiceWithPriorityForkChoice.kt | 6 +- .../dshackle/upstream/grpc/GrpcHead.kt | 2 +- 9 files changed, 86 insertions(+), 73 deletions(-) diff --git a/build.gradle b/build.gradle index e274042a..b53e93c9 100644 --- a/build.gradle +++ b/build.gradle @@ -74,6 +74,7 @@ dependencies { implementation libs.bundles.reactor implementation(libs.reactor.grpc.stub) + implementation(libs.grpc.proto.util) implementation libs.micrometer.registry.prometheus implementation libs.lettuce.core diff --git a/emerald-grpc b/emerald-grpc index fddf154c..0e10f0de 160000 --- a/emerald-grpc +++ b/emerald-grpc @@ -1 +1 @@ -Subproject commit fddf154c34b7a59bee790cc14eff9eff69e8339f +Subproject commit 0e10f0de5695658a92bf7f1436e815adf931af59 diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 5c0c4266..78ef6e89 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -44,6 +44,7 @@ etherjar-erc20 = { module = "io.emeraldpay.etherjar:etherjar-erc20", version.ref groovy = { module = "org.codehaus.groovy:groovy", version.ref = "groovy" } +grpc-proto-util = { module = "com.google.protobuf:protobuf-java-util", version.ref = "protoc" } grpc-protobuf = { module = "io.grpc:grpc-protobuf", version.ref = "grpc" } grpc-stub = { module = "io.grpc:grpc-stub", version.ref = "grpc" } grpc-netty = { module = "io.grpc:grpc-netty", version.ref = "grpc" } @@ -122,7 +123,7 @@ assertj = "org.assertj:assertj-core:3.23.1" [bundles] apache-commons = ["commons-io", "apache-commons-lang3", "apache-commons-collections4"] etherjar = ["etherjar-domain", "etherjar-hex", "etherjar-rpc-api", "etherjar-rpc-http", "etherjar-rpc-ws", "etherjar-tx", "etherjar-contract", "etherjar-erc20"] -grpc = ["grpc-protobuf", "grpc-stub", "grpc-netty"] +grpc = ["grpc-protobuf", "grpc-stub", "grpc-netty", "grpc-proto-util"] httpcomponents = ["httpcomponents-httpmime", "httpcomponents-httpclient"] jackson = ["jackson-core", "jackson-databind", "jackson-datatype-jdk8", "jackson-datatype-jsr310", "jackson-module-kotlin"] kotlin = ["kotlin-stdlib-jdk8", "kotlin-reflect"] diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 76b43fb1..c9f048ee 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -16,6 +16,7 @@ */ package io.emeraldpay.dshackle.rpc +import com.google.protobuf.util.JsonFormat import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.api.proto.ReactorBlockchainGrpc @@ -223,11 +224,18 @@ class BlockchainRpc( } override fun subscribeNodeStatus(request: Mono): Flux { - val subId = RandomStringUtils.randomAlphanumeric(8) - log.debug("Subscription for node status with id [$subId] created") - return subscribeNodeStatus.subscribe(request).subscribeOn(scheduler) - .doOnError { failMetric.increment() } - .doOnNext { log.debug("Emitted next node status to [$subId] with data [$it] ") } + return request.flatMapMany { + val subId = it.traceId.takeIf { it.isNotBlank() } ?: RandomStringUtils.randomAlphanumeric(8) + subscribeNodeStatus.subscribe(it).subscribeOn(scheduler) + .doOnError { failMetric.increment() } + .doOnNext { + log.debug( + "Emitted next node status to [$subId] with data [${ + JsonFormat.printer().omittingInsignificantWhitespace().print(it) + }]" + ) + } + } } class RequestMetrics(val chain: Chain) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt index 306ecf05..7317b01a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt @@ -26,70 +26,73 @@ class SubscribeNodeStatus( private val log = LoggerFactory.getLogger(SubscribeNodeStatus::class.java) } - fun subscribe(req: Mono): Flux = - req.flatMapMany { - val knownUpstreams = ConcurrentHashMap>() - // subscribe on head/status updates for known upstreams - val upstreamUpdates = Flux.merge( - multistreams.all() - .flatMap { ms -> - ms.getAll().map { up -> - knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort() - subscribeUpstreamUpdates(ms, up, knownUpstreams[up.getId()]!!) - } + fun subscribe(req: SubscribeNodeStatusRequest): Flux { + val knownUpstreams = ConcurrentHashMap>() + // subscribe on head/status updates for known upstreams + val upstreamUpdates = Flux.merge( + multistreams.all() + .flatMap { ms -> + ms.getAll().map { up -> + knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort() + subscribeUpstreamUpdates(ms, up, knownUpstreams[up.getId()]!!) } - ) - - // stop removed upstreams update fluxes - val removals = Flux.merge( - multistreams.all() - .map { ms -> - ms.subscribeRemovedUpstreams().mapNotNull { up -> - knownUpstreams[up.getId()]?.let { - val result = it.tryEmitNext(true) - if (result.isFailure) { - log.warn("Unable to emit event about removal of an upstream - $result") - } - knownUpstreams.remove(up.getId()) - NodeStatusResponse.newBuilder() - .setNodeId(up.getId()) - .setDescription(buildDescription(ms, up)) - .setStatus(buildStatus(UpstreamAvailability.UNAVAILABLE, up.getHead().getCurrentHeight())) - .build() + } + ) + // stop removed upstreams update fluxes + val removals = Flux.merge( + multistreams.all() + .map { ms -> + ms.subscribeRemovedUpstreams().mapNotNull { up -> + knownUpstreams[up.getId()]?.let { + val result = it.tryEmitNext(true) + if (result.isFailure) { + log.warn("Unable to emit event about removal of an upstream - $result") } - } - } - ) - - // subscribe on head/status updates for just added upstreams - val multiStreamUpdates = Flux.merge( - multistreams.all() - .map { ms -> - ms.subscribeAddedUpstreams() - .distinctUntilChanged { - it.getId() - } - .filter { - !knownUpstreams.contains(it.getId()) - } - .flatMap { - knownUpstreams[it.getId()] = Sinks.many().multicast().directBestEffort() - Flux.concat( - Mono.just( - NodeStatusResponse.newBuilder() - .setNodeId(it.getId()) - .setDescription(buildDescription(ms, it)) - .setStatus(buildStatus(it.getStatus(), it.getHead().getCurrentHeight())) - .build() - ), - subscribeUpstreamUpdates(ms, it, knownUpstreams[it.getId()]!!) + knownUpstreams.remove(up.getId()) + NodeStatusResponse.newBuilder() + .setNodeId(up.getId()) + .setDescription(buildDescription(ms, up)) + .setStatus( + buildStatus( + UpstreamAvailability.UNAVAILABLE, + up.getHead().getCurrentHeight() + ) ) - } + .build() + } } - ) + } + ) - Flux.merge(upstreamUpdates, multiStreamUpdates, removals) - } + // subscribe on head/status updates for just added upstreams + val multiStreamUpdates = Flux.merge( + multistreams.all() + .map { ms -> + ms.subscribeAddedUpstreams() + .distinctUntilChanged { + it.getId() + } + .filter { + !knownUpstreams.contains(it.getId()) + } + .flatMap { + knownUpstreams[it.getId()] = Sinks.many().multicast().directBestEffort() + Flux.concat( + Mono.just( + NodeStatusResponse.newBuilder() + .setNodeId(it.getId()) + .setDescription(buildDescription(ms, it)) + .setStatus(buildStatus(it.getStatus(), it.getHead().getCurrentHeight())) + .build() + ), + subscribeUpstreamUpdates(ms, it, knownUpstreams[it.getId()]!!) + ) + } + } + ) + + return Flux.merge(upstreamUpdates, multiStreamUpdates, removals) + } private fun subscribeUpstreamUpdates( ms: Multistream, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index b2db9781..7aa700b4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -104,8 +104,8 @@ abstract class AbstractHead @JvmOverloads constructor( val newHead = choiceResult.nwhead lastHeadUpdated = System.currentTimeMillis() when (val result = stream.tryEmitNext(newHead)) { - OK -> log.debug("New block $upstreamId ${newHead.height} ${newHead.hash} @ ${this.javaClass}") - FAIL_ZERO_SUBSCRIBER -> log.debug("No subscribers $upstreamId ${this.javaClass}") + OK -> log.trace("New block $upstreamId ${newHead.height} ${newHead.hash} @ ${this.javaClass}") + FAIL_ZERO_SUBSCRIBER -> log.trace("No subscribers $upstreamId ${this.javaClass}") else -> log.warn("Failed to dispatch block $upstreamId: $result as ${this.javaClass}") } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index 3b52ecb7..5c674544 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt @@ -129,7 +129,7 @@ abstract class DefaultUpstream( Status(nLag, curr.avail, statusByLag(nLag, curr.avail)) }.also { statusStream.emitNext(it.status) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } - log.debug("Status of upstream [$id] changed to [$it], requested change lag to [$lag]") + log.trace("Status of upstream [$id] changed to [$it], requested change lag to [$lag]") } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt index da2c69c7..96b9b1eb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt @@ -27,10 +27,10 @@ class NoChoiceWithPriorityForkChoice( } override fun choose(block: BlockContainer): ForkChoice.ChoiceResult { - log.debug("Adding priority to $upstreamId block ${block.height}") + log.trace("Adding priority to $upstreamId block ${block.height}") val nwhead = head.updateAndGet { curr -> if (!filter(block)) { - log.debug("Already seen block ${block.height} from $upstreamId") + log.trace("Already seen block ${block.height} from $upstreamId") curr } else { seenBlocks.put(block.hash, true) @@ -38,7 +38,7 @@ class NoChoiceWithPriorityForkChoice( } } if (nwhead.hash == block.hash) { - log.debug("Accepted block ${block.height} from $upstreamId with $nodeRating") + log.trace("Accepted block ${block.height} from $upstreamId with $nodeRating") return ForkChoice.ChoiceResult.Updated(nwhead) } log.debug("Declined block ${block.height} from $upstreamId with $nodeRating") 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 f57099f2..fd4a7f30 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt @@ -95,7 +95,7 @@ class GrpcHead( log.error("Head subscription error. ${err.javaClass.name}:${err.message}", err) parent.setStatus(UpstreamAvailability.UNAVAILABLE) }.doOnNext { - log.info("Received block ${it.height}") + log.trace("Received block ${it.height}") } headSubscription = super.follow(blocks)