From 4875ade0e9e651c1403e135fa1dd5f13ee43b2af Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 3 Jun 2021 18:12:46 -0400 Subject: [PATCH 01/16] solution: base code for access logging --- build.gradle | 6 +- .../kotlin/io/emeraldpay/dshackle/Global.kt | 4 + .../io/emeraldpay/dshackle/GrpcServer.kt | 18 ++- .../dshackle/config/AccessLogConfig.kt | 22 +++ .../dshackle/config/AccessLogReader.kt | 27 ++++ .../emeraldpay/dshackle/config/MainConfig.kt | 2 +- .../dshackle/config/MainConfigReader.kt | 4 + .../monitoring/accesslog/AccessHandler.kt | 110 ++++++++++++++ .../monitoring/accesslog/AccessLogWriter.kt | 124 +++++++++++++++ .../dshackle/monitoring/accesslog/Events.kt | 142 ++++++++++++++++++ 10 files changed, 449 insertions(+), 10 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/AccessLogConfig.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/AccessLogReader.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriter.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt 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/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..5a4aadf8 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt @@ -0,0 +1,110 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.monitoring.accesslog + +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.Global +import io.grpc.* +import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.stereotype.Service + +@Service +class AccessHandler( + @Autowired private val accessLogWriter: AccessLogWriter +) : ServerInterceptor { + + companion object { + private val log = LoggerFactory.getLogger(AccessHandler::class.java) + } + + override fun interceptCall( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler): ServerCall.Listener { + + when (val method = call.methodDescriptor.bareMethodName) { + "NativeCall" -> { + val builder = Events.NativeCallBuilder() + return OnNativeCall( + next.startCall(OnNativeCallResponse(call, builder), headers), + builder) { logs -> + accessLogWriter.submit(logs) + } + } + else -> { + log.trace("unsupported method `{}`", method) + } + } + + // continue + return next.startCall(call, headers) + } + + + class OnNativeCall( + val next: ServerCall.Listener, + val builder: Events.NativeCallBuilder, + val done: (List) -> Unit + ) : ForwardingServerCallListener() { + + override fun onMessage(message: ReqT) { + if (message is BlockchainOuterClass.NativeCallRequest) { + val chain = message.chain + builder.withChain(chain.number) + message.itemsList.forEach { item -> + builder.onItem(item) + } + } + super.onMessage(message) + } + + override fun onCancel() { + super.onCancel() + done(builder.build()) + } + + override fun onComplete() { + super.onComplete() + done(builder.build()) + } + + override fun delegate(): ServerCall.Listener { + return next + } + } + + class OnNativeCallResponse( + val next: ServerCall, + val builder: Events.NativeCallBuilder + ) : ForwardingServerCall() { + + override fun getMethodDescriptor(): MethodDescriptor { + return next.methodDescriptor + } + + override fun delegate(): ServerCall { + return next + } + + override fun sendMessage(message: RespT) { + if (message is BlockchainOuterClass.NativeCallReplyItem) { + builder.onItemReply(message) + } + super.sendMessage(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..93340a25 --- /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 diabled") + return + } + log.info("Writing Access Log to ${filename.absolutePath}") + scheduler.schedule(runner, START_SLEEP_MS, TimeUnit.MILLISECONDS) + } + + private fun flushRunner() { + try { + flush() + } catch (t: Throwable) { + logError { + log.error("Failed to write logs. ${t.javaClass}:${t.message}") + } + } finally { + scheduler.schedule(runner, FLUSH_SLEEP_MS, TimeUnit.MILLISECONDS) + } + } + + fun submit(event: Any) { + queue.add(event) + } + + fun submit(events: List) { + 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..6dc289f1 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -0,0 +1,142 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.monitoring.accesslog + +import com.fasterxml.jackson.annotation.JsonInclude +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.grpc.Chain +import org.slf4j.LoggerFactory +import java.time.Instant +import java.util.* +import kotlin.collections.ArrayList + +class Events { + + companion object { + private val log = LoggerFactory.getLogger(Events::class.java) + } + + abstract class Base( + val method: String, + val id: UUID + ) { + val ts = Instant.now() + } + + abstract class ChainBase( + val blockchain: Chain, method: String, id: UUID + ) : Base(method, id) { + + } + + @JsonInclude(JsonInclude.Include.NON_NULL) + class NativeCall( + // info about the initial request, that may include several native calls + val request: StreamRequestDetails, + // total native calls passes within the initial request + val total: Int, + // index of the call specific for the current response + val index: Int, + val selector: String? = null, + val quorum: Long? = null, + val minAvailability: String? = null, + + val succeed: Boolean, + val rpcError: Int? = null, + val payloadSizeBytes: Long, + + blockchain: Chain, method: String, id: UUID + ) : ChainBase(blockchain, method, id) { + + } + + data class StreamRequestDetails( + val id: UUID, + val start: Instant + ) + + data class Remote( + val ips: List, + val userAgent: String + ) + + data class NativeCallItemDetails( + val method: String, + val id: Int, + val payloadSizeBytes: Long + ) + + data class NativeCallReplyDetails( + val id: Int, + val succeed: Boolean, + val replySizeBytes: Long, + val ts: Instant = Instant.now() + ) + + class NativeCallBuilder() { + + private val requestDetails = StreamRequestDetails( + UUID.randomUUID(), + Instant.now() + ) + + var chain: Int = Chain.UNSPECIFIED.id + val items = ArrayList() + val replies = HashMap() + + fun withChain(chain: Int): NativeCallBuilder { + this.chain = chain + return this + } + + fun onItem(item: BlockchainOuterClass.NativeCallItem): NativeCallBuilder { + this.items.add( + NativeCallItemDetails( + item.method, + item.id, + item.payload.size().toLong() + ) + ) + return this + } + + fun onItemReply(reply: BlockchainOuterClass.NativeCallReplyItem): NativeCallBuilder { + this.replies[reply.id] = NativeCallReplyDetails( + reply.id, + reply.succeed, + reply.payload?.size()?.toLong() ?: 0L + ) + return this + } + + fun build(): List { + val blockchain = Chain.byId(this.chain) + return items.mapIndexed { index, item -> + val reply = replies[item.id] + NativeCall( + request = requestDetails, + total = items.size, + index = index, + succeed = reply?.succeed ?: false, + blockchain = blockchain, + method = item.method, + payloadSizeBytes = item.payloadSizeBytes, + id = UUID.randomUUID() + ) + } + } + } +} \ No newline at end of file From fd26d79f6ec1185ede18f283f0ced7fad1030e9c Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sat, 26 Jun 2021 22:15:46 -0400 Subject: [PATCH 02/16] solution: save remote client details into access log --- .../monitoring/accesslog/AccessHandler.kt | 2 +- .../monitoring/accesslog/AccessLogWriter.kt | 2 +- .../dshackle/monitoring/accesslog/Events.kt | 82 ++++++- .../EventsNativeCallBuilderSpec.groovy | 207 ++++++++++++++++++ 4 files changed, 287 insertions(+), 6 deletions(-) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsNativeCallBuilderSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt index 5a4aadf8..0c207ec2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt @@ -16,7 +16,6 @@ package io.emeraldpay.dshackle.monitoring.accesslog import io.emeraldpay.api.proto.BlockchainOuterClass -import io.emeraldpay.dshackle.Global import io.grpc.* import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired @@ -39,6 +38,7 @@ class AccessHandler( when (val method = call.methodDescriptor.bareMethodName) { "NativeCall" -> { val builder = Events.NativeCallBuilder() + .start(headers, call.attributes) return OnNativeCall( next.startCall(OnNativeCallResponse(call, builder), headers), builder) { logs -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriter.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriter.kt index 93340a25..3573952d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriter.kt @@ -56,7 +56,7 @@ class AccessLogWriter( @PostConstruct fun start() { if (!config.enabled) { - log.info("Access Log is diabled") + log.info("Access Log is disabled") return } log.info("Writing Access Log to ${filename.absolutePath}") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index 6dc289f1..a26cade0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -18,10 +18,15 @@ package io.emeraldpay.dshackle.monitoring.accesslog import com.fasterxml.jackson.annotation.JsonInclude 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.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory +import java.net.InetAddress +import java.net.InetSocketAddress import java.time.Instant import java.util.* -import kotlin.collections.ArrayList class Events { @@ -65,11 +70,13 @@ class Events { data class StreamRequestDetails( val id: UUID, - val start: Instant + val start: Instant, + val remote: Remote ) data class Remote( val ips: List, + val ip: String, val userAgent: String ) @@ -88,15 +95,82 @@ class Events { class NativeCallBuilder() { - private val requestDetails = StreamRequestDetails( + 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]+") + } + + private var requestDetails = StreamRequestDetails( UUID.randomUUID(), - Instant.now() + Instant.now(), + Remote(emptyList(), "", "") ) var chain: Int = Chain.UNSPECIFIED.id val items = ArrayList() val replies = HashMap() + 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() + } + + fun start(metadata: Metadata, attributes: Attributes): NativeCallBuilder { + val userAgent = metadata.get(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER)) + ?.let(this@NativeCallBuilder::clean) + ?: "" + val ips = ArrayList() + remoteIpKeys.forEach { key -> + metadata.get(key)?.let { + it.trim().ifEmpty { null } + ?.let(this@NativeCallBuilder::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 = Remote( + ips = ips.map { it.hostAddress }, + ip = ip, + userAgent = userAgent + )) + return this + } + fun withChain(chain: Int): NativeCallBuilder { this.chain = chain return this diff --git a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsNativeCallBuilderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsNativeCallBuilderSpec.groovy new file mode 100644 index 00000000..ef92c58a --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsNativeCallBuilderSpec.groovy @@ -0,0 +1,207 @@ +/** + * 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 spock.lang.Specification + +class EventsNativeCallBuilderSpec extends Specification { + + 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 Events.NativeCallBuilder() + .start(metadata, attributes) + .withChain(Chain.ETHEREUM.id) + .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .build() + then: + act.size() == 1 + with(act[0]) { + it.request != null + it.request.remote != null + with(it.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 Events.NativeCallBuilder() + .start(metadata, attributes) + .withChain(Chain.ETHEREUM.id) + .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .build() + then: + act.size() == 1 + with(act[0].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 Events.NativeCallBuilder() + .start(metadata, attributes) + .withChain(Chain.ETHEREUM.id) + .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .build() + then: + act.size() == 1 + with(act[0].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 Events.NativeCallBuilder() + .start(metadata, attributes) + .withChain(Chain.ETHEREUM.id) + .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .build() + then: + act.size() == 1 + with(act[0].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 Events.NativeCallBuilder() + .start(metadata, attributes) + .withChain(Chain.ETHEREUM.id) + .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .build() + then: + act.size() == 1 + with(act[0].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 Events.NativeCallBuilder() + .start(metadata, attributes) + .withChain(Chain.ETHEREUM.id) + .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .build() + then: + act.size() == 1 + with(act[0].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 Events.NativeCallBuilder() + .start(metadata, attributes) + .withChain(Chain.ETHEREUM.id) + .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .build() + then: + act.size() == 1 + with(act[0].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 Events.NativeCallBuilder() + .start(metadata, attributes) + .withChain(Chain.ETHEREUM.id) + .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .build() + then: + act.size() == 1 + with(act[0].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_" + } + } +} From a541c32104c4f73e28ce91988a6b35803cb45364 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 27 Jun 2021 18:44:41 -0400 Subject: [PATCH 03/16] solution: access logging for SubscribeHead method --- .../monitoring/accesslog/AccessHandler.kt | 115 ++++++++++++++---- .../dshackle/monitoring/accesslog/Events.kt | 75 ++++++++---- ...ec.groovy => EventsBaseBuilderSpec.groovy} | 2 +- 3 files changed, 143 insertions(+), 49 deletions(-) rename src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/{EventsNativeCallBuilderSpec.groovy => EventsBaseBuilderSpec.groovy} (99%) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt index 0c207ec2..fd0ae1b9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt @@ -16,6 +16,7 @@ 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 @@ -36,14 +37,11 @@ class AccessHandler( next: ServerCallHandler): ServerCall.Listener { when (val method = call.methodDescriptor.bareMethodName) { + "SubscribeHead" -> { + return processSubscribeHead(call, headers, next) + } "NativeCall" -> { - val builder = Events.NativeCallBuilder() - .start(headers, call.attributes) - return OnNativeCall( - next.startCall(OnNativeCallResponse(call, builder), headers), - builder) { logs -> - accessLogWriter.submit(logs) - } + return processNativeCall(call, headers, next) } else -> { log.trace("unsupported method `{}`", method) @@ -54,20 +52,68 @@ class AccessHandler( return next.startCall(call, headers) } + @Suppress("UNCHECKED_CAST") + private fun processSubscribeHead( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + val builder = Events.SubscribeHeadBuilder() + .start(headers, call.attributes) + val callWrapper: ServerCall = OnSubscribeHeadResponse( + call as ServerCall, builder, accessLogWriter) as ServerCall + return OnSubscribeHead( + next.startCall(callWrapper, headers) as ServerCall.Listener, + builder + ) as ServerCall.Listener + } - class OnNativeCall( - val next: ServerCall.Listener, + @Suppress("UNCHECKED_CAST") + private fun processNativeCall( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + val builder = Events.NativeCallBuilder() + .start(headers, call.attributes) + + val callWrapper: ServerCall = OnNativeCallResponse( + call as ServerCall, builder + ) as ServerCall + return OnNativeCall( + next.startCall(callWrapper, headers) as ServerCall.Listener, + builder) { logs -> + accessLogWriter.submit(logs) + } as ServerCall.Listener + } + + class OnSubscribeHead( + val next: ServerCall.Listener, + val builder: Events.SubscribeHeadBuilder + ) : ForwardingServerCallListener() { + + override fun onMessage(message: Common.Chain) { + val chainId = message.type.number + builder.withChain(chainId) + super.onMessage(message) + } + + override fun delegate(): ServerCall.Listener { + return next + } + } + + class OnNativeCall( + val next: ServerCall.Listener, val builder: Events.NativeCallBuilder, val done: (List) -> Unit - ) : ForwardingServerCallListener() { + ) : ForwardingServerCallListener() { - override fun onMessage(message: ReqT) { - if (message is BlockchainOuterClass.NativeCallRequest) { - val chain = message.chain - builder.withChain(chain.number) - message.itemsList.forEach { item -> - builder.onItem(item) - } + override fun onMessage(message: BlockchainOuterClass.NativeCallRequest) { + val chain = message.chain + builder.withChain(chain.number) + message.itemsList.forEach { item -> + builder.onItem(item) } super.onMessage(message) } @@ -82,16 +128,14 @@ class AccessHandler( done(builder.build()) } - override fun delegate(): ServerCall.Listener { + override fun delegate(): ServerCall.Listener { return next } } - class OnNativeCallResponse( - val next: ServerCall, - val builder: Events.NativeCallBuilder + abstract class BaseCallResponse( + val next: ServerCall ) : ForwardingServerCall() { - override fun getMethodDescriptor(): MethodDescriptor { return next.methodDescriptor } @@ -101,9 +145,30 @@ class AccessHandler( } override fun sendMessage(message: RespT) { - if (message is BlockchainOuterClass.NativeCallReplyItem) { - builder.onItemReply(message) - } + super.sendMessage(message) + } + } + + class OnNativeCallResponse( + next: ServerCall, + val builder: Events.NativeCallBuilder + ) : BaseCallResponse(next) { + + override fun sendMessage(message: BlockchainOuterClass.NativeCallReplyItem) { + builder.onItemReply(message) + super.sendMessage(message) + } + } + + class OnSubscribeHeadResponse( + next: ServerCall, + val builder: Events.SubscribeHeadBuilder, + val accessLogWriter: AccessLogWriter + ) : BaseCallResponse(next) { + + override fun sendMessage(message: BlockchainOuterClass.ChainHead) { + val event = builder.onReply(message) + accessLogWriter.submit(event) super.sendMessage(message) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index a26cade0..052b7aa5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -35,20 +35,28 @@ class Events { } abstract class Base( - val method: String, val id: UUID ) { val ts = Instant.now() } abstract class ChainBase( - val blockchain: Chain, method: String, id: UUID - ) : Base(method, id) { + val blockchain: Chain, val method: String, id: UUID + ) : Base(id) - } + @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 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 @@ -62,11 +70,8 @@ class Events { val succeed: Boolean, val rpcError: Int? = null, val payloadSizeBytes: Long, - - blockchain: Chain, method: String, id: UUID - ) : ChainBase(blockchain, method, id) { - - } + val nativeCall: NativeCallItemDetails + ) : ChainBase(blockchain, "NativeCall", id) data class StreamRequestDetails( val id: UUID, @@ -93,8 +98,7 @@ class Events { val ts: Instant = Instant.now() ) - class NativeCallBuilder() { - + abstract class BaseBuilder() { companion object { private val remoteIpKeys = listOf( Metadata.Key.of("x-real-ip", Metadata.ASCII_STRING_MARSHALLER), @@ -103,15 +107,14 @@ class Events { private val invalidCharacters = Regex("[\n\t]+") } - private var requestDetails = StreamRequestDetails( + var requestDetails = StreamRequestDetails( UUID.randomUUID(), Instant.now(), Remote(emptyList(), "", "") ) - var chain: Int = Chain.UNSPECIFIED.id - val items = ArrayList() - val replies = HashMap() + var chainId: Int = Chain.UNSPECIFIED.id + var chain = Chain.UNSPECIFIED private fun toInetAddress(ip: String): InetAddress? { val isIp = Character.digit(ip[0], 16) != -1 @@ -144,15 +147,17 @@ class Events { .trim() } - fun start(metadata: Metadata, attributes: Attributes): NativeCallBuilder { + abstract protected fun getT(): T + + fun start(metadata: Metadata, attributes: Attributes): T { val userAgent = metadata.get(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER)) - ?.let(this@NativeCallBuilder::clean) + ?.let(this@BaseBuilder::clean) ?: "" val ips = ArrayList() remoteIpKeys.forEach { key -> metadata.get(key)?.let { it.trim().ifEmpty { null } - ?.let(this@NativeCallBuilder::toInetAddress) + ?.let(this@BaseBuilder::toInetAddress) ?.let(ips::add) } } @@ -168,11 +173,36 @@ class Events { ip = ip, userAgent = userAgent )) + return getT() + } + + fun withChain(chain: Int): T { + this.chainId = chain + this.chain = Chain.byId(chainId) + return getT() + } + } + + class SubscribeHeadBuilder() : BaseBuilder() { + private var index = 0 + + override fun getT(): SubscribeHeadBuilder { return this } - fun withChain(chain: Int): NativeCallBuilder { - this.chain = chain + fun onReply(resp: BlockchainOuterClass.ChainHead): SubscribeHead { + return SubscribeHead( + chain, UUID.randomUUID(), requestDetails, index++ + ) + } + } + + class NativeCallBuilder : BaseBuilder() { + + val items = ArrayList() + val replies = HashMap() + + override fun getT(): NativeCallBuilder { return this } @@ -197,7 +227,6 @@ class Events { } fun build(): List { - val blockchain = Chain.byId(this.chain) return items.mapIndexed { index, item -> val reply = replies[item.id] NativeCall( @@ -205,8 +234,8 @@ class Events { total = items.size, index = index, succeed = reply?.succeed ?: false, - blockchain = blockchain, - method = item.method, + blockchain = chain, + nativeCall = item, payloadSizeBytes = item.payloadSizeBytes, id = UUID.randomUUID() ) diff --git a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsNativeCallBuilderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy similarity index 99% rename from src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsNativeCallBuilderSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy index ef92c58a..85a55244 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsNativeCallBuilderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy @@ -22,7 +22,7 @@ import io.grpc.Grpc import io.grpc.Metadata import spock.lang.Specification -class EventsNativeCallBuilderSpec extends Specification { +class EventsBaseBuilderSpec extends Specification { def "Parse headers from direct local access"() { setup: From ec043592c48e61705d5a1894cab11375eb2049be Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 28 Jun 2021 13:58:34 -0400 Subject: [PATCH 04/16] solution: refactoring --- .../monitoring/accesslog/AccessHandler.kt | 12 +- .../dshackle/monitoring/accesslog/Events.kt | 144 -------------- .../monitoring/accesslog/EventsBuilder.kt | 180 ++++++++++++++++++ .../accesslog/EventsBaseBuilderSpec.groovy | 16 +- 4 files changed, 194 insertions(+), 158 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt index fd0ae1b9..5208dc89 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt @@ -58,7 +58,7 @@ class AccessHandler( headers: Metadata, next: ServerCallHandler ): ServerCall.Listener { - val builder = Events.SubscribeHeadBuilder() + val builder = EventsBuilder.SubscribeHead() .start(headers, call.attributes) val callWrapper: ServerCall = OnSubscribeHeadResponse( call as ServerCall, builder, accessLogWriter) as ServerCall @@ -74,7 +74,7 @@ class AccessHandler( headers: Metadata, next: ServerCallHandler ): ServerCall.Listener { - val builder = Events.NativeCallBuilder() + val builder = EventsBuilder.NativeCall() .start(headers, call.attributes) val callWrapper: ServerCall = OnNativeCallResponse( @@ -89,7 +89,7 @@ class AccessHandler( class OnSubscribeHead( val next: ServerCall.Listener, - val builder: Events.SubscribeHeadBuilder + val builder: EventsBuilder.SubscribeHead ) : ForwardingServerCallListener() { override fun onMessage(message: Common.Chain) { @@ -105,7 +105,7 @@ class AccessHandler( class OnNativeCall( val next: ServerCall.Listener, - val builder: Events.NativeCallBuilder, + val builder: EventsBuilder.NativeCall, val done: (List) -> Unit ) : ForwardingServerCallListener() { @@ -151,7 +151,7 @@ class AccessHandler( class OnNativeCallResponse( next: ServerCall, - val builder: Events.NativeCallBuilder + val builder: EventsBuilder.NativeCall ) : BaseCallResponse(next) { override fun sendMessage(message: BlockchainOuterClass.NativeCallReplyItem) { @@ -162,7 +162,7 @@ class AccessHandler( class OnSubscribeHeadResponse( next: ServerCall, - val builder: Events.SubscribeHeadBuilder, + val builder: EventsBuilder.SubscribeHead, val accessLogWriter: AccessLogWriter ) : BaseCallResponse(next) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index 052b7aa5..020453d1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -98,148 +98,4 @@ class Events { val ts: Instant = Instant.now() ) - abstract class BaseBuilder() { - 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 = StreamRequestDetails( - UUID.randomUUID(), - Instant.now(), - 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() - } - - abstract protected fun getT(): T - - fun start(metadata: Metadata, attributes: Attributes): T { - val userAgent = metadata.get(Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER)) - ?.let(this@BaseBuilder::clean) - ?: "" - val ips = ArrayList() - remoteIpKeys.forEach { key -> - metadata.get(key)?.let { - it.trim().ifEmpty { null } - ?.let(this@BaseBuilder::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 = Remote( - ips = ips.map { it.hostAddress }, - ip = ip, - userAgent = userAgent - )) - return getT() - } - - fun withChain(chain: Int): T { - this.chainId = chain - this.chain = Chain.byId(chainId) - return getT() - } - } - - class SubscribeHeadBuilder() : BaseBuilder() { - private var index = 0 - - override fun getT(): SubscribeHeadBuilder { - return this - } - - fun onReply(resp: BlockchainOuterClass.ChainHead): SubscribeHead { - return SubscribeHead( - chain, UUID.randomUUID(), requestDetails, index++ - ) - } - } - - class NativeCallBuilder : BaseBuilder() { - - val items = ArrayList() - val replies = HashMap() - - override fun getT(): NativeCallBuilder { - return this - } - - fun onItem(item: BlockchainOuterClass.NativeCallItem): NativeCallBuilder { - this.items.add( - NativeCallItemDetails( - item.method, - item.id, - item.payload.size().toLong() - ) - ) - return this - } - - fun onItemReply(reply: BlockchainOuterClass.NativeCallReplyItem): NativeCallBuilder { - this.replies[reply.id] = NativeCallReplyDetails( - reply.id, - reply.succeed, - reply.payload?.size()?.toLong() ?: 0L - ) - return this - } - - fun build(): List { - return items.mapIndexed { index, item -> - val reply = replies[item.id] - NativeCall( - request = requestDetails, - total = items.size, - index = index, - succeed = reply?.succeed ?: false, - blockchain = chain, - nativeCall = item, - payloadSizeBytes = item.payloadSizeBytes, - id = UUID.randomUUID() - ) - } - } - } } \ 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..c9e223f3 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt @@ -0,0 +1,180 @@ +/** + * 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.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) + } + + abstract class Base() { + 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() + } + + abstract protected fun getT(): T + + fun start(metadata: Metadata, attributes: Attributes): T { + 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 + )) + return getT() + } + + fun withChain(chain: Int): T { + this.chainId = chain + this.chain = Chain.byId(chainId) + return getT() + } + } + + class SubscribeHead() : Base() { + private var index = 0 + + override fun getT(): SubscribeHead { + return this + } + + fun onReply(resp: BlockchainOuterClass.ChainHead): Events.SubscribeHead { + return Events.SubscribeHead( + chain, UUID.randomUUID(), requestDetails, index++ + ) + } + } + + class NativeCall : Base() { + + val items = ArrayList() + val replies = HashMap() + + override fun getT(): NativeCall { + return this + } + + fun onItem(item: BlockchainOuterClass.NativeCallItem): NativeCall { + this.items.add( + Events.NativeCallItemDetails( + item.method, + item.id, + item.payload.size().toLong() + ) + ) + return this + } + + fun onItemReply(reply: BlockchainOuterClass.NativeCallReplyItem): NativeCall { + this.replies[reply.id] = Events.NativeCallReplyDetails( + reply.id, + reply.succeed, + reply.payload?.size()?.toLong() ?: 0L + ) + return this + } + + fun build(): List { + return items.mapIndexed { index, item -> + val reply = replies[item.id] + Events.NativeCall( + request = requestDetails, + total = items.size, + index = index, + succeed = reply?.succeed ?: false, + blockchain = chain, + nativeCall = item, + payloadSizeBytes = item.payloadSizeBytes, + id = UUID.randomUUID() + ) + } + } + } +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy index 85a55244..b72ec97e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy @@ -32,7 +32,7 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("127.0.0.1"), 2448)) .build() when: - def act = new Events.NativeCallBuilder() + def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) @@ -59,7 +59,7 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("127.0.0.1"), 2448)) .build() when: - def act = new Events.NativeCallBuilder() + def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) @@ -81,7 +81,7 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448)) .build() when: - def act = new Events.NativeCallBuilder() + def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) @@ -104,7 +104,7 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448)) .build() when: - def act = new Events.NativeCallBuilder() + def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) @@ -127,7 +127,7 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448)) .build() when: - def act = new Events.NativeCallBuilder() + def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) @@ -150,7 +150,7 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet6Address.getByName("::1"), 2448)) .build() when: - def act = new Events.NativeCallBuilder() + def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) @@ -171,7 +171,7 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448)) .build() when: - def act = new Events.NativeCallBuilder() + def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) @@ -192,7 +192,7 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448)) .build() when: - def act = new Events.NativeCallBuilder() + def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) From 0a3053d8034b5627301eb936ba4f3aa7b987d1c5 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 28 Jun 2021 16:47:41 -0400 Subject: [PATCH 05/16] solution: access logging for SubscribeBalance method --- .../monitoring/accesslog/AccessHandler.kt | 59 ++++++++++--- .../dshackle/monitoring/accesslog/Events.kt | 27 ++++-- .../monitoring/accesslog/EventsBuilder.kt | 29 ++++++- .../EventsBuilderSubscribeBalanceSpec.groovy | 83 +++++++++++++++++++ 4 files changed, 180 insertions(+), 18 deletions(-) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt index 5208dc89..e013727b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt @@ -36,20 +36,15 @@ class AccessHandler( headers: Metadata, next: ServerCallHandler): ServerCall.Listener { - when (val method = call.methodDescriptor.bareMethodName) { - "SubscribeHead" -> { - return processSubscribeHead(call, headers, next) - } - "NativeCall" -> { - return processNativeCall(call, headers, next) - } + return when (val method = call.methodDescriptor.bareMethodName) { + "SubscribeHead" -> processSubscribeHead(call, headers, next) + "SubscribeBalance" -> processSubscribeBalance(call, headers, next) + "NativeCall" -> processNativeCall(call, headers, next) else -> { log.trace("unsupported method `{}`", method) + next.startCall(call, headers) } } - - // continue - return next.startCall(call, headers) } @Suppress("UNCHECKED_CAST") @@ -68,6 +63,22 @@ class AccessHandler( ) as ServerCall.Listener } + @Suppress("UNCHECKED_CAST") + private fun processSubscribeBalance( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + val builder = EventsBuilder.SubscribeBalance() + .start(headers, call.attributes) + val callWrapper: ServerCall = OnSubscribeBalanceResponse( + call as ServerCall, builder, accessLogWriter) as ServerCall + return OnSubscribeBalance( + next.startCall(callWrapper, headers) as ServerCall.Listener, + builder + ) as ServerCall.Listener + } + @Suppress("UNCHECKED_CAST") private fun processNativeCall( call: ServerCall, @@ -103,6 +114,21 @@ class AccessHandler( } } + class OnSubscribeBalance( + val next: ServerCall.Listener, + val builder: EventsBuilder.SubscribeBalance + ) : ForwardingServerCallListener() { + + override fun onMessage(message: BlockchainOuterClass.BalanceRequest) { + builder.withRequest(message) + super.onMessage(message) + } + + override fun delegate(): ServerCall.Listener { + return next + } + } + class OnNativeCall( val next: ServerCall.Listener, val builder: EventsBuilder.NativeCall, @@ -172,4 +198,17 @@ class AccessHandler( super.sendMessage(message) } } + + class OnSubscribeBalanceResponse( + next: ServerCall, + val builder: EventsBuilder.SubscribeBalance, + val accessLogWriter: AccessLogWriter + ) : BaseCallResponse(next) { + + override fun sendMessage(message: BlockchainOuterClass.AddressBalance) { + val event = builder.onReply(message) + accessLogWriter.submit(event) + super.sendMessage(message) + } + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index 020453d1..13e211cb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -16,15 +16,8 @@ package io.emeraldpay.dshackle.monitoring.accesslog import com.fasterxml.jackson.annotation.JsonInclude -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.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory -import java.net.InetAddress -import java.net.InetSocketAddress import java.time.Instant import java.util.* @@ -53,6 +46,17 @@ class Events { val index: Int ) : ChainBase(blockchain, "SubscribeHead", id) + @JsonInclude(JsonInclude.Include.NON_NULL) + class SubscribeBalance( + blockchain: Chain, id: UUID, + // initial request details + val request: StreamRequestDetails, + val balanceRequest: BalanceRequest, + val addressBalance: AddressBalance, + // index of the current response + val index: Int + ) : ChainBase(blockchain, "SubscribeBalance", id) + @JsonInclude(JsonInclude.Include.NON_NULL) class NativeCall( blockchain: Chain, id: UUID, @@ -98,4 +102,13 @@ class Events { 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 index c9e223f3..093b5347 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt @@ -132,8 +132,35 @@ class EventsBuilder { } } - class NativeCall : Base() { + class SubscribeBalance() : Base() { + private var index = 0 + private var balanceRequest: Events.BalanceRequest? = null + override fun getT(): SubscribeBalance { + return this + } + + fun withRequest(req: BlockchainOuterClass.BalanceRequest): SubscribeBalance { + balanceRequest = Events.BalanceRequest( + req.asset.code.toUpperCase(), + req.address.addrTypeCase.name + ) + return this + } + + fun onReply(resp: BlockchainOuterClass.AddressBalance): Events.SubscribeBalance { + if (balanceRequest == null) { + throw IllegalStateException("Request is not initialized") + } + val addressBalance = Events.AddressBalance(resp.asset.code, resp.address.address) + val chain = Chain.byId(resp.asset.chain.number) + return Events.SubscribeBalance( + chain, UUID.randomUUID(), requestDetails, balanceRequest!!, addressBalance, index++ + ) + } + } + + class NativeCall : Base() { val items = ArrayList() val replies = HashMap() 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..c445c590 --- /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() + .withRequest(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() + .withRequest(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" + } +} From f2bc0f1d9b45dc07be9962ce2116eb11476b0667 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 28 Jun 2021 18:41:08 -0400 Subject: [PATCH 06/16] solution: access logging for GetBalance method --- .../dshackle/monitoring/accesslog/AccessHandler.kt | 8 +++++--- .../io/emeraldpay/dshackle/monitoring/accesslog/Events.kt | 4 ++-- .../dshackle/monitoring/accesslog/EventsBuilder.kt | 4 ++-- 3 files changed, 9 insertions(+), 7 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt index e013727b..19df1a9e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt @@ -38,7 +38,8 @@ class AccessHandler( return when (val method = call.methodDescriptor.bareMethodName) { "SubscribeHead" -> processSubscribeHead(call, headers, next) - "SubscribeBalance" -> processSubscribeBalance(call, headers, next) + "SubscribeBalance" -> processSubscribeBalance(call, headers, next, true) + "GetBalance" -> processSubscribeBalance(call, headers, next, false) "NativeCall" -> processNativeCall(call, headers, next) else -> { log.trace("unsupported method `{}`", method) @@ -67,9 +68,10 @@ class AccessHandler( private fun processSubscribeBalance( call: ServerCall, headers: Metadata, - next: ServerCallHandler + next: ServerCallHandler, + subscribe: Boolean ): ServerCall.Listener { - val builder = EventsBuilder.SubscribeBalance() + val builder = EventsBuilder.SubscribeBalance(subscribe) .start(headers, call.attributes) val callWrapper: ServerCall = OnSubscribeBalanceResponse( call as ServerCall, builder, accessLogWriter) as ServerCall diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index 13e211cb..40daaf83 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -48,14 +48,14 @@ class Events { @JsonInclude(JsonInclude.Include.NON_NULL) class SubscribeBalance( - blockchain: Chain, id: UUID, + 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, "SubscribeBalance", id) + ) : ChainBase(blockchain, if (subscribe) "SubscribeBalance" else "GetBalance", id) @JsonInclude(JsonInclude.Include.NON_NULL) class NativeCall( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt index 093b5347..b0d7d309 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt @@ -132,7 +132,7 @@ class EventsBuilder { } } - class SubscribeBalance() : Base() { + class SubscribeBalance(val subscribe: Boolean) : Base() { private var index = 0 private var balanceRequest: Events.BalanceRequest? = null @@ -155,7 +155,7 @@ class EventsBuilder { val addressBalance = Events.AddressBalance(resp.asset.code, resp.address.address) val chain = Chain.byId(resp.asset.chain.number) return Events.SubscribeBalance( - chain, UUID.randomUUID(), requestDetails, balanceRequest!!, addressBalance, index++ + chain, UUID.randomUUID(), subscribe, requestDetails, balanceRequest!!, addressBalance, index++ ) } } From 50a7177e6a81f817f23074b6295dcaddc786f35c Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sat, 3 Jul 2021 22:32:46 -0400 Subject: [PATCH 07/16] solution: access logging for SubscribeTxStatus method --- .../monitoring/accesslog/AccessHandler.kt | 45 +++++++++++++++++++ .../dshackle/monitoring/accesslog/Events.kt | 18 ++++++++ .../monitoring/accesslog/EventsBuilder.kt | 23 ++++++++++ 3 files changed, 86 insertions(+) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt index 19df1a9e..5f248f28 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt @@ -39,6 +39,7 @@ class AccessHandler( 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) else -> { @@ -81,6 +82,22 @@ class AccessHandler( ) as ServerCall.Listener } + @Suppress("UNCHECKED_CAST") + private fun processSubscribeTxStatus( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + val builder = EventsBuilder.TxStatus() + .start(headers, call.attributes) + val callWrapper: ServerCall = OnTxStatusResponse( + call as ServerCall, builder, accessLogWriter) as ServerCall + return OnSubscribeTxStatus( + next.startCall(callWrapper, headers) as ServerCall.Listener, + builder + ) as ServerCall.Listener + } + @Suppress("UNCHECKED_CAST") private fun processNativeCall( call: ServerCall, @@ -131,6 +148,21 @@ class AccessHandler( } } + class OnSubscribeTxStatus( + val next: ServerCall.Listener, + val builder: EventsBuilder.TxStatus + ) : ForwardingServerCallListener() { + + override fun onMessage(message: BlockchainOuterClass.TxStatusRequest) { + builder.withRequest(message) + super.onMessage(message) + } + + override fun delegate(): ServerCall.Listener { + return next + } + } + class OnNativeCall( val next: ServerCall.Listener, val builder: EventsBuilder.NativeCall, @@ -213,4 +245,17 @@ class AccessHandler( super.sendMessage(message) } } + + class OnTxStatusResponse( + next: ServerCall, + val builder: EventsBuilder.TxStatus, + val accessLogWriter: AccessLogWriter + ) : BaseCallResponse(next) { + + override fun sendMessage(message: BlockchainOuterClass.TxStatus) { + val event = builder.onReply(message) + accessLogWriter.submit(event) + super.sendMessage(message) + } + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index 40daaf83..dd757c25 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -57,6 +57,24 @@ class Events { 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, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt index b0d7d309..e3a88991 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt @@ -160,6 +160,29 @@ class EventsBuilder { } } + class TxStatus() : Base() { + private var index = 0 + private var txStatusRequest: Events.TxStatusRequest? = null + + fun withRequest(req: BlockchainOuterClass.TxStatusRequest): TxStatus { + this.txStatusRequest = Events.TxStatusRequest(req.txId) + return withChain(req.chainValue) + } + + fun onReply(resp: BlockchainOuterClass.TxStatus): Events.TxStatus { + return Events.TxStatus( + chain, UUID.randomUUID(), requestDetails, txStatusRequest!!, + Events.TxStatusResponse(resp.confirmations), + index++ + ) + } + + override fun getT(): TxStatus { + return this + } + + } + class NativeCall : Base() { val items = ArrayList() val replies = HashMap() From ba1e93a9656741eb0ca3563ea2a693e1b3df5636 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 4 Jul 2021 17:33:38 -0400 Subject: [PATCH 08/16] solution: access logging for Describe method --- .../monitoring/accesslog/AccessHandler.kt | 47 +++++++++++++++++++ .../dshackle/monitoring/accesslog/Events.kt | 13 +++-- .../monitoring/accesslog/EventsBuilder.kt | 15 ++++++ 3 files changed, 72 insertions(+), 3 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt index 5f248f28..0bbde482 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt @@ -42,6 +42,7 @@ class AccessHandler( "SubscribeTxStatus" -> processSubscribeTxStatus(call, headers, next) "GetBalance" -> processSubscribeBalance(call, headers, next, false) "NativeCall" -> processNativeCall(call, headers, next) + "Describe" -> processDescribe(call, headers, next) else -> { log.trace("unsupported method `{}`", method) next.startCall(call, headers) @@ -117,6 +118,25 @@ class AccessHandler( } as ServerCall.Listener } + @Suppress("UNCHECKED_CAST") + private fun processDescribe( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + val builder = EventsBuilder.Describe() + .start(headers, call.attributes) + + val callWrapper: ServerCall = OnDescribeResponse( + call as ServerCall, + builder, + accessLogWriter + ) as ServerCall + return OnDescribeRequest( + next.startCall(callWrapper, headers) as ServerCall.Listener, + builder) as ServerCall.Listener + } + class OnSubscribeHead( val next: ServerCall.Listener, val builder: EventsBuilder.SubscribeHead @@ -193,6 +213,20 @@ class AccessHandler( } } + class OnDescribeRequest( + val next: ServerCall.Listener, + val builder: EventsBuilder.Describe + ) : ForwardingServerCallListener() { + + override fun onMessage(message: BlockchainOuterClass.DescribeRequest) { + super.onMessage(message) + } + + override fun delegate(): ServerCall.Listener { + return next + } + } + abstract class BaseCallResponse( val next: ServerCall ) : ForwardingServerCall() { @@ -258,4 +292,17 @@ class AccessHandler( super.sendMessage(message) } } + + class OnDescribeResponse( + next: ServerCall, + val builder: EventsBuilder.Describe, + val accessLogWriter: AccessLogWriter + ) : BaseCallResponse(next) { + + override fun sendMessage(message: BlockchainOuterClass.DescribeResponse) { + val event = builder.onReply() + accessLogWriter.submit(event) + super.sendMessage(message) + } + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index dd757c25..5a38ae05 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -28,14 +28,15 @@ class Events { } abstract class Base( - val id: UUID + val id: UUID, + val method: String ) { val ts = Instant.now() } abstract class ChainBase( - val blockchain: Chain, val method: String, id: UUID - ) : Base(id) + val blockchain: Chain, method: String, id: UUID + ) : Base(id, method) @JsonInclude(JsonInclude.Include.NON_NULL) class SubscribeHead( @@ -95,6 +96,12 @@ class Events { val nativeCall: NativeCallItemDetails ) : ChainBase(blockchain, "NativeCall", id) + @JsonInclude(JsonInclude.Include.NON_NULL) + class Describe( + id: UUID, + val request: StreamRequestDetails + ) : Base(id, "Describe") + data class StreamRequestDetails( val id: UUID, val start: Instant, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt index e3a88991..d29218ee 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt @@ -227,4 +227,19 @@ class EventsBuilder { } } } + + class Describe : Base() { + + override fun getT(): Describe { + return this + } + + fun onReply(): Events.Describe { + return Events.Describe( + id = UUID.randomUUID(), + request = requestDetails + ) + } + } + } \ No newline at end of file From c27217a4ffaf6338e6bb015fd04f75665984ec1f Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 4 Jul 2021 18:36:31 -0400 Subject: [PATCH 09/16] problem: failing unit tests rel: [f2bc0f1d9b] --- .../accesslog/EventsBuilderSubscribeBalanceSpec.groovy | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy index c445c590..eec89d59 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy @@ -32,7 +32,7 @@ class EventsBuilderSubscribeBalanceSpec extends Specification { .setBalance("1234560000000000000") .build() when: - def act = new EventsBuilder.SubscribeBalance() + def act = new EventsBuilder.SubscribeBalance(true) .withRequest(request) .onReply(resp) then: @@ -69,7 +69,7 @@ class EventsBuilderSubscribeBalanceSpec extends Specification { .setBalance("12345600000000") .build() when: - def act = new EventsBuilder.SubscribeBalance() + def act = new EventsBuilder.SubscribeBalance(true) .withRequest(request) .onReply(resp) then: From 3ef5a4455ce3eede5b4fe22bb01abf238408b151 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 4 Jul 2021 18:37:17 -0400 Subject: [PATCH 10/16] problem: doesn't publish upstream status updates --- .../io/emeraldpay/dshackle/upstream/DefaultUpstream.kt | 10 ++++++++-- .../io/emeraldpay/dshackle/upstream/Multistream.kt | 7 ++++++- 2 files changed, 14 insertions(+), 3 deletions(-) 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 } From 6cb703ec4e84b63c3e45492d7d4893a0904f5c4c Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 4 Jul 2021 18:38:00 -0400 Subject: [PATCH 11/16] solution: access logging for SubscribeStatus method --- .../monitoring/accesslog/AccessHandler.kt | 49 ++++++++++++++++++- .../dshackle/monitoring/accesslog/Events.kt | 6 +++ .../monitoring/accesslog/EventsBuilder.kt | 15 ++++++ 3 files changed, 69 insertions(+), 1 deletion(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt index 0bbde482..210623b7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt @@ -43,8 +43,9 @@ class AccessHandler( "GetBalance" -> processSubscribeBalance(call, headers, next, false) "NativeCall" -> processNativeCall(call, headers, next) "Describe" -> processDescribe(call, headers, next) + "SubscribeStatus" -> processStatus(call, headers, next) else -> { - log.trace("unsupported method `{}`", method) + log.warn("unsupported method `{}`", method) next.startCall(call, headers) } } @@ -137,6 +138,25 @@ class AccessHandler( builder) as ServerCall.Listener } + @Suppress("UNCHECKED_CAST") + private fun processStatus( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + val builder = EventsBuilder.Status() + .start(headers, call.attributes) + + val callWrapper: ServerCall = OnStatusResponse( + call as ServerCall, + builder, + accessLogWriter + ) as ServerCall + return OnStatusRequest( + next.startCall(callWrapper, headers) as ServerCall.Listener, + builder) as ServerCall.Listener + } + class OnSubscribeHead( val next: ServerCall.Listener, val builder: EventsBuilder.SubscribeHead @@ -227,6 +247,20 @@ class AccessHandler( } } + class OnStatusRequest( + val next: ServerCall.Listener, + val builder: EventsBuilder.Status + ) : ForwardingServerCallListener() { + + override fun onMessage(message: BlockchainOuterClass.StatusRequest) { + super.onMessage(message) + } + + override fun delegate(): ServerCall.Listener { + return next + } + } + abstract class BaseCallResponse( val next: ServerCall ) : ForwardingServerCall() { @@ -305,4 +339,17 @@ class AccessHandler( super.sendMessage(message) } } + + class OnStatusResponse( + next: ServerCall, + val builder: EventsBuilder.Status, + val accessLogWriter: AccessLogWriter + ) : BaseCallResponse(next) { + + override fun sendMessage(message: BlockchainOuterClass.ChainStatus) { + val event = builder.onReply(message) + accessLogWriter.submit(event) + super.sendMessage(message) + } + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index 5a38ae05..a0f11ab6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -102,6 +102,12 @@ class Events { 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, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt index d29218ee..520fe06f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt @@ -242,4 +242,19 @@ class EventsBuilder { } } + class Status : Base() { + override fun getT(): Status { + return this + } + + fun onReply(message: BlockchainOuterClass.ChainStatus): Events.Status { + val chain = Chain.byId(message.chainValue) + return Events.Status( + blockchain = chain, + request = requestDetails, + id = UUID.randomUUID() + ) + } + } + } \ No newline at end of file From 0d144a44a9b19e27a215e021c46038a6d922c6ce Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 5 Jul 2021 21:51:15 -0400 Subject: [PATCH 12/16] problem: NativeCall replies are logged only on complete solution: add to the log at the moment of reply --- .../monitoring/accesslog/AccessHandler.kt | 31 ++++----- .../monitoring/accesslog/EventsBuilder.kt | 35 ++++------ .../accesslog/EventsBaseBuilderSpec.groovy | 68 ++++++++----------- 3 files changed, 54 insertions(+), 80 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt index 210623b7..47efb4be 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt @@ -110,13 +110,14 @@ class AccessHandler( .start(headers, call.attributes) val callWrapper: ServerCall = OnNativeCallResponse( - call as ServerCall, builder + call as ServerCall, + builder, + accessLogWriter ) as ServerCall return OnNativeCall( next.startCall(callWrapper, headers) as ServerCall.Listener, - builder) { logs -> - accessLogWriter.submit(logs) - } as ServerCall.Listener + builder + ) as ServerCall.Listener } @Suppress("UNCHECKED_CAST") @@ -205,29 +206,18 @@ class AccessHandler( class OnNativeCall( val next: ServerCall.Listener, - val builder: EventsBuilder.NativeCall, - val done: (List) -> Unit + val builder: EventsBuilder.NativeCall ) : ForwardingServerCallListener() { override fun onMessage(message: BlockchainOuterClass.NativeCallRequest) { val chain = message.chain builder.withChain(chain.number) message.itemsList.forEach { item -> - builder.onItem(item) + builder.onRequest(item) } super.onMessage(message) } - override fun onCancel() { - super.onCancel() - done(builder.build()) - } - - override fun onComplete() { - super.onComplete() - done(builder.build()) - } - override fun delegate(): ServerCall.Listener { return next } @@ -279,11 +269,14 @@ class AccessHandler( class OnNativeCallResponse( next: ServerCall, - val builder: EventsBuilder.NativeCall + val builder: EventsBuilder.NativeCall, + val accessLogWriter: AccessLogWriter ) : BaseCallResponse(next) { override fun sendMessage(message: BlockchainOuterClass.NativeCallReplyItem) { - builder.onItemReply(message) + accessLogWriter.submit( + builder.onReply(message) + ) super.sendMessage(message) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt index 520fe06f..8428c298 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt @@ -186,12 +186,13 @@ class EventsBuilder { class NativeCall : Base() { val items = ArrayList() val replies = HashMap() + private var index = 0 override fun getT(): NativeCall { return this } - fun onItem(item: BlockchainOuterClass.NativeCallItem): NativeCall { + fun onRequest(item: BlockchainOuterClass.NativeCallItem): NativeCall { this.items.add( Events.NativeCallItemDetails( item.method, @@ -202,30 +203,20 @@ class EventsBuilder { return this } - fun onItemReply(reply: BlockchainOuterClass.NativeCallReplyItem): NativeCall { - this.replies[reply.id] = Events.NativeCallReplyDetails( - reply.id, - reply.succeed, - reply.payload?.size()?.toLong() ?: 0L + fun onReply(reply: BlockchainOuterClass.NativeCallReplyItem): Events.NativeCall { + val item = items.find { it.id == reply.id }!! + return Events.NativeCall( + request = requestDetails, + total = items.size, + index = index++, + succeed = reply.succeed, + blockchain = chain, + nativeCall = item, + payloadSizeBytes = item.payloadSizeBytes, + id = UUID.randomUUID() ) - return this } - fun build(): List { - return items.mapIndexed { index, item -> - val reply = replies[item.id] - Events.NativeCall( - request = requestDetails, - total = items.size, - index = index, - succeed = reply?.succeed ?: false, - blockchain = chain, - nativeCall = item, - payloadSizeBytes = item.payloadSizeBytes, - id = UUID.randomUUID() - ) - } - } } class Describe : Base() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy index b72ec97e..49c1fce8 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy @@ -35,18 +35,15 @@ class EventsBaseBuilderSpec extends Specification { def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) - .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) - .build() + .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: - act.size() == 1 - with(act[0]) { - it.request != null - it.request.remote != null - with(it.request.remote) { - ips == ["127.0.0.1"] - userAgent == "grpc-go/1.30.0" - ip == "127.0.0.1" - } + 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" } } @@ -62,11 +59,10 @@ class EventsBaseBuilderSpec extends Specification { def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) - .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) - .build() + .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: - act.size() == 1 - with(act[0].request.remote) { + with(act.request.remote) { ips == ["30.56.100.15", "127.0.0.1"] ip == "30.56.100.15" } @@ -84,11 +80,10 @@ class EventsBaseBuilderSpec extends Specification { def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) - .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) - .build() + .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: - act.size() == 1 - with(act[0].request.remote) { + with(act.request.remote) { ips == ["192.168.1.1", "30.56.100.15"] userAgent == "grpc-go/1.30.0" ip == "30.56.100.15" @@ -107,11 +102,10 @@ class EventsBaseBuilderSpec extends Specification { def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) - .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) - .build() + .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: - act.size() == 1 - with(act[0].request.remote) { + with(act.request.remote) { ips == ["30.56.100.15"] userAgent == "grpc-go/1.30.0" ip == "30.56.100.15" @@ -130,11 +124,10 @@ class EventsBaseBuilderSpec extends Specification { def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) - .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) - .build() + .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: - act.size() == 1 - with(act[0].request.remote) { + with(act.request.remote) { ips == ["30.56.100.15"] userAgent == "grpc-go/1.30.0" ip == "30.56.100.15" @@ -153,11 +146,10 @@ class EventsBaseBuilderSpec extends Specification { def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) - .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) - .build() + .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: - act.size() == 1 - with(act[0].request.remote) { + 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" } @@ -174,11 +166,10 @@ class EventsBaseBuilderSpec extends Specification { def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) - .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) - .build() + .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: - act.size() == 1 - with(act[0].request.remote) { + with(act.request.remote) { userAgent == "grpc-go/1.30.0 xss" } } @@ -195,11 +186,10 @@ class EventsBaseBuilderSpec extends Specification { def act = new EventsBuilder.NativeCall() .start(metadata, attributes) .withChain(Chain.ETHEREUM.id) - .onItem(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) - .build() + .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: - act.size() == 1 - with(act[0].request.remote) { + 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_" } From cd9087a4255398a07e143170b94f50d9d0aaf68d Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 6 Jul 2021 14:47:42 -0400 Subject: [PATCH 13/16] solution: refactor AccessLog handler --- .../monitoring/accesslog/AccessHandler.kt | 268 +++--------------- .../monitoring/accesslog/EventsBuilder.kt | 110 ++++--- .../accesslog/EventsBaseBuilderSpec.groovy | 102 ++++--- .../EventsBuilderSubscribeBalanceSpec.groovy | 12 +- 4 files changed, 195 insertions(+), 297 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt index 47efb4be..1685eba3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandler.kt @@ -51,20 +51,31 @@ class AccessHandler( } } + 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 { - val builder = EventsBuilder.SubscribeHead() - .start(headers, call.attributes) - val callWrapper: ServerCall = OnSubscribeHeadResponse( - call as ServerCall, builder, accessLogWriter) as ServerCall - return OnSubscribeHead( - next.startCall(callWrapper, headers) as ServerCall.Listener, - builder - ) as ServerCall.Listener + return process(call, headers, next, + EventsBuilder.SubscribeHead() as EventsBuilder.RequestReply<*, ReqT, RespT> + ) } @Suppress("UNCHECKED_CAST") @@ -74,14 +85,9 @@ class AccessHandler( next: ServerCallHandler, subscribe: Boolean ): ServerCall.Listener { - val builder = EventsBuilder.SubscribeBalance(subscribe) - .start(headers, call.attributes) - val callWrapper: ServerCall = OnSubscribeBalanceResponse( - call as ServerCall, builder, accessLogWriter) as ServerCall - return OnSubscribeBalance( - next.startCall(callWrapper, headers) as ServerCall.Listener, - builder - ) as ServerCall.Listener + return process(call, headers, next, + EventsBuilder.SubscribeBalance(subscribe) as EventsBuilder.RequestReply<*, ReqT, RespT> + ) } @Suppress("UNCHECKED_CAST") @@ -90,14 +96,9 @@ class AccessHandler( headers: Metadata, next: ServerCallHandler ): ServerCall.Listener { - val builder = EventsBuilder.TxStatus() - .start(headers, call.attributes) - val callWrapper: ServerCall = OnTxStatusResponse( - call as ServerCall, builder, accessLogWriter) as ServerCall - return OnSubscribeTxStatus( - next.startCall(callWrapper, headers) as ServerCall.Listener, - builder - ) as ServerCall.Listener + return process(call, headers, next, + EventsBuilder.TxStatus() as EventsBuilder.RequestReply<*, ReqT, RespT> + ) } @Suppress("UNCHECKED_CAST") @@ -106,18 +107,9 @@ class AccessHandler( headers: Metadata, next: ServerCallHandler ): ServerCall.Listener { - val builder = EventsBuilder.NativeCall() - .start(headers, call.attributes) - - val callWrapper: ServerCall = OnNativeCallResponse( - call as ServerCall, - builder, - accessLogWriter - ) as ServerCall - return OnNativeCall( - next.startCall(callWrapper, headers) as ServerCall.Listener, - builder - ) as ServerCall.Listener + return process(call, headers, next, + EventsBuilder.NativeCall() as EventsBuilder.RequestReply<*, ReqT, RespT> + ) } @Suppress("UNCHECKED_CAST") @@ -126,17 +118,9 @@ class AccessHandler( headers: Metadata, next: ServerCallHandler ): ServerCall.Listener { - val builder = EventsBuilder.Describe() - .start(headers, call.attributes) - - val callWrapper: ServerCall = OnDescribeResponse( - call as ServerCall, - builder, - accessLogWriter - ) as ServerCall - return OnDescribeRequest( - next.startCall(callWrapper, headers) as ServerCall.Listener, - builder) as ServerCall.Listener + return process(call, headers, next, + EventsBuilder.Describe() as EventsBuilder.RequestReply<*, ReqT, RespT> + ) } @Suppress("UNCHECKED_CAST") @@ -145,115 +129,32 @@ class AccessHandler( headers: Metadata, next: ServerCallHandler ): ServerCall.Listener { - val builder = EventsBuilder.Status() - .start(headers, call.attributes) - - val callWrapper: ServerCall = OnStatusResponse( - call as ServerCall, - builder, - accessLogWriter - ) as ServerCall - return OnStatusRequest( - next.startCall(callWrapper, headers) as ServerCall.Listener, - builder) as ServerCall.Listener + return process(call, headers, next, + EventsBuilder.Status() as EventsBuilder.RequestReply<*, ReqT, RespT> + ) } - class OnSubscribeHead( - val next: ServerCall.Listener, - val builder: EventsBuilder.SubscribeHead - ) : ForwardingServerCallListener() { + open class StdCallListener>( + val next: ServerCall.Listener, + val builder: EB + ) : ForwardingServerCallListener() { - override fun onMessage(message: Common.Chain) { - val chainId = message.type.number - builder.withChain(chainId) + override fun onMessage(message: Req) { + builder.onRequest(message) super.onMessage(message) } - override fun delegate(): ServerCall.Listener { + override fun delegate(): ServerCall.Listener { return next } } - class OnSubscribeBalance( - val next: ServerCall.Listener, - val builder: EventsBuilder.SubscribeBalance - ) : ForwardingServerCallListener() { - - override fun onMessage(message: BlockchainOuterClass.BalanceRequest) { - builder.withRequest(message) - super.onMessage(message) - } - - override fun delegate(): ServerCall.Listener { - return next - } - } - - class OnSubscribeTxStatus( - val next: ServerCall.Listener, - val builder: EventsBuilder.TxStatus - ) : ForwardingServerCallListener() { - - override fun onMessage(message: BlockchainOuterClass.TxStatusRequest) { - builder.withRequest(message) - super.onMessage(message) - } - - override fun delegate(): ServerCall.Listener { - return next - } - } - - class OnNativeCall( - val next: ServerCall.Listener, - val builder: EventsBuilder.NativeCall - ) : ForwardingServerCallListener() { - - override fun onMessage(message: BlockchainOuterClass.NativeCallRequest) { - val chain = message.chain - builder.withChain(chain.number) - message.itemsList.forEach { item -> - builder.onRequest(item) - } - super.onMessage(message) - } - - override fun delegate(): ServerCall.Listener { - return next - } - } - - class OnDescribeRequest( - val next: ServerCall.Listener, - val builder: EventsBuilder.Describe - ) : ForwardingServerCallListener() { - - override fun onMessage(message: BlockchainOuterClass.DescribeRequest) { - super.onMessage(message) - } - - override fun delegate(): ServerCall.Listener { - return next - } - } - - class OnStatusRequest( - val next: ServerCall.Listener, - val builder: EventsBuilder.Status - ) : ForwardingServerCallListener() { - - override fun onMessage(message: BlockchainOuterClass.StatusRequest) { - super.onMessage(message) - } - - override fun delegate(): ServerCall.Listener { - return next - } - } - - abstract class BaseCallResponse( - val next: ServerCall + open class StdCallResponse>( + val next: ServerCall, + val builder: EB, + val accessLogWriter: AccessLogWriter ) : ForwardingServerCall() { + override fun getMethodDescriptor(): MethodDescriptor { return next.methodDescriptor } @@ -264,85 +165,10 @@ class AccessHandler( override fun sendMessage(message: RespT) { super.sendMessage(message) - } - } - - class OnNativeCallResponse( - next: ServerCall, - val builder: EventsBuilder.NativeCall, - val accessLogWriter: AccessLogWriter - ) : BaseCallResponse(next) { - - override fun sendMessage(message: BlockchainOuterClass.NativeCallReplyItem) { accessLogWriter.submit( - builder.onReply(message) + builder.onReply(message)!! ) - super.sendMessage(message) } } - class OnSubscribeHeadResponse( - next: ServerCall, - val builder: EventsBuilder.SubscribeHead, - val accessLogWriter: AccessLogWriter - ) : BaseCallResponse(next) { - - override fun sendMessage(message: BlockchainOuterClass.ChainHead) { - val event = builder.onReply(message) - accessLogWriter.submit(event) - super.sendMessage(message) - } - } - - class OnSubscribeBalanceResponse( - next: ServerCall, - val builder: EventsBuilder.SubscribeBalance, - val accessLogWriter: AccessLogWriter - ) : BaseCallResponse(next) { - - override fun sendMessage(message: BlockchainOuterClass.AddressBalance) { - val event = builder.onReply(message) - accessLogWriter.submit(event) - super.sendMessage(message) - } - } - - class OnTxStatusResponse( - next: ServerCall, - val builder: EventsBuilder.TxStatus, - val accessLogWriter: AccessLogWriter - ) : BaseCallResponse(next) { - - override fun sendMessage(message: BlockchainOuterClass.TxStatus) { - val event = builder.onReply(message) - accessLogWriter.submit(event) - super.sendMessage(message) - } - } - - class OnDescribeResponse( - next: ServerCall, - val builder: EventsBuilder.Describe, - val accessLogWriter: AccessLogWriter - ) : BaseCallResponse(next) { - - override fun sendMessage(message: BlockchainOuterClass.DescribeResponse) { - val event = builder.onReply() - accessLogWriter.submit(event) - super.sendMessage(message) - } - } - - class OnStatusResponse( - next: ServerCall, - val builder: EventsBuilder.Status, - val accessLogWriter: AccessLogWriter - ) : BaseCallResponse(next) { - - override fun sendMessage(message: BlockchainOuterClass.ChainStatus) { - val event = builder.onReply(message) - accessLogWriter.submit(event) - super.sendMessage(message) - } - } } \ 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 index 8428c298..e8a88fe1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt @@ -16,6 +16,7 @@ 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 @@ -33,7 +34,16 @@ class EventsBuilder { private val log = LoggerFactory.getLogger(EventsBuilder::class.java) } - abstract class Base() { + 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), @@ -82,9 +92,9 @@ class EventsBuilder { .trim() } - abstract protected fun getT(): T + protected abstract fun getT(): T - fun start(metadata: Metadata, attributes: Attributes): 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) ?: "" @@ -108,7 +118,6 @@ class EventsBuilder { ip = ip, userAgent = userAgent )) - return getT() } fun withChain(chain: Int): T { @@ -118,21 +127,31 @@ class EventsBuilder { } } - class SubscribeHead() : Base() { + class SubscribeHead() : + Base(), + RequestReply { + private var index = 0 override fun getT(): SubscribeHead { return this } - fun onReply(resp: BlockchainOuterClass.ChainHead): Events.SubscribeHead { + 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() { + class SubscribeBalance(val subscribe: Boolean) : + Base(), + RequestReply { + private var index = 0 private var balanceRequest: Events.BalanceRequest? = null @@ -140,39 +159,40 @@ class EventsBuilder { return this } - fun withRequest(req: BlockchainOuterClass.BalanceRequest): SubscribeBalance { + override fun onRequest(msg: BlockchainOuterClass.BalanceRequest) { balanceRequest = Events.BalanceRequest( - req.asset.code.toUpperCase(), - req.address.addrTypeCase.name + msg.asset.code.toUpperCase(), + msg.address.addrTypeCase.name ) - return this } - fun onReply(resp: BlockchainOuterClass.AddressBalance): Events.SubscribeBalance { + override fun onReply(msg: BlockchainOuterClass.AddressBalance): Events.SubscribeBalance { if (balanceRequest == null) { throw IllegalStateException("Request is not initialized") } - val addressBalance = Events.AddressBalance(resp.asset.code, resp.address.address) - val chain = Chain.byId(resp.asset.chain.number) + 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() { + class TxStatus() : + Base(), + RequestReply { private var index = 0 private var txStatusRequest: Events.TxStatusRequest? = null - fun withRequest(req: BlockchainOuterClass.TxStatusRequest): TxStatus { - this.txStatusRequest = Events.TxStatusRequest(req.txId) - return withChain(req.chainValue) + override fun onRequest(msg: BlockchainOuterClass.TxStatusRequest) { + this.txStatusRequest = Events.TxStatusRequest(msg.txId) + withChain(msg.chainValue) } - fun onReply(resp: BlockchainOuterClass.TxStatus): Events.TxStatus { + override fun onReply(msg: BlockchainOuterClass.TxStatus): Events.TxStatus { return Events.TxStatus( chain, UUID.randomUUID(), requestDetails, txStatusRequest!!, - Events.TxStatusResponse(resp.confirmations), + Events.TxStatusResponse(msg.confirmations), index++ ) } @@ -183,7 +203,9 @@ class EventsBuilder { } - class NativeCall : Base() { + class NativeCall : + Base(), + RequestReply { val items = ArrayList() val replies = HashMap() private var index = 0 @@ -192,24 +214,26 @@ class EventsBuilder { return this } - fun onRequest(item: BlockchainOuterClass.NativeCallItem): NativeCall { - this.items.add( - Events.NativeCallItemDetails( - item.method, - item.id, - item.payload.size().toLong() - ) - ) - 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() + ) + ) + } } - fun onReply(reply: BlockchainOuterClass.NativeCallReplyItem): Events.NativeCall { - val item = items.find { it.id == reply.id }!! + 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 = reply.succeed, + succeed = msg.succeed, blockchain = chain, nativeCall = item, payloadSizeBytes = item.payloadSizeBytes, @@ -219,13 +243,18 @@ class EventsBuilder { } - class Describe : Base() { + class Describe : + Base(), + RequestReply { override fun getT(): Describe { return this } - fun onReply(): Events.Describe { + override fun onRequest(msg: BlockchainOuterClass.DescribeRequest) { + } + + override fun onReply(msg: BlockchainOuterClass.DescribeResponse): Events.Describe { return Events.Describe( id = UUID.randomUUID(), request = requestDetails @@ -233,13 +262,18 @@ class EventsBuilder { } } - class Status : Base() { + class Status : + Base(), + RequestReply { override fun getT(): Status { return this } - fun onReply(message: BlockchainOuterClass.ChainStatus): Events.Status { - val chain = Chain.byId(message.chainValue) + 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, diff --git a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy index 49c1fce8..ede6ab9d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBaseBuilderSpec.groovy @@ -20,10 +20,40 @@ 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() @@ -32,10 +62,11 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("127.0.0.1"), 2448)) .build() when: - def act = new EventsBuilder.NativeCall() - .start(metadata, attributes) - .withChain(Chain.ETHEREUM.id) - .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + def act = new TestEventBuilder() + .tap { + it.start(metadata, attributes) + it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance()) + } .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: act.request != null @@ -56,10 +87,11 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("127.0.0.1"), 2448)) .build() when: - def act = new EventsBuilder.NativeCall() - .start(metadata, attributes) - .withChain(Chain.ETHEREUM.id) - .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + def act = new TestEventBuilder() + .tap { + it.start(metadata, attributes) + it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance()) + } .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: with(act.request.remote) { @@ -77,10 +109,11 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448)) .build() when: - def act = new EventsBuilder.NativeCall() - .start(metadata, attributes) - .withChain(Chain.ETHEREUM.id) - .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + def act = new TestEventBuilder() + .tap { + it.start(metadata, attributes) + it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance()) + } .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: with(act.request.remote) { @@ -99,10 +132,11 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448)) .build() when: - def act = new EventsBuilder.NativeCall() - .start(metadata, attributes) - .withChain(Chain.ETHEREUM.id) - .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + def act = new TestEventBuilder() + .tap { + it.start(metadata, attributes) + it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance()) + } .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: with(act.request.remote) { @@ -121,10 +155,11 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448)) .build() when: - def act = new EventsBuilder.NativeCall() - .start(metadata, attributes) - .withChain(Chain.ETHEREUM.id) - .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + def act = new TestEventBuilder() + .tap { + it.start(metadata, attributes) + it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance()) + } .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: with(act.request.remote) { @@ -143,10 +178,11 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet6Address.getByName("::1"), 2448)) .build() when: - def act = new EventsBuilder.NativeCall() - .start(metadata, attributes) - .withChain(Chain.ETHEREUM.id) - .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + def act = new TestEventBuilder() + .tap { + it.start(metadata, attributes) + it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance()) + } .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: with(act.request.remote) { @@ -163,10 +199,11 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448)) .build() when: - def act = new EventsBuilder.NativeCall() - .start(metadata, attributes) - .withChain(Chain.ETHEREUM.id) - .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + def act = new TestEventBuilder() + .tap { + it.start(metadata, attributes) + it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance()) + } .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: with(act.request.remote) { @@ -183,10 +220,11 @@ class EventsBaseBuilderSpec extends Specification { .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InetSocketAddress(Inet4Address.getByName("30.56.100.15"), 2448)) .build() when: - def act = new EventsBuilder.NativeCall() - .start(metadata, attributes) - .withChain(Chain.ETHEREUM.id) - .onRequest(BlockchainOuterClass.NativeCallItem.getDefaultInstance()) + def act = new TestEventBuilder() + .tap { + it.start(metadata, attributes) + it.onRequest(BlockchainOuterClass.NativeCallRequest.getDefaultInstance()) + } .onReply(BlockchainOuterClass.NativeCallReplyItem.getDefaultInstance()) then: with(act.request.remote) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy index eec89d59..75d78f5a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilderSubscribeBalanceSpec.groovy @@ -32,9 +32,9 @@ class EventsBuilderSubscribeBalanceSpec extends Specification { .setBalance("1234560000000000000") .build() when: - def act = new EventsBuilder.SubscribeBalance(true) - .withRequest(request) - .onReply(resp) + def act = new EventsBuilder.SubscribeBalance(true).tap { + it.onRequest(request) + }.onReply(resp) then: act.index == 0 act.blockchain == Chain.ETHEREUM @@ -69,9 +69,9 @@ class EventsBuilderSubscribeBalanceSpec extends Specification { .setBalance("12345600000000") .build() when: - def act = new EventsBuilder.SubscribeBalance(true) - .withRequest(request) - .onReply(resp) + def act = new EventsBuilder.SubscribeBalance(true).tap { + it.onRequest(request) + }.onReply(resp) then: act.index == 0 act.blockchain == Chain.BITCOIN From 9e49b0e13d0fe24605b421670e5d7b3e1f5b5e32 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 19 Jul 2021 21:58:17 -0400 Subject: [PATCH 14/16] solution: access log api version --- .../kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt | 1 + 1 file changed, 1 insertion(+) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index a0f11ab6..96c255a3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -31,6 +31,7 @@ class Events { val id: UUID, val method: String ) { + val version = "accesslog/v1beta" val ts = Instant.now() } From 3a12efd0a0ac9a4ffdea99948c4e3971fbd9b9c4 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 19 Jul 2021 22:19:48 -0400 Subject: [PATCH 15/16] solution: test accesslog writer --- .../accesslog/AccessLogWriterSpec.groovy | 53 +++++++++++++++++++ 1 file changed, 53 insertions(+) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriterSpec.groovy 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" + } + } +} From e8fc025966c704e49d63d8c9f134d9aece1462e1 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 19 Jul 2021 23:15:31 -0400 Subject: [PATCH 16/16] solution: docs for access logging --- docs/01-architecture-intro.adoc | 4 +- docs/02-quick-start.adoc | 2 +- docs/03-server-config.adoc | 6 +- docs/06-monitoring.adoc | 74 +++++++++++++++++++ docs/{06-methods.adoc => 07-methods.adoc} | 2 +- ...entication.adoc => 08-authentication.adoc} | 0 ...tors.adoc => 09-quorum-and-selectors.adoc} | 0 docs/{09-caching.adoc => 10-caching.adoc} | 0 ...ibraries.adoc => 11-client-libraries.adoc} | 0 docs/README.adoc | 13 ++-- docs/reference-configuration.adoc | 65 ++++++++++++---- 11 files changed, 141 insertions(+), 25 deletions(-) create mode 100644 docs/06-monitoring.adoc rename docs/{06-methods.adoc => 07-methods.adoc} (98%) rename docs/{07-authentication.adoc => 08-authentication.adoc} (100%) rename docs/{08-quorum-and-selectors.adoc => 09-quorum-and-selectors.adoc} (100%) rename docs/{09-caching.adoc => 10-caching.adoc} (100%) rename docs/{10-client-libraries.adoc => 11-client-libraries.adoc} (100%) 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