Do not propagate gRPC deadline when propagating OTel context via java…

…agent. (#5543)

* Add test for early return in gRPC pattern.

* Do not propagate gRPC deadline when propagating OTel context via javaagent.

instrumentation/grpc-1.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/grpc/v1_6/GrpcSingletons.java

@@ -20,7 +20,7 @@ public final class GrpcSingletons {
 
  public static final ServerInterceptor SERVER_INTERCEPTOR;
 
- public static final Context.Storage STORAGE = new ContextStorageBridge();
+ public static final Context.Storage STORAGE = new ContextStorageBridge(false);
 
  static {
  boolean experimentalSpanAttributes =

instrumentation/grpc-1.6/library/src/main/java/io/grpc/override/ContextStorageOverride.java

@@ -14,7 +14,7 @@
  */
 public final class ContextStorageOverride extends Context.Storage {
 
- private static final Context.Storage delegate = new ContextStorageBridge();
+ private static final Context.Storage delegate = new ContextStorageBridge(true);
 
  @Override
  public Context doAttach(Context toAttach) {

instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/internal/ContextStorageBridge.java

@@ -28,6 +28,12 @@ public final class ContextStorageBridge extends Context.Storage {
  Context.key("otel-context");
  private static final Context.Key<Scope> OTEL_SCOPE = Context.key("otel-scope");
 
+ private final boolean propagateGrpcDeadline;
+
+ public ContextStorageBridge(boolean propagateGrpcDeadline) {
+ this.propagateGrpcDeadline = propagateGrpcDeadline;
+ }
+
  @Override
  public Context doAttach(Context toAttach) {
  io.opentelemetry.context.Context otelContext = io.opentelemetry.context.Context.current();
@@ -87,6 +93,20 @@ public Context current() {
  // This context has already been previously attached and associated with an OTel context. Just
  // create a new context referring to the current OTel context to reflect the current stack.
  // The previous context is unaffected and will continue to live in its own stack.
+
+ if (!propagateGrpcDeadline) {
+ // Because we are propagating gRPC context via OpenTelemetry here, we may also propagate a
+ // deadline where it
+ // wasn't present before. Notably, this could happen with no user intention when using the
+ // javaagent which will
+ // add OpenTelemetry propagation automatically, and cause that code to fail with a deadline
+ // cancellation. While
+ // ideally we could propagate deadline as well as gRPC intended, we cannot have existing
+ // code fail because it
+ // added the javaagent and choose to fork here.
+ current = current.fork();
+ }
+
  return current.withValue(OTEL_CONTEXT, otelContext);
  }
  return current;

instrumentation/grpc-1.6/testing/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/AbstractGrpcTest.java

@@ -49,6 +49,8 @@
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Stream;
@@ -1370,6 +1372,87 @@ public void sayHello(
  SemanticAttributes.MESSAGE_ID, 2L))))));
  }
 
+ // Regression test for
+ // https:/open-telemetry/opentelemetry-java-instrumentation/issues/4169
+ @Test
+ void clientCallAfterServerCompleted() throws Exception {
+ Server backend =
+ configureServer(
+ ServerBuilder.forPort(0)
+ .addService(
+ new GreeterGrpc.GreeterImplBase() {
+ @Override
+ public void sayHello(
+ Helloworld.Request request,
+ StreamObserver<Helloworld.Response> responseObserver) {
+ responseObserver.onNext(
+ Helloworld.Response.newBuilder()
+ .setMessage(request.getName())
+ .build());
+ responseObserver.onCompleted();
+ }
+ }))
+ .build()
+ .start();
+ ManagedChannel backendChannel = createChannel(backend);
+ closer.add(() -> backendChannel.shutdownNow().awaitTermination(10, TimeUnit.SECONDS));
+ closer.add(() -> backend.shutdownNow().awaitTermination());
+ GreeterGrpc.GreeterBlockingStub backendStub = GreeterGrpc.newBlockingStub(backendChannel);
+
+ // This executor does not propagate context without the javaagent available.
+ ExecutorService executor = Executors.newFixedThreadPool(1);
+ closer.add(executor::shutdownNow);
+
+ CountDownLatch clientCallDone = new CountDownLatch(1);
+ AtomicReference<Throwable> error = new AtomicReference<>();
+
+ Server frontend =
+ configureServer(
+ ServerBuilder.forPort(0)
+ .addService(
+ new GreeterGrpc.GreeterImplBase() {
+ @Override
+ public void sayHello(
+ Helloworld.Request request,
+ StreamObserver<Helloworld.Response> responseObserver) {
+ responseObserver.onNext(
+ Helloworld.Response.newBuilder()
+ .setMessage(request.getName())
+ .build());
+ responseObserver.onCompleted();
+
+ executor.execute(
+ () -> {
+ try {
+ backendStub.sayHello(request);
+ } catch (Throwable t) {
+ error.set(t);
+ }
+ clientCallDone.countDown();
+ });
+ }
+ }))
+ .build()
+ .start();
+ ManagedChannel frontendChannel = createChannel(frontend);
+ closer.add(() -> frontendChannel.shutdownNow().awaitTermination(10, TimeUnit.SECONDS));
+ closer.add(() -> frontend.shutdownNow().awaitTermination());
+
+ GreeterGrpc.GreeterBlockingStub frontendStub = GreeterGrpc.newBlockingStub(frontendChannel);
+ frontendStub.sayHello(Helloworld.Request.newBuilder().setName("test").build());
+
+ // We don't assert on telemetry - the intention of this test is to verify that adding
+ // instrumentation, either as
+ // library or javaagent, does not cause exceptions in the business logic. The produced telemetry
+ // will be different
+ // for the two cases due to lack of context propagation in the library case, but that isn't what
+ // we're testing here.
+
+ clientCallDone.await(10, TimeUnit.SECONDS);
+
+ assertThat(error).hasValue(null);
+ }
+
  private ManagedChannel createChannel(Server server) throws Exception {
  ManagedChannelBuilder<?> channelBuilder =
  configureClient(ManagedChannelBuilder.forAddress("localhost", server.getPort()));