add build info to upstream (#162)

* add build info to upstream

* fix tests

* add build info to node status subscription

* move proto field

* update proto submodule to master

src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.rpc
 
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
+import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
@@ -35,6 +36,7 @@ class Describe(
     fun describe(requestMono: Mono<BlockchainOuterClass.DescribeRequest>): Mono<BlockchainOuterClass.DescribeResponse> {
         return requestMono.map { _ ->
             val resp = BlockchainOuterClass.DescribeResponse.newBuilder()
+            resp.buildInfoBuilder.version = Global.version
             multistreamHolder.getAvailable().forEach { chain ->
                 multistreamHolder.getUpstream(chain).let { chainUpstreams ->
                     val status = subscribeStatus.chainStatus(chain, chainUpstreams.getStatus(), chainUpstreams)

src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt

@@ -10,6 +10,7 @@ import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Service
 import reactor.core.publisher.Flux
@@ -118,8 +119,8 @@ class SubscribeNodeStatus(
         return Flux.concat(currentState, statuses)
     }
 
-    private fun buildDescription(ms: Multistream, up: Upstream): NodeDescription.Builder =
+    private fun buildDescription(ms: Multistream, up: Upstream): NodeDescription.Builder {
-        NodeDescription.newBuilder()
+        val builder = NodeDescription.newBuilder()
             .setChain(Common.ChainRef.forNumber(ms.chain.id))
             .setNodeId(up.nodeId().toInt())
             .addAllNodeLabels(
@@ -138,6 +139,13 @@ class SubscribeNodeStatus(
             )
             .addAllSupportedSubscriptions(ms.getEgressSubscription().getAvailableTopics())
             .addAllSupportedMethods(up.getMethods().getSupportedMethods())
+        (up as? GrpcUpstream)?.let {
+            it.getBuildInfo().version?.let { version ->
+                builder.nodeBuildInfoBuilder.setVersion(version)
+            }
+        }
+        return builder
+    }
 
     private fun buildStatus(status: UpstreamAvailability, height: Long?): NodeStatus.Builder =
         NodeStatus.newBuilder()

src/main/kotlin/io/emeraldpay/dshackle/upstream/BuildInfo.kt

@@ -0,0 +1,17 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+
+data class BuildInfo(var version: String? = null) {
+    fun update(buildInfo: BuildInfo): Boolean {
+        val changed = buildInfo.version != version
+        version = buildInfo.version
+        return changed
+    }
+
+    companion object {
+        fun extract(buildInfo: BlockchainOuterClass.BuildInfo): BuildInfo {
+            return BuildInfo(buildInfo.version)
+        }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt

@@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.reader.JsonRpcReader
+import io.emeraldpay.dshackle.upstream.BuildInfo
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -103,6 +104,7 @@ class BitcoinGrpcUpstream(
     private val grpcHead = GrpcHead(getId(), chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
     private val timeout = Defaults.timeout
     private var capabilities: Set<Capability> = emptySet()
+    private val buildInfo: BuildInfo = BuildInfo()
 
     override fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub {
         return remote
@@ -149,11 +151,18 @@ class BitcoinGrpcUpstream(
     override fun stop() {
     }
 
-    override fun update(conf: BlockchainOuterClass.DescribeChain): Boolean {
+    override fun getBuildInfo(): BuildInfo {
+        return buildInfo
+    }
+
+    override fun update(conf: BlockchainOuterClass.DescribeChain, buildInfo: BlockchainOuterClass.BuildInfo): Boolean {
+        val newBuildInfo = BuildInfo.extract(buildInfo)
+        val buildInfoChanged = this.buildInfo.update(newBuildInfo)
         val newCapabilities = RemoteCapabilities.extract(conf)
         conf.status?.let { status -> onStatus(status) }
-        return (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also {
+        val upstreamStatusChanged = (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also {
             capabilities = newCapabilities
         }
+        return buildInfoChanged || upstreamStatusChanged
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt

@@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.upstream.BuildInfo
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -109,6 +110,7 @@ open class EthereumGrpcUpstream(
     private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
     private val grpcHead = GrpcHead(getId(), chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
     private var capabilities: Set<Capability> = emptySet()
+    private val buildInfo: BuildInfo = BuildInfo()
 
     private val defaultReader: JsonRpcReader = client.getReader()
     var timeout = Defaults.timeout
@@ -131,12 +133,19 @@ open class EthereumGrpcUpstream(
     override fun stop() {
     }
 
-    override fun update(conf: BlockchainOuterClass.DescribeChain): Boolean {
+    override fun getBuildInfo(): BuildInfo {
+        return buildInfo
+    }
+
+    override fun update(conf: BlockchainOuterClass.DescribeChain, buildInfo: BlockchainOuterClass.BuildInfo): Boolean {
+        val newBuildInfo = BuildInfo.extract(buildInfo)
+        val buildInfoChanged = this.buildInfo.update(newBuildInfo)
         val newCapabilities = RemoteCapabilities.extract(conf)
         conf.status?.let { status -> onStatus(status) }
-        return (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also {
+        val upstreamStatusChanged = (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also {
             capabilities = newCapabilities
         }
+        return buildInfoChanged || upstreamStatusChanged
     }
 
     override fun getQuorumByLabel(): QuorumForLabels {

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt

@@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.upstream.BuildInfo
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -81,6 +82,7 @@ open class EthereumPosGrpcUpstream(
     private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
     private val grpcHead = GrpcHead(getId(), chain, this, remote, blockConverter, null, NoChoiceWithPriorityForkChoice(nodeRating, parentId))
     private var capabilities: Set<Capability> = emptySet()
+    private val buildInfo: BuildInfo = BuildInfo()
 
     private val defaultReader: JsonRpcReader = client.getReader()
     private val timeout = Defaults.timeout
@@ -96,12 +98,19 @@ open class EthereumPosGrpcUpstream(
     override fun stop() {
     }
 
-    override fun update(conf: BlockchainOuterClass.DescribeChain): Boolean {
+    override fun getBuildInfo(): BuildInfo {
+        return buildInfo
+    }
+
+    override fun update(conf: BlockchainOuterClass.DescribeChain, buildInfo: BlockchainOuterClass.BuildInfo): Boolean {
+        val newBuildInfo = BuildInfo.extract(buildInfo)
+        val buildInfoChanged = this.buildInfo.update(newBuildInfo)
         val newCapabilities = RemoteCapabilities.extract(conf)
         conf.status?.let { status -> onStatus(status) }
-        return (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also {
+        val upstreamStatusChanged = (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also {
             capabilities = newCapabilities
         }
+        return buildInfoChanged || upstreamStatusChanged
     }
 
     override fun getQuorumByLabel(): QuorumForLabels {

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.grpc
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.BlockchainOuterClass.NativeSubscribeRequest
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
+import io.emeraldpay.dshackle.upstream.BuildInfo
 import io.emeraldpay.dshackle.upstream.Upstream
 import reactor.core.publisher.Flux
 
@@ -27,7 +28,9 @@ interface GrpcUpstream : Upstream {
      * Update the configuration of the upstream with the new data.
      * Called on the first creation, and each time a new state received from upstream
      */
-    fun update(conf: BlockchainOuterClass.DescribeChain): Boolean
+    fun update(conf: BlockchainOuterClass.DescribeChain, buildInfo: BlockchainOuterClass.BuildInfo): Boolean
+
+    fun getBuildInfo(): BuildInfo
 
     fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt

@@ -142,14 +142,16 @@ class GrpcUpstreams(
     }
 
     private fun processDescription(value: DescribeResponse): Flux<UpstreamChangeEvent> {
-        log.info("Start processing grpc upstream description for $id with chains ${value.chainsList.map { it.chain.name }}")
+        val chainNames = value.chainsList.map { it.chain.name }
+        val version = value.buildInfo.version
+        log.info("Start processing grpc upstream description for $id with chains $chainNames and version $version")
         val current = value.chainsList.filter {
             Chain.byId(it.chain.number) != Chain.UNSPECIFIED
         }.mapNotNull { chainDetails ->
             try {
                 val chain = Chain.byId(chainDetails.chain.number)
                 val up = getOrCreate(chain)
-                val changed = (up.upstream as GrpcUpstream).update(chainDetails)
+                val changed = (up.upstream as GrpcUpstream).update(chainDetails, value.buildInfo)
                 up.takeUnless {
                     changed && it.type == UpstreamChangeEvent.ChangeType.REVALIDATED
                 } ?: UpstreamChangeEvent(up.chain, up.upstream, UpstreamChangeEvent.ChangeType.UPDATED)

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy

@@ -28,6 +28,7 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.test.MockGrpcServer
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.BuildInfo
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
 import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
@@ -52,6 +53,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
     )
 
     def hash = (byte)123
+    def buildInfo = new BuildInfo("v0.0.1-test")
 
     def "Subscribe to head"() {
         setup:
@@ -86,16 +88,22 @@ class EthereumGrpcUpstreamSpec extends Specification {
         })
         def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default())
         upstream.setLag(0)
-        upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
+        upstream.update(
-                .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
+                BlockchainOuterClass.DescribeChain.newBuilder()
-                .addAllSupportedMethods(["eth_getBlockByHash"])
+                        .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
-                .build())
+                        .addAllSupportedMethods(["eth_getBlockByHash"])
+                        .build(),
+                BlockchainOuterClass.BuildInfo.newBuilder()
+                        .setVersion(buildInfo.version)
+                        .build(),
+        )
         when:
         new Thread({ Thread.sleep(50); upstream.head.start() }).start()
         def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1))
         then:
         callData.chain == Chain.ETHEREUM.id
         upstream.status == UpstreamAvailability.OK
+        upstream.getBuildInfo() == buildInfo
         h.hash == BlockId.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
     }
 
@@ -144,15 +152,21 @@ class EthereumGrpcUpstreamSpec extends Specification {
         })
         def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null, ChainsConfig.ChainConfig.default())
         upstream.setLag(0)
-        upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
+        upstream.update(
-                .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
+                BlockchainOuterClass.DescribeChain.newBuilder()
-                .addAllSupportedMethods(["eth_getBlockByHash"])
+                        .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
-                .build())
+                        .addAllSupportedMethods(["eth_getBlockByHash"])
+                        .build(),
+                BlockchainOuterClass.BuildInfo.newBuilder()
+                        .setVersion(buildInfo.version)
+                        .build(),
+        )
         when:
         new Thread({ Thread.sleep(50); upstream.head.start() }).start()
         def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block(Duration.ofSeconds(2))
         then:
         upstream.status == UpstreamAvailability.OK
+        upstream.getBuildInfo() == buildInfo
         h.hash == BlockId.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
         h.height == 650246
     }
@@ -206,16 +220,22 @@ class EthereumGrpcUpstreamSpec extends Specification {
         })
         def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default())
         upstream.setLag(0)
-        upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
+        upstream.update(
-                .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
+                BlockchainOuterClass.DescribeChain.newBuilder()
-                .addAllSupportedMethods(["eth_getBlockByHash"])
+                        .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
-                .build())
+                        .addAllSupportedMethods(["eth_getBlockByHash"])
+                        .build(),
+                BlockchainOuterClass.BuildInfo.newBuilder()
+                        .setVersion(buildInfo.version)
+                        .build(),
+        )
         when:
         new Thread({ Thread.sleep(50); upstream.head.start() }).start()
         finished.get()
         def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block(Duration.ofSeconds(2))
         then:
         upstream.status == UpstreamAvailability.OK
+        upstream.getBuildInfo() == buildInfo
         h.hash == BlockId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a")
         h.height == 650247
     }