Merge pull request #171 from emeraldpay/fix/no-ws-metrics

problem: head check doesn't use upstream WS metrics

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"),
             )

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 {

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 {
+    // metrics are shared between all connections to the same WS
-        return WsConnection(uri, origin, basicAuth, rpcMetrics, upstream, validator).also { 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
             }

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)
-        connection = ethereumWsFactory.create(this, validator, metrics)
         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)

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)
+        def up = Mock(DefaultUpstream) {
-        conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null, null)
+            _ * 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)

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<TransactionRefJson>()
         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"))