simplified zipkin with ssl
This commit is contained in:
8
.gitignore
vendored
8
.gitignore
vendored
@@ -13,3 +13,11 @@ testsetup/
|
|||||||
# nix
|
# nix
|
||||||
env
|
env
|
||||||
/dshackle-cli/node_modules/
|
/dshackle-cli/node_modules/
|
||||||
|
|
||||||
|
# temp files
|
||||||
|
/test*
|
||||||
|
/test/
|
||||||
|
Test*.kt
|
||||||
|
|
||||||
|
# http-client config
|
||||||
|
http-client.env.json
|
||||||
|
|||||||
@@ -94,7 +94,7 @@ dependencies {
|
|||||||
implementation libs.bundles.jackson
|
implementation libs.bundles.jackson
|
||||||
|
|
||||||
implementation libs.bundles.apache.commons
|
implementation libs.bundles.apache.commons
|
||||||
implementation libs.bouncycastle
|
implementation libs.bundles.bouncycastle
|
||||||
implementation libs.caffeine
|
implementation libs.caffeine
|
||||||
implementation libs.javax.annotations
|
implementation libs.javax.annotations
|
||||||
|
|
||||||
|
|||||||
@@ -24,7 +24,8 @@ logstash-encoder = "net.logstash.logback:logstash-logback-encoder:7.2"
|
|||||||
|
|
||||||
janino = "org.codehaus.janino:janino:3.1.9"
|
janino = "org.codehaus.janino:janino:3.1.9"
|
||||||
|
|
||||||
bouncycastle = "org.bouncycastle:bcprov-jdk15on:1.61"
|
bouncycastle-prov = "org.bouncycastle:bcprov-jdk15on:1.61"
|
||||||
|
bouncycastle-pkix = 'org.bouncycastle:bcpkix-jdk15on:1.61'
|
||||||
|
|
||||||
caffeine = "com.github.ben-manes.caffeine:caffeine:2.8.5"
|
caffeine = "com.github.ben-manes.caffeine:caffeine:2.8.5"
|
||||||
|
|
||||||
@@ -135,6 +136,7 @@ reactor = ["reactor-core", "reactor-netty", "reactor-extra", "reactor-kotlin"]
|
|||||||
spring-framework = ["spring-boot-starter", "spring-security-core", "spring-security-web", "spring-security-config"]
|
spring-framework = ["spring-boot-starter", "spring-security-core", "spring-security-web", "spring-security-config"]
|
||||||
testcontainers = ["testcontainers", "testcontainers-ganache"]
|
testcontainers = ["testcontainers", "testcontainers-ganache"]
|
||||||
junit = ["junit-jupiter", "assertj"]
|
junit = ["junit-jupiter", "assertj"]
|
||||||
|
bouncycastle = ["bouncycastle-pkix", "bouncycastle-prov"]
|
||||||
|
|
||||||
|
|
||||||
[plugins]
|
[plugins]
|
||||||
|
|||||||
@@ -2,12 +2,13 @@ package io.emeraldpay.dshackle.config.context
|
|||||||
|
|
||||||
import brave.grpc.GrpcTracing
|
import brave.grpc.GrpcTracing
|
||||||
import brave.rpc.RpcTracing
|
import brave.rpc.RpcTracing
|
||||||
|
import brave.sampler.Sampler
|
||||||
import io.grpc.ServerInterceptor
|
import io.grpc.ServerInterceptor
|
||||||
import org.springframework.context.annotation.Bean
|
import org.springframework.context.annotation.Bean
|
||||||
import org.springframework.context.annotation.Configuration
|
import org.springframework.context.annotation.Configuration
|
||||||
|
|
||||||
@Configuration
|
@Configuration
|
||||||
open class TraceGrpcConfiguration {
|
open class TraceConfiguration {
|
||||||
|
|
||||||
@Bean
|
@Bean
|
||||||
open fun grpcTracing(rpcTracing: RpcTracing): GrpcTracing = GrpcTracing.create(rpcTracing)
|
open fun grpcTracing(rpcTracing: RpcTracing): GrpcTracing = GrpcTracing.create(rpcTracing)
|
||||||
@@ -15,4 +16,7 @@ open class TraceGrpcConfiguration {
|
|||||||
@Bean
|
@Bean
|
||||||
open fun grpcServerBraveInterceptor(grpcTracing: GrpcTracing): ServerInterceptor =
|
open fun grpcServerBraveInterceptor(grpcTracing: GrpcTracing): ServerInterceptor =
|
||||||
grpcTracing.newServerInterceptor()
|
grpcTracing.newServerInterceptor()
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
open fun defaultSampler(): Sampler = Sampler.ALWAYS_SAMPLE
|
||||||
}
|
}
|
||||||
@@ -1,62 +0,0 @@
|
|||||||
package io.emeraldpay.dshackle.config.spans
|
|
||||||
|
|
||||||
import brave.Tracer
|
|
||||||
import brave.handler.MutableSpan
|
|
||||||
import brave.handler.SpanHandler
|
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper
|
|
||||||
import com.fasterxml.jackson.module.kotlin.readValue
|
|
||||||
import io.grpc.CallOptions
|
|
||||||
import io.grpc.Channel
|
|
||||||
import io.grpc.ClientCall
|
|
||||||
import io.grpc.ClientInterceptor
|
|
||||||
import io.grpc.ForwardingClientCall.SimpleForwardingClientCall
|
|
||||||
import io.grpc.ForwardingClientCallListener
|
|
||||||
import io.grpc.Metadata
|
|
||||||
import io.grpc.Metadata.ASCII_STRING_MARSHALLER
|
|
||||||
import io.grpc.MethodDescriptor
|
|
||||||
import org.springframework.beans.factory.annotation.Qualifier
|
|
||||||
|
|
||||||
class ClientSpansInterceptor(
|
|
||||||
private val zipkinSpanHandler: SpanHandler,
|
|
||||||
private val tracer: Tracer,
|
|
||||||
@Qualifier("spanMapper")
|
|
||||||
private val spanMapper: ObjectMapper
|
|
||||||
) : ClientInterceptor {
|
|
||||||
|
|
||||||
override fun <ReqT : Any?, RespT : Any?> interceptCall(
|
|
||||||
method: MethodDescriptor<ReqT, RespT>,
|
|
||||||
callOptions: CallOptions,
|
|
||||||
next: Channel
|
|
||||||
): ClientCall<ReqT, RespT> {
|
|
||||||
return if (method.fullMethodName == "emerald.Blockchain/NativeCall") {
|
|
||||||
SpanClientCall(next.newCall(method, callOptions))
|
|
||||||
} else {
|
|
||||||
next.newCall(method, callOptions)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private inner class SpanClientCall<ReqT, RespT>(
|
|
||||||
delegate: ClientCall<ReqT, RespT>?
|
|
||||||
) : SimpleForwardingClientCall<ReqT, RespT>(delegate) {
|
|
||||||
override fun start(responseListener: Listener<RespT>, headers: Metadata) {
|
|
||||||
val spanResponseListener = SpanClientCallListener(responseListener)
|
|
||||||
super.start(spanResponseListener, headers)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private inner class SpanClientCallListener<RespT>(
|
|
||||||
delegate: ClientCall.Listener<RespT>,
|
|
||||||
) : ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(delegate) {
|
|
||||||
override fun onHeaders(headers: Metadata) {
|
|
||||||
headers[Metadata.Key.of(SPAN_HEADER, ASCII_STRING_MARSHALLER)]
|
|
||||||
?.takeIf { it.isNotBlank() }
|
|
||||||
?.let {
|
|
||||||
val spansFromProvider = spanMapper.readValue<List<MutableSpan>>(it)
|
|
||||||
spansFromProvider.forEach { span ->
|
|
||||||
zipkinSpanHandler.end(tracer.currentSpan().context(), span, SpanHandler.Cause.FINISHED)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
super.onHeaders(headers)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -3,78 +3,48 @@ package io.emeraldpay.dshackle.config.spans
|
|||||||
import brave.handler.MutableSpan
|
import brave.handler.MutableSpan
|
||||||
import brave.handler.SpanHandler
|
import brave.handler.SpanHandler
|
||||||
import brave.propagation.TraceContext
|
import brave.propagation.TraceContext
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper
|
import org.springframework.beans.factory.annotation.Value
|
||||||
import com.github.benmanes.caffeine.cache.Caffeine
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
|
||||||
import org.springframework.beans.factory.annotation.Qualifier
|
import org.springframework.stereotype.Component
|
||||||
import org.springframework.cloud.sleuth.Span
|
import java.util.concurrent.TimeUnit
|
||||||
import java.time.Duration
|
|
||||||
|
|
||||||
|
@Component
|
||||||
|
@ConditionalOnProperty(value = ["spring.zipkin.enabled"], havingValue = "true")
|
||||||
class ProviderSpanHandler(
|
class ProviderSpanHandler(
|
||||||
@Qualifier("spanMapper")
|
@Value("\${spans.collect.collect-only-errors:false}")
|
||||||
private val spanMapper: ObjectMapper,
|
private val onlyErrors: Boolean? = false,
|
||||||
private val spanExportableList: List<SpanExportable>
|
@Value("\${spans.collect.long-span-threshold}")
|
||||||
|
private val longSpanThreshold: Long? = null
|
||||||
) : SpanHandler() {
|
) : SpanHandler() {
|
||||||
private val spans = Caffeine
|
|
||||||
.newBuilder()
|
|
||||||
.expireAfterWrite(Duration.ofMinutes(5))
|
|
||||||
.build<String, MutableList<MutableSpan>>()
|
|
||||||
|
|
||||||
override fun end(context: TraceContext, span: MutableSpan, cause: Cause): Boolean {
|
override fun end(context: TraceContext?, span: MutableSpan, cause: Cause?): Boolean {
|
||||||
if (span.traceId().length > 20 && span.parentId() != null) {
|
val duration = TimeUnit.MILLISECONDS.convert(span.finishTimestamp() - span.startTimestamp(), TimeUnit.MICROSECONDS)
|
||||||
val spanList = spans.asMap().computeIfAbsent(span.parentId()) { mutableListOf() }
|
|
||||||
spanList.add(span)
|
val isError = span.tags().containsKey("error")
|
||||||
|
|
||||||
|
// If onlyErrors is true and there is an error, return true.
|
||||||
|
if (onlyErrors == true && isError) {
|
||||||
|
return true
|
||||||
}
|
}
|
||||||
return super.end(context, span, cause)
|
|
||||||
}
|
|
||||||
|
|
||||||
fun getErrorSpans(spanId: String, currentSpan: Span): String {
|
// If onlyErrors is true, there is no error, and the span is long, return true.
|
||||||
val spansInfo = SpansInfo()
|
if (onlyErrors == true && !isError && longSpanThreshold != null && duration >= longSpanThreshold) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
enrichErrorSpans(spanId, spansInfo)
|
// If onlyErrors is false, check for the time threshold condition.
|
||||||
currentSpan.end()
|
if (onlyErrors == false) {
|
||||||
currentSpan.context().parentId()?.let {
|
// If longSpanThreshold is null, return true.
|
||||||
spans.getIfPresent(it)?.let { mutableSpans ->
|
if (longSpanThreshold == null) {
|
||||||
if (mutableSpans.isNotEmpty()) {
|
return true
|
||||||
processSpanInfo(mutableSpans[0], spansInfo)
|
}
|
||||||
}
|
// If longSpanThreshold is set, only return true if the duration is >= time threshold.
|
||||||
|
else if (duration >= longSpanThreshold) {
|
||||||
|
return true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
spansInfo.spans
|
// Return false if none of the above conditions are met.
|
||||||
.map { it.parentId() }
|
return false
|
||||||
.forEach {
|
|
||||||
if (it != null) {
|
|
||||||
spans.invalidate(it)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return if (spansInfo.exportable) {
|
|
||||||
spanMapper.writeValueAsString(spansInfo.spans)
|
|
||||||
} else {
|
|
||||||
""
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun enrichErrorSpans(spanId: String, spansInfo: SpansInfo) {
|
|
||||||
val currentSpans: List<MutableSpan>? = spans.getIfPresent(spanId)
|
|
||||||
|
|
||||||
currentSpans?.forEach {
|
|
||||||
processSpanInfo(it, spansInfo)
|
|
||||||
if (spanId != it.id()) {
|
|
||||||
enrichErrorSpans(it.id(), spansInfo)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private fun processSpanInfo(span: MutableSpan, spansInfo: SpansInfo) {
|
|
||||||
spansInfo.spans.add(span)
|
|
||||||
if (spanExportableList.any { it.isExportable(span) }) {
|
|
||||||
spansInfo.exportable = true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private data class SpansInfo(
|
|
||||||
var exportable: Boolean = false,
|
|
||||||
val spans: MutableList<MutableSpan> = mutableListOf()
|
|
||||||
)
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,44 +0,0 @@
|
|||||||
package io.emeraldpay.dshackle.config.spans
|
|
||||||
|
|
||||||
import io.grpc.ForwardingServerCall.SimpleForwardingServerCall
|
|
||||||
import io.grpc.Metadata
|
|
||||||
import io.grpc.ServerCall
|
|
||||||
import io.grpc.ServerCallHandler
|
|
||||||
import io.grpc.ServerInterceptor
|
|
||||||
import org.springframework.cloud.sleuth.Tracer
|
|
||||||
|
|
||||||
class ServerSpansInterceptor(
|
|
||||||
private val tracer: Tracer,
|
|
||||||
private val providerSpanHandler: ProviderSpanHandler
|
|
||||||
) : ServerInterceptor {
|
|
||||||
override fun <ReqT : Any?, RespT : Any?> interceptCall(
|
|
||||||
call: ServerCall<ReqT, RespT>,
|
|
||||||
headers: Metadata,
|
|
||||||
next: ServerCallHandler<ReqT, RespT>
|
|
||||||
): ServerCall.Listener<ReqT> {
|
|
||||||
val serverCall = if (call.methodDescriptor.fullMethodName == "emerald.Blockchain/NativeCall") {
|
|
||||||
SpanServerCall(call)
|
|
||||||
} else {
|
|
||||||
call
|
|
||||||
}
|
|
||||||
|
|
||||||
return next.startCall(serverCall, headers)
|
|
||||||
}
|
|
||||||
|
|
||||||
private inner class SpanServerCall<ReqT, RespT>(
|
|
||||||
private val call: ServerCall<ReqT, RespT>
|
|
||||||
) : SimpleForwardingServerCall<ReqT, RespT>(call) {
|
|
||||||
override fun sendHeaders(headers: Metadata) {
|
|
||||||
tracer.currentSpan()?.let {
|
|
||||||
val parentId = it.context().parentId()
|
|
||||||
if (parentId != null) {
|
|
||||||
val spans = providerSpanHandler.getErrorSpans(it.context().spanId(), it)
|
|
||||||
if (spans.isNotBlank()) {
|
|
||||||
headers.put(Metadata.Key.of(SPAN_HEADER, Metadata.ASCII_STRING_MARSHALLER), spans)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
call.sendHeaders(headers)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,69 +0,0 @@
|
|||||||
package io.emeraldpay.dshackle.config.spans
|
|
||||||
|
|
||||||
import brave.Tracer
|
|
||||||
import brave.handler.SpanHandler
|
|
||||||
import com.fasterxml.jackson.annotation.JsonAutoDetect
|
|
||||||
import com.fasterxml.jackson.annotation.JsonInclude
|
|
||||||
import com.fasterxml.jackson.databind.DeserializationFeature
|
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper
|
|
||||||
import io.grpc.ClientInterceptor
|
|
||||||
import io.grpc.ServerInterceptor
|
|
||||||
import org.springframework.beans.factory.annotation.Qualifier
|
|
||||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean
|
|
||||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
|
|
||||||
import org.springframework.context.annotation.Bean
|
|
||||||
import org.springframework.context.annotation.Configuration
|
|
||||||
|
|
||||||
const val SPAN_HEADER = "spans"
|
|
||||||
|
|
||||||
@Configuration
|
|
||||||
@ConditionalOnProperty(value = ["spans.collect.enabled"], havingValue = "true")
|
|
||||||
open class SpanConfig {
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
open fun spanMapper(): ObjectMapper =
|
|
||||||
ObjectMapper()
|
|
||||||
.apply {
|
|
||||||
setSerializationInclusion(JsonInclude.Include.NON_DEFAULT)
|
|
||||||
configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
|
|
||||||
setVisibility(
|
|
||||||
this.serializationConfig.defaultVisibilityChecker
|
|
||||||
.withFieldVisibility(JsonAutoDetect.Visibility.ANY)
|
|
||||||
.withGetterVisibility(JsonAutoDetect.Visibility.NONE)
|
|
||||||
.withSetterVisibility(JsonAutoDetect.Visibility.NONE)
|
|
||||||
.withCreatorVisibility(JsonAutoDetect.Visibility.NONE)
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
@Configuration
|
|
||||||
@ConditionalOnBean(SpanConfig::class)
|
|
||||||
@ConditionalOnProperty(value = ["spans.collect.main.enabled"], havingValue = "true")
|
|
||||||
open class MainSpanConfig {
|
|
||||||
@Bean
|
|
||||||
open fun clientSpansInterceptor(
|
|
||||||
zipkinSpanHandler: SpanHandler,
|
|
||||||
tracer: Tracer,
|
|
||||||
@Qualifier("spanMapper")
|
|
||||||
spanMapper: ObjectMapper
|
|
||||||
): ClientInterceptor = ClientSpansInterceptor(zipkinSpanHandler, tracer, spanMapper)
|
|
||||||
}
|
|
||||||
|
|
||||||
@Configuration
|
|
||||||
@ConditionalOnBean(SpanConfig::class)
|
|
||||||
@ConditionalOnProperty(value = ["spans.collect.provider.enabled"], havingValue = "true")
|
|
||||||
open class ProviderSpanConfig {
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
open fun errorSpanHandler(
|
|
||||||
@Qualifier("spanMapper")
|
|
||||||
spanMapper: ObjectMapper,
|
|
||||||
spanExportableList: List<SpanExportable>
|
|
||||||
): ProviderSpanHandler = ProviderSpanHandler(spanMapper, spanExportableList)
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
open fun serverSpansInterceptor(
|
|
||||||
tracer: org.springframework.cloud.sleuth.Tracer,
|
|
||||||
providerSpanHandler: ProviderSpanHandler
|
|
||||||
): ServerInterceptor = ServerSpansInterceptor(tracer, providerSpanHandler)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -0,0 +1,90 @@
|
|||||||
|
package io.emeraldpay.dshackle.config.spans
|
||||||
|
|
||||||
|
import io.emeraldpay.dshackle.config.MainConfig
|
||||||
|
import org.apache.http.conn.ssl.SSLConnectionSocketFactory
|
||||||
|
import org.apache.http.impl.client.HttpClients
|
||||||
|
import org.bouncycastle.openssl.PEMParser
|
||||||
|
import org.springframework.cloud.sleuth.zipkin2.ZipkinRestTemplateCustomizer
|
||||||
|
import org.springframework.http.HttpRequest
|
||||||
|
import org.springframework.http.client.ClientHttpRequestExecution
|
||||||
|
import org.springframework.http.client.ClientHttpRequestInterceptor
|
||||||
|
import org.springframework.http.client.ClientHttpResponse
|
||||||
|
import org.springframework.http.client.HttpComponentsClientHttpRequestFactory
|
||||||
|
import org.springframework.stereotype.Component
|
||||||
|
import org.springframework.web.client.RestTemplate
|
||||||
|
import java.io.ByteArrayOutputStream
|
||||||
|
import java.io.StringReader
|
||||||
|
import java.nio.file.Files
|
||||||
|
import java.nio.file.Paths
|
||||||
|
import java.security.KeyFactory
|
||||||
|
import java.security.KeyStore
|
||||||
|
import java.security.SecureRandom
|
||||||
|
import java.security.cert.CertificateFactory
|
||||||
|
import java.security.cert.X509Certificate
|
||||||
|
import java.security.spec.PKCS8EncodedKeySpec
|
||||||
|
import java.util.zip.GZIPOutputStream
|
||||||
|
import javax.net.ssl.KeyManagerFactory
|
||||||
|
import javax.net.ssl.SSLContext
|
||||||
|
|
||||||
|
@Component
|
||||||
|
class ZipkinSSLCustomizer(private val mainConfig: MainConfig) : ZipkinRestTemplateCustomizer {
|
||||||
|
override fun customizeTemplate(restTemplate: RestTemplate): RestTemplate {
|
||||||
|
return if (mainConfig.tls?.enabled == true) {
|
||||||
|
setupSSL(mainConfig.tls!!.certificate!!, mainConfig.tls!!.key!!)
|
||||||
|
} else {
|
||||||
|
restTemplate
|
||||||
|
}.apply {
|
||||||
|
interceptors.add(0, GZipInterceptor())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun setupSSL(certPath: String, privateKeyPath: String): RestTemplate {
|
||||||
|
// Load the certificate
|
||||||
|
val certificateReader = StringReader(Files.readString(Paths.get(certPath)))
|
||||||
|
val pemObject = PEMParser(certificateReader).readPemObject()
|
||||||
|
val certificate = CertificateFactory.getInstance("X.509").generateCertificate(pemObject.content.inputStream()) as X509Certificate
|
||||||
|
// Load the private key
|
||||||
|
val privateKeyReader = StringReader(Files.readString(Paths.get(privateKeyPath)))
|
||||||
|
val pemKeyPair = PEMParser(privateKeyReader).readPemObject()
|
||||||
|
val privKeySpec = PKCS8EncodedKeySpec(pemKeyPair.content)
|
||||||
|
val privateKey = KeyFactory.getInstance("RSA").generatePrivate(privKeySpec)
|
||||||
|
|
||||||
|
// Create the key store
|
||||||
|
val keyStore = KeyStore.getInstance(KeyStore.getDefaultType()).apply {
|
||||||
|
load(null, null)
|
||||||
|
setCertificateEntry("certificate", certificate)
|
||||||
|
setKeyEntry("private-key", privateKey, null, arrayOf(certificate))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Create the SSL context
|
||||||
|
val keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()).apply {
|
||||||
|
init(keyStore, null)
|
||||||
|
}
|
||||||
|
val sslContext = SSLContext.getInstance("TLS").apply {
|
||||||
|
init(keyManagerFactory.keyManagers, null, SecureRandom())
|
||||||
|
}
|
||||||
|
|
||||||
|
// Create the HTTP client
|
||||||
|
val socketFactory = SSLConnectionSocketFactory(sslContext)
|
||||||
|
val httpClient = HttpClients.custom().setSSLSocketFactory(socketFactory).build()
|
||||||
|
|
||||||
|
// Create the request factory
|
||||||
|
val requestFactory = HttpComponentsClientHttpRequestFactory(httpClient)
|
||||||
|
|
||||||
|
// Create the RestTemplate
|
||||||
|
return RestTemplate(requestFactory)
|
||||||
|
}
|
||||||
|
|
||||||
|
private class GZipInterceptor : ClientHttpRequestInterceptor {
|
||||||
|
override fun intercept(
|
||||||
|
request: HttpRequest,
|
||||||
|
body: ByteArray,
|
||||||
|
execution: ClientHttpRequestExecution
|
||||||
|
): ClientHttpResponse {
|
||||||
|
request.headers.add("Content-Encoding", "gzip")
|
||||||
|
val gzipped = ByteArrayOutputStream()
|
||||||
|
GZIPOutputStream(gzipped).use { compressor -> compressor.write(body) }
|
||||||
|
return execution.execute(request, gzipped.toByteArray())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,34 +0,0 @@
|
|||||||
package io.emeraldpay.dshackle.config.spans
|
|
||||||
|
|
||||||
import brave.handler.MutableSpan
|
|
||||||
import io.emeraldpay.dshackle.commons.SPAN_ERROR
|
|
||||||
import io.emeraldpay.dshackle.commons.SPAN_NO_RESPONSE_MESSAGE
|
|
||||||
import org.springframework.beans.factory.annotation.Value
|
|
||||||
import org.springframework.stereotype.Component
|
|
||||||
import java.util.concurrent.TimeUnit
|
|
||||||
|
|
||||||
interface SpanExportable {
|
|
||||||
fun isExportable(span: MutableSpan): Boolean
|
|
||||||
}
|
|
||||||
|
|
||||||
@Component
|
|
||||||
class ErrorSpanExportable : SpanExportable {
|
|
||||||
override fun isExportable(span: MutableSpan): Boolean = span.tags().containsKey(SPAN_ERROR)
|
|
||||||
}
|
|
||||||
|
|
||||||
@Component
|
|
||||||
class NoResponseSpanExportable : SpanExportable {
|
|
||||||
override fun isExportable(span: MutableSpan): Boolean = span.tags().containsKey(SPAN_NO_RESPONSE_MESSAGE)
|
|
||||||
}
|
|
||||||
|
|
||||||
@Component
|
|
||||||
class LongResponseSpanExportable(
|
|
||||||
@Value("\${spans.collect.provider.long-span-threshold}")
|
|
||||||
private val longSpanThreshold: Long? = null
|
|
||||||
) : SpanExportable {
|
|
||||||
|
|
||||||
override fun isExportable(span: MutableSpan): Boolean {
|
|
||||||
val duration = span.finishTimestamp() - span.startTimestamp()
|
|
||||||
return TimeUnit.SECONDS.convert(duration, TimeUnit.MICROSECONDS) >= longSpanThreshold!!
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -3,8 +3,8 @@ spring:
|
|||||||
max-metadata-size: ${MAX_METADATA_SIZE:16384}
|
max-metadata-size: ${MAX_METADATA_SIZE:16384}
|
||||||
name: ${DRPC_APP_NAME:dshackle}
|
name: ${DRPC_APP_NAME:dshackle}
|
||||||
zipkin:
|
zipkin:
|
||||||
enabled: ${ZIPKIN_ENABLE:false}
|
enabled: ${ENABLE_COLLECT_SPANS:false}
|
||||||
base-url: ${ZIPKIN_URL:http://localhost:9411/}
|
base-url: ${ZIPKIN_URL:https://trace.drpc.dev/}
|
||||||
sleuth:
|
sleuth:
|
||||||
span-filter:
|
span-filter:
|
||||||
additional-span-name-patterns-to-ignore:
|
additional-span-name-patterns-to-ignore:
|
||||||
@@ -16,12 +16,9 @@ spring:
|
|||||||
- ".+SubscribeStatus"
|
- ".+SubscribeStatus"
|
||||||
- ".+SubscribeNodeStatus"
|
- ".+SubscribeNodeStatus"
|
||||||
- ".+Describe"
|
- ".+Describe"
|
||||||
|
- ".+ServerReflectionInfo"
|
||||||
|
|
||||||
spans:
|
spans:
|
||||||
collect:
|
collect:
|
||||||
enabled: ${ENABLE_COLLECT_SPANS:false}
|
long-span-threshold: ${LONG_SPAN_THRESHOLD:1000}
|
||||||
provider:
|
collect-only-errors: ${ONLY_ERROR_SPANS:true}
|
||||||
enabled: ${ENABLE_PROVIDER_COLLECT_SPANS:true}
|
|
||||||
long-span-threshold: ${LONG_SPAN_THRESHOLD:1}
|
|
||||||
main:
|
|
||||||
enabled: ${ENABLE_MAIN_COLLECT_SPANS:false}
|
|
||||||
@@ -1,138 +0,0 @@
|
|||||||
package io.emeraldpay.dshackle.config.spans
|
|
||||||
|
|
||||||
import brave.handler.SpanHandler
|
|
||||||
import org.junit.jupiter.api.Assertions.assertThrows
|
|
||||||
import org.junit.jupiter.api.Test
|
|
||||||
import org.junit.jupiter.api.assertDoesNotThrow
|
|
||||||
import org.junit.jupiter.api.extension.ExtendWith
|
|
||||||
import org.mockito.Mockito.mock
|
|
||||||
import org.springframework.beans.factory.NoSuchBeanDefinitionException
|
|
||||||
import org.springframework.beans.factory.annotation.Autowired
|
|
||||||
import org.springframework.boot.test.context.TestConfiguration
|
|
||||||
import org.springframework.cloud.sleuth.Tracer
|
|
||||||
import org.springframework.context.ApplicationContext
|
|
||||||
import org.springframework.context.annotation.Bean
|
|
||||||
import org.springframework.test.context.ContextConfiguration
|
|
||||||
import org.springframework.test.context.TestPropertySource
|
|
||||||
import org.springframework.test.context.junit.jupiter.SpringExtension
|
|
||||||
|
|
||||||
class CollectSpanConfigTest {
|
|
||||||
|
|
||||||
@ContextConfiguration(
|
|
||||||
classes = [SpanConfig::class],
|
|
||||||
)
|
|
||||||
@ExtendWith(SpringExtension::class)
|
|
||||||
@TestPropertySource(
|
|
||||||
properties = [
|
|
||||||
"spans.collect.enabled=false"
|
|
||||||
]
|
|
||||||
)
|
|
||||||
class SpanConfigTest {
|
|
||||||
@Autowired
|
|
||||||
private lateinit var appCtx: ApplicationContext
|
|
||||||
|
|
||||||
@Test
|
|
||||||
fun testSpanConfig() {
|
|
||||||
assertThrows(NoSuchBeanDefinitionException::class.java) {
|
|
||||||
appCtx.getBean(SpanConfig::class.java)
|
|
||||||
}
|
|
||||||
assertThrows(NoSuchBeanDefinitionException::class.java) {
|
|
||||||
appCtx.getBean(ProviderSpanHandler::class.java)
|
|
||||||
}
|
|
||||||
assertThrows(NoSuchBeanDefinitionException::class.java) {
|
|
||||||
appCtx.getBean(ServerSpansInterceptor::class.java)
|
|
||||||
}
|
|
||||||
assertThrows(NoSuchBeanDefinitionException::class.java) {
|
|
||||||
appCtx.getBean(ClientSpansInterceptor::class.java)
|
|
||||||
}
|
|
||||||
assertThrows(NoSuchBeanDefinitionException::class.java) {
|
|
||||||
appCtx.getBean("spanMapper")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@ContextConfiguration(
|
|
||||||
classes = [SpanConfig::class, SpanProviderConfigTest.Config::class],
|
|
||||||
)
|
|
||||||
@ExtendWith(SpringExtension::class)
|
|
||||||
@TestPropertySource(
|
|
||||||
properties = [
|
|
||||||
"spans.collect.enabled=true",
|
|
||||||
"spans.collect.provider.enabled=true",
|
|
||||||
"spans.collect.main.enabled=false"
|
|
||||||
]
|
|
||||||
)
|
|
||||||
class SpanProviderConfigTest {
|
|
||||||
@Autowired
|
|
||||||
private lateinit var appCtx: ApplicationContext
|
|
||||||
|
|
||||||
@Test
|
|
||||||
fun testSpanProviderConfig() {
|
|
||||||
assertDoesNotThrow {
|
|
||||||
appCtx.getBean(SpanConfig::class.java)
|
|
||||||
}
|
|
||||||
assertDoesNotThrow {
|
|
||||||
appCtx.getBean(ProviderSpanHandler::class.java)
|
|
||||||
}
|
|
||||||
assertDoesNotThrow {
|
|
||||||
appCtx.getBean(ServerSpansInterceptor::class.java)
|
|
||||||
}
|
|
||||||
assertThrows(NoSuchBeanDefinitionException::class.java) {
|
|
||||||
appCtx.getBean(ClientSpansInterceptor::class.java)
|
|
||||||
}
|
|
||||||
assertDoesNotThrow {
|
|
||||||
appCtx.getBean("spanMapper")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@TestConfiguration
|
|
||||||
open class Config {
|
|
||||||
@Bean
|
|
||||||
open fun tracer(): Tracer = mock(Tracer::class.java)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@ContextConfiguration(
|
|
||||||
classes = [SpanConfig::class, SpanMainConfigTest.Config::class],
|
|
||||||
)
|
|
||||||
@ExtendWith(SpringExtension::class)
|
|
||||||
@TestPropertySource(
|
|
||||||
properties = [
|
|
||||||
"spans.collect.enabled=true",
|
|
||||||
"spans.collect.provider.enabled=false",
|
|
||||||
"spans.collect.main.enabled=true"
|
|
||||||
]
|
|
||||||
)
|
|
||||||
class SpanMainConfigTest {
|
|
||||||
@Autowired
|
|
||||||
private lateinit var appCtx: ApplicationContext
|
|
||||||
|
|
||||||
@Test
|
|
||||||
fun testSpanMainConfig() {
|
|
||||||
assertDoesNotThrow {
|
|
||||||
appCtx.getBean(SpanConfig::class.java)
|
|
||||||
}
|
|
||||||
assertThrows(NoSuchBeanDefinitionException::class.java) {
|
|
||||||
appCtx.getBean(ProviderSpanHandler::class.java)
|
|
||||||
}
|
|
||||||
assertThrows(NoSuchBeanDefinitionException::class.java) {
|
|
||||||
appCtx.getBean(ServerSpansInterceptor::class.java)
|
|
||||||
}
|
|
||||||
assertDoesNotThrow {
|
|
||||||
appCtx.getBean(ClientSpansInterceptor::class.java)
|
|
||||||
}
|
|
||||||
assertDoesNotThrow {
|
|
||||||
appCtx.getBean("spanMapper")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@TestConfiguration
|
|
||||||
open class Config {
|
|
||||||
@Bean
|
|
||||||
open fun tracer(): brave.Tracer = mock(brave.Tracer::class.java)
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
open fun zipkinSpanHandler(): SpanHandler = mock(SpanHandler::class.java)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,128 +1,85 @@
|
|||||||
package io.emeraldpay.dshackle.config.spans
|
package io.emeraldpay.dshackle.config.spans
|
||||||
|
|
||||||
import brave.handler.MutableSpan
|
import brave.handler.MutableSpan
|
||||||
import brave.handler.SpanHandler
|
|
||||||
import brave.propagation.TraceContext
|
|
||||||
import com.fasterxml.jackson.module.kotlin.readValue
|
|
||||||
import io.emeraldpay.dshackle.commons.SPAN_ERROR
|
|
||||||
import io.emeraldpay.dshackle.commons.SPAN_NO_RESPONSE_MESSAGE
|
|
||||||
import org.junit.jupiter.api.Assertions.assertEquals
|
import org.junit.jupiter.api.Assertions.assertEquals
|
||||||
import org.junit.jupiter.api.Assertions.assertTrue
|
|
||||||
import org.junit.jupiter.api.Test
|
|
||||||
import org.junit.jupiter.params.ParameterizedTest
|
import org.junit.jupiter.params.ParameterizedTest
|
||||||
|
import org.junit.jupiter.params.provider.Arguments
|
||||||
import org.junit.jupiter.params.provider.MethodSource
|
import org.junit.jupiter.params.provider.MethodSource
|
||||||
import org.mockito.Mockito
|
import java.util.stream.Stream
|
||||||
import org.mockito.Mockito.`when`
|
|
||||||
import org.springframework.cloud.sleuth.Span
|
|
||||||
import org.springframework.cloud.sleuth.brave.bridge.BraveTraceContext
|
|
||||||
|
|
||||||
class ProviderSpanHandlerTest {
|
class ProviderSpanHandlerTest {
|
||||||
private val mapper = SpanConfig().spanMapper()
|
|
||||||
private val spanExportableList = listOf(
|
|
||||||
ErrorSpanExportable(), NoResponseSpanExportable(), LongResponseSpanExportable(10)
|
|
||||||
)
|
|
||||||
private val ctx = TraceContext.newBuilder()
|
|
||||||
.traceId(1223324)
|
|
||||||
.spanId(234235)
|
|
||||||
.build()
|
|
||||||
|
|
||||||
@Test
|
|
||||||
fun `span with length of traceId less than 20 is not collected`() {
|
|
||||||
val spanId = "f7e83f2b69ec684d"
|
|
||||||
val currentSpan = Mockito.mock(Span::class.java)
|
|
||||||
val handler = spanHandler()
|
|
||||||
|
|
||||||
`when`(currentSpan.context()).thenReturn(BraveTraceContext(ctx))
|
|
||||||
|
|
||||||
handler.end(ctx, span("f7e83f2b69ec684d", spanId), SpanHandler.Cause.FINISHED)
|
|
||||||
|
|
||||||
val result = handler.getErrorSpans(spanId, currentSpan)
|
|
||||||
assertEquals("", result)
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
fun `span without parenId is not collected`() {
|
|
||||||
val spanId = "f7e83f2b69ec684d"
|
|
||||||
val currentSpan = Mockito.mock(Span::class.java)
|
|
||||||
val handler = spanHandler()
|
|
||||||
|
|
||||||
`when`(currentSpan.context()).thenReturn(BraveTraceContext(ctx))
|
|
||||||
|
|
||||||
handler.end(ctx, span("6666632728347823749827349723985", spanId), SpanHandler.Cause.FINISHED)
|
|
||||||
|
|
||||||
val result = handler.getErrorSpans(spanId, currentSpan)
|
|
||||||
assertEquals("", result)
|
|
||||||
}
|
|
||||||
|
|
||||||
@ParameterizedTest
|
|
||||||
@MethodSource("spans")
|
|
||||||
fun `span with length of traceId greater than 20 and with parentId is collected`(span: MutableSpan) {
|
|
||||||
val currentSpan = Mockito.mock(Span::class.java)
|
|
||||||
val handler = spanHandler()
|
|
||||||
|
|
||||||
`when`(currentSpan.context()).thenReturn(BraveTraceContext(ctx))
|
|
||||||
|
|
||||||
handler.end(ctx, span, SpanHandler.Cause.FINISHED)
|
|
||||||
|
|
||||||
val result = handler.getErrorSpans("f7e83f2b69ec682d", currentSpan)
|
|
||||||
val collectedSpans = mapper.readValue<List<MutableSpan>>(result)
|
|
||||||
assertTrue(collectedSpans.size == 1)
|
|
||||||
assertEquals(span, collectedSpans[0])
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
fun `span without error tag is not collected`() {
|
|
||||||
val spanId = "f7e83f2b69ec684d"
|
|
||||||
val currentSpan = Mockito.mock(Span::class.java)
|
|
||||||
val handler = spanHandler()
|
|
||||||
val span = span("6666632728347823749827349723985", spanId)
|
|
||||||
.apply {
|
|
||||||
parentId("f7e83f2b69ec682d")
|
|
||||||
removeTag(SPAN_ERROR)
|
|
||||||
}
|
|
||||||
|
|
||||||
`when`(currentSpan.context()).thenReturn(BraveTraceContext(ctx))
|
|
||||||
|
|
||||||
handler.end(ctx, span, SpanHandler.Cause.FINISHED)
|
|
||||||
|
|
||||||
val result = handler.getErrorSpans("f7e83f2b69ec682d", currentSpan)
|
|
||||||
assertEquals("", result)
|
|
||||||
}
|
|
||||||
|
|
||||||
private fun span(traceId: String, spanId: String) = MutableSpan()
|
|
||||||
.apply {
|
|
||||||
traceId(traceId)
|
|
||||||
id(spanId)
|
|
||||||
tag(SPAN_ERROR, "true")
|
|
||||||
}
|
|
||||||
|
|
||||||
companion object {
|
companion object {
|
||||||
@JvmStatic
|
@JvmStatic
|
||||||
fun spans() = listOf(
|
fun data(): Stream<Arguments> {
|
||||||
MutableSpan()
|
return Stream.of(
|
||||||
.apply {
|
Arguments.of(
|
||||||
traceId("6666632728347823749827349723985")
|
false, null,
|
||||||
id("f7e83f2b69ec682d")
|
MutableSpan().apply {
|
||||||
tag(SPAN_ERROR, "true")
|
startTimestamp(0)
|
||||||
parentId("f7e83f2b69ec682d")
|
finishTimestamp(1000000)
|
||||||
},
|
},
|
||||||
MutableSpan()
|
true
|
||||||
.apply {
|
),
|
||||||
traceId("6666632728347823749827349723985")
|
Arguments.of(
|
||||||
id("f7e83f2b69ec111d")
|
false, 1000L,
|
||||||
parentId("f7e83f2b69ec682d")
|
MutableSpan().apply {
|
||||||
tag(SPAN_NO_RESPONSE_MESSAGE, "noResp")
|
startTimestamp(0)
|
||||||
},
|
finishTimestamp(999999)
|
||||||
MutableSpan()
|
},
|
||||||
.apply {
|
false
|
||||||
traceId("6666632728347823749827349723985")
|
),
|
||||||
id("f7e83f2b69ec111d")
|
Arguments.of(
|
||||||
parentId("f7e83f2b69ec682d")
|
false, 1000L,
|
||||||
startTimestamp(3034272)
|
MutableSpan().apply {
|
||||||
finishTimestamp(15034272)
|
startTimestamp(0)
|
||||||
}
|
finishTimestamp(1000000)
|
||||||
)
|
},
|
||||||
|
true
|
||||||
|
),
|
||||||
|
Arguments.of(
|
||||||
|
true, null,
|
||||||
|
MutableSpan().apply {
|
||||||
|
startTimestamp(0)
|
||||||
|
finishTimestamp(1000000)
|
||||||
|
},
|
||||||
|
false
|
||||||
|
),
|
||||||
|
Arguments.of(
|
||||||
|
true, null,
|
||||||
|
MutableSpan().apply {
|
||||||
|
startTimestamp(0)
|
||||||
|
finishTimestamp(1000000)
|
||||||
|
tag("error", "true")
|
||||||
|
},
|
||||||
|
true
|
||||||
|
),
|
||||||
|
Arguments.of(
|
||||||
|
false, null,
|
||||||
|
MutableSpan().apply {
|
||||||
|
startTimestamp(0)
|
||||||
|
finishTimestamp(1000000)
|
||||||
|
tag("error", "true")
|
||||||
|
},
|
||||||
|
true
|
||||||
|
),
|
||||||
|
Arguments.of(
|
||||||
|
true, 1000L,
|
||||||
|
MutableSpan().apply {
|
||||||
|
startTimestamp(0)
|
||||||
|
finishTimestamp(1000000)
|
||||||
|
},
|
||||||
|
true
|
||||||
|
),
|
||||||
|
)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun spanHandler() = ProviderSpanHandler(mapper, spanExportableList)
|
@ParameterizedTest
|
||||||
|
@MethodSource("data")
|
||||||
|
fun end(onlyErrors: Boolean?, timeThreshold: Long?, span: MutableSpan, expected: Boolean) {
|
||||||
|
val handler = ProviderSpanHandler(onlyErrors, timeThreshold)
|
||||||
|
val res = handler.end(null, span, null)
|
||||||
|
assertEquals(expected, res)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user