diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index c5290b4e..bf261a9c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/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 { 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() { + } + enum class UpstreamType private constructor(vararg code: String) { ETHEREUM_JSON_RPC("ethereum"), DSHACKLE("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 bd0155ab..7a65d69d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/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() - upstream.id = getValueAsString(upNode, "id") - upstream.provider = getValueAsString(upNode, "provider") - upstream.chain = getValueAsString(upNode, "chain") + readUpstreamCommon(upNode, upstream) + readUpstreamEthereum(upNode, upstream) 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() - upstream.id = getValueAsString(upNode, "id") - upstream.provider = getValueAsString(upNode, "provider") + readUpstreamCommon(upNode, upstream) + 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) { + 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) { + 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) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index 31901d30..01e55783 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt +++ b/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.ConfiguredUpstreams -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.dshackle.upstream.* 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()) { - val status = subscribeStatus.chainStatus(chain, ups) - val targets = chainUpstreams.getSupportedTargets() - val chainDescription = BlockchainOuterClass.DescribeChain.newBuilder() - .setChain(Common.ChainRef.forNumber(chain.id)) - .addAllSupportedTargets(targets) - .setStatus(status) - .build() - resp.addChains(chainDescription) + ups.forEach { up -> + val nodes = NodeDetailsList() + if (up is EthereumUpstream) { + nodes.add(up.node) + } else if (up is GrpcUpstream) { + nodes.add(up.getNodes()) + } + nodes.getNodes().forEach { node -> + 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() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt index ade00099..e8665c47 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt @@ -34,7 +34,7 @@ class SubscribeStatus( fun chainStatus(chain: Chain, ups: List): BlockchainOuterClass.ChainStatus { val available = ups.map { u -> u.getStatus() - }.min()!! + }.min() ?: UpstreamAvailability.UNAVAILABLE val quorum = ups.filter { it.getStatus() > UpstreamAvailability.UNAVAILABLE }.count() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index a3b16170..92290155 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/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() @@ -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))) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index 900562c7..32be2967 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt index 92da29b8..22fde950 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt +++ b/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>(null) private val streamBlocks: TopicProcessor> = TopicProcessor.create() - private var status = AtomicReference(UpstreamAvailability.UNAVAILABLE) + private val status = AtomicReference(UpstreamAvailability.UNAVAILABLE) + private val nodes = AtomicReference(NodeDetailsList()) private val head = Head(this) private val api: EthereumApi private val statusStream: TopicProcessor = 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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NodeDetailsList.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/NodeDetailsList.kt new file mode 100644 index 00000000..09e65d9b --- /dev/null +++ b/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() + + 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 { + return Collections.unmodifiableList(nodes) + } + + class NodeDetails(val quorum: Int, val labels: UpstreamsConfig.Labels) { + companion object { + } + } + +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 0fa80bef..d48b3cb2 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/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" + } + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/NodeDetailsListSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/NodeDetailsListSpec.groovy new file mode 100644 index 00000000..1b3ddade --- /dev/null +++ b/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 values) { + UpstreamsConfig.Labels result = new UpstreamsConfig.Labels() + values.entrySet().forEach { + result.put(it.key, it.value) + } + return result + } +} diff --git a/src/test/resources/upstreams-basic.yaml b/src/test/resources/upstreams-basic.yaml index f7028cbe..24b74db1 100644 --- a/src/test/resources/upstreams-basic.yaml +++ b/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: diff --git a/src/test/resources/upstreams-ds.yaml b/src/test/resources/upstreams-ds.yaml index cc2df79e..4575a5a4 100644 --- a/src/test/resources/upstreams-ds.yaml +++ b/src/test/resources/upstreams-ds.yaml @@ -9,7 +9,6 @@ defaultOptions: upstreams: - id: remote - provider: dshackle connection: grpc: host: "10.2.0.15" diff --git a/src/test/resources/upstreams-labels.yaml b/src/test/resources/upstreams-labels.yaml new file mode 100644 index 00000000..dbfd9061 --- /dev/null +++ b/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" \ No newline at end of file