From a3a89ab9203501a4508c3ff4e573f6e5aaef2c65 Mon Sep 17 00:00:00 2001 From: Vadim Vlasov Date: Mon, 20 Mar 2023 20:20:30 +0800 Subject: [PATCH] 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 --- .../io/emeraldpay/dshackle/rpc/Describe.kt | 2 + .../dshackle/rpc/SubscribeNodeStatus.kt | 12 ++++- .../emeraldpay/dshackle/upstream/BuildInfo.kt | 17 +++++++ .../upstream/grpc/BitcoinGrpcUpstream.kt | 13 +++++- .../upstream/grpc/EthereumGrpcUpstream.kt | 13 +++++- .../upstream/grpc/EthereumPosGrpcUpstream.kt | 13 +++++- .../dshackle/upstream/grpc/GrpcUpstream.kt | 5 ++- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 6 ++- .../grpc/EthereumGrpcUpstreamSpec.groovy | 44 ++++++++++++++----- 9 files changed, 102 insertions(+), 23 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/BuildInfo.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index fd7828ee..509dc088 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt +++ b/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): Mono { 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) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt index f267688a..8d3ef6c6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt +++ b/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 = - NodeDescription.newBuilder() + private fun buildDescription(ms: Multistream, up: Upstream): NodeDescription.Builder { + 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() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/BuildInfo.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/BuildInfo.kt new file mode 100644 index 00000000..bce6e9c7 --- /dev/null +++ b/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) + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index 2b4020c7..fbafa702 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/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 = 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 } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 02cbc7ab..1ea00a97 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/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 = 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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt index 8f48f509..96c8e4da 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt +++ b/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 = 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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt index 877d1c0e..856a38a3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt +++ b/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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index 85e7e5ba..4391503a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -142,14 +142,16 @@ class GrpcUpstreams( } private fun processDescription(value: DescribeResponse): Flux { - 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) 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 fa0754c9..d140dc45 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy +++ b/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() - .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) - .addAllSupportedMethods(["eth_getBlockByHash"]) - .build()) + upstream.update( + BlockchainOuterClass.DescribeChain.newBuilder() + .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) + .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() - .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) - .addAllSupportedMethods(["eth_getBlockByHash"]) - .build()) + upstream.update( + BlockchainOuterClass.DescribeChain.newBuilder() + .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) + .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() - .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) - .addAllSupportedMethods(["eth_getBlockByHash"]) - .build()) + upstream.update( + BlockchainOuterClass.DescribeChain.newBuilder() + .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) + .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 }