Only reset server context if there is a leak.

instrumentation-api/src/main/java/io/opentelemetry/instrumentation/api/tracer/BaseTracer.java

@@ -298,10 +298,19 @@ protected Throwable unwrapThrowable(Throwable throwable) {
   public <C> Context extract(C carrier, TextMapGetter<C> getter) {
     ContextPropagationDebug.debugContextLeakIfEnabled();
 
-    // Using Context.root() here may be quite unexpected, but the reason is simple.
-    // We want either span context extracted from the carrier or invalid one.
-    // We DO NOT want any span context potentially lingering in the current context.
-    return propagators.getTextMapPropagator().extract(Context.root(), carrier, getter);
+    Context parent = Context.current();
+    if (Span.fromContextOrNull(parent) != null) {
+      // A span has leaked from another thread.
+      // We want either span context extracted from the carrier or invalid one.
+      // We DO NOT want any span context potentially lingering in the current context.
+      // We reset to the root context, which may not always be appropriate (e.g., a framework added
+      // an item to the context before we create a span) but it is safer than removing all the
+      // possible spans that instrumentation may have added and such frameworks as of now do not
+      // have leaks.
+      parent = Context.root();
+    }
+
+    return propagators.getTextMapPropagator().extract(parent, carrier, getter);
   }
 
   /**

instrumentation/grpc-1.5/javaagent/grpc-1.5-javaagent.gradle

@@ -33,6 +33,7 @@ dependencies {
 
   testLibrary group: 'io.grpc', name: 'grpc-netty', version: grpcVersion
   testLibrary group: 'io.grpc', name: 'grpc-protobuf', version: grpcVersion
+  testLibrary group: 'io.grpc', name: 'grpc-services', version: grpcVersion
   testLibrary group: 'io.grpc', name: 'grpc-stub', version: grpcVersion
 
   testImplementation project(':instrumentation:grpc-1.5:testing')

instrumentation/grpc-1.5/library/grpc-1.5-library.gradle

@@ -7,6 +7,7 @@ dependencies {
 
   testLibrary group: 'io.grpc', name: 'grpc-netty', version: grpcVersion
   testLibrary group: 'io.grpc', name: 'grpc-protobuf', version: grpcVersion
+  testLibrary group: 'io.grpc', name: 'grpc-services', version: grpcVersion
   testLibrary group: 'io.grpc', name: 'grpc-stub', version: grpcVersion
 
   testImplementation deps.assertj

instrumentation/grpc-1.5/testing/grpc-1.5-testing.gradle

@@ -25,6 +25,7 @@ dependencies {
 
   api group: 'io.grpc', name: 'grpc-core', version: grpcVersion
   api group: 'io.grpc', name: 'grpc-protobuf', version: grpcVersion
+  api group: 'io.grpc', name: 'grpc-services', version: grpcVersion
   api group: 'io.grpc', name: 'grpc-stub', version: grpcVersion
 
   implementation group: 'javax.annotation', name: 'javax.annotation-api', version: '1.3.2'

instrumentation/grpc-1.5/testing/src/main/groovy/io/opentelemetry/instrumentation/grpc/v1_5/AbstractGrpcTest.groovy

@@ -31,6 +31,10 @@ import io.grpc.ServerCallHandler
 import io.grpc.ServerInterceptor
 import io.grpc.Status
 import io.grpc.StatusRuntimeException
+import io.grpc.protobuf.services.ProtoReflectionService
+import io.grpc.reflection.v1alpha.ServerReflectionGrpc
+import io.grpc.reflection.v1alpha.ServerReflectionRequest
+import io.grpc.reflection.v1alpha.ServerReflectionResponse
 import io.grpc.stub.StreamObserver
 import io.opentelemetry.instrumentation.test.InstrumentationSpecification
 import io.opentelemetry.instrumentation.test.utils.PortUtils
@@ -508,4 +512,99 @@ abstract class AbstractGrpcTest extends InstrumentationSpecification {
     channel?.shutdownNow()?.awaitTermination(10, TimeUnit.SECONDS)
     server?.shutdownNow()?.awaitTermination()
   }
+
+  def "test reflection service"() {
+    setup:
+    def service = ProtoReflectionService.newInstance()
+    def port = PortUtils.findOpenPort()
+    Server server = configureServer(ServerBuilder.forPort(port).addService(service)).build().start()
+    ManagedChannelBuilder channelBuilder = configureClient(ManagedChannelBuilder.forAddress("localhost", port))
+
+    // Depending on the version of gRPC usePlainText may or may not take an argument.
+    try {
+      channelBuilder.usePlaintext()
+    } catch (MissingMethodException e) {
+      channelBuilder.usePlaintext(true)
+    }
+    ManagedChannel channel = channelBuilder.build()
+    ServerReflectionGrpc.ServerReflectionStub client = ServerReflectionGrpc.newStub(channel)
+
+    when:
+    AtomicReference<Throwable> error = new AtomicReference<>()
+    AtomicReference<ServerReflectionResponse> response = new AtomicReference<>()
+    CountDownLatch latch = new CountDownLatch(1)
+    def request = client.serverReflectionInfo(new StreamObserver<ServerReflectionResponse>() {
+      @Override
+      void onNext(ServerReflectionResponse serverReflectionResponse) {
+        response.set(serverReflectionResponse)
+      }
+
+      @Override
+      void onError(Throwable throwable) {
+        error.set(throwable)
+        latch.countDown()
+      }
+
+      @Override
+      void onCompleted() {
+        latch.countDown()
+      }
+    })
+
+    request.onNext(ServerReflectionRequest.newBuilder()
+      .setListServices("The content will not be checked?")
+      .build())
+    request.onCompleted()
+
+    latch.await(10, TimeUnit.SECONDS)
+
+    then:
+    error.get() == null
+    response.get().listServicesResponse.getService(0).name == "grpc.reflection.v1alpha.ServerReflection"
+
+    assertTraces(1) {
+      trace(0, 2) {
+        span(0) {
+          name "grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo"
+          kind CLIENT
+          hasNoParent()
+          event(0) {
+            eventName "message"
+            attributes {
+              "message.type" "SENT"
+              "message.id" 1
+            }
+          }
+          attributes {
+            "${SemanticAttributes.RPC_SYSTEM.key}" "grpc"
+            "${SemanticAttributes.RPC_SERVICE.key}" "grpc.reflection.v1alpha.ServerReflection"
+            "${SemanticAttributes.RPC_METHOD.key}" "ServerReflectionInfo"
+          }
+        }
+        span(1) {
+          name "grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo"
+          kind SERVER
+          childOf span(0)
+          event(0) {
+            eventName "message"
+            attributes {
+              "message.type" "RECEIVED"
+              "message.id" 1
+            }
+          }
+          attributes {
+            "${SemanticAttributes.RPC_SYSTEM.key}" "grpc"
+            "${SemanticAttributes.RPC_SERVICE.key}" "grpc.reflection.v1alpha.ServerReflection"
+            "${SemanticAttributes.RPC_METHOD.key}" "ServerReflectionInfo"
+            "${SemanticAttributes.NET_PEER_IP.key}" "127.0.0.1"
+            "${SemanticAttributes.NET_PEER_PORT.key}" Long
+          }
+        }
+      }
+    }
+
+    cleanup:
+    channel?.shutdownNow()?.awaitTermination(10, TimeUnit.SECONDS)
+    server?.shutdownNow()?.awaitTermination()
+  }
 }