From 8b7888eb8cb41144329521f1141cee20672fd4dd Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sat, 31 Aug 2019 21:01:38 -0400 Subject: [PATCH] problem: grpc upstream can change set of provided upstreams solution: periodically recheck and update current list --- .../dshackle/config/UpstreamsConfigReader.kt | 73 ++++++++------ .../dshackle/upstream/AggregatedUpstream.kt | 6 +- .../dshackle/upstream/ChainUpstreams.kt | 20 +++- .../dshackle/upstream/ConfiguredUpstreams.kt | 93 +++++------------- .../dshackle/upstream/CurrentUpstreams.kt | 98 +++++++++++++++++++ .../emeraldpay/dshackle/upstream/Upstream.kt | 1 + .../dshackle/upstream/UpstreamChange.kt | 31 ++++++ .../emeraldpay/dshackle/upstream/Upstreams.kt | 1 - .../upstream/ethereum/EthereumUpstream.kt | 7 +- .../dshackle/upstream/grpc/GrpcUpstream.kt | 24 +++-- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 67 ++++++++----- .../config/UpstreamsConfigReaderSpec.groovy | 31 ++++++ .../dshackle/test/EthereumUpstreamMock.groovy | 10 +- .../dshackle/test/UpstreamsMock.groovy | 1 - .../upstream/AggregatedUpstreamSpec.groovy | 43 ++++++++ .../upstream/CurrentUpstreamsSpec.groovy | 56 +++++++++++ .../upstream/FilteringApiIteratorSpec.groovy | 1 + .../upstream/grpc/GrpcUpstreamSpec.groovy | 6 +- src/test/resources/upstreams-no-id.yaml | 26 +++++ 19 files changed, 453 insertions(+), 142 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamChange.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy create mode 100644 src/test/resources/upstreams-no-id.yaml diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index 421be80e..88553a5d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -50,55 +50,72 @@ class UpstreamsConfigReader { } config.upstreams = ArrayList>() - getList(configNode, "upstreams")?.value?.forEach { upNode -> + getList(configNode, "upstreams")?.value?.forEachIndexed { pos, upNode -> val connNode = getMapping(upNode, "connection") if (hasAny(connNode, "ethereum")) { val connConfigNode = getMapping(connNode, "ethereum")!! val upstream = UpstreamsConfig.Upstream() readUpstreamCommon(upNode, upstream) readUpstreamEthereum(upNode, upstream) - config.upstreams.add(upstream) - val connection = UpstreamsConfig.EthereumConnection() - upstream.connection = connection - getMapping(connConfigNode, "rpc")?.let { node -> - getValueAsString(node, "url")?.let { url -> - val http = UpstreamsConfig.HttpEndpoint(URI(url)) - connection.rpc = http - http.basicAuth = readBasicAuth(node) - http.tls = readTls(node) - } - } - getMapping(connConfigNode, "ws")?.let { node -> - getValueAsString(node, "url")?.let { url -> - val ws = UpstreamsConfig.WsEndpoint(URI(url)) - connection.ws = ws - getValueAsString(node, "origin")?.let { origin -> - ws.origin = URI(origin) + if (isValid(upstream)) { + config.upstreams.add(upstream) + val connection = UpstreamsConfig.EthereumConnection() + upstream.connection = connection + getMapping(connConfigNode, "rpc")?.let { node -> + getValueAsString(node, "url")?.let { url -> + val http = UpstreamsConfig.HttpEndpoint(URI(url)) + connection.rpc = http + http.basicAuth = readBasicAuth(node) + http.tls = readTls(node) } - ws.basicAuth = readBasicAuth(node) } + getMapping(connConfigNode, "ws")?.let { node -> + getValueAsString(node, "url")?.let { url -> + val ws = UpstreamsConfig.WsEndpoint(URI(url)) + connection.ws = ws + getValueAsString(node, "origin")?.let { origin -> + ws.origin = URI(origin) + } + ws.basicAuth = readBasicAuth(node) + } + } + } else { + log.error("Upstream at #0 has invalid configuration") } } else if (hasAny(connNode, "grpc")) { val connConfigNode = getMapping(connNode, "grpc")!! val upstream = UpstreamsConfig.Upstream() readUpstreamCommon(upNode, upstream) readUpstreamGrpc(upNode, upstream) - config.upstreams.add(upstream) - val connection = UpstreamsConfig.GrpcConnection() - upstream.connection = connection - getValueAsString(connConfigNode, "host")?.let { - connection.host = it + if (isValid(upstream)) { + config.upstreams.add(upstream) + val connection = UpstreamsConfig.GrpcConnection() + upstream.connection = connection + getValueAsString(connConfigNode, "host")?.let { + connection.host = it + } + getValueAsInt(connConfigNode, "port")?.let { + connection.port = it + } + connection.auth = readTls(connConfigNode) + } else { + log.error("Upstream at #0 has invalid configuration") } - getValueAsInt(connConfigNode, "port")?.let { - connection.port = it - } - connection.auth = readTls(connConfigNode) } } return config } + fun isValid(upstream: UpstreamsConfig.Upstream<*>): Boolean { + val id = upstream.id + if (id == null || id.length < 3 || !id.matches(Regex("[a-zA-Z][a-zA-Z0-9_-]+[a-zA-Z0-9]"))) { + log.warn("Invalid id: $id") + return false + } + return true + } + internal fun readUpstreamCommon(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) { upstream.id = getValueAsString(upNode, "id") upstream.options = tryReadOptions(upNode) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt index 764ed0a9..fde1e178 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt @@ -47,13 +47,13 @@ abstract class AggregatedUpstream( ) var cache: CachingEthereumApi = CachingEthereumApi.empty() private val reconfigLock = ReentrantLock() - private var callMethods: CallMethods = DirectCallMethods() + private var callMethods: CallMethods? = null abstract fun getAll(): List abstract fun addUpstream(upstream: Upstream) abstract fun getApis(matcher: Selector.Matcher): Iterator - fun reconfigure() { + fun onUpstreamsUpdated() { reconfigLock.withLock { getAll().map { it.getMethods() }.let { callMethods = AggregatedCallMethods(it) @@ -83,7 +83,7 @@ abstract class AggregatedUpstream( } override fun getMethods(): CallMethods { - return callMethods + return callMethods ?: throw IllegalStateException("Methods are not initialized yet") } class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index 6657438d..93f067c6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -35,12 +35,19 @@ open class ChainUpstreams ( private val log = LoggerFactory.getLogger(ChainUpstreams::class.java) private var seq = 0 - private var head: EthereumHead? + private var head: EthereumHead? = null private var lagObserver: HeadLagObserver? = null private var subscription: Disposable? = null init { - head = updateHead() + if (upstreams.size > 0) { + head = updateHead() + onUpstreamsUpdated() + } + } + + override fun getId(): String { + return "!all:${chain.chainCode}" } override fun isRunning(): Boolean { @@ -99,7 +106,14 @@ open class ChainUpstreams ( override fun addUpstream(upstream: Upstream) { upstreams.add(upstream) head = updateHead() - reconfigure() + onUpstreamsUpdated() + } + + fun removeUpstream(id: String) { + if (upstreams.removeIf { it.getId() == id }) { + head = updateHead() + onUpstreamsUpdated() + } } override fun getApis(matcher: Selector.Matcher): Iterator { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index aeb3f713..d0874dc5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -43,13 +43,11 @@ import kotlin.collections.HashMap @Repository open class ConfiguredUpstreams( @Autowired val env: Environment, - @Autowired private val objectMapper: ObjectMapper -) : Upstreams { + @Autowired private val objectMapper: ObjectMapper, + @Autowired private val currentUpstreams: CurrentUpstreams +) { private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java) - private val chainMapping = ConcurrentHashMap() - private val chainsBus = TopicProcessor.create() - private val callTargets = HashMap() private val chainNames = mapOf( "ethereum" to Chain.ETHEREUM, @@ -67,7 +65,7 @@ open class ConfiguredUpstreams( config.upstreams.forEach { up -> if (up.connection is UpstreamsConfig.GrpcConnection) { - buildGrpcUpstream(up.connection as UpstreamsConfig.GrpcConnection) + buildGrpcUpstream(up as UpstreamsConfig.Upstream) } else { val chain = chainNames[up.chain] if (chain == null) { @@ -127,12 +125,12 @@ open class ConfiguredUpstreams( var rpcApi: DirectEthereumApi? = null val urls = ArrayList() val methods = if (config.methods != null) { - ManagedCallMethods(getDefaultMethods(chain), + ManagedCallMethods(currentUpstreams.getDefaultMethods(chain), config.methods!!.enabled.map { it.name }.toSet(), config.methods!!.disabled.map { it.name }.toSet() ) } else { - getDefaultMethods(chain) + currentUpstreams.getDefaultMethods(chain) } conn.rpc?.let { endpoint -> val rpcTransport = DefaultRpcTransport(endpoint.url) @@ -168,75 +166,32 @@ open class ConfiguredUpstreams( } log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") - val ethereumUpstream = EthereumUpstream(chain, rpcApi!!, wsApi, options, + val ethereumUpstream = EthereumUpstream( + config.id!!, + chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, config.labels), methods) ethereumUpstream.start() - addUpstream(chain, ethereumUpstream) + currentUpstreams.update(UpstreamChange(chain, ethereumUpstream, UpstreamChange.ChangeType.ADDED)) } } - private fun buildGrpcUpstream(up: UpstreamsConfig.GrpcConnection) { - val endpoint = up - val ds = GrpcUpstreams( - endpoint.host!!, - endpoint.port ?: 443, - objectMapper, - up.auth - ) - log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}") - ds.start() - .subscribe { - log.info("Subscribed to ${it.t1} through gRPC at ${endpoint.host}:${endpoint.port}") - addUpstream(it.t1, it.t2) - } - } - - override fun getUpstream(chain: Chain): AggregatedUpstream? { - return chainMapping[chain] - } - - override fun addUpstream(chain: Chain, up: Upstream): ChainUpstreams { - val current = chainMapping[chain] - if (current == null) { - val created = ChainUpstreams(chain, ArrayList(), objectMapper) - created.addUpstream(up) - created.start() - chainMapping[chain] = created - chainsBus.onNext(chain) - return created - } else { - current.addUpstream(up) - } - return current - } - - @Scheduled(fixedRate = 15000) - fun printStatuses() { - chainMapping.forEach { it.value.printStatus() } - } - - override fun getAvailable(): List { - return Collections.unmodifiableList(chainMapping.keys.toList()) - } - - override fun observeChains(): Flux { - return Flux.merge( - Flux.fromIterable(getAvailable()), - Flux.from(chainsBus) + private fun buildGrpcUpstream(config: UpstreamsConfig.Upstream) { + val endpoint = config.connection!! + val ds = GrpcUpstreams( + config.id!!, + endpoint.host!!, + endpoint.port ?: 443, + objectMapper, + endpoint.auth ) + log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}") + ds.start() + .doOnNext { + log.info("Chain ${it.chain} has ${it.type} through gRPC at ${endpoint.host}:${endpoint.port}") + } + .subscribe(currentUpstreams::update) } - override fun getDefaultMethods(chain: Chain): CallMethods { - var current = callTargets[chain] - if (current == null) { - current = QuorumBasedMethods(objectMapper, chain) - callTargets[chain] = current - } - return current - } - override fun isAvailable(chain: Chain): Boolean { - return chainMapping.containsKey(chain) && callTargets.containsKey(chain) - } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt new file mode 100644 index 00000000..a65a3da0 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt @@ -0,0 +1,98 @@ +/** + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream + +import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.grpc.Chain +import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.scheduling.annotation.Scheduled +import org.springframework.stereotype.Repository +import reactor.core.publisher.Flux +import reactor.core.publisher.TopicProcessor +import java.util.* +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.locks.ReentrantLock +import kotlin.concurrent.withLock + +@Repository +class CurrentUpstreams( + @Autowired private val objectMapper: ObjectMapper +): Upstreams { + + private val log = LoggerFactory.getLogger(CurrentUpstreams::class.java) + + private val chainMapping = ConcurrentHashMap() + private val chainsBus = TopicProcessor.create() + private val callTargets = HashMap() + private val updateLock = ReentrantLock() + + fun update(change: UpstreamChange) { + updateLock.withLock { + val chain = change.chain + val up = change.upstream + val current = chainMapping[chain] + if (change.type == UpstreamChange.ChangeType.REMOVED) { + current?.removeUpstream(up.getId()) + log.info("Upstream ${change.upstream.getId()} with chain $chain has been removed") + } else { + if (current == null) { + val created = ChainUpstreams(chain, ArrayList(), objectMapper) + created.addUpstream(up) + created.start() + chainMapping[chain] = created + chainsBus.onNext(chain) + } else { + current.addUpstream(up) + } + log.info("Upstream ${change.upstream.getId()} with chain $chain has been added") + } + } + } + + override fun getUpstream(chain: Chain): AggregatedUpstream? { + return chainMapping[chain] + } + + @Scheduled(fixedRate = 15000) + fun printStatuses() { + chainMapping.forEach { it.value.printStatus() } + } + + override fun getAvailable(): List { + return Collections.unmodifiableList(chainMapping.keys.toList()) + } + + override fun observeChains(): Flux { + return Flux.merge( + Flux.fromIterable(getAvailable()), + Flux.from(chainsBus) + ) + } + + override fun getDefaultMethods(chain: Chain): CallMethods { + var current = callTargets[chain] + if (current == null) { + current = QuorumBasedMethods(objectMapper, chain) + callTargets[chain] = current + } + return current + } + + override fun isAvailable(chain: Chain): Boolean { + return chainMapping.containsKey(chain) && callTargets.containsKey(chain) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index b2c3a703..0ef63858 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -31,4 +31,5 @@ interface Upstream { fun getLag(): Long fun getLabels(): Collection fun getMethods(): CallMethods + fun getId(): String } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamChange.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamChange.kt new file mode 100644 index 00000000..889670bc --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamChange.kt @@ -0,0 +1,31 @@ +/** + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.grpc.Chain + +class UpstreamChange( + val chain: Chain, + val upstream: Upstream, + val type: ChangeType +) { + enum class ChangeType { + ADDED, + REVALIDATED, + STALE, + REMOVED, + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt index 99ad5d9d..c823b0e2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt @@ -19,7 +19,6 @@ import io.emeraldpay.grpc.Chain import reactor.core.publisher.Flux interface Upstreams { - fun addUpstream(chain: Chain, up: Upstream): AggregatedUpstream fun getUpstream(chain: Chain): AggregatedUpstream? fun getAvailable(): List fun observeChains(): Flux diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt index 836422b5..07453196 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -23,6 +23,7 @@ import org.springframework.context.Lifecycle import reactor.core.Disposable open class EthereumUpstream( + private val id: String, val chain: Chain, private val api: DirectEthereumApi, private val ethereumWs: EthereumWs? = null, @@ -31,7 +32,7 @@ open class EthereumUpstream( private val targets: CallMethods ): DefaultUpstream(), Lifecycle { - constructor(chain: Chain, api: DirectEthereumApi): this(chain, api, null, + constructor(id: String, chain: Chain, api: DirectEthereumApi): this(id, chain, api, null, UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()), DirectCallMethods()) @@ -45,6 +46,10 @@ open class EthereumUpstream( api.upstream = this } + override fun getId(): String { + return id + } + override fun start() { log.info("Configured for ${chain.chainName}") 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 a3b0fc09..45220ee4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt @@ -45,6 +45,7 @@ import java.util.function.Function import kotlin.collections.ArrayList open class GrpcUpstream( + private val parentId: String, private val chain: Chain, private val client: ReactorBlockchainGrpc.ReactorBlockchainStub, private val objectMapper: ObjectMapper @@ -58,12 +59,13 @@ open class GrpcUpstream( private val streamBlocks: TopicProcessor> = TopicProcessor.create() private val nodes = AtomicReference(NodeDetailsList()) private val head = Head(this) - private var targets: CallMethods = DirectCallMethods() + private var targets: CallMethods? = null private val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper) private var headSubscription: Disposable? = null open fun createApi(matcher: Selector.Matcher): DirectEthereumApi { + val targets = this.getMethods() val rpcClient = DefaultRpcClient(grpcTransport.withLabels(Selector.extractLabels(matcher))) return DirectEthereumApi(rpcClient, objectMapper, targets).let { it.upstream = this @@ -71,7 +73,12 @@ open class GrpcUpstream( } } + override fun getId(): String { + return "$parentId/${chain.chainCode}" + } + override fun start() { + if (this.isRunning) return val chainRef = Common.Chain.newBuilder() .setTypeValue(chain.id) .build() @@ -116,7 +123,12 @@ open class GrpcUpstream( .executeAndConvert(Commands.eth().getBlock(it.hash)) .timeout(Duration.ofSeconds(5), Mono.error(Exception("Timeout requesting block from upstream"))) .doOnError { t -> - log.warn("Failed to download block data", t) + val msg = "Failed to download block data for chain $chain" + if (t is RpcException) { + log.warn("$msg. Message: ${t.message}") + } else { + log.error(msg, t) + } } } .onErrorContinue { err, _ -> @@ -134,9 +146,9 @@ open class GrpcUpstream( targets = DirectCallMethods(conf.supportedMethodsList.toSet()) val nodes = NodeDetailsList() val allLabels = ArrayList() - conf.nodesList.forEach { node -> - val node = NodeDetailsList.NodeDetails(node.quorum, - node.labelsList.let { provided -> + conf.nodesList.forEach { remoteNode -> + val node = NodeDetailsList.NodeDetails(remoteNode.quorum, + remoteNode.labelsList.let { provided -> val labels = UpstreamsConfig.Labels() provided.forEach { labels[it.name] = it.value @@ -171,7 +183,7 @@ open class GrpcUpstream( } override fun getMethods(): CallMethods { - return targets + return targets ?: throw IllegalStateException("Upstream is not initialized yet") } override fun isAvailable(): Boolean { 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 b3c19129..7b64fce2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -19,7 +19,7 @@ import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.dshackle.upstream.UpstreamChange import io.emeraldpay.grpc.Chain import io.grpc.ManagedChannelBuilder import io.grpc.netty.NettyChannelBuilder @@ -27,15 +27,14 @@ import io.netty.handler.ssl.* import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory import reactor.core.publisher.Flux -import reactor.core.publisher.toFlux -import reactor.util.function.Tuple2 -import reactor.util.function.Tuples import java.io.File +import java.time.Duration import java.util.* import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.withLock class GrpcUpstreams( + private val id: String, private val host: String, private val port: Int, private val objectMapper: ObjectMapper, @@ -44,39 +43,32 @@ class GrpcUpstreams( private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java) private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null - private var known = HashMap() + private val known = HashMap() private val lock = ReentrantLock() - fun start(): Flux> { + fun start(): Flux { val channel: ManagedChannelBuilder<*> = if (auth != null && StringUtils.isNotEmpty(auth.ca)) { NettyChannelBuilder.forAddress(host, port) .useTransportSecurity() .sslContext(withTls(auth)) } else { - log.warn("Using insecure connection for $host:$port") + log.warn("Using insecure connection to $host:$port") ManagedChannelBuilder.forAddress(host, port) .usePlaintext() } val client = ReactorBlockchainGrpc.newReactorStub(channel.build()) this.client = client - val loaded = client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build()) - .map { value -> - val chains = ArrayList() - value.chainsList.filter { - Chain.byId(it.chain.number) != Chain.UNSPECIFIED - }.map { chainDetails -> - val chain = Chain.byId(chainDetails.chain.number) - val up = getOrCreate(chain) - up.init(chainDetails) - chains.add(chain) - Tuples.of(chain, up) - } - }.flatMapMany { - it.toFlux() + + val updates = Flux.interval(Duration.ZERO, Duration.ofMinutes(1)) + .flatMap { + client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build()) + }.flatMap { value -> + processDescription(value) }.doOnError { t -> log.error("Failed to get description from $host:$port", t) } + //TODO subscribe only after receiving details client.subscribeStatus(BlockchainOuterClass.StatusRequest.newBuilder().build()) .subscribe { value -> @@ -85,7 +77,30 @@ class GrpcUpstreams( known[chain]?.onStatus(value) } } - return loaded + return updates + } + + fun processDescription(value: BlockchainOuterClass.DescribeResponse): Flux { + val current = value.chainsList.filter { + Chain.byId(it.chain.number) != Chain.UNSPECIFIED + }.map { chainDetails -> + val chain = Chain.byId(chainDetails.chain.number) + val up = getOrCreate(chain) + (up.upstream as GrpcUpstream).init(chainDetails) + up + } + + val added = current.filter { + it.type == UpstreamChange.ChangeType.ADDED + } + + val removed = known.filterNot { kv -> + val stillCurrent = current.any { c -> c.chain == kv.key } + stillCurrent + }.map { + UpstreamChange(it.key, known.remove(it.key)!!, UpstreamChange.ChangeType.REMOVED) + } + return Flux.fromIterable(removed + added) } internal fun withTls(auth: UpstreamsConfig.TlsAuth): SslContext { @@ -106,16 +121,16 @@ class GrpcUpstreams( return sslContext.build() } - fun getOrCreate(chain: Chain): GrpcUpstream { + fun getOrCreate(chain: Chain): UpstreamChange { lock.withLock { val current = known[chain] return if (current == null) { - val created = GrpcUpstream(chain, client!!, objectMapper) + val created = GrpcUpstream(id, chain, client!!, objectMapper) known[chain] = created created.start() - created + UpstreamChange(chain, created, UpstreamChange.ChangeType.ADDED) } else { - current + UpstreamChange(chain, current, UpstreamChange.ChangeType.REVALIDATED) } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 9d0f7653..045af98a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -15,6 +15,10 @@ */ package io.emeraldpay.dshackle.config +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream +import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.rpc.RpcClient import spock.lang.Specification class UpstreamsConfigReaderSpec extends Specification { @@ -194,4 +198,31 @@ class UpstreamsConfigReaderSpec extends Specification { } } } + + def "Parse config with invalid ids"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("upstreams-no-id.yaml") + when: + def act = reader.read(config) + then: + act != null + act.upstreams.size() == 1 + with(act.upstreams.get(0)) { + id == "test" + } + } + + def "Invalidate wrong ids"() { + expect: + !reader.isValid(new UpstreamsConfig.Upstream(id: id)) + where: + id << ["", null, "a", "ab", "!ab", "foo bar", "foo@bar", "123test", "_test", "test/test"] + } + + def "Accept good ids"() { + expect: + reader.isValid(new UpstreamsConfig.Upstream(id: id)) + where: + id << ["test", "test_test", "test-test", "test123", "test1test", "foo_bar_12"] + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy index 40794718..bb438605 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy @@ -36,8 +36,16 @@ class EthereumUpstreamMock extends EthereumUpstream { this(chain, api, new QuorumBasedMethods(TestingCommons.objectMapper(), chain)) } + EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull DirectEthereumApi api) { + this(id, chain, api, new QuorumBasedMethods(TestingCommons.objectMapper(), chain)) + } + EthereumUpstreamMock(@NotNull Chain chain, @NotNull DirectEthereumApi api, CallMethods methods) { - super(chain, api, null, + this("test", chain, api, methods) + } + + EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull DirectEthereumApi api, CallMethods methods) { + super(id, chain, api, null, UpstreamsConfig.Options.getDefaults(), new NodeDetailsList.NodeDetails(1, new UpstreamsConfig.Labels()), methods) setLag(0) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy index 5107ac9f..46712da9 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy @@ -38,7 +38,6 @@ class UpstreamsMock implements Upstreams { addUpstream(chain2, up2) } - @Override AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull Upstream up) { if (!upstreams.containsKey(chain)) { upstreams[chain] = new ChainUpstreams(chain, [up], TestingCommons.objectMapper()) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy new file mode 100644 index 00000000..a9712946 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy @@ -0,0 +1,43 @@ +/** + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.quorum.AlwaysQuorum +import io.emeraldpay.dshackle.test.EthereumUpstreamMock +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi +import io.emeraldpay.grpc.Chain +import spock.lang.Specification + +class AggregatedUpstreamSpec extends Specification { + + def "Aggregates methods"() { + setup: + def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, Stub(DirectEthereumApi), new DirectCallMethods(["eth_test1", "eth_test2"])) + def up2 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, Stub(DirectEthereumApi), new DirectCallMethods(["eth_test2", "eth_test3"])) + def aggr = new ChainUpstreams(Chain.ETHEREUM, [up1, up2], TestingCommons.objectMapper()) + when: + aggr.onUpstreamsUpdated() + def act = aggr.getMethods() + then: + act.isAllowed("eth_test1") + act.isAllowed("eth_test2") + act.isAllowed("eth_test3") + act.getQuorumFor("eth_test1") instanceof AlwaysQuorum + act.getQuorumFor("eth_test2") instanceof AlwaysQuorum + act.getQuorumFor("eth_test3") instanceof AlwaysQuorum + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy new file mode 100644 index 00000000..dbd9e2e1 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy @@ -0,0 +1,56 @@ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.test.EthereumUpstreamMock +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream +import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.rpc.RpcClient +import spock.lang.Specification + +class CurrentUpstreamsSpec extends Specification { + + def "add upstream"() { + setup: + def current = new CurrentUpstreams(TestingCommons.objectMapper()) + def up = new EthereumUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) + when: + current.update(new UpstreamChange(Chain.ETHEREUM, up, UpstreamChange.ChangeType.ADDED)) + then: + current.getAvailable() == [Chain.ETHEREUM] + current.getUpstream(Chain.ETHEREUM).getAll()[0] == up + } + + def "add multiple upstreams"() { + setup: + def current = new CurrentUpstreams(TestingCommons.objectMapper()) + def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) + def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api(Stub(RpcClient))) + def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) + when: + current.update(new UpstreamChange(Chain.ETHEREUM, up1, UpstreamChange.ChangeType.ADDED)) + current.update(new UpstreamChange(Chain.ETHEREUM_CLASSIC, up2, UpstreamChange.ChangeType.ADDED)) + current.update(new UpstreamChange(Chain.ETHEREUM, up3, UpstreamChange.ChangeType.ADDED)) + then: + current.getAvailable().toSet() == [Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC].toSet() + current.getUpstream(Chain.ETHEREUM).getAll().toSet() == [up1, up3].toSet() + current.getUpstream(Chain.ETHEREUM_CLASSIC).getAll().toSet() == [up2].toSet() + } + + def "remove upstream"() { + setup: + def current = new CurrentUpstreams(TestingCommons.objectMapper()) + def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) + def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api(Stub(RpcClient))) + def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) + def up1_del = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) + when: + current.update(new UpstreamChange(Chain.ETHEREUM, up1, UpstreamChange.ChangeType.ADDED)) + current.update(new UpstreamChange(Chain.ETHEREUM_CLASSIC, up2, UpstreamChange.ChangeType.ADDED)) + current.update(new UpstreamChange(Chain.ETHEREUM, up3, UpstreamChange.ChangeType.ADDED)) + current.update(new UpstreamChange(Chain.ETHEREUM, up1_del, UpstreamChange.ChangeType.REMOVED)) + then: + current.getAvailable().toSet() == [Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC].toSet() + current.getUpstream(Chain.ETHEREUM).getAll().toSet() == [up3].toSet() + current.getUpstream(Chain.ETHEREUM_CLASSIC).getAll().toSet() == [up2].toSet() + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy index f09252cc..925abf7f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy @@ -40,6 +40,7 @@ class FilteringApiIteratorSpec extends Specification { [test: "baz"] ].collect { new EthereumUpstream( + "test", Chain.ETHEREUM, new DirectEthereumApi(rpcClient, objectMapper, ethereumTargets), (EthereumWs) null, diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy index d46d2292..27d6badb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy @@ -70,7 +70,7 @@ class GrpcUpstreamSpec extends Specification { ) } }) - def upstream = new GrpcUpstream(chain, client, objectMapper) + def upstream = new GrpcUpstream("test", chain, client, objectMapper) upstream.setLag(0) upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() .addAllSupportedMethods(["eth_getBlockByHash"]) @@ -129,7 +129,7 @@ class GrpcUpstreamSpec extends Specification { finished.complete(true) } }) - def upstream = new GrpcUpstream(chain, client, objectMapper) + def upstream = new GrpcUpstream("test", chain, client, objectMapper) upstream.setLag(0) upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() .addAllSupportedMethods(["eth_getBlockByHash"]) @@ -189,7 +189,7 @@ class GrpcUpstreamSpec extends Specification { finished.complete(true) } }) - def upstream = new GrpcUpstream(chain, client, objectMapper) + def upstream = new GrpcUpstream("test", chain, client, objectMapper) upstream.setLag(0) upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() .addAllSupportedMethods(["eth_getBlockByHash"]) diff --git a/src/test/resources/upstreams-no-id.yaml b/src/test/resources/upstreams-no-id.yaml new file mode 100644 index 00000000..e299b5e3 --- /dev/null +++ b/src/test/resources/upstreams-no-id.yaml @@ -0,0 +1,26 @@ +version: v1 + +upstreams: + - chain: ethereum + connection: + ethereum: + rpc: + url: "http://localhost:8545" + - chain: ethereum + id: test + connection: + ethereum: + rpc: + url: "http://localhost:8545" + - chain: ethereum + id: test/test + connection: + ethereum: + rpc: + url: "http://localhost:8545" + - chain: ethereum + id: !test + connection: + ethereum: + rpc: + url: "http://localhost:8545"