solution: introduce labels for upstreams

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.config
 import java.net.URI
 import java.util.*
 import kotlin.collections.ArrayList
+import kotlin.collections.HashMap
 
 class UpstreamsConfig {
     var version: String? = null
@@ -50,10 +51,10 @@ class UpstreamsConfig {
     class Upstream<T : UpstreamConnection> {
         var id: String? = null
         var chain: String? = null
-        var provider: String? = null
         var options: Options? = null
         var isEnabled = true
         var connection: T? = null
+        val labels = Labels()
     }
 
     open class UpstreamConnection
@@ -91,6 +92,10 @@ class UpstreamsConfig {
         var key: String? = null
     }
 
+    //TODO make it unmodifiable after initial load
+    class Labels: HashMap<String, String>() {
+    }
+
     enum class UpstreamType private constructor(vararg code: String) {
         ETHEREUM_JSON_RPC("ethereum"),
         DSHACKLE("dshackle", "grpc"),

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -1,11 +1,13 @@
 package io.emeraldpay.dshackle.config
 
+import org.apache.commons.lang3.StringUtils
 import org.slf4j.LoggerFactory
 import org.yaml.snakeyaml.Yaml
 import org.yaml.snakeyaml.nodes.CollectionNode
 import org.yaml.snakeyaml.nodes.MappingNode
 import org.yaml.snakeyaml.nodes.Node
 import org.yaml.snakeyaml.nodes.ScalarNode
+import reactor.util.function.Tuples
 import java.io.InputStream
 import java.io.InputStreamReader
 import java.lang.IllegalArgumentException
@@ -44,9 +46,8 @@ class UpstreamsConfigReader {
             if (hasAny(connNode, "ethereum")) {
                 val connConfigNode = getMapping(connNode, "ethereum")!!
                 val upstream = UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>()
-                upstream.id = getValueAsString(upNode, "id")
+                readUpstreamCommon(upNode, upstream)
-                upstream.provider = getValueAsString(upNode, "provider")
+                readUpstreamEthereum(upNode, upstream)
-                upstream.chain = getValueAsString(upNode, "chain")
                 config.upstreams.add(upstream)
                 val connection = UpstreamsConfig.EthereumConnection()
                 upstream.connection = connection
@@ -70,8 +71,8 @@ class UpstreamsConfigReader {
             } else if (hasAny(connNode, "grpc")) {
                 val connConfigNode = getMapping(connNode, "grpc")!!
                 val upstream = UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>()
-                upstream.id = getValueAsString(upNode, "id")
+                readUpstreamCommon(upNode, upstream)
-                upstream.provider = getValueAsString(upNode, "provider")
+                readUpstreamGrpc(upNode, upstream)
                 config.upstreams.add(upstream)
                 val connection = UpstreamsConfig.GrpcConnection()
                 upstream.connection = connection
@@ -88,6 +89,35 @@ class UpstreamsConfigReader {
         return config
     }
 
+    internal fun readUpstreamCommon(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) {
+        upstream.id = getValueAsString(upNode, "id")
+    }
+
+    internal fun readUpstreamGrpc(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>) {
+        if (hasAny(upNode, "labels")) {
+            log.warn("Labels are not applied to gRPC upstream")
+        }
+        if (hasAny(upNode, "chain")) {
+            log.warn("Chain is not applied to gRPC upstream")
+        }
+    }
+
+    internal fun readUpstreamEthereum(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>) {
+        upstream.chain = getValueAsString(upNode, "chain")
+        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
+                        }
+            }
+        }
+    }
+
     private fun readAuth(authNode: MappingNode?): UpstreamsConfig.Auth? {
         return getValueAsString(authNode, "type")?.let {
             return when (it) {

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

@@ -2,10 +2,7 @@ package io.emeraldpay.dshackle.rpc
 
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
-import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import io.emeraldpay.dshackle.upstream.*
-import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams
-import io.emeraldpay.dshackle.upstream.Upstream
-import io.emeraldpay.dshackle.upstream.Upstreams
 import io.emeraldpay.grpc.Chain
 import io.grpc.stub.StreamObserver
 import org.springframework.beans.factory.annotation.Autowired
@@ -23,18 +20,34 @@ class Describe(
             val resp = BlockchainOuterClass.DescribeResponse.newBuilder()
             upstreams.getAvailable().forEach { chain ->
                 upstreams.getUpstream(chain)?.let { chainUpstreams ->
+                    val status = subscribeStatus.chainStatus(chain, chainUpstreams.getAll())
+                    val targets = chainUpstreams.getSupportedTargets()
+                    val chainDescription = BlockchainOuterClass.DescribeChain.newBuilder()
+                            .setChain(Common.ChainRef.forNumber(chain.id))
+                            .addAllSupportedTargets(targets)
+                            .setStatus(status)
                     chainUpstreams.getAll().let { ups ->
-                        if (ups.isNotEmpty()) {
+                        ups.forEach { up ->
-                            val status = subscribeStatus.chainStatus(chain, ups)
+                            val nodes = NodeDetailsList()
-                            val targets = chainUpstreams.getSupportedTargets()
+                            if (up is EthereumUpstream) {
-                            val chainDescription = BlockchainOuterClass.DescribeChain.newBuilder()
+                                nodes.add(up.node)
-                                    .setChain(Common.ChainRef.forNumber(chain.id))
+                            } else if (up is GrpcUpstream) {
-                                    .addAllSupportedTargets(targets)
+                                nodes.add(up.getNodes())
-                                    .setStatus(status)
+                            }
-                                    .build()
+                            nodes.getNodes().forEach { node ->
-                            resp.addChains(chainDescription)
+                                val nodeDetails = BlockchainOuterClass.NodeDetails.newBuilder()
+                                        .setQuorum(node.quorum)
+                                        .addAllLabels(node.labels.entries.map { label ->
+                                            BlockchainOuterClass.Label.newBuilder()
+                                                    .setName(label.key)
+                                                    .setValue(label.value)
+                                                    .build()
+                                        })
+                                chainDescription.addNodes(nodeDetails)
+                            }
                         }
                     }
+                    resp.addChains(chainDescription.build())
                 }
             }
             resp.build()

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

@@ -34,7 +34,7 @@ class SubscribeStatus(
     fun chainStatus(chain: Chain, ups: List<Upstream>): BlockchainOuterClass.ChainStatus {
         val available = ups.map { u ->
             u.getStatus()
-        }.min()!!
+        }.min() ?: UpstreamAvailability.UNAVAILABLE
         val quorum = ups.filter {
             it.getStatus() > UpstreamAvailability.UNAVAILABLE
         }.count()

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

@@ -12,8 +12,6 @@ import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.core.env.Environment
 import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Repository
-import reactor.core.publisher.Flux
-import reactor.core.publisher.TopicProcessor
 import reactor.core.publisher.toFlux
 import java.io.File
 import java.net.URI
@@ -47,11 +45,11 @@ open class ConfiguredUpstreams(
             val options = (up.options ?: UpstreamsConfig.Options())
                     .merge(UpstreamsConfig.Options.getDefaults())
 
-            if (up.provider == "dshackle") {
+            if (up.connection is UpstreamsConfig.GrpcConnection) {
                 buildGrpcUpstream(up.connection as UpstreamsConfig.GrpcConnection, options)
             } else {
                 val chain = chainNames[up.chain] ?: return
-                buildEthereumUpstream(up.connection as UpstreamsConfig.EthereumConnection, chain, options)
+                buildEthereumUpstream(up.connection as UpstreamsConfig.EthereumConnection, chain, options, up.labels)
             }
         }
     }
@@ -93,7 +91,8 @@ open class ConfiguredUpstreams(
 
     private fun buildEthereumUpstream(up: UpstreamsConfig.EthereumConnection,
                                       chain: Chain,
-                                      options: UpstreamsConfig.Options) {
+                                      options: UpstreamsConfig.Options,
+                                      labels: UpstreamsConfig.Labels) {
         var rpcApi: EthereumApi? = null
         var wsApi: EthereumWs? = null
         val urls = ArrayList<URI>()
@@ -115,7 +114,7 @@ open class ConfiguredUpstreams(
         }
         if (rpcApi != null) {
             log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
-            getOrCreateUpstream(chain).addUpstream(EthereumUpstream(chain, rpcApi!!, wsApi, options))
+            getOrCreateUpstream(chain).addUpstream(EthereumUpstream(chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels)))
         }
     }
 

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

@@ -13,7 +13,8 @@ class EthereumUpstream(
         val chain: Chain,
         private val api: EthereumApi,
         private val ethereumWs: EthereumWs? = null,
-        private val options: UpstreamsConfig.Options
+        private val options: UpstreamsConfig.Options,
+        val node: NodeDetailsList.NodeDetails
 ): Upstream {
 
     override fun getSupportedTargets(): Set<String> {

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

@@ -23,7 +23,6 @@ import java.time.Duration
 import java.util.*
 import java.util.concurrent.atomic.AtomicReference
 import java.util.function.Function
-import kotlin.collections.ArrayList
 
 open class GrpcUpstream(
         private val chain: Chain,
@@ -39,7 +38,8 @@ open class GrpcUpstream(
 
     private val headBlock = AtomicReference<BlockJson<TransactionId>>(null)
     private val streamBlocks: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create()
-    private var status = AtomicReference<UpstreamAvailability>(UpstreamAvailability.UNAVAILABLE)
+    private val status = AtomicReference<UpstreamAvailability>(UpstreamAvailability.UNAVAILABLE)
+    private val nodes = AtomicReference<NodeDetailsList>(NodeDetailsList())
     private val head = Head(this)
     private val api: EthereumApi
     private val statusStream: TopicProcessor<UpstreamAvailability> = TopicProcessor.create()
@@ -97,6 +97,18 @@ open class GrpcUpstream(
 
     fun init(conf: BlockchainOuterClass.DescribeChain) {
         supportedMethods.addAll(conf.supportedTargetsList)
+        val nodes = NodeDetailsList()
+        conf.nodesList.forEach { node ->
+            val node = NodeDetailsList.NodeDetails(node.quorum,
+                    node.labelsList.let { provided ->
+                        val labels = UpstreamsConfig.Labels()
+                        provided.forEach { labels.put(it.name, it.value) }
+                        labels
+                    }
+            )
+            nodes.add(node)
+        }
+        this.nodes.set(nodes)
         conf.status?.let { status -> onStatus(status) }
     }
 
@@ -112,6 +124,11 @@ open class GrpcUpstream(
         status.set(value)
         statusStream.onNext(value)
     }
+
+    fun getNodes(): NodeDetailsList {
+        return nodes.get()
+    }
+
     // ------------------------------------------------------------------------------------------
 
     override fun getSupportedTargets(): Set<String> {

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

@@ -0,0 +1,43 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import java.util.*
+import java.util.concurrent.locks.ReentrantReadWriteLock
+import kotlin.collections.ArrayList
+import kotlin.concurrent.read
+import kotlin.concurrent.write
+
+class NodeDetailsList {
+
+    private val lock = ReentrantReadWriteLock()
+    private val nodes = ArrayList<NodeDetails>()
+
+    fun add(node: NodeDetails) {
+        lock.read {
+            val existing = nodes.find { it.labels == node.labels }
+            lock.write {
+                if (existing != null) {
+                    val merged = NodeDetails(existing.quorum + node.quorum, existing.labels)
+                    nodes.remove(existing)
+                    nodes.add(merged)
+                } else {
+                    nodes.add(node)
+                }
+            }
+        }
+    }
+
+    fun add(nodes: NodeDetailsList) {
+        nodes.nodes.forEach { node -> this.add(node) }
+    }
+
+    fun getNodes(): List<NodeDetails> {
+        return Collections.unmodifiableList(nodes)
+    }
+
+    class NodeDetails(val quorum: Int, val labels: UpstreamsConfig.Labels) {
+        companion object {
+        }
+    }
+
+}

src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy

@@ -26,7 +26,6 @@ class UpstreamsConfigReaderSpec extends Specification {
         with(act.upstreams.get(0)) {
             id == "local"
             chain == "ethereum"
-            provider == "geth"
             connection instanceof UpstreamsConfig.EthereumConnection
             with((UpstreamsConfig.EthereumConnection)connection) {
                 rpc != null
@@ -38,7 +37,6 @@ class UpstreamsConfigReaderSpec extends Specification {
         with(act.upstreams.get(1)) {
             id == "infura"
             chain == "ethereum"
-            provider == "infura"
             connection instanceof UpstreamsConfig.EthereumConnection
             with((UpstreamsConfig.EthereumConnection)connection) {
                 rpc.url == new URI("https://mainnet.infura.io/v3/fa28c968191849c1aff541ad1d8511f2")
@@ -63,7 +61,6 @@ class UpstreamsConfigReaderSpec extends Specification {
         act.upstreams.size() == 1
         with(act.upstreams.get(0)) {
             id == "remote"
-            provider == "dshackle"
             connection instanceof UpstreamsConfig.GrpcConnection
             with((UpstreamsConfig.GrpcConnection)connection) {
                 host == "10.2.0.15"
@@ -76,4 +73,24 @@ class UpstreamsConfigReaderSpec extends Specification {
             }
         }
     }
+
+    def "Parse config with labels"() {
+        setup:
+        def config = this.class.getClassLoader().getResourceAsStream("upstreams-labels.yaml")
+        when:
+        def act = reader.read(config)
+        then:
+        act != null
+        act.upstreams.size() == 2
+        with(act.upstreams.get(0)) {
+            connection instanceof UpstreamsConfig.GrpcConnection
+            labels.isEmpty()
+        }
+        with(act.upstreams.get(1)) {
+            !labels.isEmpty()
+            labels.size() == 2
+            labels["fullnode"] == "true"
+            labels["api"] == "geth"
+        }
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/NodeDetailsListSpec.groovy

@@ -0,0 +1,119 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import org.codehaus.groovy.runtime.DefaultGroovyMethods
+import spock.lang.Specification
+
+class NodeDetailsListSpec extends Specification {
+
+    def "Adds new node"() {
+        setup:
+        def list = new NodeDetailsList()
+        when:
+        list.add(new NodeDetailsList.NodeDetails(1, asLabels([foo: "bar"])))
+        then:
+        list.nodes.size() == 1
+        with(list.nodes.get(0)) {
+            quorum == 1
+            DefaultGroovyMethods.equals(labels, [foo: "bar"])
+        }
+
+        when:
+        list.add(new NodeDetailsList.NodeDetails(2, asLabels([foo: "not-bar"])))
+        then:
+        list.nodes.size() == 2
+        with(list.nodes.get(0)) {
+            quorum == 1
+            DefaultGroovyMethods.equals(labels, [foo: "bar"])
+        }
+        with(list.nodes.get(1)) {
+            quorum == 2
+            DefaultGroovyMethods.equals(labels, [foo: "not-bar"])
+        }
+
+        when:
+        list.add(new NodeDetailsList.NodeDetails(1, asLabels([foo: "bar", baz: "baz"])))
+        then:
+        list.nodes.size() == 3
+        with(list.nodes.get(0)) {
+            quorum == 1
+            DefaultGroovyMethods.equals(labels, [foo: "bar"])
+        }
+        with(list.nodes.get(1)) {
+            quorum == 2
+            DefaultGroovyMethods.equals(labels, [foo: "not-bar"])
+        }
+        with(list.nodes.get(2)) {
+            quorum == 1
+            DefaultGroovyMethods.equals(labels, [foo: "bar", baz: "baz"])
+        }
+    }
+
+    def "Updates existing node"() {
+        setup:
+        def list = new NodeDetailsList()
+        list.add(new NodeDetailsList.NodeDetails(1, asLabels([foo: "bar"])))
+        list.add(new NodeDetailsList.NodeDetails(1, asLabels([baz: "true"])))
+
+        when:
+        list.add(new NodeDetailsList.NodeDetails(2, asLabels([baz: "true"])))
+        then:
+        list.nodes.size() == 2
+        with(list.nodes.get(0)) {
+            quorum == 1
+            DefaultGroovyMethods.equals(labels, [foo: "bar"])
+        }
+        with(list.nodes.get(1)) {
+            quorum == 3
+            DefaultGroovyMethods.equals(labels, [baz: "true"])
+        }
+    }
+
+    def "Applies all from another list"() {
+        setup:
+        def list1 = new NodeDetailsList()
+        list1.add(new NodeDetailsList.NodeDetails(1, asLabels([foo: "bar"])))
+        list1.add(new NodeDetailsList.NodeDetails(2, asLabels([baz: "true"])))
+        list1.add(new NodeDetailsList.NodeDetails(3, asLabels([baz: "true", bar: "bar"])))
+
+        def list2 = new NodeDetailsList()
+        list1.add(new NodeDetailsList.NodeDetails(4, asLabels([foo: "bar"])))
+        list1.add(new NodeDetailsList.NodeDetails(5, asLabels([baz: "true", bar: "bar"])))
+        list1.add(new NodeDetailsList.NodeDetails(6, asLabels([bar: "bar"])))
+
+        def list = new NodeDetailsList()
+        when:
+        list.add(list1)
+        list.add(list2)
+        def nodes = list.nodes.toSorted { it.quorum }
+        then:
+        list.nodes.size() == 4
+        with(nodes.get(0)) {
+            quorum == 2
+            DefaultGroovyMethods.equals(labels, [baz: "true"])
+        }
+        with(nodes.get(1)) {
+            quorum == 1 + 4
+            DefaultGroovyMethods.equals(labels, [foo: "bar"])
+        }
+        with(nodes.get(2)) {
+            quorum == 6
+            DefaultGroovyMethods.equals(labels, [bar: "bar"])
+        }
+        with(nodes.get(3)) {
+            quorum == 3 + 5
+            DefaultGroovyMethods.equals(labels, [baz: "true", bar: "bar"])
+        }
+    }
+
+    // --------
+
+
+    UpstreamsConfig.Labels asLabels(Map<String, String> values) {
+        UpstreamsConfig.Labels result = new UpstreamsConfig.Labels()
+        values.entrySet().forEach {
+            result.put(it.key, it.value)
+        }
+        return result
+    }
+}

src/test/resources/upstreams-basic.yaml

@@ -10,7 +10,6 @@ defaultOptions:
 upstreams:
   - id: local
     chain: ethereum
-    provider: geth
     connection:
       ethereum:
         rpc:
@@ -20,7 +19,6 @@ upstreams:
           origin: "http://localhost"
   - id: infura
     chain: ethereum
-    provider: infura
     connection:
       ethereum:
         rpc:

src/test/resources/upstreams-ds.yaml

@@ -9,7 +9,6 @@ defaultOptions:
 
 upstreams:
   - id: remote
-    provider: dshackle
     connection:
       grpc:
         host: "10.2.0.15"

src/test/resources/upstreams-labels.yaml

@@ -0,0 +1,31 @@
+version: v1
+
+defaultOptions:
+  - chains:
+      - ethereum
+    options:
+      disable-syncing: true
+      min-peers: 3
+
+upstreams:
+  - id: remote
+    connection:
+      grpc:
+        host: "10.2.0.15"
+        auth:
+          type: tls
+          ca: /etc/ca.myservice.com.crt
+          certificate: /etc/client1.myservice.com.crt
+          key: /etc/client1.myservice.com.key
+  - id: local
+    chain: ethereum
+    labels:
+      api: geth
+      fullnode: true
+    connection:
+      ethereum:
+        rpc:
+          url: "http://localhost:8545"
+        ws:
+          url: "ws://localhost:8546"
+          origin: "http://localhost"