Merge pull request #93 from emeraldpay/feat/accesslog

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'

docs/01-architecture-intro.adoc

@@ -29,7 +29,7 @@ And for a request:
 - Is it for concrete data (_block #100_) or the latest (_get balance_)?
 - Is result a static value (_just block_) or may vary depending on network and node (_latest nonce_)?
 - Does it need to be repeated over multiple nodes (_broadcast transaction_)?
-- Request can also specify which subset of nodes should be able to execute the request by selecting node _Labels_ (see "link:08-quorum-and-selectors.adoc[Quorum and Selectors]")
+- Request can also specify which subset of nodes should be able to execute the request by selecting node _Labels_ (see "link:09-quorum-and-selectors.adoc[Quorum and Selectors]")
 
 Based on these factors, Dshackle executes the request on a most optimal node.
 For most of the simple requests, it just gets a node that is synchronized to the point that must have a response for that particular request.
@@ -43,7 +43,7 @@ It provides:
 - Routes only realy/alive upstreams, i.e., synchronized and with enough peers
 - Load Balancing
 - Request Retry on upstream errors
-- Local Caching (memory and Redis, see link:09-caching.adoc[Caching])
+- Local Caching (memory and Redis, see link:10-caching.adoc[Caching])
 - Broadcasting and Quorum for requests
 
 === gRPC protocol

docs/02-quick-start.adoc

@@ -174,4 +174,4 @@ grpcurl -import-path ./proto/ -proto blockchain.proto -d '{"asset": {"chain": "1
 ...
 ----
 
-See other enhanced methods in the link:06-methods.adoc[Documentation for Enhanced Methods]
+See other enhanced methods in the link:07-methods.adoc[Documentation for Enhanced Methods]

docs/03-server-config.adoc

@@ -48,7 +48,8 @@ a| `port: 12449`
 
 a| `tls`
 |
-| TLS configuration for gRPC. See link:07-authentication.adoc[Authentication] for details
+| TLS configuration for gRPC.
+See link:08-authentication.adoc[Authentication] for details
 
 a| `proxy`
 |
@@ -121,7 +122,8 @@ a| `enabled: true`
 
 a| `tls`
 |
-| TLS configuration for proxy. See link:07-authentication.adoc[Authentication] for details
+| TLS configuration for proxy.
+See link:08-authentication.adoc[Authentication] for details
 
 a| `routes`
 |

docs/06-monitoring.adoc

@@ -0,0 +1,74 @@
+= Logging & Monitoring
+
+== Access / Request Log
+
+Dshackle can log all requests to a file in JSON format.
+Or https://jsonlines.org/[JSON Lines] to be more precise, i.e., a test file where each line is a JSON.
+
+NOTE: By default, the access log is disabled.
+
+To enable access log add following configuration:
+
+[source,yaml]
+----
+accessLog:
+  enabled: true
+  filename: /var/log/dshackle/access_log.jsonl
+----
+
+`filename` is optional, and the default value is `access_log.jsonl` (i.e., in the current directory).
+
+Since a single request may contain multiple replies (ex., a batch call, or subscribe to the head blocks) the Dshackle logging is based on replies.
+The access log file contains details per each response send from the server, and each of them refers to original request details.
+
+The access log contains the JSON lines similar to:
+
+[source,json]
+----
+{
+  "version":"accesslog/v1beta",
+  "ts":"2021-07-20T01:53:33.174645Z",
+  "id":"578d83db-cf53-4ef8-b73e-3f1cc0a67e96",
+  "method":"NativeCall",
+  "blockchain":"ETHEREUM",
+  "total":2,
+  "index":0,
+  "succeed":true,
+  "request":{
+    "id":"513b9b49-b472-4c83-b4b7-58dd2aabe9f6",
+    "start":"2021-07-20T01:53:33.086946Z",
+    "remote":{
+      "ips":["127.0.0.1", "10.0.5.102", "172.217.8.78"],
+      "ip":"172.217.8.78",
+      "userAgent":"grpc-node-js/1.1.8"
+    }
+  },
+  "nativeCall":{
+     "method":"eth_blockNumber",
+     "id":2,
+     "payloadSizeBytes":2
+  }
+}
+----
+
+.Where:
+- `ts` timestamp of the reply
+- `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
+- `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
+- `request` original request details
+** `id` uniq id of the request; all replied to the same request have same id
+** `start` when request was received
+** `remote` remote details
+*** `ips` list of all recognized IPs (including headers such as `X-Real-IP` and `X-Forwarded-For`)
+*** `ip` a single ip, that likely represent a real IP of the remote
+*** `userAgent` user agent
+- `nativeCall` details of the individual Native Call request
+** `method` method name terms of Blockchain API
+** `id` request id provided in the original request
+** `payloadSizeBytes` size of the original _individual_ request (for JSON RPC it's size of the `params` value)
+
+

docs/07-methods.adoc

@@ -170,4 +170,4 @@ message TxStatus {
 
 === gRPC Client Libraries
 
-See link:10-client-libraries.adoc[Client Libraries] documentation.
+See link:11-client-libraries.adoc[Client Libraries] documentation.

docs/README.adoc

@@ -3,7 +3,7 @@
 == What is Dshackle
 
 Dshackle is a L7 Load Balancer for Blockchain APIs with automatic discovery, health checking, secure access, TLS with
-client authentication, and many other features. It can be configured as an edge proxy, middle proxy or API gateway.
+client authentication, and many other features.It can be configured as an edge proxy, middle proxy or API gateway.
 
 Dshackle provided a high level aggregated API on top of several underlying upstreams (blockchain nodes or providers,
 such as Geth, Parity, Infura, etc), automatically verifies their availability and the current status of the network,
@@ -38,11 +38,12 @@ Main goals:
 . link:03-server-config.adoc[Server Configuration]
 . link:04-upstream-config.adoc[Upstreams Configuration]
 . link:05-start.adoc[How to launch a server]
-. link:06-methods.adoc[API methods]
+. link:06-monitoring.adoc[Logging & Monitoring]
-. link:07-authentication.adoc[Authentication]
+. link:07-methods.adoc[API methods]
-. link:08-quorum-and-selectors.adoc[Quorum and Selectors]
+. link:08-authentication.adoc[Authentication]
-. link:09-caching.adoc[Caching]
+. link:09-quorum-and-selectors.adoc[Quorum and Selectors]
-. link:10-client-libraries.adoc[Client Libraries]
+. link:10-caching.adoc[Caching]
+. link:11-client-libraries.adoc[Client Libraries]
 
 == Reference
 

docs/reference-configuration.adoc

@@ -58,6 +58,10 @@ tokens:
     type: ERC-20
     address: 0xdac17f958d2ee523a2206206994597c13d831ec7
 
+accessLog:
+  enabled: true
+  filename: /var/log/dshackle/access_log.jsonl
+
 cluster:
   defaults:
     - chains:
@@ -142,24 +146,33 @@ cluster:
 
 | `tls`
 |
-| Setup TLS configuration for the gRPC server. See <<tls>> section
+| Setup TLS configuration for the gRPC server.
+See <<tls>> section
 
 | `proxy`
 |
-| Setup HTTP proxy that emulates all standard JSON RPC requests. See <<proxy>> section
+| Setup HTTP proxy that emulates all standard JSON RPC requests.
+See <<proxy>> section
+
+| `accessLog`
+|
+| Configure access logging.
+See <<accessLog>> section
+
 
 | `tokens`
 |
-| Configure tokens for tracking balance. See <<tokens>> section
+| Configure tokens for tracking balance.
-
+See <<tokens>> section
 
 | `cache`
 |
-| Caching configuration. See <<cache>> section.
+| Caching configuration.
+See <<cache>> section.
 
 | `cluster`
 |
-| Setup connection to remote nodes. See <<cluster>> section
+| Setup connection to remote nodes.See <<cluster>> section
 
 |===
 
@@ -247,7 +260,7 @@ proxy:
 
 | `routes`
 |
-a| Routing paths for Proxy. The proxy will handle requests as `https://${HOST}:${PORT}/${ROUTE_ID}` (or `http://` if TLS is not enabled)
+a| Routing paths for Proxy.The proxy will handle requests as `https://${HOST}:${PORT}/${ROUTE_ID}` (or `http://` if TLS is not enabled)
 |===
 
 .Route config
@@ -265,6 +278,31 @@ a| Routing paths for Proxy. The proxy will handle requests as `https://${HOST}:$
 
 |===
 
+[#accessLog]
+== Access Log config
+
+[source,yaml]
+----
+accessLog:
+  enabled: true
+  filename: /var/log/dshackle/access_log.jsonl
+----
+
+.Access Log config
+[cols="2a,3a,7"]
+|===
+| Option | Default | Description
+
+| `enabled`
+| `false`
+| Enable/Disable Access logging
+
+| `filename`
+| `access_log.jsonl`
+| Path to the access log file
+
+|===
+
 [#tokens]
 == Tokens config
 
@@ -284,7 +322,7 @@ tokens:
 ----
 
 Tokens config enables tracking of a balance amount in the configured tokens.
-After making the configuration above you can request balance (`GetBalance`), or subscribe to balance changes (`SubscribeBalance`), using link:06-methods.adoc[enhanced protocol]
+After making the configuration above you can request balance (`GetBalance`), or subscribe to balance changes (`SubscribeBalance`), using link:07-methods.adoc[enhanced protocol]
 
 .Token config
 [cols="2a,7"]
@@ -301,7 +339,7 @@ After making the configuration above you can request balance (`GetBalance`), or
 | Name of the token, used for balance response as asset code (as converted to UPPERCASE)
 
 | `type`
-| Type of token. Only `ERC-20` is supported at this moment
+| Type of token.Only `ERC-20` is supported at this moment
 
 | `address`
 | Address of the deployed contract
@@ -446,7 +484,7 @@ Accepted types: `bitcoin`, `bitcoin-testnet`, `ethereum`, `ethereum-classic`, `k
 | no
 | Key-Value pairs that are assigned to the upstream.
 Used to select an upstream per-request.
-See link:08-quorum-and-selectors.adoc[Quorum and Selectors]
+See link:09-quorum-and-selectors.adoc[Quorum and Selectors]
 
 | `methods`
 | no
@@ -538,9 +576,10 @@ It's more effective, easier to secure connection, and allows to build a distribu
 | Address to connect to
 
 | `tls`
-a| TLC configuration for the connection. In general it's an optional configuration, but it's strongly recommended. Also
+a| TLC configuration for the connection.
-HTTP2 + gRPC is designed to be used with TLS, and some of the related software is unable to use it without TLS. +
+In general it's an optional configuration, but it's strongly recommended.
-  See link:07-authentication.adoc[Authentication] docs and <<tls>>.
+Also HTTP2 + gRPC is designed to be used with TLS, and some of the related software is unable to use it without TLS. +
+See link:08-authentication.adoc[Authentication] docs and <<tls>>.
 
 | `tls.ca`
 | Path to x509 certificate to verify remote server

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,174 @@
+/**
+ * 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.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(
+        @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> {
+
+        return when (val method = call.methodDescriptor.bareMethodName) {
+            "SubscribeHead" -> processSubscribeHead(call, headers, next)
+            "SubscribeBalance" -> processSubscribeBalance(call, headers, next, true)
+            "SubscribeTxStatus" -> processSubscribeTxStatus(call, headers, next)
+            "GetBalance" -> processSubscribeBalance(call, headers, next, false)
+            "NativeCall" -> processNativeCall(call, headers, next)
+            "Describe" -> processDescribe(call, headers, next)
+            "SubscribeStatus" -> processStatus(call, headers, next)
+            else -> {
+                log.warn("unsupported method `{}`", method)
+                next.startCall(call, headers)
+            }
+        }
+    }
+
+    private fun <ReqT : Any, RespT : Any, E> process(
+            call: ServerCall<ReqT, RespT>,
+            headers: Metadata,
+            next: ServerCallHandler<ReqT, RespT>,
+            builder: EventsBuilder.RequestReply<E, ReqT, RespT>
+    ): ServerCall.Listener<ReqT> {
+        builder.start(headers, call.attributes)
+        val callWrapper: ServerCall<ReqT, RespT> = StdCallResponse(
+                call, builder, accessLogWriter
+        )
+        return StdCallListener(
+                next.startCall(callWrapper, headers),
+                builder
+        )
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    private fun <ReqT : Any, RespT : Any> processSubscribeHead(
+            call: ServerCall<ReqT, RespT>,
+            headers: Metadata,
+            next: ServerCallHandler<ReqT, RespT>
+    ): ServerCall.Listener<ReqT> {
+        return process(call, headers, next,
+                EventsBuilder.SubscribeHead() as EventsBuilder.RequestReply<*, ReqT, RespT>
+        )
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    private fun <ReqT : Any, RespT : Any> processSubscribeBalance(
+            call: ServerCall<ReqT, RespT>,
+            headers: Metadata,
+            next: ServerCallHandler<ReqT, RespT>,
+            subscribe: Boolean
+    ): ServerCall.Listener<ReqT> {
+        return process(call, headers, next,
+                EventsBuilder.SubscribeBalance(subscribe) as EventsBuilder.RequestReply<*, ReqT, RespT>
+        )
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    private fun <ReqT : Any, RespT : Any> processSubscribeTxStatus(
+            call: ServerCall<ReqT, RespT>,
+            headers: Metadata,
+            next: ServerCallHandler<ReqT, RespT>
+    ): ServerCall.Listener<ReqT> {
+        return process(call, headers, next,
+                EventsBuilder.TxStatus() as EventsBuilder.RequestReply<*, ReqT, RespT>
+        )
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    private fun <ReqT : Any, RespT : Any> processNativeCall(
+            call: ServerCall<ReqT, RespT>,
+            headers: Metadata,
+            next: ServerCallHandler<ReqT, RespT>
+    ): ServerCall.Listener<ReqT> {
+        return process(call, headers, next,
+                EventsBuilder.NativeCall() as EventsBuilder.RequestReply<*, ReqT, RespT>
+        )
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    private fun <ReqT : Any, RespT : Any> processDescribe(
+            call: ServerCall<ReqT, RespT>,
+            headers: Metadata,
+            next: ServerCallHandler<ReqT, RespT>
+    ): ServerCall.Listener<ReqT> {
+        return process(call, headers, next,
+                EventsBuilder.Describe() as EventsBuilder.RequestReply<*, ReqT, RespT>
+        )
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    private fun <ReqT : Any, RespT : Any> processStatus(
+            call: ServerCall<ReqT, RespT>,
+            headers: Metadata,
+            next: ServerCallHandler<ReqT, RespT>
+    ): ServerCall.Listener<ReqT> {
+        return process(call, headers, next,
+                EventsBuilder.Status() as EventsBuilder.RequestReply<*, ReqT, RespT>
+        )
+    }
+
+    open class StdCallListener<Req, EB : EventsBuilder.RequestReply<*, Req, *>>(
+            val next: ServerCall.Listener<Req>,
+            val builder: EB
+    ) : ForwardingServerCallListener<Req>() {
+
+        override fun onMessage(message: Req) {
+            builder.onRequest(message)
+            super.onMessage(message)
+        }
+
+        override fun delegate(): ServerCall.Listener<Req> {
+            return next
+        }
+    }
+
+    open class StdCallResponse<ReqT : Any, RespT : Any, EB : EventsBuilder.RequestReply<*, ReqT, RespT>>(
+            val next: ServerCall<ReqT, RespT>,
+            val builder: EB,
+            val accessLogWriter: AccessLogWriter
+    ) : 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) {
+            super.sendMessage(message)
+            accessLogWriter.submit(
+                    builder.onReply(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 disabled")
+            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,146 @@
+/**
+ * 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.grpc.Chain
+import org.slf4j.LoggerFactory
+import java.time.Instant
+import java.util.*
+
+class Events {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(Events::class.java)
+    }
+
+    abstract class Base(
+            val id: UUID,
+            val method: String
+    ) {
+        val version = "accesslog/v1beta"
+        val ts = Instant.now()
+    }
+
+    abstract class ChainBase(
+            val blockchain: Chain, method: String, id: UUID
+    ) : Base(id, method)
+
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    class SubscribeHead(
+            blockchain: Chain, id: UUID,
+            // initial request details
+            val request: StreamRequestDetails,
+            // index of the current response
+            val index: Int
+    ) : ChainBase(blockchain, "SubscribeHead", id)
+
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    class SubscribeBalance(
+            blockchain: Chain, id: UUID, subscribe: Boolean,
+            // initial request details
+            val request: StreamRequestDetails,
+            val balanceRequest: BalanceRequest,
+            val addressBalance: AddressBalance,
+            // index of the current response
+            val index: Int
+    ) : ChainBase(blockchain, if (subscribe) "SubscribeBalance" else "GetBalance", id)
+
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    class TxStatus(
+            blockchain: Chain, id: UUID,
+            val request: StreamRequestDetails,
+            val txStatusRequest: TxStatusRequest,
+            val txStatus: TxStatusResponse,
+            // index of the current response
+            val index: Int
+    ) : ChainBase(blockchain, "SubscribeTxStatus", id)
+
+    data class TxStatusRequest(
+            val txId: String
+    )
+
+    data class TxStatusResponse(
+            val confirmations: Int
+    )
+
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    class NativeCall(
+            blockchain: Chain, id: UUID,
+
+            // 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,
+            val nativeCall: NativeCallItemDetails
+    ) : ChainBase(blockchain, "NativeCall", id)
+
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    class Describe(
+            id: UUID,
+            val request: StreamRequestDetails
+    ) : Base(id, "Describe")
+
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    class Status(
+            blockchain: Chain, id: UUID,
+            val request: StreamRequestDetails
+    ) : ChainBase(blockchain, "Status", id)
+
+    data class StreamRequestDetails(
+            val id: UUID,
+            val start: Instant,
+            val remote: Remote
+    )
+
+    data class Remote(
+            val ips: List<String>,
+            val ip: 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()
+    )
+
+    data class BalanceRequest(
+            val asset: String,
+            val addressType: String
+    )
+
+    data class AddressBalance(
+            val asset: String,
+            val address: String
+    )
+}

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

@@ -0,0 +1,285 @@
+/**
+ * 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.api.proto.Common
+import io.emeraldpay.grpc.Chain
+import io.grpc.Attributes
+import io.grpc.Grpc
+import io.grpc.Metadata
+import org.apache.commons.lang3.StringUtils
+import org.slf4j.LoggerFactory
+import java.net.InetAddress
+import java.net.InetSocketAddress
+import java.time.Instant
+import java.util.*
+
+class EventsBuilder {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(EventsBuilder::class.java)
+    }
+
+    interface StartingRequest {
+        fun start(metadata: Metadata, attributes: Attributes)
+    }
+
+    interface RequestReply<E, Req, Resp> : StartingRequest {
+        fun onRequest(msg: Req)
+        fun onReply(msg: Resp): E
+    }
+
+    abstract class Base<T>() : StartingRequest {
+        companion object {
+            private val remoteIpKeys = listOf(
+                    Metadata.Key.of("x-real-ip", Metadata.ASCII_STRING_MARSHALLER),
+                    Metadata.Key.of("x-forwarded-for", Metadata.ASCII_STRING_MARSHALLER)
+            )
+            private val invalidCharacters = Regex("[\n\t]+")
+        }
+
+        var requestDetails = Events.StreamRequestDetails(
+                UUID.randomUUID(),
+                Instant.now(),
+                Events.Remote(emptyList(), "", "")
+        )
+
+        var chainId: Int = Chain.UNSPECIFIED.id
+        var chain = Chain.UNSPECIFIED
+
+        private fun toInetAddress(ip: String): InetAddress? {
+            val isIp = Character.digit(ip[0], 16) != -1
+            if (!isIp) {
+                return null
+            }
+            return try {
+                InetAddress.getByName(ip)
+            } catch (t: Throwable) {
+                null
+            }
+        }
+
+        private fun findBestIp(ips: List<InetAddress>): InetAddress? {
+            // check if a real remote address is provided, otherwise use any local address
+            return ips.sortedWith(kotlin.Comparator { a, b ->
+                val aLocal = a.isLoopbackAddress || a.isSiteLocalAddress
+                val bLocal = b.isLoopbackAddress || b.isSiteLocalAddress
+                when {
+                    aLocal && bLocal -> 0
+                    aLocal -> 1
+                    else -> -1
+                }
+            }).firstOrNull()
+        }
+
+        private fun clean(s: String): String {
+            return StringUtils.truncate(s, 128)
+                    .replace(invalidCharacters, " ")
+                    .trim()
+        }
+
+        protected abstract fun getT(): T
+
+        override fun start(metadata: Metadata, attributes: Attributes) {
+            val userAgent = metadata.get(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER))
+                    ?.let(this@Base::clean)
+                    ?: ""
+            val ips = ArrayList<InetAddress>()
+            remoteIpKeys.forEach { key ->
+                metadata.get(key)?.let {
+                    it.trim().ifEmpty { null }
+                            ?.let(this@Base::toInetAddress)
+                            ?.let(ips::add)
+                }
+            }
+            attributes.get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR)?.let { addr ->
+                if (addr is InetSocketAddress) {
+                    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)
+            return getT()
+        }
+    }
+
+    class SubscribeHead() :
+            Base<SubscribeHead>(),
+            RequestReply<Events.SubscribeHead, Common.Chain, BlockchainOuterClass.ChainHead> {
+
+        private var index = 0
+
+        override fun getT(): SubscribeHead {
+            return this
+        }
+
+        override fun onRequest(msg: Common.Chain) {
+            withChain(msg.type.number)
+        }
+
+        override fun onReply(msg: BlockchainOuterClass.ChainHead): Events.SubscribeHead {
+            return Events.SubscribeHead(
+                    chain, UUID.randomUUID(), requestDetails, index++
+            )
+        }
+    }
+
+    class SubscribeBalance(val subscribe: Boolean) :
+            Base<SubscribeBalance>(),
+            RequestReply<Events.SubscribeBalance, BlockchainOuterClass.BalanceRequest, BlockchainOuterClass.AddressBalance> {
+
+        private var index = 0
+        private var balanceRequest: Events.BalanceRequest? = null
+
+        override fun getT(): SubscribeBalance {
+            return this
+        }
+
+        override fun onRequest(msg: BlockchainOuterClass.BalanceRequest) {
+            balanceRequest = Events.BalanceRequest(
+                    msg.asset.code.toUpperCase(),
+                    msg.address.addrTypeCase.name
+            )
+        }
+
+        override fun onReply(msg: BlockchainOuterClass.AddressBalance): Events.SubscribeBalance {
+            if (balanceRequest == null) {
+                throw IllegalStateException("Request is not initialized")
+            }
+            val addressBalance = Events.AddressBalance(msg.asset.code, msg.address.address)
+            val chain = Chain.byId(msg.asset.chain.number)
+            return Events.SubscribeBalance(
+                    chain, UUID.randomUUID(), subscribe, requestDetails, balanceRequest!!, addressBalance, index++
+            )
+        }
+    }
+
+    class TxStatus() :
+            Base<TxStatus>(),
+            RequestReply<Events.TxStatus, BlockchainOuterClass.TxStatusRequest, BlockchainOuterClass.TxStatus> {
+        private var index = 0
+        private var txStatusRequest: Events.TxStatusRequest? = null
+
+        override fun onRequest(msg: BlockchainOuterClass.TxStatusRequest) {
+            this.txStatusRequest = Events.TxStatusRequest(msg.txId)
+            withChain(msg.chainValue)
+        }
+
+        override fun onReply(msg: BlockchainOuterClass.TxStatus): Events.TxStatus {
+            return Events.TxStatus(
+                    chain, UUID.randomUUID(), requestDetails, txStatusRequest!!,
+                    Events.TxStatusResponse(msg.confirmations),
+                    index++
+            )
+        }
+
+        override fun getT(): TxStatus {
+            return this
+        }
+
+    }
+
+    class NativeCall :
+            Base<NativeCall>(),
+            RequestReply<Events.NativeCall, BlockchainOuterClass.NativeCallRequest, BlockchainOuterClass.NativeCallReplyItem> {
+        val items = ArrayList<Events.NativeCallItemDetails>()
+        val replies = HashMap<Int, Events.NativeCallReplyDetails>()
+        private var index = 0
+
+        override fun getT(): NativeCall {
+            return this
+        }
+
+        override fun onRequest(msg: BlockchainOuterClass.NativeCallRequest) {
+            withChain(msg.chain.number)
+            msg.itemsList.forEach { item ->
+                this.items.add(
+                        Events.NativeCallItemDetails(
+                                item.method,
+                                item.id,
+                                item.payload.size().toLong()
+                        )
+                )
+            }
+        }
+
+        override fun onReply(msg: BlockchainOuterClass.NativeCallReplyItem): Events.NativeCall {
+            val item = items.find { it.id == msg.id }!!
+            return Events.NativeCall(
+                    request = requestDetails,
+                    total = items.size,
+                    index = index++,
+                    succeed = msg.succeed,
+                    blockchain = chain,
+                    nativeCall = item,
+                    payloadSizeBytes = item.payloadSizeBytes,
+                    id = UUID.randomUUID()
+            )
+        }
+
+    }
+
+    class Describe :
+            Base<Describe>(),
+            RequestReply<Events.Describe, BlockchainOuterClass.DescribeRequest, BlockchainOuterClass.DescribeResponse> {
+
+        override fun getT(): Describe {
+            return this
+        }
+
+        override fun onRequest(msg: BlockchainOuterClass.DescribeRequest) {
+        }
+
+        override fun onReply(msg: BlockchainOuterClass.DescribeResponse): Events.Describe {
+            return Events.Describe(
+                    id = UUID.randomUUID(),
+                    request = requestDetails
+            )
+        }
+    }
+
+    class Status :
+            Base<Status>(),
+            RequestReply<Events.Status, BlockchainOuterClass.StatusRequest, BlockchainOuterClass.ChainStatus> {
+        override fun getT(): Status {
+            return this
+        }
+
+        override fun onRequest(msg: BlockchainOuterClass.StatusRequest) {
+        }
+
+        override fun onReply(msg: BlockchainOuterClass.ChainStatus): Events.Status {
+            val chain = Chain.byId(msg.chainValue)
+            return Events.Status(
+                    blockchain = chain,
+                    request = requestDetails,
+                    id = UUID.randomUUID()
+            )
+        }
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import reactor.core.publisher.Flux
+import reactor.core.publisher.Sinks
 import reactor.extra.processor.TopicProcessor
 import java.util.concurrent.atomic.AtomicReference
 
@@ -41,7 +42,9 @@ abstract class DefaultUpstream(
             this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, node)
 
     private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail)))
-    private val statusStream: TopicProcessor<UpstreamAvailability> = TopicProcessor.create()
+    private val statusStream = Sinks.many()
+            .multicast()
+            .directBestEffort<UpstreamAvailability>()
 
     override fun isAvailable(): Boolean {
         return getStatus() == UpstreamAvailability.OK
@@ -63,6 +66,7 @@ abstract class DefaultUpstream(
         status.updateAndGet { curr ->
             Status(curr.lag, avail, statusByLag(curr.lag, avail))
         }
+        statusStream.tryEmitNext(status.get().status)
     }
 
     fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability {
@@ -76,7 +80,8 @@ abstract class DefaultUpstream(
     }
 
     override fun observeStatus(): Flux<UpstreamAvailability> {
-        return Flux.from(statusStream)
+        return statusStream.asFlux()
+                .distinctUntilChanged()
     }
 
     override fun setLag(lag: Long) {
@@ -86,6 +91,7 @@ abstract class DefaultUpstream(
             status.updateAndGet { curr ->
                 Status(lag, curr.avail, statusByLag(lag, curr.avail))
             }
+            statusStream.tryEmitNext(status.get().status)
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt

@@ -168,7 +168,12 @@ abstract class Multistream(
     }
 
     override fun observeStatus(): Flux<UpstreamAvailability> {
-        val upstreamsFluxes = getAll().map { up -> up.observeStatus().map { UpstreamStatus(up, it) } }
+        val upstreamsFluxes = getAll().map { up ->
+            Flux.concat(
+                    Mono.just(up.getStatus()),
+                    up.observeStatus()
+            ).map { UpstreamStatus(up, it) }
+        }
         return Flux.merge(upstreamsFluxes)
                 .filter(FilterBestAvailability())
                 .map { it.status }

src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriterSpec.groovy

@@ -0,0 +1,53 @@
+package io.emeraldpay.dshackle.monitoring.accesslog
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.config.AccessLogConfig
+import io.emeraldpay.dshackle.config.MainConfig
+import io.emeraldpay.grpc.Chain
+import spock.lang.Specification
+
+import java.time.Instant
+
+class AccessLogWriterSpec extends Specification {
+
+    def "writes log event"() {
+        setup:
+        File dir = File.createTempDir("dshackle-test-")
+        File accessLog = new File(dir, "accesslog.jsonl")
+        println("Write access log to $accessLog.absolutePath")
+        MainConfig config = new MainConfig()
+        config.accessLogConfig = new AccessLogConfig(true).tap {
+            it.filename = accessLog.absolutePath
+        }
+        AccessLogWriter logWriter = new AccessLogWriter(config)
+
+        when:
+        def event = new Events.Status(
+                Chain.ETHEREUM, UUID.fromString("9d8ecbf3-12fb-49cf-af9d-949a1050a000"),
+                new Events.StreamRequestDetails(
+                        UUID.fromString("9d8ecbf3-12fb-49cf-af9d-949a1050a000"),
+                        Instant.ofEpochMilli(1626746880123),
+                        new Events.Remote(
+                                ["127.0.0.1", "172.217.8.78"], "172.217.8.78", "UnitTest"
+                        )
+                )
+        )
+        logWriter.submit([event])
+        logWriter.flush()
+        def act = accessLog.readLines()
+        then:
+        act.size() == 1
+        with(act[0]) {
+            def json = Global.objectMapper.readValue(it, Map)
+            json["version"] == "accesslog/v1beta"
+            json["id"] == "9d8ecbf3-12fb-49cf-af9d-949a1050a000"
+            json["method"] == "Status"
+            json["blockchain"] == "ETHEREUM"
+            json["request"]["start"] == "2021-07-20T02:08:00.123Z"
+            json["request"]["id"] == "9d8ecbf3-12fb-49cf-af9d-949a1050a000"
+            json["request"]["remote"]["ip"] == "172.217.8.78"
+            json["request"]["remote"]["userAgent"] == "UnitTest"
+        }
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy

@@ -0,0 +1,235 @@
+/**
+ * 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.grpc.Chain
+import io.grpc.Attributes
+import io.grpc.Grpc
+import io.grpc.Metadata
+import org.jetbrains.annotations.NotNull
+import org.junit.validator.TestClassValidator
+import spock.lang.Specification
+
+class EventsBaseBuilderSpec extends Specification {
+
+    class TestEvent extends Events.Base {
+        Events.StreamRequestDetails request
+
+        TestEvent(Events.StreamRequestDetails request) {
+            super(UUID.randomUUID(), "TEST")
+            this.request = request
+        }
+    }
+
+    class TestEventBuilder extends EventsBuilder.Base<TestEventBuilder>
+            implements EventsBuilder.RequestReply<TestEvent, BlockchainOuterClass.NativeCallRequest, BlockchainOuterClass.NativeCallReplyItem> {
+
+        @Override
+        protected TestEventBuilder getT() {
+            return this
+        }
+
+        @Override
+        void onRequest(BlockchainOuterClass.NativeCallRequest msg) {
+
+        }
+
+        @Override
+        TestEvent onReply(BlockchainOuterClass.NativeCallReplyItem msg) {
+            return new TestEvent(requestDetails)
+        }
+    }
+
+    def "Parse headers from direct local access"() {
+        setup:
+        def metadata = new Metadata()
+        metadata.put(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER), "grpc-go/1.30.0")
+        def attributes = Attributes.newBuilder()
+                .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("127.0.0.1"), 2448))
+                .build()
+        when:
+        def act = new TestEventBuilder()
+                .tap {
+                    it.start(metadata, attributes)
+                    it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance())
+                }
+                .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance())
+        then:
+        act.request != null
+        act.request.remote != null
+        with(act.request.remote) {
+            ips == ["127.0.0.1"]
+            userAgent == "grpc-go/1.30.0"
+            ip == "127.0.0.1"
+        }
+    }
+
+    def "Extracts real remote ip"() {
+        setup:
+        def metadata = new Metadata()
+        metadata.put(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER), "grpc-go/1.30.0")
+        metadata.put(Metadata.Key.of("x-real-ip", Metadata.ASCII_STRING_MARSHALLER), "30.56.100.15")
+        def attributes = Attributes.newBuilder()
+                .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("127.0.0.1"), 2448))
+                .build()
+        when:
+        def act = new TestEventBuilder()
+                .tap {
+                    it.start(metadata, attributes)
+                    it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance())
+                }
+                .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance())
+        then:
+        with(act.request.remote) {
+            ips == ["30.56.100.15", "127.0.0.1"]
+            ip == "30.56.100.15"
+        }
+    }
+
+    def "Ignores remote ip header if already connected from remote"() {
+        setup:
+        def metadata = new Metadata()
+        metadata.put(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER), "grpc-go/1.30.0")
+        metadata.put(Metadata.Key.of("x-real-ip", Metadata.ASCII_STRING_MARSHALLER), "192.168.1.1")
+        def attributes = Attributes.newBuilder()
+                .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448))
+                .build()
+        when:
+        def act = new TestEventBuilder()
+                .tap {
+                    it.start(metadata, attributes)
+                    it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance())
+                }
+                .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance())
+        then:
+        with(act.request.remote) {
+            ips == ["192.168.1.1", "30.56.100.15"]
+            userAgent == "grpc-go/1.30.0"
+            ip == "30.56.100.15"
+        }
+    }
+
+    def "Ignores invalid ip header"() {
+        setup:
+        def metadata = new Metadata()
+        metadata.put(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER), "grpc-go/1.30.0")
+        metadata.put(Metadata.Key.of("x-real-ip", Metadata.ASCII_STRING_MARSHALLER), "271.194.19.1")
+        def attributes = Attributes.newBuilder()
+                .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448))
+                .build()
+        when:
+        def act = new TestEventBuilder()
+                .tap {
+                    it.start(metadata, attributes)
+                    it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance())
+                }
+                .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance())
+        then:
+        with(act.request.remote) {
+            ips == ["30.56.100.15"]
+            userAgent == "grpc-go/1.30.0"
+            ip == "30.56.100.15"
+        }
+    }
+
+    def "Ignores host addr in ip header"() {
+        setup:
+        def metadata = new Metadata()
+        metadata.put(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER), "grpc-go/1.30.0")
+        metadata.put(Metadata.Key.of("x-real-ip", Metadata.ASCII_STRING_MARSHALLER), "google.com")
+        def attributes = Attributes.newBuilder()
+                .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448))
+                .build()
+        when:
+        def act = new TestEventBuilder()
+                .tap {
+                    it.start(metadata, attributes)
+                    it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance())
+                }
+                .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance())
+        then:
+        with(act.request.remote) {
+            ips == ["30.56.100.15"]
+            userAgent == "grpc-go/1.30.0"
+            ip == "30.56.100.15"
+        }
+    }
+
+    def "Extracts ipv6 addresses"() {
+        setup:
+        def metadata = new Metadata()
+        metadata.put(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER), "grpc-go/1.30.0")
+        metadata.put(Metadata.Key.of("x-real-ip", Metadata.ASCII_STRING_MARSHALLER), "2001:0db8:0000:0000:0000:ff00:0042:8329")
+        def attributes = Attributes.newBuilder()
+                .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet6Address.getByName("::1"), 2448))
+                .build()
+        when:
+        def act = new TestEventBuilder()
+                .tap {
+                    it.start(metadata, attributes)
+                    it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance())
+                }
+                .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance())
+        then:
+        with(act.request.remote) {
+            ips == ["2001:db8:0:0:0:ff00:42:8329", "0:0:0:0:0:0:0:1"]
+            ip == "2001:db8:0:0:0:ff00:42:8329"
+        }
+    }
+
+    def "Cleans up user agent"() {
+        setup:
+        def metadata = new Metadata()
+        metadata.put(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER), "grpc-go/1.30.0\nxss\n\r")
+        def attributes = Attributes.newBuilder()
+                .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448))
+                .build()
+        when:
+        def act = new TestEventBuilder()
+                .tap {
+                    it.start(metadata, attributes)
+                    it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance())
+                }
+                .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance())
+        then:
+        with(act.request.remote) {
+            userAgent == "grpc-go/1.30.0 xss"
+        }
+    }
+
+    def "Truncates up user agent to 128 characters max"() {
+        setup:
+        def metadata = new Metadata()
+        metadata.put(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER),
+                "0123456_1_0123456_2_0123456_3_0123456_4_0123456_5_0123456_6_0123456_7_0123456_8_0123456_9_0123456_0_0123456_1_0123456_2_0123456_3_0123456_4_0123456_5")
+        def attributes = Attributes.newBuilder()
+                .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448))
+                .build()
+        when:
+        def act = new TestEventBuilder()
+                .tap {
+                    it.start(metadata, attributes)
+                    it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance())
+                }
+                .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance())
+        then:
+        with(act.request.remote) {
+            userAgent.length() == 128
+            userAgent == "0123456_1_0123456_2_0123456_3_0123456_4_0123456_5_0123456_6_0123456_7_0123456_8_0123456_9_0123456_0_0123456_1_0123456_2_0123456_"
+        }
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy

@@ -0,0 +1,83 @@
+package io.emeraldpay.dshackle.monitoring.accesslog
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.Common
+import io.emeraldpay.grpc.Chain
+import spock.lang.Specification
+
+class EventsBuilderSubscribeBalanceSpec extends Specification {
+
+    def "Basic ethereum event"() {
+        setup:
+        def request = BlockchainOuterClass.BalanceRequest.newBuilder()
+                .setAddress(
+                        Common.AnyAddress.newBuilder()
+                                .setAddressSingle(
+                                        Common.SingleAddress.newBuilder()
+                                                .setAddress("0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D")
+                                )
+                )
+                .setAsset(
+                        Common.Asset.newBuilder()
+                                .setChainValue(100)
+                                .setCode("ETHER")
+                )
+                .build()
+        def resp = BlockchainOuterClass.AddressBalance.newBuilder()
+                .setAddress(Common.SingleAddress.newBuilder()
+                        .setAddress("0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D"))
+                .setAsset(Common.Asset.newBuilder()
+                        .setChainValue(100)
+                        .setCode("ETHER"))
+                .setBalance("1234560000000000000")
+                .build()
+        when:
+        def act = new EventsBuilder.SubscribeBalance(true).tap {
+            it.onRequest(request)
+        }.onReply(resp)
+        then:
+        act.index == 0
+        act.blockchain == Chain.ETHEREUM
+        act.balanceRequest.asset == "ETHER"
+        act.balanceRequest.addressType == "ADDRESS_SINGLE"
+        act.addressBalance.asset == "ETHER"
+        act.addressBalance.address == "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D"
+    }
+
+    def "Basic bitcoin event"() {
+        setup:
+        def request = BlockchainOuterClass.BalanceRequest.newBuilder()
+                .setAddress(
+                        Common.AnyAddress.newBuilder()
+                                .setAddressSingle(
+                                        Common.SingleAddress.newBuilder()
+                                                .setAddress("1NDyJtNTjmwk5xPNhjgAMu4HDHigtobu1s")
+                                )
+                )
+                .setAsset(
+                        Common.Asset.newBuilder()
+                                .setChainValue(1)
+                                .setCode("BTC")
+                )
+                .build()
+        def resp = BlockchainOuterClass.AddressBalance.newBuilder()
+                .setAddress(Common.SingleAddress.newBuilder()
+                        .setAddress("1NDyJtNTjmwk5xPNhjgAMu4HDHigtobu1s"))
+                .setAsset(Common.Asset.newBuilder()
+                        .setChainValue(1)
+                        .setCode("BTC"))
+                .setBalance("12345600000000")
+                .build()
+        when:
+        def act = new EventsBuilder.SubscribeBalance(true).tap {
+            it.onRequest(request)
+        }.onReply(resp)
+        then:
+        act.index == 0
+        act.blockchain == Chain.BITCOIN
+        act.balanceRequest.asset == "BTC"
+        act.balanceRequest.addressType == "ADDRESS_SINGLE"
+        act.addressBalance.asset == "BTC"
+        act.addressBalance.address == "1NDyJtNTjmwk5xPNhjgAMu4HDHigtobu1s"
+    }
+}