Fix http.server.active_requests metric with async requests (#11638)

Co-authored-by: Helen <56097766+heyams@users.noreply.github.com>
Co-authored-by: Trask Stalnaker <trask.stalnaker@gmail.com>

instrumentation-api/src/main/java/io/opentelemetry/instrumentation/api/instrumenter/Instrumenter.java

@@ -11,6 +11,7 @@
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.api.trace.Tracer;
 import io.opentelemetry.context.Context;
+import io.opentelemetry.context.ContextKey;
 import io.opentelemetry.instrumentation.api.internal.HttpRouteState;
 import io.opentelemetry.instrumentation.api.internal.InstrumenterAccess;
 import io.opentelemetry.instrumentation.api.internal.InstrumenterUtil;
@@ -41,6 +42,9 @@
  */
 public class Instrumenter<REQUEST, RESPONSE> {
 
+  private static final ContextKey<OperationListener[]> START_OPERATION_LISTENERS =
+      ContextKey.named("instrumenter-start-operation-listeners");
+
   /**
    * Returns a new {@link InstrumenterBuilder}.
    *
@@ -74,6 +78,7 @@ public static <REQUEST, RESPONSE> InstrumenterBuilder<REQUEST, RESPONSE> builder
   private final ContextCustomizer<? super REQUEST>[] contextCustomizers;
   private final OperationListener[] operationListeners;
   private final ErrorCauseExtractor errorCauseExtractor;
+  private final boolean propagateOperationListenersToOnEnd;
   private final boolean enabled;
   private final SpanSuppressor spanSuppressor;
 
@@ -89,6 +94,7 @@ public static <REQUEST, RESPONSE> InstrumenterBuilder<REQUEST, RESPONSE> builder
     this.contextCustomizers = builder.contextCustomizers.toArray(new ContextCustomizer[0]);
     this.operationListeners = builder.buildOperationListeners().toArray(new OperationListener[0]);
     this.errorCauseExtractor = builder.errorCauseExtractor;
+    this.propagateOperationListenersToOnEnd = builder.propagateOperationListenersToOnEnd;
     this.enabled = builder.enabled;
     this.spanSuppressor = builder.buildSpanSuppressor();
   }
@@ -198,6 +204,15 @@ private Context doStart(Context parentContext, REQUEST request, @Nullable Instan
         context = operationListeners[i].onStart(context, attributes, startNanos);
       }
     }
+    if (propagateOperationListenersToOnEnd || context.get(START_OPERATION_LISTENERS) != null) {
+      // when start and end are not called on the same instrumenter we need to use the operation
+      // listeners that were used during start in end to correctly handle metrics like
+      // http.server.active_requests that is recorded both in start and end
+      //
+      // need to also add when there is already START_OPERATION_LISTENERS, otherwise this
+      // instrumenter will call its parent's operation listeners in doEnd
+      context = context.with(START_OPERATION_LISTENERS, operationListeners);
+    }
 
     if (localRoot) {
       context = LocalRootSpan.store(context, span);
@@ -228,6 +243,10 @@ private void doEnd(
     }
     span.setAllAttributes(attributes);
 
+    OperationListener[] operationListeners = context.get(START_OPERATION_LISTENERS);
+    if (operationListeners == null) {
+      operationListeners = this.operationListeners;
+    }
     if (operationListeners.length != 0) {
       long endNanos = getNanos(endTime);
       for (int i = operationListeners.length - 1; i >= 0; i--) {

instrumentation-api/src/main/java/io/opentelemetry/instrumentation/api/instrumenter/InstrumenterBuilder.java

@@ -66,6 +66,7 @@ public final class InstrumenterBuilder<REQUEST, RESPONSE> {
   SpanStatusExtractor<? super REQUEST, ? super RESPONSE> spanStatusExtractor =
       SpanStatusExtractor.getDefault();
   ErrorCauseExtractor errorCauseExtractor = ErrorCauseExtractor.getDefault();
+  boolean propagateOperationListenersToOnEnd = false;
   boolean enabled = true;
 
   InstrumenterBuilder(
@@ -370,6 +371,10 @@ private Set<SpanKey> getSpanKeysFromAttributesExtractors() {
         .collect(Collectors.toSet());
   }
 
+  private void propagateOperationListenersToOnEnd() {
+    propagateOperationListenersToOnEnd = true;
+  }
+
   private interface InstrumenterConstructor<RQ, RS> {
     Instrumenter<RQ, RS> create(InstrumenterBuilder<RQ, RS> builder);
 
@@ -406,6 +411,12 @@ public <RQ, RS> Instrumenter<RQ, RS> buildDownstreamInstrumenter(
               SpanKindExtractor<RQ> spanKindExtractor) {
             return builder.buildDownstreamInstrumenter(setter, spanKindExtractor);
           }
+
+          @Override
+          public <RQ, RS> void propagateOperationListenersToOnEnd(
+              InstrumenterBuilder<RQ, RS> builder) {
+            builder.propagateOperationListenersToOnEnd();
+          }
         });
   }
 }

instrumentation-api/src/main/java/io/opentelemetry/instrumentation/api/internal/InstrumenterBuilderAccess.java

@@ -26,4 +26,7 @@ <REQUEST, RESPONSE> Instrumenter<REQUEST, RESPONSE> buildDownstreamInstrumenter(
       InstrumenterBuilder<REQUEST, RESPONSE> builder,
       TextMapSetter<REQUEST> setter,
       SpanKindExtractor<REQUEST> spanKindExtractor);
+
+  <REQUEST, RESPONSE> void propagateOperationListenersToOnEnd(
+      InstrumenterBuilder<REQUEST, RESPONSE> builder);
 }

instrumentation-api/src/main/java/io/opentelemetry/instrumentation/api/internal/InstrumenterUtil.java

@@ -67,5 +67,11 @@ public static <REQUEST, RESPONSE> Instrumenter<REQUEST, RESPONSE> buildDownstrea
         builder, setter, spanKindExtractor);
   }
 
+  public static <REQUEST, RESPONSE> void propagateOperationListenersToOnEnd(
+      InstrumenterBuilder<REQUEST, RESPONSE> builder) {
+    // instrumenterBuilderAccess is guaranteed to be non-null here
+    instrumenterBuilderAccess.propagateOperationListenersToOnEnd(builder);
+  }
+
   private InstrumenterUtil() {}
 }

instrumentation/jetty/jetty-11.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/jetty/v11_0/Jetty11Singletons.java

@@ -24,6 +24,7 @@ public final class Jetty11Singletons {
           ServletInstrumenterBuilder.<HttpServletRequest, HttpServletResponse>create()
               .addContextCustomizer(
                   (context, request, attributes) -> new AppServerBridge.Builder().init(context))
+              .propagateOperationListenersToOnEnd()
               .build(INSTRUMENTATION_NAME, Servlet5Accessor.INSTANCE);
 
   private static final JettyHelper<HttpServletRequest, HttpServletResponse> HELPER =

instrumentation/jetty/jetty-8.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/jetty/v8_0/Jetty8Singletons.java

@@ -24,6 +24,7 @@ public final class Jetty8Singletons {
           ServletInstrumenterBuilder.<HttpServletRequest, HttpServletResponse>create()
               .addContextCustomizer(
                   (context, request, attributes) -> new AppServerBridge.Builder().init(context))
+              .propagateOperationListenersToOnEnd()
               .build(INSTRUMENTATION_NAME, Servlet3Accessor.INSTANCE);
 
   private static final JettyHelper<HttpServletRequest, HttpServletResponse> HELPER =

instrumentation/liberty/liberty-20.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/liberty/LibertySingletons.java

@@ -24,6 +24,7 @@ public final class LibertySingletons {
               .addContextCustomizer(
                   (context, request, attributes) ->
                       new AppServerBridge.Builder().recordException().init(context))
+              .propagateOperationListenersToOnEnd()
               .build(INSTRUMENTATION_NAME, Servlet3Accessor.INSTANCE);
 
   private static final LibertyHelper<HttpServletRequest, HttpServletResponse> HELPER =

instrumentation/servlet/servlet-3.0/javaagent/src/test/groovy/JettyServlet3Test.groovy

@@ -149,13 +149,6 @@ class JettyServlet3TestAsync extends JettyServlet3Test {
   boolean isAsyncTest() {
     true
   }
-
-  @Override
-  String getMetricsInstrumentationName() {
-    // with async requests the span is started in one instrumentation (server instrumentation)
-    // but ended from another (servlet instrumentation)
-    "io.opentelemetry.servlet-3.0"
-  }
 }
 
 class JettyServlet3TestFakeAsync extends JettyServlet3Test {
@@ -164,13 +157,6 @@ class JettyServlet3TestFakeAsync extends JettyServlet3Test {
   Class<Servlet> servlet() {
     TestServlet3.FakeAsync
   }
-
-  @Override
-  String getMetricsInstrumentationName() {
-    // with async requests the span is started in one instrumentation (server instrumentation)
-    // but ended from another (servlet instrumentation)
-    "io.opentelemetry.servlet-3.0"
-  }
 }
 
 class JettyServlet3TestForward extends JettyDispatchTest {
@@ -247,13 +233,6 @@ class JettyServlet3TestDispatchImmediate extends JettyDispatchTest {
     true
   }
 
-  @Override
-  String getMetricsInstrumentationName() {
-    // with async requests the span is started in one instrumentation (server instrumentation)
-    // but ended from another (servlet instrumentation)
-    "io.opentelemetry.servlet-3.0"
-  }
-
   @Override
   protected void setupServlets(ServletContextHandler context) {
     super.setupServlets(context)
@@ -283,13 +262,6 @@ class JettyServlet3TestDispatchAsync extends JettyDispatchTest {
     true
   }
 
-  @Override
-  String getMetricsInstrumentationName() {
-    // with async requests the span is started in one instrumentation (server instrumentation)
-    // but ended from another (servlet instrumentation)
-    "io.opentelemetry.servlet-3.0"
-  }
-
   @Override
   protected void setupServlets(ServletContextHandler context) {
     super.setupServlets(context)

instrumentation/servlet/servlet-5.0/javaagent/src/test/groovy/JettyServlet5Test.groovy

@@ -123,13 +123,6 @@ class JettyServlet5TestAsync extends JettyServlet5Test {
   boolean errorEndpointUsesSendError() {
     false
   }
-
-  @Override
-  String getMetricsInstrumentationName() {
-    // with async requests the span is started in one instrumentation (server instrumentation)
-    // but ended from another (servlet instrumentation)
-    "io.opentelemetry.servlet-5.0"
-  }
 }
 
 @IgnoreIf({ !jvm.java11Compatible })
@@ -139,13 +132,6 @@ class JettyServlet5TestFakeAsync extends JettyServlet5Test {
   Class<Servlet> servlet() {
     TestServlet5.FakeAsync
   }
-
-  @Override
-  String getMetricsInstrumentationName() {
-    // with async requests the span is started in one instrumentation (server instrumentation)
-    // but ended from another (servlet instrumentation)
-    "io.opentelemetry.servlet-5.0"
-  }
 }
 
 @IgnoreIf({ !jvm.java11Compatible })
@@ -237,13 +223,6 @@ class JettyServlet5TestDispatchImmediate extends JettyDispatchTest {
     addServlet(context, "/dispatch" + INDEXED_CHILD.path, TestServlet5.DispatchImmediate)
     addServlet(context, "/dispatch/recursive", TestServlet5.DispatchRecursive)
   }
-
-  @Override
-  String getMetricsInstrumentationName() {
-    // with async requests the span is started in one instrumentation (server instrumentation)
-    // but ended from another (servlet instrumentation)
-    "io.opentelemetry.servlet-5.0"
-  }
 }
 
 @IgnoreIf({ !jvm.java11Compatible })
@@ -275,13 +254,6 @@ class JettyServlet5TestDispatchAsync extends JettyDispatchTest {
   boolean errorEndpointUsesSendError() {
     false
   }
-
-  @Override
-  String getMetricsInstrumentationName() {
-    // with async requests the span is started in one instrumentation (server instrumentation)
-    // but ended from another (servlet instrumentation)
-    "io.opentelemetry.servlet-5.0"
-  }
 }
 
 abstract class JettyDispatchTest extends JettyServlet5Test {

instrumentation/servlet/servlet-common/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/servlet/ServletInstrumenterBuilder.java

@@ -14,6 +14,7 @@
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
 import io.opentelemetry.instrumentation.api.instrumenter.SpanNameExtractor;
+import io.opentelemetry.instrumentation.api.internal.InstrumenterUtil;
 import io.opentelemetry.instrumentation.api.semconv.http.HttpServerAttributesExtractor;
 import io.opentelemetry.instrumentation.api.semconv.http.HttpServerAttributesGetter;
 import io.opentelemetry.instrumentation.api.semconv.http.HttpServerMetrics;
@@ -31,6 +32,8 @@ private ServletInstrumenterBuilder() {}
   private final List<ContextCustomizer<? super ServletRequestContext<REQUEST>>> contextCustomizers =
       new ArrayList<>();
 
+  private boolean propagateOperationListenersToOnEnd;
+
   public static <REQUEST, RESPONSE> ServletInstrumenterBuilder<REQUEST, RESPONSE> create() {
     return new ServletInstrumenterBuilder<>();
   }
@@ -42,6 +45,12 @@ public ServletInstrumenterBuilder<REQUEST, RESPONSE> addContextCustomizer(
     return this;
   }
 
+  @CanIgnoreReturnValue
+  public ServletInstrumenterBuilder<REQUEST, RESPONSE> propagateOperationListenersToOnEnd() {
+    propagateOperationListenersToOnEnd = true;
+    return this;
+  }
+
   public Instrumenter<ServletRequestContext<REQUEST>, ServletResponseContext<RESPONSE>> build(
       String instrumentationName,
       ServletAccessor<REQUEST, RESPONSE> accessor,
@@ -85,6 +94,9 @@ public Instrumenter<ServletRequestContext<REQUEST>, ServletResponseContext<RESPO
           .addAttributesExtractor(HttpExperimentalAttributesExtractor.create(httpAttributesGetter))
           .addOperationMetrics(HttpServerExperimentalMetrics.get());
     }
+    if (propagateOperationListenersToOnEnd) {
+      InstrumenterUtil.propagateOperationListenersToOnEnd(builder);
+    }
     return builder.buildServerInstrumenter(new ServletRequestGetter<>(accessor));
   }
 

instrumentation/tomcat/tomcat-10.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/tomcat/v10_0/TomcatAsyncTest.java

@@ -113,10 +113,6 @@ protected void configure(HttpServerTestOptions options) {
     options.setExpectedException(new ServletException(EXCEPTION.getBody()));
 
     options.setHasResponseSpan(endpoint -> endpoint == NOT_FOUND || endpoint == REDIRECT);
-
-    // with async requests the span is started in one instrumentation (server instrumentation)
-    // but ended from another (servlet instrumentation)
-    options.setMetricsInstrumentationName(() -> "io.opentelemetry.servlet-5.0");
   }
 
   @Override

instrumentation/tomcat/tomcat-7.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/tomcat/v7_0/TomcatAsyncTest.java

@@ -108,10 +108,6 @@ protected void configure(HttpServerTestOptions options) {
     options.setExpectedException(new ServletException(EXCEPTION.getBody()));
 
     options.setHasResponseSpan(endpoint -> endpoint == NOT_FOUND || endpoint == REDIRECT);
-
-    // with async requests the span is started in one instrumentation (server instrumentation)
-    // but ended from another (servlet instrumentation)
-    options.setMetricsInstrumentationName(() -> "io.opentelemetry.servlet-3.0");
   }
 
   @Override

instrumentation/tomcat/tomcat-common/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/tomcat/common/TomcatInstrumenterFactory.java

@@ -10,6 +10,7 @@
 import io.opentelemetry.instrumentation.api.incubator.semconv.http.HttpServerExperimentalMetrics;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
+import io.opentelemetry.instrumentation.api.internal.InstrumenterUtil;
 import io.opentelemetry.instrumentation.api.semconv.http.HttpServerAttributesExtractor;
 import io.opentelemetry.instrumentation.api.semconv.http.HttpServerMetrics;
 import io.opentelemetry.instrumentation.api.semconv.http.HttpServerRoute;
@@ -61,6 +62,7 @@ public static <REQUEST, RESPONSE> Instrumenter<Request, Response> create(
           .addAttributesExtractor(HttpExperimentalAttributesExtractor.create(httpAttributesGetter))
           .addOperationMetrics(HttpServerExperimentalMetrics.get());
     }
+    InstrumenterUtil.propagateOperationListenersToOnEnd(builder);
     return builder.buildServerInstrumenter(TomcatRequestGetter.INSTANCE);
   }
 }

testing-common/src/main/groovy/io/opentelemetry/instrumentation/test/base/HttpServerTest.groovy

@@ -173,10 +173,6 @@ abstract class HttpServerTest<SERVER> extends InstrumentationSpecification imple
     return true
   }
 
-  String getMetricsInstrumentationName() {
-    null
-  }
-
   /** A list of additional HTTP server span attributes extracted by the instrumentation per URI. */
   Set<AttributeKey<?>> httpAttributes(ServerEndpoint endpoint) {
     [
@@ -237,9 +233,6 @@ abstract class HttpServerTest<SERVER> extends InstrumentationSpecification imple
       options.sockPeerAddr = { endpoint ->
         HttpServerTest.this.sockPeerAddr(endpoint)
       }
-      options.metricsInstrumentationName = {
-        HttpServerTest.this.getMetricsInstrumentationName()
-      }
       options.responseCodeOnNonStandardHttpMethod = getResponseCodeOnNonStandardHttpMethod()
 
       options.testRedirect = testRedirect()

testing-common/src/main/java/io/opentelemetry/instrumentation/testing/junit/http/AbstractHttpServerTest.java

@@ -353,12 +353,8 @@ void httpServerMetrics() {
               spanData -> assertServerSpan(assertThat(spanData), method, SUCCESS, SUCCESS.status));
         });
 
-    String metricsInstrumentationName = options.metricsInstrumentationName.get();
-    if (metricsInstrumentationName == null) {
-      metricsInstrumentationName = instrumentationName.get();
-    }
     testing.waitAndAssertMetrics(
-        metricsInstrumentationName,
+        instrumentationName.get(),
         "http.server.request.duration",
         metrics ->
             metrics.anySatisfy(

testing-common/src/main/java/io/opentelemetry/instrumentation/testing/junit/http/HttpServerTestOptions.java

@@ -19,7 +19,6 @@
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Predicate;
-import java.util.function.Supplier;
 import javax.annotation.Nullable;
 
 public final class HttpServerTestOptions {
@@ -42,7 +41,6 @@ public final class HttpServerTestOptions {
   Function<ServerEndpoint, String> sockPeerAddr = unused -> "127.0.0.1";
   String contextPath = "";
   Throwable expectedException = new Exception(EXCEPTION.body);
-  Supplier<String> metricsInstrumentationName = () -> null;
   // we're calling /success in the test, and most servers respond with 200 anyway
   int responseCodeOnNonStandardHttpMethod = ServerEndpoint.SUCCESS.status;
 
@@ -108,13 +106,6 @@ public HttpServerTestOptions setExpectedException(Throwable expectedException) {
     return this;
   }
 
-  @CanIgnoreReturnValue
-  public HttpServerTestOptions setMetricsInstrumentationName(
-      Supplier<String> metricsInstrumentationName) {
-    this.metricsInstrumentationName = metricsInstrumentationName;
-    return this;
-  }
-
   @CanIgnoreReturnValue
   public HttpServerTestOptions setResponseCodeOnNonStandardHttpMethod(
       int responseCodeOnNonStandardHttpMethod) {