diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index d0038558..4be0ba84 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -106,6 +106,7 @@ open class UpstreamsConfig { class EthereumConnection : RpcConnection() { var ws: WsEndpoint? = null + var preferHttp: Boolean = false } class BitcoinConnection : RpcConnection() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 488efe06..81e0d55f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -28,6 +28,7 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsUpstream import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcHttpClient import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -184,11 +185,6 @@ open class ConfiguredUpstreams( chain: Chain, options: UpstreamsConfig.Options) { val conn = config.connection!! - val directApi: Reader? = buildHttpClient(config) - if (directApi == null) { - log.warn("Upstream doesn't have API configuration") - return - } val urls = ArrayList() val methods = buildMethods(config, chain) @@ -209,13 +205,29 @@ open class ConfiguredUpstreams( } log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") - val ethereumUpstream = EthereumRpcUpstream( - config.id!!, - chain, directApi, wsFactoryApi, - options, config.role, - QuorumForLabels.QuorumItem(1, config.labels), - methods - ) + val ethereumUpstream = if (wsFactoryApi != null && !conn.preferHttp) { + EthereumWsUpstream( + config.id!!, + chain, wsFactoryApi, + options, config.role, + QuorumForLabels.QuorumItem(1, config.labels), + methods + ) + } else { + val directApi: Reader? = buildHttpClient(config) + if (directApi == null) { + log.warn("Upstream doesn't have API configuration") + return + } + EthereumRpcUpstream( + config.id!!, + chain, directApi, wsFactoryApi, + options, config.role, + QuorumForLabels.QuorumItem(1, config.labels), + methods + ) + } + ethereumUpstream.start() currentUpstreams.update(UpstreamChange(chain, ethereumUpstream, UpstreamChange.ChangeType.ADDED)) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt index 1844fa49..2bae1815 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt @@ -38,11 +38,6 @@ open class EthereumRpcUpstream( private val head: Head = this.createHead() private var validatorSubscription: Disposable? = null - private val capabilities = if (options.providesBalance != false) { - setOf(Capability.RPC, Capability.BALANCE) - } else { - setOf(Capability.RPC) - } override fun setCaches(caches: Caches) { if (head is CachesEnabled) { @@ -79,7 +74,7 @@ open class EthereumRpcUpstream( open fun createHead(): Head { return if (ethereumWsFactory != null) { - val ws = ethereumWsFactory.create().apply { + val ws = ethereumWsFactory.create(null).apply { connect() } val wsHead = EthereumWsHead(ws).apply { @@ -108,14 +103,6 @@ open class EthereumRpcUpstream( return directReader } - override fun getLabels(): Collection { - return listOf(node.labels) - } - - override fun getCapabilities(): Set { - return capabilities - } - override fun isGrpc(): Boolean { return false } 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 dbaaaf18..f629373e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -26,5 +26,20 @@ abstract class EthereumUpstream( options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?, - node: QuorumForLabels.QuorumItem? -) : DefaultUpstream(id, options, role, targets, node) \ No newline at end of file + private val node: QuorumForLabels.QuorumItem? +) : DefaultUpstream(id, options, role, targets, node) { + + private val capabilities = if (options.providesBalance != false) { + setOf(Capability.RPC, Capability.BALANCE) + } else { + setOf(Capability.RPC) + } + + override fun getCapabilities(): Set { + return capabilities + } + + override fun getLabels(): Collection { + return node?.let { listOf(it.labels) } ?: emptyList() + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt index 37977e31..3ff2883e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser +import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import io.netty.buffer.ByteBuf @@ -58,14 +59,15 @@ class EthereumWsFactory( var basicAuth: AuthConfig.ClientBasicAuth? = null - fun create(): EthereumWs { - return EthereumWs(uri, origin, basicAuth) + fun create(rpcMetrics: RpcMetrics?): EthereumWs { + return EthereumWs(uri, origin, basicAuth, rpcMetrics) } class EthereumWs( private val uri: URI, private val origin: URI, - private val basicAuth: AuthConfig.ClientBasicAuth? + private val basicAuth: AuthConfig.ClientBasicAuth?, + private val rpcMetrics: RpcMetrics? ) : AutoCloseable { companion object { @@ -255,12 +257,13 @@ class EthereumWsFactory( fun call(originalRequest: JsonRpcRequest): Mono { return Mono.fromCallable { + val startTime = System.nanoTime() // use an internal id sequence, to avoid id conflicts with user calls val internalId = sendIdSeq.getAndIncrement() val originalId = originalRequest.id - Tuples.of(originalRequest.copy(id = internalId), originalId) + Tuples.of(originalRequest.copy(id = internalId), originalId, startTime) }.flatMap { request -> - waitForResponse(request.t1, request.t2) + waitForResponse(request.t1, request.t2, request.t3) } } @@ -274,7 +277,7 @@ class EthereumWsFactory( } } - fun waitForResponse(request: JsonRpcRequest, originalId: Int): Mono { + fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono { val expectedId = request.id.toLong() return Mono.just(request) .flatMap { @@ -283,6 +286,12 @@ class EthereumWsFactory( .filter { resp -> resp.id.asNumber() == expectedId } .take(1) .singleOrEmpty() + .doOnNext { + rpcMetrics?.timer?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS) + } + .doOnError { + rpcMetrics?.errors?.increment() + } .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt new file mode 100644 index 00000000..cd1163a8 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt @@ -0,0 +1,120 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * 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.ethereum + +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.startup.QuorumForLabels +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient +import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics +import io.emeraldpay.grpc.Chain +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.Metrics +import io.micrometer.core.instrument.Tag +import io.micrometer.core.instrument.Timer +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable + +class EthereumWsUpstream( + id: String, + val chain: Chain, + ethereumWsFactory: EthereumWsFactory, + options: UpstreamsConfig.Options, + role: UpstreamsConfig.UpstreamRole, + node: QuorumForLabels.QuorumItem, + targets: CallMethods +) : EthereumUpstream(id, options, role, targets, node), Upstream, Lifecycle { + + companion object { + private val log = LoggerFactory.getLogger(EthereumWsUpstream::class.java) + } + + private val head: EthereumWsHead + private val connection: EthereumWsFactory.EthereumWs + private val api: JsonRpcWsClient + + private var validatorSubscription: Disposable? = null + + init { + val metricsTags = listOf( + Tag.of("upstream", id), + // UNSPECIFIED shouldn't happen too + Tag.of("chain", chain.chainCode) + ) + val metrics = RpcMetrics( + Timer.builder("upstream.ws.conn") + .description("Request time through a WebSocket JSON RPC connection") + .tags(metricsTags) + .publishPercentileHistogram() + .register(Metrics.globalRegistry), + Counter.builder("upstream.ws.err") + .description("Errors received on request through WebSocket JSON RPC connection") + .tags(metricsTags) + .register(Metrics.globalRegistry) + ) + + connection = ethereumWsFactory.create(metrics) + head = EthereumWsHead(connection) + api = JsonRpcWsClient(connection) + } + + override fun getHead(): Head { + return head + } + + override fun getApi(): Reader { + return api + } + + override fun isGrpc(): Boolean { + return false + } + + @Suppress("UNCHECKED_CAST") + override fun cast(selfType: Class): T { + if (!selfType.isAssignableFrom(this.javaClass)) { + throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") + } + return this as T + } + + override fun start() { + connection.connect() + head.start() + + log.debug("Start validation for upstream ${this.getId()}") + val validator = EthereumUpstreamValidator(this, getOptions()) + validatorSubscription = validator.start() + .subscribe(this::setStatus) + } + + override fun stop() { + validatorSubscription?.dispose() + validatorSubscription = null + head.stop() + connection.close() + } + + override fun isRunning(): Boolean { + return head.isRunning + } +} \ 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 5c1a553d..54608f81 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -74,6 +74,32 @@ class UpstreamsConfigReaderSpec extends Specification { } } + def "Parse websocket-only config"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("upstreams-ws-only.yaml") + when: + def act = reader.read(config) + then: + act != null + act.upstreams.size() == 1 + with(act.upstreams.get(0)) { + id == "local" + chain == "ethereum" + connection instanceof UpstreamsConfig.EthereumConnection + with((UpstreamsConfig.EthereumConnection) connection) { + rpc == null + ws != null + ws.url == new URI("ws://localhost:8546") + ws.basicAuth != null + with(ws.basicAuth) { + username == "9c199ad8f281f20154fc258fe41a6814" + password == "258fe4149c199ad8f2811a68f20154fc" + } + } + } + } + + def "Parse bitcoin upstreams"() { setup: def config = this.class.getClassLoader().getResourceAsStream("upstreams-bitcoin.yaml") diff --git a/src/test/resources/upstreams-ws-only.yaml b/src/test/resources/upstreams-ws-only.yaml new file mode 100644 index 00000000..decb2ca2 --- /dev/null +++ b/src/test/resources/upstreams-ws-only.yaml @@ -0,0 +1,13 @@ +version: v1 + +upstreams: + - id: local + chain: ethereum + connection: + ethereum: + ws: + url: "ws://localhost:8546" + origin: "http://localhost" + basic-auth: + username: 9c199ad8f281f20154fc258fe41a6814 + password: 258fe4149c199ad8f2811a68f20154fc \ No newline at end of file