Merge pull request #11 from p2p-org/node-status-sub-id

id for subscription for node statuses

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

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"]

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<BlockchainOuterClass.SubscribeNodeStatusRequest>): Flux<BlockchainOuterClass.NodeStatusResponse> {
-        val subId = RandomStringUtils.randomAlphanumeric(8)
+        return request.flatMapMany {
-        log.debug("Subscription for node status with id [$subId] created")
+            val subId = it.traceId.takeIf { it.isNotBlank() } ?: RandomStringUtils.randomAlphanumeric(8)
-        return subscribeNodeStatus.subscribe(request).subscribeOn(scheduler)
+            subscribeNodeStatus.subscribe(it).subscribeOn(scheduler)
-            .doOnError { failMetric.increment() }
+                .doOnError { failMetric.increment() }
-            .doOnNext { log.debug("Emitted next node status to [$subId] with data [$it] ") }
+                .doOnNext {
+                    log.debug(
+                        "Emitted next node status to [$subId] with data [${
+                        JsonFormat.printer().omittingInsignificantWhitespace().print(it)
+                        }]"
+                    )
+                }
+        }
     }
 
     class RequestMetrics(val chain: Chain) {

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<SubscribeNodeStatusRequest>): Flux<NodeStatusResponse> =
+    fun subscribe(req: SubscribeNodeStatusRequest): Flux<NodeStatusResponse> {
-        req.flatMapMany {
+        val knownUpstreams = ConcurrentHashMap<String, Sinks.Many<Boolean>>()
-            val knownUpstreams = ConcurrentHashMap<String, Sinks.Many<Boolean>>()
+        // subscribe on head/status updates for known upstreams
-            // subscribe on head/status updates for known upstreams
+        val upstreamUpdates = Flux.merge(
-            val upstreamUpdates = Flux.merge(
+            multistreams.all()
-                multistreams.all()
+                .flatMap { ms ->
-                    .flatMap { ms ->
+                    ms.getAll().map { up ->
-                        ms.getAll().map { up ->
+                        knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort<Boolean>()
-                            knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort<Boolean>()
+                        subscribeUpstreamUpdates(ms, up, knownUpstreams[up.getId()]!!)
-                            subscribeUpstreamUpdates(ms, up, knownUpstreams[up.getId()]!!)
-                        }
                     }
-            )
+                }
-
+        )
-            // stop removed upstreams update fluxes
+        // stop removed upstreams update fluxes
-            val removals = Flux.merge(
+        val removals = Flux.merge(
-                multistreams.all()
+            multistreams.all()
-                    .map { ms ->
+                .map { ms ->
-                        ms.subscribeRemovedUpstreams().mapNotNull { up ->
+                    ms.subscribeRemovedUpstreams().mapNotNull { up ->
-                            knownUpstreams[up.getId()]?.let {
+                        knownUpstreams[up.getId()]?.let {
-                                val result = it.tryEmitNext(true)
+                            val result = it.tryEmitNext(true)
-                                if (result.isFailure) {
+                            if (result.isFailure) {
-                                    log.warn("Unable to emit event about removal of an upstream - $result")
+                                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()
                             }
-                        }
+                            knownUpstreams.remove(up.getId())
-                    }
+                            NodeStatusResponse.newBuilder()
-            )
+                                .setNodeId(up.getId())
-
+                                .setDescription(buildDescription(ms, up))
-            // subscribe on head/status updates for just added upstreams
+                                .setStatus(
-            val multiStreamUpdates = Flux.merge(
+                                    buildStatus(
-                multistreams.all()
+                                        UpstreamAvailability.UNAVAILABLE,
-                    .map { ms ->
+                                        up.getHead().getCurrentHeight()
-                        ms.subscribeAddedUpstreams()
+                                    )
-                            .distinctUntilChanged {
-                                it.getId()
-                            }
-                            .filter {
-                                !knownUpstreams.contains(it.getId())
-                            }
-                            .flatMap {
-                                knownUpstreams[it.getId()] = Sinks.many().multicast().directBestEffort<Boolean>()
-                                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()]!!)
                                 )
-                            }
+                                .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<Boolean>()
+                            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,

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}")
+                                OK -> log.trace("New block $upstreamId ${newHead.height} ${newHead.hash} @ ${this.javaClass}")
-                                FAIL_ZERO_SUBSCRIBER -> log.debug("No subscribers $upstreamId ${this.javaClass}")
+                                FAIL_ZERO_SUBSCRIBER -> log.trace("No subscribers $upstreamId ${this.javaClass}")
                                 else -> log.warn("Failed to dispatch block $upstreamId: $result as ${this.javaClass}")
                             }
                         }

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]")
             }
         }
     }

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")

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)