solution: access log for JSON RPC proxy

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

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<io.grpc.BindableService>,
         @Autowired val mainConfig: MainConfig,
         @Autowired val tlsSetup: TlsSetup,
-        @Autowired val accessHandler: AccessHandler
+        @Autowired val accessHandler: AccessHandlerGrpc
 ) {
 
     private val log = LoggerFactory.getLogger(GrpcServer::class.java)

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()
     }
 

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 <ReqT : Any, RespT : Any> interceptCall(

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<NativeCall.CallResult>()
+        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)
+            }
+        }
+    }
+}

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
+            val blockchain: Chain, method: String, id: UUID, channel: Channel
-    ) : Base(id, method)
+    ) : 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,

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<E, Req, Resp> : StartingRequest {
+    interface StartingHttp1Request {
+        fun start(request: HttpServerRequest)
+    }
+
+    interface RequestReply<E, Req, Resp> : StartingHttp2Request {
         fun onRequest(msg: Req)
         fun onReply(msg: Resp): E
     }
 
-    abstract class Base<T>() : StartingRequest {
+    abstract class Base<T>() : 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<InetAddress>()
+            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 :

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<String> {
+    fun execute(chain: Common.ChainRef, call: ProxyCall, handler: AccessHandlerHttp.RequestHandler): Publisher<String> {
         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<ByteArray>): Flux<ByteBuf> {
+    fun processRequest(chain: Common.ChainRef, request: Mono<ByteArray>, handler: AccessHandlerHttp.RequestHandler): Flux<ByteBuf> {
         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<HttpServerRequest, HttpServerResponse, Publisher<Void>> {
         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)
         }

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

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)
+    }
 }