From 4680a4490c91cf34bf0a8fea90ab8318dc249a8b Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 21 Jun 2019 16:10:01 -0400 Subject: [PATCH] solution: verify upstream status --- .../{Upstreams.java => UpstreamsConfig.java} | 35 +++++++++++++++-- .../dshackle/config/Configuration.kt | 4 -- .../dshackle/config/UpstreamsConfigReader.kt | 17 ++++++++ .../dshackle/config/UpstreamsReader.kt | 17 -------- .../dshackle/upstream/ChainConnect.kt | 18 +++++++++ .../dshackle/upstream/EthereumWs.kt | 9 ++++- .../emeraldpay/dshackle/upstream/Upstream.kt | 21 ++++++++-- .../dshackle/upstream/UpstreamAvailability.kt | 11 ++++++ .../dshackle/upstream/UpstreamValidator.kt | 39 +++++++++++++++++++ .../emeraldpay/dshackle/upstream/Upstreams.kt | 35 ++++++++++++++--- ...roovy => UpstreamsConfigReaderSpec.groovy} | 14 +++---- 11 files changed, 177 insertions(+), 43 deletions(-) rename src/main/java/io/emeraldpay/dshackle/config/{Upstreams.java => UpstreamsConfig.java} (88%) delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/Configuration.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsReader.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamAvailability.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt rename src/test/groovy/io/emeraldpay/dshackle/config/{UpstreamsReaderSpec.groovy => UpstreamsConfigReaderSpec.groovy} (77%) diff --git a/src/main/java/io/emeraldpay/dshackle/config/Upstreams.java b/src/main/java/io/emeraldpay/dshackle/config/UpstreamsConfig.java similarity index 88% rename from src/main/java/io/emeraldpay/dshackle/config/Upstreams.java rename to src/main/java/io/emeraldpay/dshackle/config/UpstreamsConfig.java index 21e4facc..e36d6bb9 100644 --- a/src/main/java/io/emeraldpay/dshackle/config/Upstreams.java +++ b/src/main/java/io/emeraldpay/dshackle/config/UpstreamsConfig.java @@ -18,7 +18,7 @@ import java.util.Collections; import java.util.List; import java.util.Optional; -public class Upstreams { +public class UpstreamsConfig { private String version; private List defaultOptions; @@ -49,9 +49,9 @@ public class Upstreams { } public static class Options { - private Boolean disableSyncing = true; - private Integer minPeers = 1; - private Integer quorum = 1; + private Boolean disableSyncing; + private Integer minPeers; + private Integer quorum; public Boolean getDisableSyncing() { return disableSyncing; @@ -66,6 +66,9 @@ public class Upstreams { } public void setMinPeers(Integer minPeers) { + if (minPeers < 0) { + throw new IllegalArgumentException("minPeers must be positive number"); + } this.minPeers = minPeers; } @@ -74,8 +77,31 @@ public class Upstreams { } public void setQuorum(Integer quorum) { + if (quorum < 0) { + throw new IllegalArgumentException("quorum must be positive number"); + } this.quorum = quorum; } + + public Options merge(Options additional) { + if (additional == null) { + return this; + } + Options copy = new Options(); + copy.setDisableSyncing(this.disableSyncing != null ? this.disableSyncing : additional.disableSyncing); + copy.setMinPeers(this.minPeers != null ? this.minPeers : additional.minPeers); + copy.setQuorum(this.quorum != null ? this.quorum : additional.quorum); + return copy; + } + + public static Options getDefaults() { + Options options = new Options(); + options.setDisableSyncing(true); + options.setMinPeers(1); + options.setQuorum(1); + return options; + } + } public static class OptionsYaml extends TypeDescription { @@ -154,6 +180,7 @@ public class Upstreams { this.endpoints = endpoints; } + @Nullable public Options getOptions() { return options; } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/Configuration.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/Configuration.kt deleted file mode 100644 index 14204953..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/Configuration.kt +++ /dev/null @@ -1,4 +0,0 @@ -package io.emeraldpay.dshackle.config - -class Configuration { -} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt new file mode 100644 index 00000000..0fa3738d --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -0,0 +1,17 @@ +package io.emeraldpay.dshackle.config + +import org.yaml.snakeyaml.Yaml +import java.io.InputStream + +class UpstreamsConfigReader { + + fun read(input: InputStream): UpstreamsConfig { + val yaml = Yaml() + yaml.addTypeDescription(UpstreamsConfig.EndpointTypeYaml()) + yaml.addTypeDescription(UpstreamsConfig.OptionsYaml()) + yaml.addTypeDescription(UpstreamsConfig.AuthYaml()) + return yaml.loadAs(input, UpstreamsConfig::class.java) + } + + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsReader.kt deleted file mode 100644 index ae644724..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsReader.kt +++ /dev/null @@ -1,17 +0,0 @@ -package io.emeraldpay.dshackle.config - -import org.yaml.snakeyaml.Yaml -import java.io.InputStream - -class UpstreamsReader { - - fun read(input: InputStream): Upstreams { - val yaml = Yaml() - yaml.addTypeDescription(Upstreams.EndpointTypeYaml()) - yaml.addTypeDescription(Upstreams.OptionsYaml()) - yaml.addTypeDescription(Upstreams.AuthYaml()) - return yaml.loadAs(input, Upstreams::class.java) - } - - -} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainConnect.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainConnect.kt index ce51f3f9..91d3915b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainConnect.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainConnect.kt @@ -1,13 +1,18 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import org.slf4j.LoggerFactory import java.lang.IllegalStateException +import java.time.Duration class ChainConnect( val chain: Chain, val upstreams: List ) { + private val log = LoggerFactory.getLogger(ChainConnect::class.java) private var seq = 0 val head: EthereumHead = if (upstreams.size == 1) { @@ -29,6 +34,19 @@ class ChainConnect( return QuorumApi(upstreams, 1, seq) } + fun printStatus() { + var height: Long = -1 + try { + height = head.getHead().block(Duration.ofSeconds(1))?.number ?: -1 + } catch (e: Exception) { } + val statuses = upstreams.map { it.getStatus() } + .groupBy { it } + .map { "${it.key.name}/${it.value.size}" } + .joinToString(",") + + log.info("State of ${chain.chainCode}: height=$height, status=$statuses") + } + class SingleApi( val quorumApi: QuorumApi ): Iterator { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt index 714c2541..76dbb015 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt @@ -20,12 +20,17 @@ class EthereumWs( .builder>() .name("new-blocks") .build() - private val head = AtomicReference>(null); + private val head = AtomicReference>(null) fun connect() { log.info("Connecting to WebSocket: $uri") val client = WebsocketClient() - client.connect(uri, origin) + try { + client.connect(uri, origin) + } catch (e: Exception) { + log.error("Failed to connect to websocket at $uri. Error: ${e.message}") + return + } client.onNewBlock { topic.onNext(it) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index 329ea5c1..01cf6e69 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -1,14 +1,15 @@ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.json.BlockJson -import io.infinitape.etherjar.rpc.json.TransactionJson import org.slf4j.LoggerFactory +import java.util.concurrent.atomic.AtomicReference class Upstream( val chain: Chain, val api: EthereumApi, - private val ethereumWs: EthereumWs? = null + private val ethereumWs: EthereumWs? = null, + private val options: UpstreamsConfig.Options ) { private val log = LoggerFactory.getLogger(Upstream::class.java) @@ -21,11 +22,23 @@ class Upstream( } } + val validator = UpstreamValidator(this, options) + private val status = AtomicReference(UpstreamAvailability.UNAVAILABLE) + init { log.info("Configured for ${chain.chainName}") + + validator.start() + .subscribe { + status.set(it) + } } fun isAvailable(): Boolean { - return true + return status.get() == UpstreamAvailability.OK + } + + fun getStatus(): UpstreamAvailability { + return status.get() } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamAvailability.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamAvailability.kt new file mode 100644 index 00000000..1d7214e7 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamAvailability.kt @@ -0,0 +1,11 @@ +package io.emeraldpay.dshackle.upstream + +enum class UpstreamAvailability { + + OK, + IMMATURE, + SYNCING, + LAGGING, + UNAVAILABLE + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt new file mode 100644 index 00000000..1c9b530e --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt @@ -0,0 +1,39 @@ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.infinitape.etherjar.rpc.Batch +import io.infinitape.etherjar.rpc.Commands +import reactor.core.publisher.Flux +import java.time.Duration +import java.util.concurrent.TimeUnit + +class UpstreamValidator( + private val upstream: Upstream, + private val options: UpstreamsConfig.Options +) { + + fun validate(): UpstreamAvailability { + val batch = Batch() + val peerCount = batch.add(Commands.net().peerCount()) + val syncing = batch.add(Commands.eth().syncing()) + try { + upstream.api.execute(batch).get(5, TimeUnit.SECONDS) + if (syncing.get().isSyncing) { + return UpstreamAvailability.SYNCING + } + if (peerCount.get() < options.minPeers) { + return UpstreamAvailability.IMMATURE + } + return UpstreamAvailability.OK + } catch (e: Throwable) { + return UpstreamAvailability.UNAVAILABLE + } + } + + fun start(): Flux { + return Flux.interval(Duration.ofSeconds(15)) + .map { + validate() + }.onErrorContinue { _, _ -> UpstreamAvailability.UNAVAILABLE } + } +} \ 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 1d3e4b05..9f81b0a0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt @@ -1,7 +1,8 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper -import io.emeraldpay.dshackle.config.UpstreamsReader +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.config.UpstreamsConfigReader import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.DefaultRpcClient import io.infinitape.etherjar.rpc.transport.DefaultRpcTransport @@ -9,6 +10,7 @@ import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory 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 java.io.File import java.net.URI @@ -45,25 +47,40 @@ class Upstreams( System.exit(1) } log.info("Read upstream configuration from ${upstreamConfig.path}") - val reader = UpstreamsReader() + val reader = UpstreamsConfigReader() val config = reader.read(upstreamConfig.inputStream()) val groups = HashMap>() + val defaultOptions = HashMap() + config.defaultOptions.forEach { df -> + df.chains.forEach { chainName -> + chainNames[chainName]?.let { chain -> + var current = defaultOptions[chain] + if (current == null) { + current = df.options + } else { + current = current.merge(df.options) + } + defaultOptions[chain] = current + } + } + } + config.upstreams.forEach { up -> val chain = chainNames[up.chain] ?: return@forEach var rpcApi: EthereumApi? = null var wsApi: EthereumWs? = null val urls = ArrayList() up.endpoints.forEach { endpoint -> - if (endpoint.type == io.emeraldpay.dshackle.config.Upstreams.EndpointType.JSON_RPC) { + if (endpoint.type == UpstreamsConfig.EndpointType.JSON_RPC) { rpcApi = EthereumApi( DefaultRpcClient(DefaultRpcTransport(endpoint.url)), objectMapper, chain ) } - if (endpoint.type == io.emeraldpay.dshackle.config.Upstreams.EndpointType.WEBSOCKET) { + if (endpoint.type == UpstreamsConfig.EndpointType.WEBSOCKET) { wsApi = EthereumWs( endpoint.url, endpoint.origin ?: URI("http://localhost") @@ -72,10 +89,13 @@ class Upstreams( } urls.add(endpoint.url) } + val options = (up.options ?: UpstreamsConfig.Options()) + .merge(defaultOptions[chain]) + .merge(UpstreamsConfig.Options.getDefaults()) if (rpcApi != null) { log.info("Info using ${chain.chainName} upstream, at ${urls.joinToString()}") val current = groups[chain] ?: ArrayList() - current.add(Upstream(chain, rpcApi!!, wsApi)) + current.add(Upstream(chain, rpcApi!!, wsApi, options)) groups[chain] = current } } @@ -87,4 +107,9 @@ class Upstreams( fun ethereumUpstream(chain: Chain): ChainConnect? { return chainMapping[chain] } + + @Scheduled(fixedRate = 15000) + fun printStatuses() { + chainMapping.forEach { it.value.printStatus() } + } } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy similarity index 77% rename from src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsReaderSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 3146a42d..1d038d37 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -2,9 +2,9 @@ package io.emeraldpay.dshackle.config import spock.lang.Specification -class UpstreamsReaderSpec extends Specification { +class UpstreamsConfigReaderSpec extends Specification { - UpstreamsReader reader = new UpstreamsReader() + UpstreamsConfigReader reader = new UpstreamsConfigReader() def "Parse standard config"() { setup: @@ -30,11 +30,11 @@ class UpstreamsReaderSpec extends Specification { provider == "geth" endpoints.size() == 2 with(endpoints.get(0)) { - type == Upstreams.EndpointType.JSON_RPC + type == UpstreamsConfig.EndpointType.JSON_RPC url == new URI("http://localhost:8545") } with(endpoints.get(1)) { - type == Upstreams.EndpointType.WEBSOCKET + type == UpstreamsConfig.EndpointType.WEBSOCKET url == new URI("ws://localhost:8546") } } @@ -44,10 +44,10 @@ class UpstreamsReaderSpec extends Specification { provider == "infura" endpoints.size() == 1 with(endpoints.get(0)) { - type == Upstreams.EndpointType.JSON_RPC + type == UpstreamsConfig.EndpointType.JSON_RPC url == new URI("https://mainnet.infura.io/v3/fa28c968191849c1aff541ad1d8511f2") - auth instanceof Upstreams.BasicAuth - with((Upstreams.BasicAuth)auth) { + auth instanceof UpstreamsConfig.BasicAuth + with((UpstreamsConfig.BasicAuth)auth) { key == "4fc258fe41a68149c199ad8f281f2015" } }