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

problem: doesn't increment error metrics for WS calls

Browse Source 
rel: #150
This commit is contained in:
Igor Artamonov
2022-03-16 22:14:37 -04:00
parent eac2bbb2d5
commit 10960aa99f
2 changed files with 25 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt
View File

@@ -46,7 +46,7 @@ class WebsocketHandler(
nativeCall: NativeCall,
private val nativeSubscribe: NativeSubscribe,
private val accessHandler: AccessHandlerHttp.HandlerFactory,
requestMetrics: ProxyServer.RequestMetricsFactory,
private val requestMetrics: ProxyServer.RequestMetricsFactory,
) : BaseHandler(writeRpcJson, nativeCall, requestMetrics) {
companion object {
@@ -68,7 +68,7 @@ class WebsocketHandler(
val requests: Flux<RequestJson<Any>> = req.aggregateFrames()
.receiveFrames()
.map { ByteBufInputStream(it.content()).readAllBytes() }
.flatMap(this@WebsocketHandler::parseRequest)
.flatMap { parseRequest(it, routeConfig.blockchain) }
val eventHandler = accessHandler.start(req, routeConfig.blockchain)
@@ -80,12 +80,13 @@ class WebsocketHandler(
}
}
fun parseRequest(data: ByteArray): Mono<RequestJson<Any>> {
fun parseRequest(data: ByteArray, blockchain: Chain): Mono<RequestJson<Any>> {
// try to parse JSON call. If received an invalid value just silently ignore it, that's what other Ethereum servers do
try {
val type = readRpcJson.getType(data)
// WS is not supposed to have batches, so ignore them too
if (type != ProxyCall.RpcType.SINGLE) {
requestMetrics.get(blockchain, "batch").errorMetric.increment()
return Mono.empty()
}
val items = readRpcJson.extract(type, data)
@@ -96,8 +97,14 @@ class WebsocketHandler(
return Mono
.just(items.first())
.map(readRpcJson.jsonExtractor)
.onErrorResume { Mono.empty() }
.onErrorResume {
log.debug("Failed to process request JSON with: ${it.javaClass} ${it.message}")
requestMetrics.get(blockchain, "invalid_method").errorMetric.increment()
Mono.empty()
}
} catch (t: Throwable) {
log.warn("Unhandled exception processing request message: ${t.javaClass} ${t.message}")
requestMetrics.get(blockchain, "invalid_method").errorMetric.increment()
return Mono.empty()
}
}
@@ -143,6 +150,7 @@ class WebsocketHandler(
eventHandler.onResponse(it.length.toLong())
}
} else {
requestMetrics.get(blockchain, "eth_subscribe").errorMetric.increment()
// TODO should it produce a 404 to the AccessLog?
Mono.empty()
}

17
src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy
View File

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.rpc.NativeCall
import io.emeraldpay.dshackle.rpc.NativeSubscribe
import io.emeraldpay.etherjar.rpc.json.RequestJson
import io.emeraldpay.grpc.Chain
import io.micrometer.core.instrument.Counter
import reactor.core.publisher.Flux
import reactor.core.publisher.Sinks
import spock.lang.Specification
@@ -37,7 +38,7 @@ class WebsocketHandlerSpec extends Specification {
new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory)
)
when:
def act = handler.parseRequest('{"id": 5, "jsonrpc": "2.0", "method": "eth_getBlockByNumber", "params": ["0x100001", false]}'.bytes)
def act = handler.parseRequest('{"id": 5, "jsonrpc": "2.0", "method": "eth_getBlockByNumber", "params": ["0x100001", false]}'.bytes, Chain.ETHEREUM)
.block(Duration.ofSeconds(1))
then:
@@ -48,11 +49,19 @@ class WebsocketHandlerSpec extends Specification {
def "Parse to empty an invalid request"() {
setup:
Counter errorMetric = Mock(Counter) {
1 * increment()
}
ProxyServer.RequestMetricsFactory metrics = Mock(ProxyServer.RequestMetricsFactory) {
1 * get(Chain.ETHEREUM, "invalid_method") >> Mock(ProxyServer.RequestMetrics) {
1 * it.errorMetric >> errorMetric
}
}
def handler = new WebsocketHandler(
new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory)
new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, metrics
)
when:
def act = handler.parseRequest('hello world'.bytes)
def act = handler.parseRequest('hello world'.bytes, Chain.ETHEREUM)
.block(Duration.ofSeconds(1))
then:
@@ -66,7 +75,7 @@ class WebsocketHandlerSpec extends Specification {
new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory)
)
when:
def act = handler.parseRequest("[$req1]".bytes)
def act = handler.parseRequest("[$req1]".bytes, Chain.ETHEREUM)
.block(Duration.ofSeconds(1))
then: