WIP feat(otel): Enable OTel semantics tests for Java weblog applications

tests/test_otel_http_semantics.py

@@ -184,10 +184,6 @@ def _span_is_error(span: HttpSpan) -> bool:
     return _status_code(status.get("code", StatusCode.STATUS_CODE_UNSET.value)) == StatusCode.STATUS_CODE_ERROR.value
 
 
-def _span_is_sampled(span: HttpSpan) -> bool:
-    return bool(int(span.get("flags", 0)) & 1)
-
-
 def _iter_otlp_spans(request: HttpResponse) -> Iterator[dict[str, Any]]:
     """Yield every span in the OTLP payloads associated with ``request``, each one once.
 
@@ -625,12 +621,12 @@ def test_status_500_is_error(self) -> None:
         )
 
     def setup_status_3xx_not_error(self) -> None:
-        self.response = _distant_call("http://weblog:7777/status?code=302")
+        self.response = _distant_call("http://weblog:7777/status?code=304")
 
     def test_status_3xx_not_error(self) -> None:
         span = _client_span(self.response)
-        _assert_int_attribute(span, "http.response.status_code", 302)
-        assert not _span_is_error(span), "HTTP 302 span must not be marked as an error"
+        _assert_int_attribute(span, "http.response.status_code", 304)
+        assert not _span_is_error(span), "HTTP 304 span must not be marked as an error"
         assert "error.type" not in _attributes(span)
 
     def setup_url_full_credential_redaction(self) -> None:
@@ -785,9 +781,6 @@ def test_otel_span_name_is_available_to_sampling_rules(self) -> None:
         assert trace_id, "the request-correlated matched span carries no traceId"
         matched_trace = [span for span in _iter_otlp_spans(self.matched_response) if span.get("traceId") == trace_id]
         assert matched_trace, "the matched trace must contain at least one exported span"
-        assert all(_span_is_sampled(span) for span in matched_trace), (
-            "every span in the trace kept by the HTTP* resource rule must carry the sampled flag"
-        )
 
     def setup_server_span_name_is_available_before_client_sampling(self) -> None:
         self.distant_call_response = _distant_call()
@@ -800,5 +793,3 @@ def test_server_span_name_is_available_before_client_sampling(self) -> None:
         assert _attributes(server_span).get("_dd.rule_psr") == 1, (
             "the GET* rule must keep the server trace whether or not the route was resolved before client sampling"
         )
-        assert _span_is_sampled(server_span)
-        assert _span_is_sampled(client_span), "sampling the server trace must keep its outbound HTTP client span"

utils/_context/_scenarios/otel_semantics.py

@@ -102,6 +102,7 @@ def __init__(self) -> None:
             weblog_env={
                 "DD_TRACE_OTEL_SEMANTICS_ENABLED": "true",
                 "DD_TRACE_OTEL_ENABLED": "true",
+                "DD_METRICS_OTEL_ENABLED": "true",
                 "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf",
                 "OTEL_TRACES_EXPORTER": "otlp",
                 "OTEL_TRACES_SPAN_METRICS_ENABLED": "true",
@@ -124,6 +125,9 @@ class OtelSemanticsOtlpSamplingRules(DdTraceEndToEndScenario):
     def __init__(self) -> None:
         super().__init__(
             "OTEL_SEMANTICS_OTLP_SAMPLING_RULES",
+            # Security findings can force-keep a trace before ordinary sampling rules run.
+            appsec_enabled=False,
+            iast_enabled=False,
             weblog_env={
                 "DD_TRACE_OTEL_SEMANTICS_ENABLED": "true",
                 "DD_TRACE_OTEL_ENABLED": "true",

utils/build/docker/java/akka-http/src/main/scala/com/datadoghq/akka_http/AppSecRoutes.scala

@@ -208,15 +208,15 @@ object AppSecRoutes {
                 span.setTag(if (decision == "keep") DDTags.MANUAL_KEEP else DDTags.MANUAL_DROP, true)
               }
               // Call downstream so that tests can assert on the sampling decision that gets propagated
-              complete(StatusCodes.OK, makeDistantCall("http://localhost:7777/"))(Marshaller.futureMarshaller(jsonMarshaller))
+              complete(StatusCodes.OK, makeDistantCall("http://localhost:7777/", "GET"))(Marshaller.futureMarshaller(jsonMarshaller))
             }
           }
         }
       } ~
       path("make_distant_call") {
         get {
-          parameter("url") { url =>
-            complete(StatusCodes.OK, makeDistantCall(url))(Marshaller.futureMarshaller(jsonMarshaller))
+          parameters("url", "method".withDefault("GET")) { (url, method) =>
+            complete(StatusCodes.OK, makeDistantCall(url, method))(Marshaller.futureMarshaller(jsonMarshaller))
           }
         }
       } ~
@@ -365,8 +365,9 @@ object AppSecRoutes {
                                   response_headers: Map[String, String]
                                 )
 
-  private def makeDistantCall(url: String): Future[DistantCallResponse] = {
-    val request = HttpRequest(uri = url)
+  private def makeDistantCall(url: String, method: String): Future[DistantCallResponse] = {
+    val httpMethod = HttpMethods.getForKey(method).getOrElse(HttpMethod.custom(method))
+    val request = HttpRequest(method = httpMethod, uri = url)
     val requestHeaders = request.headers.map(h => (h.name(), h.value())).toMap
 
     Http().singleRequest(request).map { response =>

utils/build/docker/java/akka-http/src/main/scala/com/datadoghq/akka_http/Main.scala

@@ -4,6 +4,7 @@ import akka.http.scaladsl.Http
 import akka.http.scaladsl.model._
 import akka.http.scaladsl.server.Directives._
 import akka.http.scaladsl.server.Route
+import akka.http.scaladsl.settings.ServerSettings
 
 import com.datadoghq.system_tests.iast.infra.{LdapServer, SqlServer}
 import org.slf4j.LoggerFactory
@@ -14,8 +15,12 @@ import scala.util.{Failure, Success, Try}
 
 object Main extends App {
 
+  // Accept an extension method so HTTP semantics tests reach request instrumentation.
+  private val settings = ServerSettings(system)
+  private val parserSettings = settings.parserSettings.withCustomMethods(HttpMethod.custom("PROPFIND"))
+
   private val bindingFuture: Future[Http.ServerBinding] =
-    Http().newServerAt("0.0.0.0", 7777).bindFlow(MainRoutes.route ~ AppSecRoutes.route ~ IastRoutes.route ~ RaspRoutes.route ~ HealthcheckRoutes.route)
+    Http().newServerAt("0.0.0.0", 7777).withSettings(settings.withParserSettings(parserSettings)).bindFlow(MainRoutes.route ~ AppSecRoutes.route ~ IastRoutes.route ~ RaspRoutes.route ~ HealthcheckRoutes.route)
 
   LoggerFactory.getLogger(this.getClass).info("Server online at port 7777")
 }

utils/build/docker/java/jersey-grizzly2/src/main/java/com/datadoghq/jersey/MyResource.java

@@ -413,37 +413,34 @@ public Response traceManualKeepDrop(@QueryParam("decision") String decision) thr
         }
 
         // Call downstream so that tests can assert on the sampling decision that gets propagated
-        String result = new ObjectMapper().writeValueAsString(make_distant_call("http://localhost:7777/"));
+        String result = new ObjectMapper().writeValueAsString(make_distant_call("http://localhost:7777/", "GET"));
 
         return Response.ok(result).type(MediaType.APPLICATION_JSON).build();
     }
 
     @GET
     @Path("/make_distant_call")
-    public DistantCallResponse make_distant_call(@QueryParam("url") String url) throws Exception {
-        URL urlObject = new URL(url);
-
-        HttpURLConnection con = (HttpURLConnection) urlObject.openConnection();
-        con.setRequestMethod("GET");
+    public DistantCallResponse make_distant_call(
+            @QueryParam("url") String url,
+            @DefaultValue("GET") @QueryParam("method") String method) throws Exception {
+        java.net.http.HttpRequest request = java.net.http.HttpRequest.newBuilder()
+                .uri(java.net.URI.create(url))
+                .method(method, java.net.http.HttpRequest.BodyPublishers.noBody())
+                .build();
+        java.net.http.HttpResponse<Void> response = java.net.http.HttpClient.newHttpClient().send(
+                request,
+                java.net.http.HttpResponse.BodyHandlers.discarding());
 
         // Save request headers
         HashMap<String, String> request_headers = new HashMap<String, String>();
-        for (Map.Entry<String, List<String>> header: con.getRequestProperties().entrySet()) {
-            if (header.getKey() == null) {
-                continue;
-            }
-
+        for (Map.Entry<String, List<String>> header: request.headers().map().entrySet()) {
             request_headers.put(header.getKey(), header.getValue().get(0));
         }
 
         // Save response headers and status code
-        int status_code = con.getResponseCode();
+        int status_code = response.statusCode();
         HashMap<String, String> response_headers = new HashMap<String, String>();
-        for (Map.Entry<String, List<String>> header: con.getHeaderFields().entrySet()) {
-            if (header.getKey() == null) {
-                continue;
-            }
-
+        for (Map.Entry<String, List<String>> header: response.headers().map().entrySet()) {
             response_headers.put(header.getKey(), header.getValue().get(0));
         }
 

utils/build/docker/java/play/app/SystemTestsAkkaHttpServerProvider.scala

@@ -0,0 +1,12 @@
+import akka.http.scaladsl.model.HttpMethod
+import akka.http.scaladsl.settings.ParserSettings
+import play.core.server.{AkkaHttpServer, AkkaHttpServerProvider, ServerProvider}
+
+/** Accept an extension method so HTTP semantics tests reach request instrumentation. */
+class SystemTestsAkkaHttpServerProvider extends AkkaHttpServerProvider {
+  override def createServer(context: ServerProvider.Context): AkkaHttpServer =
+    new AkkaHttpServer(AkkaHttpServer.Context.fromServerProviderContext(context)) {
+      override protected def createParserSettings(): ParserSettings =
+        super.createParserSettings().withCustomMethods(HttpMethod.custom("PROPFIND"))
+    }
+}

utils/build/docker/java/play/app/controllers/AppSecController.scala

@@ -193,16 +193,16 @@ class AppSecController @Inject()(cc: MessagesControllerComponents, ws: WSClient,
         span.setTag(if (decision == "keep") DDTags.MANUAL_KEEP else DDTags.MANUAL_DROP, true)
       }
       // Call downstream so that tests can assert on the sampling decision that gets propagated
-      makeDistantCall("http://localhost:7777/")
+      makeDistantCall("http://localhost:7777/", "GET")
     }
   }
 
-  def distantCall(url: String) = Action.async {
-    makeDistantCall(url)
+  def distantCall(url: String, method: String) = Action.async {
+    makeDistantCall(url, method)
   }
 
-  private def makeDistantCall(url: String): Future[Result] = {
-    val remoteReq: WSRequest = ws.url(url).withMethod("GET")
+  private def makeDistantCall(url: String, method: String): Future[Result] = {
+    val remoteReq: WSRequest = ws.url(url).withMethod(method)
 
     // we need to break the abstraction to be able to get to the request headers
     val ahcRequest: AHCRequest = remoteReq.asInstanceOf[AhcWSRequest].underlying.buildRequest()

utils/build/docker/java/play/conf/application.conf

@@ -1,2 +1,3 @@
 play.http.secret.key="7Yf2WqExHc0LqpDBYzVeHnns656b6Iw6ekgApxvWXOLyLZzaAGsepCecumbk6qh"
 http.port=7777
+play.server.provider="SystemTestsAkkaHttpServerProvider"

utils/build/docker/java/play/conf/routes

@@ -13,7 +13,7 @@ GET  /waf/                      controllers.AppSecController.waf
 GET  /waf/*segments            controllers.AppSecController.params(segments: Seq[String])
 POST /waf                      controllers.AppSecController.wafPost
 GET  /trace/manual_keep_drop   controllers.AppSecController.traceManualKeepDrop(decision: String)
-GET  /make_distant_call        controllers.AppSecController.distantCall(url: String)
+GET  /make_distant_call        controllers.AppSecController.distantCall(url: String, method: String ?= "GET")
 GET  /status                   controllers.AppSecController.status(code: Int)
 GET  /stats-unique             controllers.AppSecController.statsUnique(code: Option[Int])
 GET  /users                    controllers.AppSecController.users(user: String)

utils/build/docker/java/play/pom.xml

@@ -49,7 +49,6 @@
             <groupId>com.typesafe.play</groupId>
             <artifactId>play-akka-http-server_2.13</artifactId>
             <version>${play2.version}</version>
-            <scope>runtime</scope>
         </dependency>
 
         <dependency>

utils/build/docker/java/ratpack/src/main/java/com/datadoghq/ratpack/Main.java

@@ -231,30 +231,27 @@ public static void main(String[] args) throws Exception {
                             .get("make_distant_call", ctx -> {
                                 final Promise<String> res = Blocking.get(() -> {
                                     String url = ctx.getRequest().getQueryParams().get("url");
-
-                                    URL urlObject = new URL(url);
-
-                                    HttpURLConnection con = (HttpURLConnection) urlObject.openConnection();
-                                    con.setRequestMethod("GET");
+                                    String method = Optional.ofNullable(
+                                            ctx.getRequest().getQueryParams().get("method"))
+                                            .orElse("GET");
+                                    java.net.http.HttpRequest request = java.net.http.HttpRequest.newBuilder()
+                                            .uri(java.net.URI.create(url))
+                                            .method(method, java.net.http.HttpRequest.BodyPublishers.noBody())
+                                            .build();
+                                    java.net.http.HttpResponse<Void> response = java.net.http.HttpClient
+                                            .newHttpClient()
+                                            .send(request, java.net.http.HttpResponse.BodyHandlers.discarding());
 
                                     // Save request headers
                                     HashMap<String, String> request_headers = new HashMap<String, String>();
-                                    for (Map.Entry<String, List<String>> header : con.getRequestProperties().entrySet()) {
-                                        if (header.getKey() == null) {
-                                            continue;
-                                        }
-
+                                    for (Map.Entry<String, List<String>> header : request.headers().map().entrySet()) {
                                         request_headers.put(header.getKey(), header.getValue().get(0));
                                     }
 
                                     // Save response headers and status code
-                                    int status_code = con.getResponseCode();
+                                    int status_code = response.statusCode();
                                     HashMap<String, String> response_headers = new HashMap<String, String>();
-                                    for (Map.Entry<String, List<String>> header : con.getHeaderFields().entrySet()) {
-                                        if (header.getKey() == null) {
-                                            continue;
-                                        }
-
+                                    for (Map.Entry<String, List<String>> header : response.headers().map().entrySet()) {
                                         response_headers.put(header.getKey(), header.getValue().get(0));
                                     }
 
@@ -483,4 +480,3 @@ public static final class DistantCallResponse {
 
     public static final DataSource DATA_SOURCE = new SqlServer().start();
 }
-

utils/build/docker/java/resteasy-netty3/src/main/java/com/datadoghq/resteasy/MyResource.java

@@ -482,37 +482,34 @@ public Response traceManualKeepDrop(@QueryParam("decision") String decision) thr
         }
 
         // Call downstream so that tests can assert on the sampling decision that gets propagated
-        String result = new ObjectMapper().writeValueAsString(make_distant_call("http://localhost:7777/"));
+        String result = new ObjectMapper().writeValueAsString(make_distant_call("http://localhost:7777/", "GET"));
 
         return Response.ok(result).type(MediaType.APPLICATION_JSON).build();
     }
 
     @GET
     @Path("/make_distant_call")
-    public DistantCallResponse make_distant_call(@QueryParam("url") String url) throws Exception {
-        URL urlObject = new URL(url);
-
-        HttpURLConnection con = (HttpURLConnection) urlObject.openConnection();
-        con.setRequestMethod("GET");
+    public DistantCallResponse make_distant_call(
+            @QueryParam("url") String url,
+            @DefaultValue("GET") @QueryParam("method") String method) throws Exception {
+        java.net.http.HttpRequest request = java.net.http.HttpRequest.newBuilder()
+                .uri(java.net.URI.create(url))
+                .method(method, java.net.http.HttpRequest.BodyPublishers.noBody())
+                .build();
+        java.net.http.HttpResponse<Void> response = java.net.http.HttpClient.newHttpClient().send(
+                request,
+                java.net.http.HttpResponse.BodyHandlers.discarding());
 
         // Save request headers
         HashMap<String, String> request_headers = new HashMap<String, String>();
-        for (Map.Entry<String, List<String>> header: con.getRequestProperties().entrySet()) {
-            if (header.getKey() == null) {
-                continue;
-            }
-
+        for (Map.Entry<String, List<String>> header: request.headers().map().entrySet()) {
             request_headers.put(header.getKey(), header.getValue().get(0));
         }
 
         // Save response headers and status code
-        int status_code = con.getResponseCode();
+        int status_code = response.statusCode();
         HashMap<String, String> response_headers = new HashMap<String, String>();
-        for (Map.Entry<String, List<String>> header: con.getHeaderFields().entrySet()) {
-            if (header.getKey() == null) {
-                continue;
-            }
-
+        for (Map.Entry<String, List<String>> header: response.headers().map().entrySet()) {
             response_headers.put(header.getKey(), header.getValue().get(0));
         }
 

utils/build/docker/java/spring-boot-3-native/src/main/java/com/datadoghq/springbootnative/WebController.java

@@ -189,34 +189,31 @@ ResponseEntity<?> traceManualKeepDrop(@RequestParam String decision) throws Exce
     }
 
     // Call downstream so that tests can assert on the sampling decision that gets propagated
-    return ResponseEntity.ok(make_distant_call("http://localhost:7777/"));
+    return ResponseEntity.ok(make_distant_call("http://localhost:7777/", "GET"));
   }
 
   @RequestMapping("/make_distant_call")
-  DistantCallResponse make_distant_call(@RequestParam String url) throws Exception {
-    URL urlObject = new URL(url);
-
-    HttpURLConnection con = (HttpURLConnection) urlObject.openConnection();
-    con.setRequestMethod("GET");
+  DistantCallResponse make_distant_call(
+      @RequestParam String url,
+      @RequestParam(defaultValue = "GET") String method) throws Exception {
+    java.net.http.HttpRequest request = java.net.http.HttpRequest.newBuilder()
+        .uri(java.net.URI.create(url))
+        .method(method, java.net.http.HttpRequest.BodyPublishers.noBody())
+        .build();
+    java.net.http.HttpResponse<Void> response = java.net.http.HttpClient.newHttpClient().send(
+        request,
+        java.net.http.HttpResponse.BodyHandlers.discarding());
 
     // Save request headers
     HashMap<String, String> request_headers = new HashMap<String, String>();
-    for (Map.Entry<String, List<String>> header: con.getRequestProperties().entrySet()) {
-      if (header.getKey() == null) {
-        continue;
-      }
-
+    for (Map.Entry<String, List<String>> header: request.headers().map().entrySet()) {
       request_headers.put(header.getKey(), header.getValue().get(0));
     }
 
     // Save response headers and status code
-    int status_code = con.getResponseCode();
+    int status_code = response.statusCode();
     HashMap<String, String> response_headers = new HashMap<String, String>();
-    for (Map.Entry<String, List<String>> header: con.getHeaderFields().entrySet()) {
-        if (header.getKey() == null) {
-          continue;
-        }
-
+    for (Map.Entry<String, List<String>> header: response.headers().map().entrySet()) {
       response_headers.put(header.getKey(), header.getValue().get(0));
     }