From b9866fe19a6bc9cc1e5e9b3df0bd98a64cdbaf24 Mon Sep 17 00:00:00 2001 From: Maxksim Fomenkov Date: Wed, 17 Aug 2022 13:45:14 +0300 Subject: [PATCH] Support gRPC upstream labels --- docs/reference-configuration.adoc | 6 ++ .../dshackle/config/UpstreamsConfigReader.kt | 39 ++++---- .../dshackle/config/YamlConfigReader.kt | 36 +++---- .../dshackle/startup/ConfiguredUpstreams.kt | 3 +- .../connectors/EthereumWsConnector.kt | 5 - .../upstream/grpc/BitcoinGrpcUpstream.kt | 5 +- .../upstream/grpc/EthereumGrpcUpstream.kt | 5 +- .../upstream/grpc/EthereumPosGrpcUpstream.kt | 11 ++- .../upstream/grpc/GrpcUpstreamStatus.kt | 22 ++--- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 9 +- .../config/UpstreamsConfigReaderSpec.groovy | 3 +- .../grpc/EthereumGrpcUpstreamSpec.groovy | 6 +- .../grpc/GrpcUpstreamStatusSpec.groovy | 96 ++++++++++++++++++- src/test/resources/upstreams-labels.yaml | 2 + 14 files changed, 166 insertions(+), 82 deletions(-) diff --git a/docs/reference-configuration.adoc b/docs/reference-configuration.adoc index 4a61982e..0d691617 100644 --- a/docs/reference-configuration.adoc +++ b/docs/reference-configuration.adoc @@ -829,6 +829,8 @@ It's more effective, easier to secure connection, and allows to build a distribu [source,yaml] ---- - id: test1 + labels: + provider: some connection: grpc: host: eu-api.mycompany.com @@ -848,6 +850,10 @@ It's more effective, easier to secure connection, and allows to build a distribu | yes | Per-cluster identifier of an upstream +| `labels` +| no +| Defines the labels can be used for the proper upstream instance selection. Overrides the labels retrieved by the `describe` method + | `connection.grpc` | yes | Connection configuration for Dshackle gRPC diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index b13d9a02..2dc43a68 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -20,8 +20,6 @@ import io.emeraldpay.dshackle.FileResolver import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory import org.yaml.snakeyaml.nodes.MappingNode -import org.yaml.snakeyaml.nodes.ScalarNode -import reactor.util.function.Tuples import java.io.InputStream import java.net.URI import java.time.Duration @@ -83,7 +81,7 @@ class UpstreamsConfigReader( } } - getList(input, "upstreams")?.value?.forEachIndexed { _, upNode -> + getList(input, "upstreams")?.value?.forEach { upNode -> val connNode = getMapping(upNode, "connection") if (hasAny(connNode, "ethereum")) { readUpstream(config, upNode) { @@ -211,7 +209,11 @@ class UpstreamsConfigReader( return connection } - private fun readUpstream(config: UpstreamsConfig, upNode: MappingNode, connFactory: () -> T) { + private fun readUpstream( + config: UpstreamsConfig, + upNode: MappingNode, + connFactory: () -> T + ) { val upstream = UpstreamsConfig.Upstream() readUpstreamCommon(upNode, upstream) readUpstreamStandard(upNode, upstream) @@ -241,18 +243,21 @@ class UpstreamsConfigReader( getValueAsBool(upNode, "enabled")?.let { upstream.isEnabled = it } + if (hasAny(upNode, "labels")) { + getMapping(upNode, "labels")?.let { labels -> + labels.value.stream() + .map { it.keyNode.valueAsString() to it.valueNode.valueAsString() } + .filter { StringUtils.isNotBlank(it.first) && StringUtils.isNotBlank(it.second) } + .forEach { + upstream.labels[it.first!!.trim()] = it.second!!.trim() + } + } + } } internal fun readUpstreamGrpc( upNode: MappingNode, ) { - // Dshackle gRPC connection dispatches requests to different upstreams, which may - // be on different blockchains, and each may have different set of labels. - // So the labels and chains assigned to the gRPC connection make no sense. - if (hasAny(upNode, "labels")) { - // Actual labels from underlying upstreams are handled by GrpcUpstreamStatus - log.warn("Labels should be not applied to gRPC upstream") - } if (hasAny(upNode, "chain")) { log.warn("Chain should be not applied to gRPC upstream") } @@ -272,18 +277,6 @@ class UpstreamsConfigReader( log.warn("Unsupported role `$name` for upstream ${upstream.id}") } } - if (hasAny(upNode, "labels")) { - getMapping(upNode, "labels")?.let { labels -> - labels.value.stream() - .filter { n -> n.keyNode is ScalarNode && n.valueNode is ScalarNode } - .map { n -> Tuples.of((n.keyNode as ScalarNode).value, (n.valueNode as ScalarNode).value) } - .map { kv -> Tuples.of(kv.t1.trim(), kv.t2.trim()) } - .filter { kv -> StringUtils.isNotEmpty(kv.t1) && StringUtils.isNotEmpty(kv.t2) } - .forEach { kv -> - upstream.labels[kv.t1] = kv.t2 - } - } - } } internal fun tryReadOptions(upNode: MappingNode): UpstreamsConfig.Options? { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt index c358f999..aa256d60 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt @@ -37,18 +37,12 @@ abstract class YamlConfigReader { return asMappingNode(yaml.compose(InputStreamReader(input))) } - protected fun hasAny(mappingNode: MappingNode?, key: String): Boolean { - if (mappingNode == null) { - return false - } - return mappingNode.value - .stream() - .filter { n -> n.keyNode is ScalarNode } - .filter { n -> - val sn = n.keyNode as ScalarNode - key == sn.value - }.count() > 0 - } + protected fun hasAny(mappingNode: MappingNode?, key: String): Boolean = + mappingNode?.let { node -> + node.value + .stream() + .anyMatch { it.keyNode.valueAsString() == key } + } ?: false @Suppress("UNCHECKED_CAST") private fun getValue(mappingNode: MappingNode?, key: String, type: Class): T? { @@ -57,21 +51,15 @@ abstract class YamlConfigReader { } return mappingNode.value .stream() - .filter { n -> n.keyNode is ScalarNode && type.isAssignableFrom(n.valueNode.javaClass) } - .filter { n -> - val sn = n.keyNode as ScalarNode - key == sn.value - } + .filter { type.isAssignableFrom(it.valueNode.javaClass) } + .filter { it.keyNode.valueAsString() == key } .map { n -> n.valueNode as T } - .findFirst().let { - if (it.isPresent) { - it.get() - } else { - null - } - } + .findFirst() + .orElse(null) } + fun Node.valueAsString(): String? = if (this is ScalarNode) this.value else null + protected fun getMapping(mappingNode: MappingNode?, key: String): MappingNode? { return getValue(mappingNode, key, MappingNode::class.java) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 819527ff..b9bff2e4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -256,7 +256,8 @@ open class ConfiguredUpstreams( endpoint.port, endpoint.auth, fileResolver, - endpoint.upstreamRating + endpoint.upstreamRating, + config.labels ).apply { timeout = options.timeout } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt index 369bb564..ec670e7f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt @@ -11,12 +11,7 @@ import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient -import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics import io.emeraldpay.grpc.Chain -import io.micrometer.core.instrument.Counter -import io.micrometer.core.instrument.Metrics -import io.micrometer.core.instrument.Tag -import io.micrometer.core.instrument.Timer class EthereumWsConnector( wsFactory: EthereumWsFactory, 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 4d77df23..0117efe0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -50,7 +50,8 @@ class BitcoinGrpcUpstream( role: UpstreamsConfig.UpstreamRole, chain: Chain, val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, - private val client: JsonRpcGrpcClient + private val client: JsonRpcGrpcClient, + overrideLabels: UpstreamsConfig.Labels? ) : BitcoinUpstream( "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}", chain, @@ -95,7 +96,7 @@ class BitcoinGrpcUpstream( } } } - private val upstreamStatus = GrpcUpstreamStatus() + private val upstreamStatus = GrpcUpstreamStatus(overrideLabels) private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice()) var timeout = Defaults.timeout private var capabilities: Set = emptySet() 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 db01eaea..fd0674ff 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -49,7 +49,8 @@ open class EthereumGrpcUpstream( role: UpstreamsConfig.UpstreamRole, private val chain: Chain, private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, - private val client: JsonRpcGrpcClient + private val client: JsonRpcGrpcClient, + overrideLabels: UpstreamsConfig.Labels? ) : EthereumUpstream( "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}", UpstreamsConfig.Options.getDefaults(), @@ -93,7 +94,7 @@ open class EthereumGrpcUpstream( } private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java) - private val upstreamStatus = GrpcUpstreamStatus() + private val upstreamStatus = GrpcUpstreamStatus(overrideLabels) private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice()) private var capabilities: Set = emptySet() 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 1f0e7807..256f0a4f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt @@ -24,7 +24,11 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels -import io.emeraldpay.dshackle.upstream.* +import io.emeraldpay.dshackle.upstream.Capability +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice @@ -50,7 +54,8 @@ open class EthereumPosGrpcUpstream( private val chain: Chain, private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, client: JsonRpcGrpcClient, - nodeRating: Int + nodeRating: Int, + overrideLabels: UpstreamsConfig.Labels? ) : EthereumPosUpstream( "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}", UpstreamsConfig.Options.getDefaults(), @@ -94,7 +99,7 @@ open class EthereumPosGrpcUpstream( } private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java) - private val upstreamStatus = GrpcUpstreamStatus() + private val upstreamStatus = GrpcUpstreamStatus(overrideLabels) private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, NoChoiceWithPriorityForkChoice(nodeRating)) private var capabilities: Set = emptySet() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatus.kt index d64348b3..335f97cc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatus.kt @@ -24,14 +24,15 @@ import org.slf4j.LoggerFactory import java.util.Collections import java.util.concurrent.atomic.AtomicReference -class GrpcUpstreamStatus { - +class GrpcUpstreamStatus( + private val overrideLabels: UpstreamsConfig.Labels? +) { companion object { private val log = LoggerFactory.getLogger(GrpcUpstreamStatus::class.java) } private val allLabels: AtomicReference> = AtomicReference(emptyList()) - private val nodes = AtomicReference(QuorumForLabels()) + private val nodes = AtomicReference(QuorumForLabels()) private var targets: CallMethods? = null fun update(conf: BlockchainOuterClass.DescribeChain) { @@ -39,18 +40,17 @@ class GrpcUpstreamStatus { val updateNodes = QuorumForLabels() conf.nodesList.forEach { remoteNode -> + val labels = UpstreamsConfig.Labels() + remoteNode.labelsList.forEach { + labels[it.name] = it.value + } + overrideLabels?.let { labels.putAll(it) } val node = QuorumForLabels.QuorumItem( remoteNode.quorum, - remoteNode.labelsList.let { provided -> - val labels = UpstreamsConfig.Labels() - provided.forEach { - labels[it.name] = it.value - } - updateLabels.add(labels) - labels - } + labels ) updateNodes.add(node) + updateLabels.add(labels) } this.nodes.set(updateNodes) 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 01a6d68b..0fafb099 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -57,7 +57,8 @@ class GrpcUpstreams( private val port: Int, private val auth: AuthConfig.ClientTlsAuth? = null, private val fileResolver: FileResolver, - private val nodeRating: Int + private val nodeRating: Int, + private val labels: UpstreamsConfig.Labels ) { private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java) @@ -205,7 +206,7 @@ class GrpcUpstreams( val current = known[chain] return if (current == null) { val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics) - val created = EthereumGrpcUpstream(id, role, chain, client!!, rpcClient) + val created = EthereumGrpcUpstream(id, role, chain, client!!, rpcClient, labels) created.timeout = this.timeout known[chain] = created created.start() @@ -221,7 +222,7 @@ class GrpcUpstreams( val current = known[chain] return if (current == null) { val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics) - val created = EthereumPosGrpcUpstream(id, role, chain, client!!, rpcClient, nodeRating) + val created = EthereumPosGrpcUpstream(id, role, chain, client!!, rpcClient, nodeRating, labels) created.timeout = this.timeout known[chain] = created created.start() @@ -237,7 +238,7 @@ class GrpcUpstreams( val current = known[chain] return if (current == null) { val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics) - val created = BitcoinGrpcUpstream(id, role, chain, client!!, rpcClient) + val created = BitcoinGrpcUpstream(id, role, chain, client!!, rpcClient, labels) created.timeout = this.timeout known[chain] = created created.start() diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 99b81f93..2ffdb0fb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -233,7 +233,8 @@ class UpstreamsConfigReaderSpec extends Specification { act.upstreams.size() == 2 with(act.upstreams.get(0)) { connection instanceof UpstreamsConfig.GrpcConnection - labels.isEmpty() + labels.size() == 1 + labels["provider"] == "some_service" } with(act.upstreams.get(1)) { !labels.isEmpty() 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 b98904a1..5418684b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy @@ -81,7 +81,7 @@ class EthereumGrpcUpstreamSpec extends Specification { ) } }) - def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics)) + def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null) upstream.setLag(0) upstream.update(BlockchainOuterClass.DescribeChain.newBuilder() .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) @@ -139,7 +139,7 @@ class EthereumGrpcUpstreamSpec extends Specification { ) } }) - def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics)) + def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null) upstream.setLag(0) upstream.update(BlockchainOuterClass.DescribeChain.newBuilder() .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) @@ -201,7 +201,7 @@ class EthereumGrpcUpstreamSpec extends Specification { finished.complete(true) } }) - def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics)) + def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null) upstream.setLag(0) upstream.update(BlockchainOuterClass.DescribeChain.newBuilder() .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatusSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatusSpec.groovy index 2f1df05b..ad98480e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatusSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatusSpec.groovy @@ -25,7 +25,7 @@ class GrpcUpstreamStatusSpec extends Specification { def "Updates with new labels"() { setup: - def status = new GrpcUpstreamStatus() + def status = new GrpcUpstreamStatus(null) when: status.update( BlockchainOuterClass.DescribeChain.newBuilder() @@ -86,9 +86,74 @@ class GrpcUpstreamStatusSpec extends Specification { ] } + def "Updates with new labels override"() { + setup: + def status = new GrpcUpstreamStatus(UpstreamsConfig.Labels.fromMap([fix: "value"])) + when: + status.update( + BlockchainOuterClass.DescribeChain.newBuilder() + .addNodes( + BlockchainOuterClass.NodeDetails.newBuilder() + .setQuorum(1) + .addLabels( + BlockchainOuterClass.Label.newBuilder().setName("test").setValue("foo") + ) + ) + .build() + ) + def act = status.getLabels() + then: + act.toList() == [ + UpstreamsConfig.Labels.fromMap([test: "foo", fix: "value"]) + ] + + // replace with new value + when: + status.update( + BlockchainOuterClass.DescribeChain.newBuilder() + .addNodes( + BlockchainOuterClass.NodeDetails.newBuilder() + .setQuorum(1) + .addLabels( + BlockchainOuterClass.Label.newBuilder().setName("test").setValue("bar") + ).addLabels( + BlockchainOuterClass.Label.newBuilder().setName("fix").setValue("val") + ) + ) + .build() + ) + act = status.getLabels() + then: + act.toList() == [ + UpstreamsConfig.Labels.fromMap([test: "bar", fix: "value"]) + ] + + // more values + when: + status.update( + BlockchainOuterClass.DescribeChain.newBuilder() + .addNodes( + BlockchainOuterClass.NodeDetails.newBuilder() + .setQuorum(1) + .addLabels( + BlockchainOuterClass.Label.newBuilder().setName("test1").setValue("bar") + ) + .addLabels( + BlockchainOuterClass.Label.newBuilder().setName("test2").setValue("baz") + ) + ) + .build() + ) + act = status.getLabels() + then: + act.toList() == [ + UpstreamsConfig.Labels.fromMap([test1: "bar", test2: "baz", fix: "value"]) + ] + } + def "Updates with new nodes"() { setup: - def status = new GrpcUpstreamStatus() + def status = new GrpcUpstreamStatus(null) when: status.update( BlockchainOuterClass.DescribeChain.newBuilder() @@ -108,9 +173,34 @@ class GrpcUpstreamStatusSpec extends Specification { } } + def "Updates with new nodes override labels"() { + setup: + def status = new GrpcUpstreamStatus(UpstreamsConfig.Labels.fromMap([fix: "value"])) + when: + status.update( + BlockchainOuterClass.DescribeChain.newBuilder() + .addNodes( + BlockchainOuterClass.NodeDetails.newBuilder() + .setQuorum(1) + .addLabels( + BlockchainOuterClass.Label.newBuilder().setName("test").setValue("foo") + ) + .addLabels( + BlockchainOuterClass.Label.newBuilder().setName("fix").setValue("val") + ) + ) + .build() + ) + def act = status.getNodes() + then: + act == new QuorumForLabels().tap { + it.add(new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap([test: "foo", fix: "value"]))) + } + } + def "Updates with methods"() { setup: - def status = new GrpcUpstreamStatus() + def status = new GrpcUpstreamStatus(null) when: status.update( BlockchainOuterClass.DescribeChain.newBuilder() diff --git a/src/test/resources/upstreams-labels.yaml b/src/test/resources/upstreams-labels.yaml index 8523ae30..c8156382 100644 --- a/src/test/resources/upstreams-labels.yaml +++ b/src/test/resources/upstreams-labels.yaml @@ -8,6 +8,8 @@ defaultOptions: upstreams: - id: remote + labels: + provider: some_service connection: grpc: host: "10.2.0.15"