From bd6d6aa6bbdbb2fb027b4ae991569f58d88ef0b8 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 10 Sep 2019 22:36:20 -0400 Subject: [PATCH] solution: configurable timeout --- README.adoc | 2 +- docs/04-upstream-config.adoc | 27 ++++++++++++------- .../UpstreamServices.kt => Defaults.kt} | 14 +++------- .../dshackle/config/UpstreamsConfig.kt | 3 +++ .../dshackle/config/UpstreamsConfigReader.kt | 4 +++ .../emeraldpay/dshackle/rpc/TrackAddress.kt | 3 ++- .../dshackle/upstream/ConfiguredUpstreams.kt | 14 +++++++--- .../dshackle/upstream/UpstreamValidator.kt | 3 ++- .../upstream/ethereum/DirectEthereumApi.kt | 3 ++- .../upstream/ethereum/EthereumRpcHead.kt | 5 ++-- .../dshackle/upstream/ethereum/EthereumWs.kt | 4 +-- .../dshackle/upstream/grpc/GrpcUpstream.kt | 6 +++-- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 5 +++- 13 files changed, 58 insertions(+), 35 deletions(-) rename src/main/kotlin/io/emeraldpay/dshackle/{upstream/UpstreamServices.kt => Defaults.kt} (55%) diff --git a/README.adoc b/README.adoc index 516ce751..eeb0ccbe 100644 --- a/README.adoc +++ b/README.adoc @@ -45,7 +45,7 @@ upstreams: Which sets the following: -- application listen on 0.0.0.0:2448 +- application listen on 0.0.0.0:2449 - TLS security is disabled (_don't use in production!_) - read upstreams configuration from file `upstreams.yaml` in the current directory diff --git a/docs/04-upstream-config.adoc b/docs/04-upstream-config.adoc index 6ce9f4b8..cfdaf47d 100644 --- a/docs/04-upstream-config.adoc +++ b/docs/04-upstream-config.adoc @@ -86,9 +86,15 @@ In the example above we have: Options (default or as part of upstream config): -- `disable-validation` - if `true` then Dshackle would not try to verify status of the upstream (useful for a trusted cloud - provider such as Infura, but is not recommended for an own node) -- `min-peers` - do not use upstream with less than specified connected peers +[cols="2,1,5a"] +|=== +| Option | Default | Description + +| `disable-validation` | false | if `true` then Dshackle will not try to verify status of the upstream (could be useful for a trusted cloud + provider such as Infura, but disabling it is not recommended for a normal node) +| `min-peers` | 3 | specify minimum amount of connected peers, Dshackle will not use upstream with less than specified number +| `timeout` | 60 | timeout in seconds after which request to the upstream will be discarded (and may be retried on an another upstream) +|=== === Connection type @@ -153,12 +159,13 @@ upstreams: - name: eth_getBlockByNumber ---- -Such configuration option allows to execute methods `trace_transaction` and also disables `eth_getBlockByNumber` on that -particular upstream. If a client tries to execute method `trace_transaction` it will be executed on that upstream, or -another upstream will have such method enabled. +Such configuration option allows to execute method `trace_transaction` and also disables `eth_getBlockByNumber` on that +particular upstream. If a client requests to execute method `trace_transaction` then it will be scheduled to that upstream (or +any upstream with such method enabled). -Together with label `archive: true` it's possible to specify during execution that a client wants to execute method only -on an archive node. +NOTE: It's especially useful when used together with upstream labels. If an archive upstream has label `archive: true` it's +possible to specify that the client wants to execute method `trace_transaction` only on an archive node(s), which has +complete historical data for tracing. === Authentication @@ -167,7 +174,9 @@ on an archive node. All connection types can use TLS secured connection, with optional client certificate authentication: - `ca` path to certificate required from remote server -- optional `certificate` and `key` for client authentication. Please note that `key` is encoded with _PKCS 8_ +- optional `certificate` and `key` for client authentication. + +NOTE: Please note that `key` must be encoded with _PKCS 8_ ==== Basic Authentication diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamServices.kt b/src/main/kotlin/io/emeraldpay/dshackle/Defaults.kt similarity index 55% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamServices.kt rename to src/main/kotlin/io/emeraldpay/dshackle/Defaults.kt index ad4658da..0f19e9b4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamServices.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Defaults.kt @@ -13,21 +13,13 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package io.emeraldpay.dshackle.upstream +package io.emeraldpay.dshackle -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono import java.time.Duration -class UpstreamServices { +class Defaults { companion object { - fun onceOk(up: Upstream, waitTime: Duration = Duration.ofSeconds(15)): Mono { - return Flux.concat(Flux.just(up.getStatus()), up.observeStatus()) - .timeout(waitTime, Mono.just(UpstreamAvailability.UNAVAILABLE)) - .filter { it == UpstreamAvailability.OK } - .next() - .hasElement() - } + val timeout = Duration.ofSeconds(60) } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 42462938..e4537db9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -15,7 +15,9 @@ */ package io.emeraldpay.dshackle.config +import io.emeraldpay.dshackle.Defaults import java.net.URI +import java.time.Duration import java.util.* import kotlin.collections.ArrayList import kotlin.collections.HashMap @@ -27,6 +29,7 @@ class UpstreamsConfig { open class Options { var disableValidation: Boolean? = null + var timeout = Defaults.timeout var minPeers: Int? = 1 set(minPeers) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index 88553a5d..0339401b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -27,6 +27,7 @@ import java.io.InputStream import java.io.InputStreamReader import java.lang.IllegalArgumentException import java.net.URI +import java.time.Duration class UpstreamsConfigReader { @@ -185,6 +186,9 @@ class UpstreamsConfigReader { getValueAsInt(values, "min-peers")?.let { options.minPeers = it } + getValueAsInt(values, "timeout")?.let { + options.timeout = Duration.ofSeconds(it.toLong()) + } getValueAsBool(values, "disable-validation")?.let { options.disableValidation = it } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt index e89d190a..ef092d71 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain @@ -169,7 +170,7 @@ class TrackAddress( val up = upstreams.getUpstream(addr.chain) ?: return Mono.error(Exception("Unsupported chain: ${addr.chain}")) return up.getApi(Selector.empty) .flatMap { api -> api.executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST)) } - .timeout(Duration.ofSeconds(15)) + .timeout(Defaults.timeout) } private fun updateBalances(chain: Chain, group: List): Flux { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index 5a65307e..dbd70d8b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -63,7 +63,8 @@ open class ConfiguredUpstreams( config.upstreams.forEach { up -> if (up.connection is UpstreamsConfig.GrpcConnection) { - buildGrpcUpstream(up as UpstreamsConfig.Upstream) + val options = up.options ?: UpstreamsConfig.Options() + buildGrpcUpstream(up as UpstreamsConfig.Upstream, options) } else { val chain = chainNames[up.chain] if (chain == null) { @@ -145,7 +146,10 @@ open class ConfiguredUpstreams( rpcClient, objectMapper, methods - ) + ).apply { + timeout = options.timeout + } + urls.add(endpoint.url) } if (rpcApi != null) { @@ -174,7 +178,7 @@ open class ConfiguredUpstreams( } } - private fun buildGrpcUpstream(config: UpstreamsConfig.Upstream) { + private fun buildGrpcUpstream(config: UpstreamsConfig.Upstream, options: UpstreamsConfig.Options) { val endpoint = config.connection!! val ds = GrpcUpstreams( config.id!!, @@ -183,7 +187,9 @@ open class ConfiguredUpstreams( objectMapper, endpoint.auth, fileResolver - ) + ).apply { + timeout = options.timeout + } log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}") ds.start() .doOnNext { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt index a553f35b..8f4a5288 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.infinitape.etherjar.rpc.Batch @@ -36,7 +37,7 @@ class UpstreamValidator( return ethereumUpstream.getApi(Selector.empty) .map { api -> api.rpcClient.execute(batch) } .flatMap { Mono.fromCompletionStage(it) } - .timeout(Duration.ofSeconds(10)) + .timeout(Defaults.timeout) .map { if (syncing.get().isSyncing) { UpstreamAvailability.SYNCING diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt index ce28dca9..a0d62dfe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt @@ -16,6 +16,7 @@ package io.emeraldpay.dshackle.upstream.ethereum import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.upstream.CallMethods import io.infinitape.etherjar.rpc.RpcCall import io.infinitape.etherjar.rpc.RpcClient @@ -31,7 +32,7 @@ open class DirectEthereumApi( val targets: CallMethods ): EthereumApi(objectMapper) { - private val timeout = Duration.ofSeconds(5) + var timeout = Defaults.timeout private val log = LoggerFactory.getLogger(EthereumApi::class.java) override fun execute(id: Int, method: String, params: List): Mono { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt index 6acece47..d8b8ac9c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.Defaults import io.infinitape.etherjar.rpc.Batch import io.infinitape.etherjar.rpc.Commands import org.slf4j.LoggerFactory @@ -39,13 +40,13 @@ class EthereumRpcHead( val batch = Batch() val f = batch.add(Commands.eth().blockNumber) api.rpcClient.execute(batch) - Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5), Mono.empty()) + Mono.fromCompletionStage(f).timeout(Defaults.timeout, Mono.empty()) } .flatMap { val batch = Batch() val f = batch.add(Commands.eth().getBlock(it)) api.rpcClient.execute(batch) - Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5), Mono.empty()) + Mono.fromCompletionStage(f).timeout(Defaults.timeout, Mono.empty()) } .onErrorContinue { err, _ -> log.debug("RPC error ${err.message}") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWs.kt index 7578007d..9e3fe945 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWs.kt @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.config.UpstreamsConfig import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.Commands @@ -62,7 +63,7 @@ class EthereumWs( .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(250)) .apply(n) } - .timeout(Duration.ofSeconds(5), Mono.empty()) + .timeout(Defaults.timeout, Mono.empty()) .subscribe(topic::onNext) } else { topic.onNext(it) @@ -73,6 +74,5 @@ class EthereumWs( fun getFlux(): Flux> { return Flux.from(this.topic) .onBackpressureLatest() - .sample(Duration.ofMillis(100)) } } \ No newline at end of file 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 3aa18469..42a9ac69 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt @@ -20,6 +20,7 @@ import com.salesforce.reactorgrpc.GrpcRetry import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.api.proto.ReactorBlockchainGrpc +import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.ethereum.DefaultEthereumHead @@ -61,9 +62,10 @@ open class GrpcUpstream( private val nodes = AtomicReference(NodeDetailsList()) private val head = DefaultEthereumHead() private var targets: CallMethods? = null - private var headSubscription: Disposable? = null + var timeout = Defaults.timeout + open fun createApi(matcher: Selector.Matcher): DirectEthereumApi { val targets = this.getMethods() val transport = Selector.extractLabels(matcher)?.let { selector -> @@ -123,7 +125,7 @@ open class GrpcUpstream( }.flatMap { getApi(Selector.EmptyMatcher()) .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(it.hash)) } - .timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Timeout from upstream"))) + .timeout(timeout, Mono.error(TimeoutException("Timeout from upstream"))) .doOnError { t -> setStatus(UpstreamAvailability.UNAVAILABLE) val msg = "Failed to download block data for chain $chain on $parentId" 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 513d2a1d..092140cc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.grpc import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.ReactorBlockchainGrpc +import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.UpstreamAvailability @@ -50,10 +51,11 @@ class GrpcUpstreams( ) { private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java) + var timeout = Defaults.timeout + private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null private val known = HashMap() private val lock = ReentrantLock() - private var grpcTransport: EmeraldGrpcTransport? = null fun start(): Flux { @@ -162,6 +164,7 @@ class GrpcUpstreams( val current = known[chain] return if (current == null) { val created = GrpcUpstream(id, chain, client!!, objectMapper, grpcTransport!!.copyForChain(chain)) + created.timeout = this.timeout known[chain] = created created.start() UpstreamChange(chain, created, UpstreamChange.ChangeType.ADDED)