Distributed tracing support with Micrometer tracing

[oburgosm]

Since spring cloud sleuth doesn't work with Spring Boot 3.X, and the core of this project was moved to Micrometer Tracing which hasn't support natively for gRPC tracing, ¿is there a plan to support Micrometer tracing?

[jvmlet]

Hi, yes, sure. Metrics are already exported via micrometer. PR with tracing interceptor is welcome as well.

[jvmlet]

Another thought - may be open telemetry integration is better and faster solution ? It can be integrated as-is with almost zero efforts ?

[o-shevchenko]

I want to add tracing as well. I'm trying to use:

implementation("io.micrometer:micrometer-tracing-bridge-otel")
implementation("io.opentelemetry:opentelemetry-exporter-otlp")

@GRpcService(interceptors = [LogInterceptor::class, ObservationGrpcServerInterceptor::class])
class MyGrpcService

But I get:

Caused by: org.springframework.beans.factory.BeanCreationException: Failed to create interceptor instance.
    at org.lognet.springboot.grpc.GRpcServerRunner.lambda$bindInterceptors$3(GRpcServerRunner.java:131)
    at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
    at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:992)
    at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
    at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
    at java.base/java.util.stream.StreamSpliterators$WrappingSpliterator.forEachRemaining(StreamSpliterators.java:310)
    at java.base/java.util.stream.Streams$ConcatSpliterator.forEachRemaining(Streams.java:735)
    at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
    at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
    at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921)
    at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
    at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682)
    at org.lognet.springboot.grpc.GRpcServerRunner.bindInterceptors(GRpcServerRunner.java:142)
    at org.lognet.springboot.grpc.GRpcServerRunner.lambda$start$0(GRpcServerRunner.java:74)
    at java.base/java.util.HashMap.forEach(HashMap.java:1421)
    at org.lognet.springboot.grpc.GRpcServerRunner.start(GRpcServerRunner.java:68)
    ... 19 common frames omitted
Caused by: java.lang.InstantiationException: io.micrometer.core.instrument.binder.grpc.ObservationGrpcServerInterceptor
    at java.base/java.lang.Class.newInstance(Class.java:639)
    at org.lognet.springboot.grpc.GRpcServerRunner.lambda$bindInterceptors$3(GRpcServerRunner.java:129)
    ... 34 common frames omitted
Caused by: java.lang.NoSuchMethodException: io.micrometer.core.instrument.binder.grpc.ObservationGrpcServerInterceptor.<init>()
    at java.base/java.lang.Class.getConstructor0(Class.java:3585)
    at java.base/java.lang.Class.newInstance(Class.java:626)
    ... 35 common frames omitted

Is there any interceptor we can use to add distributed tracing context? Log example:

2023-08-30T13:23:44.864+03:00 trace_id=62922d23a87ae5a5eba4f8b10614b737 span_id=adfc11a38f569193

[o-shevchenko]

The same issue with TracingServerInterceptor from implementation("io.opentracing.contrib:opentracing-grpc:0.2.3")

Caused by: org.springframework.beans.factory.BeanCreationException: Failed to create interceptor instance.
    at org.lognet.springboot.grpc.GRpcServerRunner.lambda$bindInterceptors$3(GRpcServerRunner.java:131)
    at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
    at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:992)
    at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
    at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
    at java.base/java.util.stream.StreamSpliterators$WrappingSpliterator.forEachRemaining(StreamSpliterators.java:310)
    at java.base/java.util.stream.Streams$ConcatSpliterator.forEachRemaining(Streams.java:735)
    at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
    at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
    at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921)
    at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
    at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682)
    at org.lognet.springboot.grpc.GRpcServerRunner.bindInterceptors(GRpcServerRunner.java:142)
    at org.lognet.springboot.grpc.GRpcServerRunner.lambda$start$0(GRpcServerRunner.java:74)
    at java.base/java.util.HashMap.forEach(HashMap.java:1421)
    at org.lognet.springboot.grpc.GRpcServerRunner.start(GRpcServerRunner.java:68)
    ... 19 common frames omitted
Caused by: java.lang.InstantiationException: io.opentracing.contrib.grpc.TracingServerInterceptor
    at java.base/java.lang.Class.newInstance(Class.java:639)
    at org.lognet.springboot.grpc.GRpcServerRunner.lambda$bindInterceptors$3(GRpcServerRunner.java:129)
    ... 34 common frames omitted
Caused by: java.lang.NoSuchMethodException: io.opentracing.contrib.grpc.TracingServerInterceptor.<init>()
    at java.base/java.lang.Class.getConstructor0(Class.java:3585)
    at java.base/java.lang.Class.newInstance(Class.java:626)
    ... 35 common frames omitted

[o-shevchenko]

I was able to create such a tracing implemented interceptors: Client:

import io.grpc.CallOptions
import io.grpc.Channel
import io.grpc.ClientCall
import io.grpc.ClientInterceptor
import io.grpc.ForwardingClientCall
import io.grpc.Metadata
import io.grpc.MethodDescriptor
import io.opentelemetry.api.trace.Span
import io.opentelemetry.api.trace.TraceId

class TracingClientInterceptor : ClientInterceptor {
    override fun <ReqT, RespT> interceptCall(
        method: MethodDescriptor<ReqT, RespT>,
        callOptions: CallOptions,
        next: Channel
    ): ClientCall<ReqT, RespT> {
        return object : ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
            override fun start(responseListener: ClientCall.Listener<RespT>, headers: Metadata) {
                val currentSpan = Span.current()
                val traceId = currentSpan.spanContext.traceId
                val spanId = currentSpan.spanContext.spanId

                if (traceId != TraceId.getInvalid()) {
                    headers.put(Metadata.Key.of("traceid", Metadata.ASCII_STRING_MARSHALLER), traceId)
                    headers.put(Metadata.Key.of("spanid", Metadata.ASCII_STRING_MARSHALLER), spanId)
                }
                super.start(responseListener, headers)
            }
        }
    }
}

...
val managedChannel = ManagedChannelBuilder.forAddress(host, port)
            .usePlaintext()
            .enableRetry()
            .maxInboundMessageSize(maxInboundMessageSize)
            .maxInboundMetadataSize(maxInboundMetadataSize)
            .idleTimeout(idleTimeout, TimeUnit.MINUTES)
            .keepAliveTime(keepAliveTime, TimeUnit.MINUTES)
            .keepAliveTimeout(keepAliveTimeout, TimeUnit.MINUTES)
            .defaultLoadBalancingPolicy(defaultLoadBalancingPolicy)
            .maxRetryAttempts(maxRetryAttempts)
            .intercept(TracingClientInterceptor())
            .build()
            ....

   Server:
   import io.grpc.Metadata
import io.grpc.ServerCall
import io.grpc.ServerCallHandler
import io.grpc.ServerInterceptor
import io.opentelemetry.api.trace.Span
import io.opentelemetry.api.trace.SpanBuilder
import io.opentelemetry.api.trace.SpanContext
import io.opentelemetry.api.trace.TraceFlags
import io.opentelemetry.api.trace.TraceState
import io.opentelemetry.api.trace.Tracer
import io.opentelemetry.context.Context
import io.opentelemetry.context.Scope

class TracingServerInterceptor(private val tracer: Tracer) : ServerInterceptor {

    override fun <ReqT, RespT> interceptCall(
        call: ServerCall<ReqT, RespT>,
        headers: Metadata,
        next: ServerCallHandler<ReqT, RespT>
    ): ServerCall.Listener<ReqT> {
        val traceId = headers.get(Metadata.Key.of("traceid", Metadata.ASCII_STRING_MARSHALLER))
        val spanId = headers.get(Metadata.Key.of("spanid", Metadata.ASCII_STRING_MARSHALLER))

        val spanContext = createSpanContext(traceId, spanId)
        val spanBuilder: SpanBuilder = tracer.spanBuilder("span").setParent(spanContext)
        val span: Span = spanBuilder.startSpan()
        val scopedContext: Context = Context.current().with(span)

        val scope: Scope = scopedContext.makeCurrent()

        try {
            return next.startCall(call, headers)
        } finally {
            scope.close()
            span.end()
        }
    }

    private fun createSpanContext(traceId: String?, spanId: String?): Context {
        val spanContext = SpanContext.createFromRemoteParent(
            traceId ?: "",
            spanId ?: "",
            TraceFlags.getDefault(),
            TraceState.getDefault()
        )
        return Context.current().with(Span.wrap(spanContext))
    }
}

@Configuration
@EnableConfigurationProperties(OtlpProperties::class)
class OtelConfiguration {
    @Bean
    fun otlpExporter(properties: OtlpProperties): OtlpGrpcSpanExporter {
        val builder = OtlpGrpcSpanExporter.builder().setEndpoint(properties.endpoint)
        return builder.build()
    }

    @Bean
    fun tracer(otlpExporter: OtlpGrpcSpanExporter): Tracer {
        val tracerProvider = SdkTracerProvider.builder()
            .addSpanProcessor(BatchSpanProcessor.builder(otlpExporter).build())
            .build()

        val openTelemetry = OpenTelemetrySdk.builder()
            .setTracerProvider(tracerProvider)
            .buildAndRegisterGlobal()

        return openTelemetry.tracerProvider.get("service")
    }
}

import io.opentelemetry.api.trace.Tracer
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration

@Configuration
class GrpcConfig {

    @Bean
    fun tracingServerInterceptor(tracer: Tracer): TracingServerInterceptor {
        return TracingServerInterceptor(tracer)
    }
}

...
@GRpcService(interceptors = [LogInterceptor::class, TracingServerInterceptor::class])
...         

[o-shevchenko]

Now I trace in my logs:

2023-09-07T12:27:30.952+03:00 trace_id=c9f81f73c2a61bd688789abf5e26125d span_id=4ba91946733299cc INFO 75489 --- [ault-executor-0] c.d.c.b.s.a.rpc.config.LogInterceptor    : service.MyService/Run
2023-09-07T12:27:32.905+03:00 trace_id= span_id= WARN 75489 --- [atcher-worker-1] o.l.s.grpc.FailureHandlingSupport        : Handled exception NotAuthorizedException call as Status{code=UNAUTHENTICATED

Not sure how to integrate it with

@GRpcServiceAdvice
class GrpcExceptionHandler {

[jvmlet]

Great, glad to hear that. Do you mind to PR this? I'll check how to hookup the exception handler then

[o-shevchenko]

yeah, I can open a PR with such interceptors

[o-shevchenko]

Opened PR: https://github.com/LogNet/grpc-spring-boot-starter/pull/376

[vicmosin]

@jvmlet hey, can we merge the PR? Would be nice to have it already

[molszowy-te]

@jvmlet, is it going to be released anytime soon ?

[jvmlet]

It is going to be released, but please be patient with the current disaster in Israel....

[vicmosin]

@jvmlet I am sorry I didn't know you are from there. We stand with Israel, the humanity will win.

[jvmlet]

@vicmosin , thanks (writing from the shelter)

[o-shevchenko]

Stay safe and stay strong @jvmlet 🇮🇱

[vicmosin]

@o-shevchenko I saw the PR was closed, any hints how can it be achieved at the end?