solution: base code for access logging

build.gradle

@@ -110,8 +110,10 @@ dependencies {
     implementation 'org.yaml:snakeyaml:1.24'
     implementation 'org.apache.httpcomponents:httpmime:4.5.8'
     implementation 'org.apache.httpcomponents:httpclient:4.5.8'
-    implementation 'com.fasterxml.jackson.core:jackson-core:2.9.8'
+    implementation "com.fasterxml.jackson.core:jackson-core:$jacksonVersion"
-    implementation 'com.fasterxml.jackson.core:jackson-databind:2.9.8'
+    implementation "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
+    implementation "com.fasterxml.jackson.datatype:jackson-datatype-jdk8:$jacksonVersion"
+    implementation "com.fasterxml.jackson.datatype:jackson-datatype-jsr310:$jacksonVersion"
     implementation 'commons-io:commons-io:2.6'
     implementation 'org.apache.commons:commons-lang3:3.9'
     implementation 'org.apache.commons:commons-collections4:4.3'

src/main/kotlin/io/emeraldpay/dshackle/Global.kt

@@ -19,6 +19,8 @@ import com.fasterxml.jackson.core.Version
 import com.fasterxml.jackson.databind.DeserializationFeature
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.databind.module.SimpleModule
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
 import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspent
 import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspentDeserializer
 import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent
@@ -49,6 +51,8 @@ class Global {
 
             val objectMapper = ObjectMapper()
             objectMapper.registerModule(module)
+            objectMapper.registerModule(Jdk8Module())
+            objectMapper.registerModule(JavaTimeModule())
             objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
             objectMapper
                     .setDateFormat(SimpleDateFormat("yyyy-MM-dd\'T\'HH:mm:ss.SSS"))

src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt

@@ -17,16 +17,11 @@
 package io.emeraldpay.dshackle
 
 import io.emeraldpay.dshackle.config.MainConfig
+import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandler
 import io.grpc.*
-import io.grpc.netty.GrpcSslContexts
 import io.grpc.netty.NettyServerBuilder
-import io.netty.handler.ssl.ClientAuth
-import io.netty.handler.ssl.SslContext
-import org.apache.commons.lang3.StringUtils
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
-import org.springframework.core.env.Environment
-import org.springframework.core.io.ResourceLoader
 import org.springframework.stereotype.Service
 import java.net.InetSocketAddress
 import javax.annotation.PostConstruct
@@ -36,7 +31,8 @@ import javax.annotation.PreDestroy
 open class GrpcServer(
         @Autowired val rpcs: List<io.grpc.BindableService>,
         @Autowired val mainConfig: MainConfig,
-        @Autowired val tlsSetup: TlsSetup
+        @Autowired val tlsSetup: TlsSetup,
+        @Autowired val accessHandler: AccessHandler
 ) {
 
     private val log = LoggerFactory.getLogger(GrpcServer::class.java)
@@ -50,6 +46,14 @@ open class GrpcServer(
         log.info("Listening Native gRPC on ${mainConfig.host}:${mainConfig.port}")
         val serverBuilder = NettyServerBuilder
                 .forAddress(InetSocketAddress(mainConfig.host, mainConfig.port))
+                .let {
+                    if (mainConfig.accessLogConfig.enabled) {
+                        it.intercept(accessHandler)
+                    } else {
+                        it
+                    }
+                }
+
         tlsSetup.setupServer("Native gRPC", mainConfig.tls, true)?.let {
             serverBuilder.sslContext(it)
         }

src/main/kotlin/io/emeraldpay/dshackle/config/AccessLogConfig.kt

@@ -0,0 +1,22 @@
+package io.emeraldpay.dshackle.config
+
+class AccessLogConfig(
+        val enabled: Boolean = false
+) {
+
+    var filename: String = "./access_log.jsonl"
+
+    companion object {
+
+        fun default(): AccessLogConfig {
+            return disabled()
+        }
+
+        fun disabled(): AccessLogConfig {
+            return AccessLogConfig(
+                    enabled = false
+            )
+        }
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/config/AccessLogReader.kt

@@ -0,0 +1,27 @@
+package io.emeraldpay.dshackle.config
+
+import org.slf4j.LoggerFactory
+import org.yaml.snakeyaml.nodes.MappingNode
+
+class AccessLogReader : YamlConfigReader(), ConfigReader<AccessLogConfig> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(AccessLogReader::class.java)
+    }
+
+    override fun read(input: MappingNode?): AccessLogConfig {
+        return getMapping(input, "accessLog")?.let { node ->
+            val enabled = getValueAsBool(node, "enabled") ?: false
+            if (!enabled) {
+                AccessLogConfig.disabled()
+            } else {
+                val config = AccessLogConfig(true)
+                getValueAsString(node, "filename")?.let {
+                    config.filename = it
+                }
+                config
+            }
+        } ?: AccessLogConfig.default()
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/config/MainConfig.kt

@@ -24,5 +24,5 @@ class MainConfig {
     var upstreams: UpstreamsConfig? = null
     var tokens: TokensConfig? = null
     var monitoring: MonitoringConfig = MonitoringConfig.default()
-
+    var accessLogConfig: AccessLogConfig = AccessLogConfig.default()
 }

src/main/kotlin/io/emeraldpay/dshackle/config/MainConfigReader.kt

@@ -34,6 +34,7 @@ class MainConfigReader(
     private val cacheConfigReader = CacheConfigReader()
     private val tokensConfigReader = TokensConfigReader()
     private val monitoringConfigReader = MonitoringConfigReader()
+    private val accessLogReader = AccessLogReader()
 
     fun read(input: InputStream): MainConfig? {
         val configNode = readNode(input)
@@ -67,6 +68,9 @@ class MainConfigReader(
         monitoringConfigReader.read(input).let {
             config.monitoring = it
         }
+        accessLogReader.read(input).let {
+            config.accessLogConfig = it
+        }
         return config
     }
 

src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt

@@ -0,0 +1,110 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.monitoring.accesslog
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.dshackle.Global
+import io.grpc.*
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.stereotype.Service
+
+@Service
+class AccessHandler(
+        @Autowired private val accessLogWriter: AccessLogWriter
+) : ServerInterceptor {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(AccessHandler::class.java)
+    }
+
+    override fun <ReqT : Any, RespT : Any> interceptCall(
+            call: ServerCall<ReqT, RespT>,
+            headers: Metadata,
+            next: ServerCallHandler<ReqT, RespT>): ServerCall.Listener<ReqT> {
+
+        when (val method = call.methodDescriptor.bareMethodName) {
+            "NativeCall" -> {
+                val builder = Events.NativeCallBuilder()
+                return OnNativeCall<ReqT, RespT>(
+                        next.startCall(OnNativeCallResponse(call, builder), headers),
+                        builder) { logs ->
+                    accessLogWriter.submit(logs)
+                }
+            }
+            else -> {
+                log.trace("unsupported method `{}`", method)
+            }
+        }
+
+        // continue
+        return next.startCall(call, headers)
+    }
+
+
+    class OnNativeCall<ReqT : Any, RespT : Any>(
+            val next: ServerCall.Listener<ReqT>,
+            val builder: Events.NativeCallBuilder,
+            val done: (List<Events.NativeCall>) -> Unit
+    ) : ForwardingServerCallListener<ReqT>() {
+
+        override fun onMessage(message: ReqT) {
+            if (message is BlockchainOuterClass.NativeCallRequest) {
+                val chain = message.chain
+                builder.withChain(chain.number)
+                message.itemsList.forEach { item ->
+                    builder.onItem(item)
+                }
+            }
+            super.onMessage(message)
+        }
+
+        override fun onCancel() {
+            super.onCancel()
+            done(builder.build())
+        }
+
+        override fun onComplete() {
+            super.onComplete()
+            done(builder.build())
+        }
+
+        override fun delegate(): ServerCall.Listener<ReqT> {
+            return next
+        }
+    }
+
+    class OnNativeCallResponse<ReqT : Any, RespT : Any>(
+            val next: ServerCall<ReqT, RespT>,
+            val builder: Events.NativeCallBuilder
+    ) : ForwardingServerCall<ReqT, RespT>() {
+
+        override fun getMethodDescriptor(): MethodDescriptor<ReqT, RespT> {
+            return next.methodDescriptor
+        }
+
+        override fun delegate(): ServerCall<ReqT, RespT> {
+            return next
+        }
+
+        override fun sendMessage(message: RespT) {
+            if (message is BlockchainOuterClass.NativeCallReplyItem) {
+                builder.onItemReply(message)
+            }
+            super.sendMessage(message)
+        }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriter.kt

@@ -0,0 +1,124 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.monitoring.accesslog
+
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.config.MainConfig
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.stereotype.Repository
+import java.io.*
+import java.time.Duration
+import java.time.Instant
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import javax.annotation.PostConstruct
+
+@Repository
+class AccessLogWriter(
+        @Autowired mainConfig: MainConfig
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(AccessLogWriter::class.java)
+        private const val WRITE_BATCH_LIMIT = 5000
+        private const val FLUSH_SLEEP_MS = 500L
+        private const val START_SLEEP_MS = 2000L
+        private val NL = "\n".toByteArray()
+    }
+
+    private val config = mainConfig.accessLogConfig
+    private val filename = File(config.filename)
+    private val scheduler = Executors.newSingleThreadScheduledExecutor()
+
+    private val queue = ConcurrentLinkedQueue<Any>()
+    private val objectMapper = Global.objectMapper
+    private var lastErrorAt: Instant = Instant.ofEpochMilli(0)
+
+    private val runner = Runnable {
+        flushRunner()
+    }
+
+    @PostConstruct
+    fun start() {
+        if (!config.enabled) {
+            log.info("Access Log is diabled")
+            return
+        }
+        log.info("Writing Access Log to ${filename.absolutePath}")
+        scheduler.schedule(runner, START_SLEEP_MS, TimeUnit.MILLISECONDS)
+    }
+
+    private fun flushRunner() {
+        try {
+            flush()
+        } catch (t: Throwable) {
+            logError {
+                log.error("Failed to write logs. ${t.javaClass}:${t.message}")
+            }
+        } finally {
+            scheduler.schedule(runner, FLUSH_SLEEP_MS, TimeUnit.MILLISECONDS)
+        }
+    }
+
+    fun submit(event: Any) {
+        queue.add(event)
+    }
+
+    fun submit(events: List<Any>) {
+        queue.addAll(events)
+    }
+
+    fun logError(m: () -> Unit) {
+        val now = Instant.now()
+        if (lastErrorAt.isBefore(now - Duration.ofMinutes(1))) {
+            lastErrorAt = now
+            m()
+        }
+    }
+
+    protected fun flush() {
+        if (!filename.exists()) {
+            if (!filename.createNewFile()) {
+                logError {
+                    log.error("Cannot create Access Log file at ${filename.absolutePath}")
+                }
+                return
+            }
+        }
+        BufferedOutputStream(FileOutputStream(filename, true)).use { wrt ->
+            var limit = WRITE_BATCH_LIMIT
+            while (limit > 0) {
+                limit--
+                val next = queue.poll() ?: return
+                val bytes: ByteArray? = try {
+                    objectMapper.writeValueAsBytes(next)
+                } catch (t: Throwable) {
+                    logError {
+                        log.warn("Failed to write an access log line. ${t.message}")
+                    }
+                    null
+                }
+                if (bytes != null && bytes.isNotEmpty()) {
+                    wrt.write(bytes)
+                    wrt.write(NL, 0, 1)
+                }
+            }
+        }
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt

@@ -0,0 +1,142 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.monitoring.accesslog
+
+import com.fasterxml.jackson.annotation.JsonInclude
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.grpc.Chain
+import org.slf4j.LoggerFactory
+import java.time.Instant
+import java.util.*
+import kotlin.collections.ArrayList
+
+class Events {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(Events::class.java)
+    }
+
+    abstract class Base(
+            val method: String,
+            val id: UUID
+    ) {
+        val ts = Instant.now()
+    }
+
+    abstract class ChainBase(
+            val blockchain: Chain, method: String, id: UUID
+    ) : Base(method, id) {
+
+    }
+
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    class NativeCall(
+            // info about the initial request, that may include several native calls
+            val request: StreamRequestDetails,
+            // total native calls passes within the initial request
+            val total: Int,
+            // index of the call specific for the current response
+            val index: Int,
+            val selector: String? = null,
+            val quorum: Long? = null,
+            val minAvailability: String? = null,
+
+            val succeed: Boolean,
+            val rpcError: Int? = null,
+            val payloadSizeBytes: Long,
+
+            blockchain: Chain, method: String, id: UUID
+    ) : ChainBase(blockchain, method, id) {
+
+    }
+
+    data class StreamRequestDetails(
+            val id: UUID,
+            val start: Instant
+    )
+
+    data class Remote(
+            val ips: List<String>,
+            val userAgent: String
+    )
+
+    data class NativeCallItemDetails(
+            val method: String,
+            val id: Int,
+            val payloadSizeBytes: Long
+    )
+
+    data class NativeCallReplyDetails(
+            val id: Int,
+            val succeed: Boolean,
+            val replySizeBytes: Long,
+            val ts: Instant = Instant.now()
+    )
+
+    class NativeCallBuilder() {
+
+        private val requestDetails = StreamRequestDetails(
+                UUID.randomUUID(),
+                Instant.now()
+        )
+
+        var chain: Int = Chain.UNSPECIFIED.id
+        val items = ArrayList<NativeCallItemDetails>()
+        val replies = HashMap<Int, NativeCallReplyDetails>()
+
+        fun withChain(chain: Int): NativeCallBuilder {
+            this.chain = chain
+            return this
+        }
+
+        fun onItem(item: BlockchainOuterClass.NativeCallItem): NativeCallBuilder {
+            this.items.add(
+                    NativeCallItemDetails(
+                            item.method,
+                            item.id,
+                            item.payload.size().toLong()
+                    )
+            )
+            return this
+        }
+
+        fun onItemReply(reply: BlockchainOuterClass.NativeCallReplyItem): NativeCallBuilder {
+            this.replies[reply.id] = NativeCallReplyDetails(
+                    reply.id,
+                    reply.succeed,
+                    reply.payload?.size()?.toLong() ?: 0L
+            )
+            return this
+        }
+
+        fun build(): List<NativeCall> {
+            val blockchain = Chain.byId(this.chain)
+            return items.mapIndexed { index, item ->
+                val reply = replies[item.id]
+                NativeCall(
+                        request = requestDetails,
+                        total = items.size,
+                        index = index,
+                        succeed = reply?.succeed ?: false,
+                        blockchain = blockchain,
+                        method = item.method,
+                        payloadSizeBytes = item.payloadSizeBytes,
+                        id = UUID.randomUUID()
+                )
+            }
+        }
+    }
+}