StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

problem: head check doesn't use upstream WS metrics

Browse Source
This commit is contained in:
Igor Artamonov
2022-08-05 21:26:35 -04:00
parent a26db98376
commit a34da8b4a3
6 changed files with 53 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
View File

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

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
View File

@@ -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 {

37
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt
View File

@@ -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
}

25
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt
View File

@@ -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)

11
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy
View File

@@ -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)

17
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy
View File

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