Merge pull request #3 from p2p-org/support_grpc_labels

Support gRPC upstream labels

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

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<MappingNode>(input, "upstreams")?.value?.forEachIndexed { _, upNode ->
+        getList<MappingNode>(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 <T : UpstreamsConfig.UpstreamConnection> readUpstream(config: UpstreamsConfig, upNode: MappingNode, connFactory: () -> T) {
+    private fun <T : UpstreamsConfig.UpstreamConnection> readUpstream(
+        config: UpstreamsConfig,
+        upNode: MappingNode,
+        connFactory: () -> T
+    ) {
         val upstream = UpstreamsConfig.Upstream<T>()
         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? {

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 {
+    protected fun hasAny(mappingNode: MappingNode?, key: String): Boolean =
-        if (mappingNode == null) {
+        mappingNode?.let { node ->
-            return false
+            node.value
-        }
+                .stream()
-        return mappingNode.value
+                .anyMatch { it.keyNode.valueAsString() == key }
-            .stream()
+        } ?: false
-            .filter { n -> n.keyNode is ScalarNode }
-            .filter { n ->
-                val sn = n.keyNode as ScalarNode
-                key == sn.value
-            }.count() > 0
-    }
 
     @Suppress("UNCHECKED_CAST")
     private fun <T> getValue(mappingNode: MappingNode?, key: String, type: Class<T>): T? {
@@ -57,21 +51,15 @@ abstract class YamlConfigReader {
         }
         return mappingNode.value
             .stream()
-            .filter { n -> n.keyNode is ScalarNode && type.isAssignableFrom(n.valueNode.javaClass) }
+            .filter { type.isAssignableFrom(it.valueNode.javaClass) }
-            .filter { n ->
+            .filter { it.keyNode.valueAsString() == key }
-                val sn = n.keyNode as ScalarNode
-                key == sn.value
-            }
             .map { n -> n.valueNode as T }
-            .findFirst().let {
+            .findFirst()
-                if (it.isPresent) {
+            .orElse(null)
-                    it.get()
-                } else {
-                    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)
     }

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
         }

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,

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<Capability> = emptySet()

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<Capability> = emptySet()
 

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<Capability> = emptySet()
 

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<Collection<UpstreamsConfig.Labels>> = AtomicReference(emptyList())
-    private val nodes = AtomicReference<QuorumForLabels>(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 ->
+                labels
-                    val labels = UpstreamsConfig.Labels()
-                    provided.forEach {
-                        labels[it.name] = it.value
-                    }
-                    updateLabels.add(labels)
-                    labels
-                }
             )
             updateNodes.add(node)
+            updateLabels.add(labels)
         }
 
         this.nodes.set(updateNodes)

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

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

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

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

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"