diff --git a/docs/06-monitoring.adoc b/docs/06-monitoring.adoc index 92612fa3..2b4aa9ff 100644 --- a/docs/06-monitoring.adoc +++ b/docs/06-monitoring.adoc @@ -30,6 +30,7 @@ The access log contains the JSON lines similar to: "ts":"2021-07-20T01:53:33.174645Z", "id":"578d83db-cf53-4ef8-b73e-3f1cc0a67e96", "method":"NativeCall", + "channel":"GRPC", "blockchain":"ETHEREUM", "total":2, "index":0, @@ -56,6 +57,7 @@ The access log contains the JSON lines similar to: - `id` uniq id of the reply - `method` Dshackle method which was called (i.e., not a Blockchain API method, see `nativeCall` details) - `blockchain` blockchain code +- `channel` access channel (`GRPC` for native Dshackle calls, `JSONRPC` for JSON RPC HTTP Proxy) - `total` how many requests in the batch (available only for a `NativeCall` call) - `index` current index (i.e. count) of the reply to the original request - `succeed` if call succeeded, in terms of Blockchain API diff --git a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt index 10563348..62a09f4a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt @@ -17,7 +17,7 @@ package io.emeraldpay.dshackle import io.emeraldpay.dshackle.config.MainConfig -import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandler +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerGrpc import io.grpc.* import io.grpc.netty.NettyServerBuilder import org.slf4j.LoggerFactory @@ -32,7 +32,7 @@ open class GrpcServer( @Autowired val rpcs: List, @Autowired val mainConfig: MainConfig, @Autowired val tlsSetup: TlsSetup, - @Autowired val accessHandler: AccessHandler + @Autowired val accessHandler: AccessHandlerGrpc ) { private val log = LoggerFactory.getLogger(GrpcServer::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt b/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt index 069369b1..fc70c2c8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle import io.emeraldpay.dshackle.config.MainConfig import io.emeraldpay.dshackle.config.ProxyConfig +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp import io.emeraldpay.dshackle.proxy.ProxyServer import io.emeraldpay.dshackle.proxy.ReadRpcJson import io.emeraldpay.dshackle.proxy.WriteRpcJson @@ -37,7 +38,8 @@ class ProxyStarter( @Autowired private val readRpcJson: ReadRpcJson, @Autowired private val writeRpcJson: WriteRpcJson, @Autowired private val nativeCall: NativeCall, - @Autowired private val tlsSetup: TlsSetup + @Autowired private val tlsSetup: TlsSetup, + @Autowired private val accessHandlerHttp: AccessHandlerHttp ) { companion object { @@ -51,7 +53,7 @@ class ProxyStarter( log.debug("Proxy server is not configured") return } - val server = ProxyServer(config, readRpcJson, writeRpcJson, nativeCall, tlsSetup) + val server = ProxyServer(config, readRpcJson, writeRpcJson, nativeCall, tlsSetup, accessHandlerHttp.factory) server.start() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt similarity index 97% rename from src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt rename to src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt index 1685eba3..2e5d0377 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt @@ -15,20 +15,18 @@ */ package io.emeraldpay.dshackle.monitoring.accesslog -import io.emeraldpay.api.proto.BlockchainOuterClass -import io.emeraldpay.api.proto.Common import io.grpc.* import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @Service -class AccessHandler( +class AccessHandlerGrpc( @Autowired private val accessLogWriter: AccessLogWriter ) : ServerInterceptor { companion object { - private val log = LoggerFactory.getLogger(AccessHandler::class.java) + private val log = LoggerFactory.getLogger(AccessHandlerGrpc::class.java) } override fun interceptCall( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerHttp.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerHttp.kt new file mode 100644 index 00000000..68daa1b4 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerHttp.kt @@ -0,0 +1,111 @@ +package io.emeraldpay.dshackle.monitoring.accesslog + +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.config.MainConfig +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.grpc.Chain +import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.stereotype.Service +import reactor.netty.http.server.HttpServerRequest +import java.time.Instant +import java.util.concurrent.locks.ReentrantLock +import kotlin.concurrent.withLock + +/** + * Access Log handler for JSON RPC proxy + * + * @see io.emeraldpay.dshackle.proxy.ProxyServer + */ +@Service +class AccessHandlerHttp( + @Autowired private val mainConfig: MainConfig, + @Autowired accessLogWriter: AccessLogWriter +) { + + companion object { + private val log = LoggerFactory.getLogger(AccessHandlerHttp::class.java) + } + + /** + * Use factory since we need a different behaviour for situation when log is configured and when is not + */ + val factory: HandlerFactory = if (mainConfig.accessLogConfig.enabled) { + StandardFactory(accessLogWriter) + } else { + NoOpFactory() + } + + interface HandlerFactory { + fun create(req: HttpServerRequest, blockchain: Chain): RequestHandler + } + + class NoOpFactory() : HandlerFactory { + override fun create(req: HttpServerRequest, blockchain: Chain): RequestHandler { + return NoOpHandler() + } + } + + class StandardFactory(val accessLogWriter: AccessLogWriter) : HandlerFactory { + override fun create(req: HttpServerRequest, blockchain: Chain): RequestHandler { + return StandardHandler(accessLogWriter, req, blockchain) + } + } + + interface RequestHandler { + fun close() + fun onRequest(request: BlockchainOuterClass.NativeCallRequest) + fun onResponse(callResult: NativeCall.CallResult) + } + + class NoOpHandler : RequestHandler { + override fun close() { + } + + override fun onRequest(request: BlockchainOuterClass.NativeCallRequest) { + } + + override fun onResponse(callResult: NativeCall.CallResult) { + } + } + + class StandardHandler( + private val accessLogWriter: AccessLogWriter, + private val httpRequest: HttpServerRequest, + private val blockchain: Chain + ) : RequestHandler { + + private var request: BlockchainOuterClass.NativeCallRequest? = null + private val responses = ArrayList() + private val updateLock = ReentrantLock() + + override fun close() { + if (request == null) { + return + } + val responseTime = Instant.now() + val builder = EventsBuilder.NativeCall() + builder.withChain(blockchain.id) + builder.start(httpRequest) + builder.onRequest(request!!) + responses + .map { + builder.onReply(it, Events.Channel.JSONRPC).also { item -> + //since for JSON RPC you get a single response then the timestamp of all items included in it must have the same timestamp + item.ts = responseTime + } + } + .let(accessLogWriter::submit) + } + + override fun onRequest(request: BlockchainOuterClass.NativeCallRequest) { + this.request = request + } + + override fun onResponse(callResult: NativeCall.CallResult) { + updateLock.withLock { + responses.add(callResult) + } + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index 96c255a3..39045728 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -27,17 +27,22 @@ class Events { private val log = LoggerFactory.getLogger(Events::class.java) } + enum class Channel { + GRPC, JSONRPC + } + abstract class Base( val id: UUID, - val method: String + val method: String, + val channel: Channel ) { val version = "accesslog/v1beta" - val ts = Instant.now() + var ts = Instant.now() } abstract class ChainBase( - val blockchain: Chain, method: String, id: UUID - ) : Base(id, method) + val blockchain: Chain, method: String, id: UUID, channel: Channel + ) : Base(id, method, channel) @JsonInclude(JsonInclude.Include.NON_NULL) class SubscribeHead( @@ -46,7 +51,7 @@ class Events { val request: StreamRequestDetails, // index of the current response val index: Int - ) : ChainBase(blockchain, "SubscribeHead", id) + ) : ChainBase(blockchain, "SubscribeHead", id, Channel.GRPC) @JsonInclude(JsonInclude.Include.NON_NULL) class SubscribeBalance( @@ -57,7 +62,7 @@ class Events { val addressBalance: AddressBalance, // index of the current response val index: Int - ) : ChainBase(blockchain, if (subscribe) "SubscribeBalance" else "GetBalance", id) + ) : ChainBase(blockchain, if (subscribe) "SubscribeBalance" else "GetBalance", id, Channel.GRPC) @JsonInclude(JsonInclude.Include.NON_NULL) class TxStatus( @@ -67,7 +72,7 @@ class Events { val txStatus: TxStatusResponse, // index of the current response val index: Int - ) : ChainBase(blockchain, "SubscribeTxStatus", id) + ) : ChainBase(blockchain, "SubscribeTxStatus", id, Channel.GRPC) data class TxStatusRequest( val txId: String @@ -79,7 +84,7 @@ class Events { @JsonInclude(JsonInclude.Include.NON_NULL) class NativeCall( - blockchain: Chain, id: UUID, + blockchain: Chain, id: UUID, channel: Channel, // info about the initial request, that may include several native calls val request: StreamRequestDetails, @@ -95,19 +100,19 @@ class Events { val rpcError: Int? = null, val payloadSizeBytes: Long, val nativeCall: NativeCallItemDetails - ) : ChainBase(blockchain, "NativeCall", id) + ) : ChainBase(blockchain, "NativeCall", id, channel) @JsonInclude(JsonInclude.Include.NON_NULL) class Describe( id: UUID, val request: StreamRequestDetails - ) : Base(id, "Describe") + ) : Base(id, "Describe", Channel.GRPC) @JsonInclude(JsonInclude.Include.NON_NULL) class Status( blockchain: Chain, id: UUID, val request: StreamRequestDetails - ) : ChainBase(blockchain, "Status", id) + ) : ChainBase(blockchain, "Status", id, Channel.GRPC) data class StreamRequestDetails( val id: UUID, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt index e8a88fe1..cf1be6d3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt @@ -23,6 +23,7 @@ import io.grpc.Grpc import io.grpc.Metadata import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory +import reactor.netty.http.server.HttpServerRequest import java.net.InetAddress import java.net.InetSocketAddress import java.time.Instant @@ -34,17 +35,25 @@ class EventsBuilder { private val log = LoggerFactory.getLogger(EventsBuilder::class.java) } - interface StartingRequest { + interface StartingHttp2Request { fun start(metadata: Metadata, attributes: Attributes) } - interface RequestReply : StartingRequest { + interface StartingHttp1Request { + fun start(request: HttpServerRequest) + } + + interface RequestReply : StartingHttp2Request { fun onRequest(msg: Req) fun onReply(msg: Resp): E } - abstract class Base() : StartingRequest { + abstract class Base() : StartingHttp2Request, StartingHttp1Request { companion object { + private val remoteIpHeaders = listOf( + "x-real-ip", + "x-forwarded-for" + ) private val remoteIpKeys = listOf( Metadata.Key.of("x-real-ip", Metadata.ASCII_STRING_MARSHALLER), Metadata.Key.of("x-forwarded-for", Metadata.ASCII_STRING_MARSHALLER) @@ -120,6 +129,31 @@ class EventsBuilder { )) } + override fun start(request: HttpServerRequest) { + val headers = request.requestHeaders() + val userAgent = headers.get("user-agent") + ?.let(this@Base::clean) + ?: "" + val ips = ArrayList() + remoteIpHeaders.forEach { key -> + headers.get(key)?.let { + it.trim().ifEmpty { null } + ?.let(this@Base::toInetAddress) + ?.let(ips::add) + } + } + request.remoteAddress()?.let { addr -> + ips.add(addr.address) + } + val ip = findBestIp(ips)?.hostAddress ?: "" + this.requestDetails = this.requestDetails + .copy(remote = Events.Remote( + ips = ips.map { it.hostAddress }, + ip = ip, + userAgent = userAgent + )) + } + fun withChain(chain: Int): T { this.chainId = chain this.chain = Chain.byId(chainId) @@ -237,10 +271,26 @@ class EventsBuilder { blockchain = chain, nativeCall = item, payloadSizeBytes = item.payloadSizeBytes, - id = UUID.randomUUID() + id = UUID.randomUUID(), + channel = Events.Channel.GRPC ) } + fun onReply(reply: io.emeraldpay.dshackle.rpc.NativeCall.CallResult, + channel: Events.Channel): Events.NativeCall { + val item = items.find { it.id == reply.id }!! + return Events.NativeCall( + request = requestDetails, + total = items.size, + index = index++, + succeed = !reply.isError(), + blockchain = chain, + nativeCall = item, + payloadSizeBytes = item.payloadSizeBytes, + id = UUID.randomUUID(), + channel = channel + ) + } } class Describe : diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt index b39f0cac..0268c8da 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.ChainValue import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.TlsSetup import io.emeraldpay.dshackle.config.ProxyConfig +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp import io.emeraldpay.dshackle.rpc.NativeCall import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain @@ -33,7 +34,6 @@ import io.netty.buffer.ByteBuf import io.netty.buffer.Unpooled import io.netty.channel.ChannelHandler import io.netty.channel.ChannelHandlerContext -import io.netty.channel.ChannelOption import org.reactivestreams.Publisher import org.slf4j.LoggerFactory import org.springframework.http.HttpHeaders @@ -55,7 +55,8 @@ class ProxyServer( private val readRpcJson: ReadRpcJson, private val writeRpcJson: WriteRpcJson, private val nativeCall: NativeCall, - private val tlsSetup: TlsSetup + private val tlsSetup: TlsSetup, + private val accessHandler: AccessHandlerHttp.HandlerFactory ) { companion object { @@ -113,13 +114,15 @@ class ProxyServer( } } - fun execute(chain: Common.ChainRef, call: ProxyCall): Publisher { + fun execute(chain: Common.ChainRef, call: ProxyCall, handler: AccessHandlerHttp.RequestHandler): Publisher { val request = BlockchainOuterClass.NativeCallRequest.newBuilder() .setChain(chain) .addAllItems(call.items) .build() + handler.onRequest(request) val jsons = nativeCall .nativeCallResult(Mono.just(request)) + .doOnNext { handler.onResponse(it) } .transform(writeRpcJson.toJsons(call)) return if (call.type == ProxyCall.RpcType.SINGLE) { jsons.next() @@ -128,13 +131,13 @@ class ProxyServer( } } - fun processRequest(chain: Common.ChainRef, request: Mono): Flux { + fun processRequest(chain: Common.ChainRef, request: Mono, handler: AccessHandlerHttp.RequestHandler): Flux { val metrics = chainMetrics.get(chain) val startTime = System.currentTimeMillis() metrics.requestMetric.increment() return request .map(readRpcJson) - .flatMapMany { call -> execute(chain, call) } + .flatMapMany { call -> execute(chain, call, handler) } .doOnNext { metrics.callMetric.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS) } @@ -153,10 +156,14 @@ class ProxyServer( fun proxy(routeConfig: ProxyConfig.Route): BiFunction> { val chain = Common.ChainRef.forNumber(routeConfig.blockchain.id) return BiFunction { req, resp -> + // handle access events + val eventHandler = accessHandler.create(req, routeConfig.blockchain) val request = req.receive() .aggregate() .asByteArray() - val results = processRequest(chain, request) + val results = processRequest(chain, request, eventHandler) + // make sure that the access log handler is closed at the end, so it can render the logs + .doFinally { eventHandler.close() } resp.addHeader(HttpHeaders.CONTENT_TYPE, "application/json") .send(results) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy index ede6ab9d..a8375ac6 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy @@ -30,7 +30,7 @@ class EventsBaseBuilderSpec extends Specification { Events.StreamRequestDetails request TestEvent(Events.StreamRequestDetails request) { - super(UUID.randomUUID(), "TEST") + super(UUID.randomUUID(), "TEST", Events.Channel.GRPC) this.request = request } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy index 76ecf537..682123a3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy @@ -16,10 +16,12 @@ */ package io.emeraldpay.dshackle.proxy +import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.TlsSetup import io.emeraldpay.dshackle.config.ProxyConfig +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp import io.emeraldpay.dshackle.rpc.NativeCall import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse @@ -48,7 +50,8 @@ class ProxyServerSpec extends Specification { new ReadRpcJson(), writeRpcJson, nativeCall, - new TlsSetup(TestingCommons.fileResolver()) + new TlsSetup(TestingCommons.fileResolver()), + new AccessHandlerHttp.NoOpFactory() ) def call = new ProxyCall(ProxyCall.RpcType.SINGLE) @@ -59,7 +62,7 @@ class ProxyServerSpec extends Specification { .build() ) when: - def act = server.execute(Common.ChainRef.CHAIN_ETHEREUM, call) + def act = server.execute(Common.ChainRef.CHAIN_ETHEREUM, call, new AccessHandlerHttp.NoOpHandler()) then: 1 * nativeCall.nativeCallResult(_) >> Flux.just(new NativeCall.CallResult(1, "".bytes, null)) @@ -77,10 +80,11 @@ class ProxyServerSpec extends Specification { def server = new ProxyServer( Stub(ProxyConfig), read, - Stub(WriteRpcJson), Stub(NativeCall), Stub(TlsSetup) + Stub(WriteRpcJson), Stub(NativeCall), Stub(TlsSetup), + new AccessHandlerHttp.NoOpFactory() ) when: - def act = server.processRequest(Common.ChainRef.CHAIN_ETHEREUM, Mono.just("".bytes)) + def act = server.processRequest(Common.ChainRef.CHAIN_ETHEREUM, Mono.just("".bytes), new AccessHandlerHttp.NoOpHandler()) .map { new String(it.array()) } then: StepVerifier.create(act) @@ -88,4 +92,44 @@ class ProxyServerSpec extends Specification { .expectComplete() .verify(Duration.ofSeconds(1)) } + + def "Calls access log handler"() { + setup: + def reqItem = BlockchainOuterClass.NativeCallItem.newBuilder() + .setId(1) + .setMethod("test_test") + .setPayload(ByteString.copyFromUtf8("[]")) + .build() + def respItem = new NativeCall.CallResult(1, "100".bytes, null) + def req = BlockchainOuterClass.NativeCallRequest.newBuilder() + .setChain(Common.ChainRef.CHAIN_ETHEREUM) + .addItems(reqItem) + .build() + + + ReadRpcJson read = Mock(ReadRpcJson) { + 1 * apply(_) >> new ProxyCall(ProxyCall.RpcType.SINGLE).tap { it.items.add(reqItem) } + } + NativeCall nativeCall = Mock(NativeCall) { + 1 * nativeCallResult(_) >> Flux.fromIterable([respItem]) + } + def handler = Mock(AccessHandlerHttp.RequestHandler.class) + + def server = new ProxyServer( + Stub(ProxyConfig), + read, + new WriteRpcJson(), + nativeCall, + Stub(TlsSetup), + new AccessHandlerHttp.NoOpFactory() + ) + + when: + server.processRequest(Common.ChainRef.CHAIN_ETHEREUM, Mono.just("".bytes), handler) + .blockLast() + + then: + 1 * handler.onRequest(req) + 1 * handler.onResponse(respItem) + } }