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: