Micrometer 1 10 reactor fixed (#258)

without this change we're trying to clear thread locals and scopes by iterating over all scopes

with this change we always add a scope (either from a span or a null-span scope if there's no span). That way we don't mess up the scope hierarchy

Closes gh-256

build.gradle

@@ -303,6 +303,8 @@ subprojects {
 
 	repositories {
 		mavenCentral()
+		mavenLocal()
+		maven { url "https://repo.spring.io/snapshot" } // for reactor snapshots
 	}
 
 	def check = tasks.findByName('check')

dependencies.gradle

@@ -2,7 +2,7 @@ def VERSIONS = [
 	'javax.servlet:javax.servlet-api:latest.release',
 	'jakarta.platform:jakarta.jakartaee-web-api:9.+',
 
-	'io.micrometer:context-propagation:1.0.+',
+	'io.micrometer:context-propagation:1.0.3-SNAPSHOT',
 
 	'aopalliance:aopalliance:1.0',
 	// logging
@@ -36,7 +36,7 @@ def PLATFORM_VERSIONS = [
 	// opentelemetry-instrumentation-api dependency above with this
 	'io.opentelemetry:opentelemetry-bom:1.19.+',
 	'io.opentelemetry:opentelemetry-bom-alpha:1.19.+',
-	'io.projectreactor:reactor-bom:2022.0.+', // latest release for tests
+	'io.projectreactor:reactor-bom:2022.0.8-SNAPSHOT', // latest release for tests
 	'org.junit:junit-bom:5.9.+'
 ]
 

micrometer-tracing-bridges/micrometer-tracing-bridge-brave/src/main/java/io/micrometer/tracing/brave/bridge/BraveCurrentTraceContext.java

@@ -15,7 +15,6 @@
  */
 package io.micrometer.tracing.brave.bridge;
 
-import brave.propagation.ThreadLocalCurrentTraceContext;
 import io.micrometer.tracing.CurrentTraceContext;
 import io.micrometer.tracing.TraceContext;
 
@@ -76,10 +75,6 @@ public Scope newScope(TraceContext context) {
 
     @Override
     public Scope maybeScope(TraceContext context) {
-        if (context == null && delegate instanceof ThreadLocalCurrentTraceContext) {
-            ((ThreadLocalCurrentTraceContext) delegate).clear();
-            return Scope.NOOP;
-        }
         return new BraveScope(
                 this.delegate.maybeScope(io.micrometer.tracing.brave.bridge.BraveTraceContext.toBrave(context)));
     }

micrometer-tracing-bridges/micrometer-tracing-bridge-brave/src/test/java/io/micrometer/tracing/brave/contextpropagation/ScopesTests.java

@@ -34,7 +34,6 @@
 import io.micrometer.tracing.brave.bridge.BraveTracer;
 import io.micrometer.tracing.handler.DefaultTracingObservationHandler;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Hooks;
 import reactor.core.publisher.Mono;
@@ -71,9 +70,9 @@ void setup() {
         observationRegistry.observationConfig().observationHandler(new DefaultTracingObservationHandler(this.tracer));
 
         Hooks.enableAutomaticContextPropagation();
+        ObservationThreadLocalAccessor.getInstance().setObservationRegistry(observationRegistry);
     }
 
-    @Disabled("TODO: Bug")
     @Test
     void should_open_and_close_scopes_with_reactor() {
         Observation obs1 = Observation.start("1", observationRegistry);

micrometer-tracing-bridges/micrometer-tracing-bridge-otel/src/main/java/io/micrometer/tracing/otel/bridge/OtelCurrentTraceContext.java

@@ -48,6 +48,13 @@ public TraceContext context() {
         return new OtelTraceContext(currentSpan);
     }
 
+    /**
+     * Since OpenTelemetry works on statics, and we would like to pass the tracing
+     * information on the {@link TraceContext} we are checking what we have currently in
+     * ThreadLocal and what was passed on {@link TraceContext}.
+     * @param context span to place into scope or {@code null} to clear the scope
+     * @return scope that always must be closed
+     */
     @Override
     public Scope newScope(TraceContext context) {
         OtelTraceContext otelTraceContext = (OtelTraceContext) context;
@@ -56,30 +63,37 @@ public Scope newScope(TraceContext context) {
         }
         Context current = Context.current();
         Context old = otelTraceContext.context();
-
-        Span currentSpan = Span.fromContext(current);
-        Span oldSpan = Span.fromContext(otelTraceContext.context());
+        // Check if there's a span in the static OTel context
+        Span spanFromCurrentCtx = Span.fromContext(current);
+        // Check if there's a span in the ctx attached to TraceContext
+        Span spanFromCtxOnNewSpan = Span.fromContext(otelTraceContext.context());
         SpanContext spanContext = otelTraceContext.delegate;
-        boolean sameSpan = currentSpan.getSpanContext().equals(oldSpan.getSpanContext())
-                && currentSpan.getSpanContext().equals(spanContext);
+        boolean sameSpan = spanFromCurrentCtx.getSpanContext().equals(spanFromCtxOnNewSpan.getSpanContext())
+                && spanFromCurrentCtx.getSpanContext().equals(spanContext);
         SpanFromSpanContext fromContext = new SpanFromSpanContext(((OtelTraceContext) context).span, spanContext,
                 otelTraceContext);
-
         Baggage currentBaggage = Baggage.fromContext(current);
         Baggage oldBaggage = Baggage.fromContext(old);
         boolean sameBaggage = sameBaggage(currentBaggage, oldBaggage);
-
         if (sameSpan && sameBaggage) {
             return io.opentelemetry.context.Scope::noop;
         }
+        Baggage updatedBaggage = mergeBaggage(currentBaggage, oldBaggage);
+        Context newContext = old.with(fromContext).with(updatedBaggage);
+        io.opentelemetry.context.Scope attach = newContext.makeCurrent();
+        otelTraceContext.updateContext(newContext);
+        return () -> {
+            otelTraceContext.updateContext(old);
+            attach.close();
+        };
+    }
 
+    private static Baggage mergeBaggage(Baggage currentBaggage, Baggage oldBaggage) {
         BaggageBuilder baggageBuilder = currentBaggage.toBuilder();
         oldBaggage.forEach(
                 (key, baggageEntry) -> baggageBuilder.put(key, baggageEntry.getValue(), baggageEntry.getMetadata()));
         Baggage updatedBaggage = baggageBuilder.build();
-
-        io.opentelemetry.context.Scope attach = old.with(fromContext).with(updatedBaggage).makeCurrent();
-        return attach::close;
+        return updatedBaggage;
     }
 
     private boolean sameBaggage(Baggage currentBaggage, Baggage oldBaggage) {
@@ -89,7 +103,7 @@ private boolean sameBaggage(Baggage currentBaggage, Baggage oldBaggage) {
     @Override
     public Scope maybeScope(TraceContext context) {
         if (context == null) {
-            io.opentelemetry.context.Scope scope = Context.current().with(Span.getInvalid()).makeCurrent();
+            io.opentelemetry.context.Scope scope = Context.root().makeCurrent();
             return scope::close;
         }
         return newScope(context);

micrometer-tracing-bridges/micrometer-tracing-bridge-otel/src/main/java/io/micrometer/tracing/otel/bridge/OtelTracer.java

@@ -73,7 +73,8 @@ public Span nextSpan(Span parent) {
         AtomicReference<Context> context = otelSpan.context().context;
         Context otelContext = context.get();
         Scope scope = null;
-        if (otelContext != null) {
+        if (otelContext != null && Context.current() != otelContext) { // This shouldn't
+                                                                       // happen
             scope = otelContext.makeCurrent();
         }
         try {

micrometer-tracing-bridges/micrometer-tracing-bridge-otel/src/test/java/io/micrometer/tracing/otel/contextpropagation/ScopesTests.java

@@ -33,7 +33,6 @@
 import io.opentelemetry.sdk.trace.SdkTracerProvider;
 import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Hooks;
 import reactor.core.publisher.Mono;
@@ -74,9 +73,9 @@ void setup() {
         observationRegistry.observationConfig().observationHandler(handler);
 
         Hooks.enableAutomaticContextPropagation();
+        ObservationThreadLocalAccessor.getInstance().setObservationRegistry(observationRegistry);
     }
 
-    @Disabled("TODO: Bug")
     @Test
     void should_open_and_close_scopes_with_reactor() {
         Observation obs1 = Observation.start("1", observationRegistry);