diff --git a/build.gradle b/build.gradle index 09a746ad..e0dfb311 100644 --- a/build.gradle +++ b/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-databind:2.9.8' + implementation "com.fasterxml.jackson.core:jackson-core:$jacksonVersion" + 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' diff --git a/docs/01-architecture-intro.adoc b/docs/01-architecture-intro.adoc index cf0c334e..e5c8cbf0 100644 --- a/docs/01-architecture-intro.adoc +++ b/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 diff --git a/docs/02-quick-start.adoc b/docs/02-quick-start.adoc index 40cde176..78b87c60 100644 --- a/docs/02-quick-start.adoc +++ b/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] diff --git a/docs/03-server-config.adoc b/docs/03-server-config.adoc index e2ed9b4d..b1e45fc9 100644 --- a/docs/03-server-config.adoc +++ b/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` | diff --git a/docs/06-monitoring.adoc b/docs/06-monitoring.adoc new file mode 100644 index 00000000..9ef6d8b1 --- /dev/null +++ b/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) + + diff --git a/docs/06-methods.adoc b/docs/07-methods.adoc similarity index 98% rename from docs/06-methods.adoc rename to docs/07-methods.adoc index a5936743..8ffdae78 100644 --- a/docs/06-methods.adoc +++ b/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. diff --git a/docs/07-authentication.adoc b/docs/08-authentication.adoc similarity index 100% rename from docs/07-authentication.adoc rename to docs/08-authentication.adoc diff --git a/docs/08-quorum-and-selectors.adoc b/docs/09-quorum-and-selectors.adoc similarity index 100% rename from docs/08-quorum-and-selectors.adoc rename to docs/09-quorum-and-selectors.adoc diff --git a/docs/09-caching.adoc b/docs/10-caching.adoc similarity index 100% rename from docs/09-caching.adoc rename to docs/10-caching.adoc diff --git a/docs/10-client-libraries.adoc b/docs/11-client-libraries.adoc similarity index 100% rename from docs/10-client-libraries.adoc rename to docs/11-client-libraries.adoc diff --git a/docs/README.adoc b/docs/README.adoc index 5aa5cdae..9c452b2f 100644 --- a/docs/README.adoc +++ b/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:07-authentication.adoc[Authentication] -. link:08-quorum-and-selectors.adoc[Quorum and Selectors] -. link:09-caching.adoc[Caching] -. link:10-client-libraries.adoc[Client Libraries] +. link:06-monitoring.adoc[Logging & Monitoring] +. link:07-methods.adoc[API methods] +. link:08-authentication.adoc[Authentication] +. link:09-quorum-and-selectors.adoc[Quorum and Selectors] +. link:10-caching.adoc[Caching] +. link:11-client-libraries.adoc[Client Libraries] == Reference diff --git a/docs/reference-configuration.adoc b/docs/reference-configuration.adoc index 3d992060..d8d615c9 100644 --- a/docs/reference-configuration.adoc +++ b/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 <> section +| Setup TLS configuration for the gRPC server. +See <> section | `proxy` | -| Setup HTTP proxy that emulates all standard JSON RPC requests. See <> section +| Setup HTTP proxy that emulates all standard JSON RPC requests. +See <> section + +| `accessLog` +| +| Configure access logging. +See <> section + | `tokens` | -| Configure tokens for tracking balance. See <> section - +| Configure tokens for tracking balance. +See <> section | `cache` | -| Caching configuration. See <> section. +| Caching configuration. +See <> section. | `cluster` | -| Setup connection to remote nodes. See <> section +| Setup connection to remote nodes.See <> 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 -HTTP2 + gRPC is designed to be used with TLS, and some of the related software is unable to use it without TLS. + - See link:07-authentication.adoc[Authentication] docs and <>. +a| TLC configuration for the connection. +In general it's an optional configuration, but it's strongly recommended. +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.ca` | Path to x509 certificate to verify remote server diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt index 9f21ca54..dfdc9aa6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt +++ b/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")) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt index 3580f1e2..10563348 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt +++ b/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, @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) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/AccessLogConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/AccessLogConfig.kt new file mode 100644 index 00000000..5dde856c --- /dev/null +++ b/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 + ) + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/AccessLogReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/AccessLogReader.kt new file mode 100644 index 00000000..7174efe9 --- /dev/null +++ b/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 { + + 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() + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfig.kt index 456b44d6..637fd352 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfig.kt +++ b/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() } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfigReader.kt index 84c82119..bed43f8c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfigReader.kt +++ b/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 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt new file mode 100644 index 00000000..1685eba3 --- /dev/null +++ b/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 interceptCall( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler): ServerCall.Listener { + + 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 process( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler, + builder: EventsBuilder.RequestReply + ): ServerCall.Listener { + builder.start(headers, call.attributes) + val callWrapper: ServerCall = StdCallResponse( + call, builder, accessLogWriter + ) + return StdCallListener( + next.startCall(callWrapper, headers), + builder + ) + } + + @Suppress("UNCHECKED_CAST") + private fun processSubscribeHead( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + return process(call, headers, next, + EventsBuilder.SubscribeHead() as EventsBuilder.RequestReply<*, ReqT, RespT> + ) + } + + @Suppress("UNCHECKED_CAST") + private fun processSubscribeBalance( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler, + subscribe: Boolean + ): ServerCall.Listener { + return process(call, headers, next, + EventsBuilder.SubscribeBalance(subscribe) as EventsBuilder.RequestReply<*, ReqT, RespT> + ) + } + + @Suppress("UNCHECKED_CAST") + private fun processSubscribeTxStatus( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + return process(call, headers, next, + EventsBuilder.TxStatus() as EventsBuilder.RequestReply<*, ReqT, RespT> + ) + } + + @Suppress("UNCHECKED_CAST") + private fun processNativeCall( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + return process(call, headers, next, + EventsBuilder.NativeCall() as EventsBuilder.RequestReply<*, ReqT, RespT> + ) + } + + @Suppress("UNCHECKED_CAST") + private fun processDescribe( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + return process(call, headers, next, + EventsBuilder.Describe() as EventsBuilder.RequestReply<*, ReqT, RespT> + ) + } + + @Suppress("UNCHECKED_CAST") + private fun processStatus( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + return process(call, headers, next, + EventsBuilder.Status() as EventsBuilder.RequestReply<*, ReqT, RespT> + ) + } + + open class StdCallListener>( + val next: ServerCall.Listener, + val builder: EB + ) : ForwardingServerCallListener() { + + override fun onMessage(message: Req) { + builder.onRequest(message) + super.onMessage(message) + } + + override fun delegate(): ServerCall.Listener { + return next + } + } + + open class StdCallResponse>( + val next: ServerCall, + val builder: EB, + val accessLogWriter: AccessLogWriter + ) : ForwardingServerCall() { + + override fun getMethodDescriptor(): MethodDescriptor { + return next.methodDescriptor + } + + override fun delegate(): ServerCall { + return next + } + + override fun sendMessage(message: RespT) { + super.sendMessage(message) + accessLogWriter.submit( + builder.onReply(message)!! + ) + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriter.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriter.kt new file mode 100644 index 00000000..3573952d --- /dev/null +++ b/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() + 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) { + 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) + } + } + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt new file mode 100644 index 00000000..96c255a3 --- /dev/null +++ b/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, + 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 + ) +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt new file mode 100644 index 00000000..e8a88fe1 --- /dev/null +++ b/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 : StartingRequest { + fun onRequest(msg: Req) + fun onReply(msg: Resp): E + } + + abstract class Base() : 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? { + // 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() + 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(), + RequestReply { + + 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(), + RequestReply { + + 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(), + RequestReply { + 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(), + RequestReply { + val items = ArrayList() + val replies = HashMap() + 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(), + RequestReply { + + 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(), + RequestReply { + 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() + ) + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index f443c2f9..aba5ab05 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/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 = TopicProcessor.create() + private val statusStream = Sinks.many() + .multicast() + .directBestEffort() 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 { - 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) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 5fe53e10..3122ada9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -168,7 +168,12 @@ abstract class Multistream( } override fun observeStatus(): Flux { - 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 } diff --git a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriterSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriterSpec.groovy new file mode 100644 index 00000000..bde9ce73 --- /dev/null +++ b/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" + } + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy new file mode 100644 index 00000000..ede6ab9d --- /dev/null +++ b/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 + implements EventsBuilder.RequestReply { + + @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_" + } + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy new file mode 100644 index 00000000..75d78f5a --- /dev/null +++ b/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" + } +}