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