diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 33e05ebc..e76a486f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -200,6 +200,7 @@ open class ConfiguredUpstreams( val wsFactoryApi: EthereumWsFactory? = conn.ws?.let { endpoint -> val wsApi = EthereumWsFactory( + config.id!!, chain, endpoint.url, endpoint.origin ?: URI("http://localhost"), ) 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 51bed8d7..771820b0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt @@ -79,7 +79,7 @@ open class EthereumRpcUpstream( open fun createHead(): Head { return if (ethereumWsFactory != null) { // do not set upstream to the WS, since it doesn't control the RPC upstream - val ws = ethereumWsFactory.create(null, null, null).apply { + val ws = ethereumWsFactory.create(null, null).apply { connect() } val wsHead = EthereumWsHead(ws).apply { 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 da7aa412..00def0bf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -20,18 +20,49 @@ import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.DefaultUpstream 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 java.net.URI class EthereumWsFactory( + private val id: String, + private val chain: Chain, private val uri: URI, - private val origin: URI + private val origin: URI, ) { var basicAuth: AuthConfig.ClientBasicAuth? = null var config: UpstreamsConfig.WsEndpoint? = null - fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?, rpcMetrics: RpcMetrics?): WsConnection { - return WsConnection(uri, origin, basicAuth, rpcMetrics, upstream, validator).also { ws -> + // metrics are shared between all connections to the same WS + private val metrics: RpcMetrics = run { + val metricsTags = listOf( + Tag.of("upstream", id), + // UNSPECIFIED shouldn't happen too + Tag.of("chain", chain.chainCode) + ) + + 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.fail") + .description("Number of failures of WebSocket JSON RPC requests") + .tags(metricsTags) + .register(Metrics.globalRegistry) + ) + } + + fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?): WsConnection { + require(upstream == null || upstream.getId() == id) { + "Creating instance for different upstream. ${upstream?.getId()} != id" + } + return WsConnection(uri, origin, basicAuth, metrics, upstream, validator).also { ws -> config?.frameSize?.let { ws.frameSize = it } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt index f2d85e74..c721e3e0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt @@ -26,12 +26,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcSwitchClient 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 @@ -59,26 +54,8 @@ class EthereumWsUpstream( private val validator: EthereumUpstreamValidator 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.fail") - .description("Number of failures of WebSocket JSON RPC requests") - .tags(metricsTags) - .register(Metrics.globalRegistry) - ) - validator = EthereumUpstreamValidator(this, getOptions()) - - connection = ethereumWsFactory.create(this, validator, metrics) + connection = ethereumWsFactory.create(this, validator) head = EthereumWsHead(connection) // Sometimes the server may close the WebSocket connection during the execution of a call, for example if the response // is too large for WebSockets Frame (and Geth is unable to split messages into separate frames) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy index 9f16a8a2..eb3f22c0 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy @@ -4,6 +4,7 @@ import io.emeraldpay.dshackle.test.MockWSServer import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.grpc.Chain import reactor.test.StepVerifier import spock.lang.Shared import spock.lang.Specification @@ -30,7 +31,7 @@ class WsConnectionRealSpec extends Specification { server = new MockWSServer(port) server.start() Thread.sleep(SLEEP) - conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, null, null) + conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, null) } def cleanup() { @@ -109,8 +110,10 @@ class WsConnectionRealSpec extends Specification { def "Gets UNAVAIL status right after disconnect"() { setup: - def up = Mock(DefaultUpstream) - conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null, null) + def up = Mock(DefaultUpstream) { + _ * getId() >> "test" + } + conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null) when: conn.connect() conn.reconnectIntervalSeconds = 10 @@ -125,7 +128,7 @@ class WsConnectionRealSpec extends Specification { def "Validates after connect"() { setup: def validator = Mock(EthereumUpstreamValidator) - conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, validator, null) + conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, validator) when: conn.connect() Thread.sleep(100) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy index 3b4b5edb..b8996287 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy @@ -25,6 +25,7 @@ import io.emeraldpay.etherjar.rpc.RpcResponseError import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson +import io.emeraldpay.grpc.Chain import reactor.core.publisher.Flux import reactor.test.StepVerifier import spock.lang.Specification @@ -37,7 +38,7 @@ class WsConnectionSpec extends Specification { def "Fetch block"() { setup: - def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) + def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost")) def block = new BlockJson() block.number = 100 @@ -53,7 +54,7 @@ class WsConnectionSpec extends Specification { def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create(null, null, null) + def ws = wsf.create(null, null) apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block) @@ -71,10 +72,10 @@ class WsConnectionSpec extends Specification { def "Makes a RPC call"() { setup: - def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) + def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create(null, null, null) + def ws = wsf.create(null, null) def tx = new TransactionJson().tap { hash = TransactionId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200") @@ -96,10 +97,10 @@ class WsConnectionSpec extends Specification { def "Makes a RPC call - return null"() { setup: - def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) + def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create(null, null, null) + def ws = wsf.create(null, null) apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], null) @@ -119,10 +120,10 @@ class WsConnectionSpec extends Specification { def "Makes a RPC call - return error"() { setup: - def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) + def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create(null, null, null) + def ws = wsf.create(null, null) apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "test"))