test header capture (#895)

Author: SylvainJuge

common/src/main/java/co/elastic/otel/common/ElasticAttributes.java

@@ -22,13 +22,6 @@
 import java.util.List;
 
 public interface ElasticAttributes {
-  AttributeKey<Long> SELF_TIME = AttributeKey.longKey("elastic.span.self_time");
-  AttributeKey<String> LOCAL_ROOT_ID = AttributeKey.stringKey("elastic.span.local_root.id");
-  AttributeKey<String> LOCAL_ROOT_NAME = AttributeKey.stringKey("elastic.local_root.name");
-  AttributeKey<String> LOCAL_ROOT_TYPE = AttributeKey.stringKey("elastic.local_root.type");
-  AttributeKey<Boolean> IS_LOCAL_ROOT = AttributeKey.booleanKey("elastic.span.is_local_root");
-  AttributeKey<String> SPAN_TYPE = AttributeKey.stringKey("elastic.span.type");
-  AttributeKey<String> SPAN_SUBTYPE = AttributeKey.stringKey("elastic.span.subtype");
 
   /**
    * Marker attribute for inferred spans. Does not have the elastic-prefix anymore because it has

custom/breakdown-metrics.md

@@ -23,26 +23,41 @@
 
 import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest;
 import io.opentelemetry.proto.trace.v1.Span;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
 class AgentFeaturesSmokeTest extends TestAppSmokeTest {
 
   @BeforeAll
-  public static void start() {
+  static void start() {
     startTestApp(
-        (container) -> container.addEnv("ELASTIC_OTEL_JAVA_SPAN_STACKTRACE_MIN_DURATION", "0ms"));
+        (container) -> {
+          // capture span stacktrace for any duration
+          container.addEnv("ELASTIC_OTEL_JAVA_SPAN_STACKTRACE_MIN_DURATION", "0ms");
+          // Capture HTTP request/response headers on server side
+          // header key should not be case-sensitive in config
+          container.addEnv("OTEL_INSTRUMENTATION_HTTP_SERVER_CAPTURE_REQUEST_HEADERS", "hello");
+          container.addEnv(
+              "OTEL_INSTRUMENTATION_HTTP_SERVER_CAPTURE_RESPONSE_HEADERS", "Content-length,Date");
+          // Capture messaging headers
+          // Header name IS case-sensitive, syntax may be limited by implementation, for example JMS
+          // requires it to be a valid java identifier.
+          container.addEnv(
+              "OTEL_INSTRUMENTATION_MESSAGING_EXPERIMENTAL_CAPTURE_HEADERS", "My_Header");
+        });
   }
 
   @AfterAll
-  public static void end() {
+  static void end() {
     stopApp();
   }
 
   @Test
-  public void healthcheck() throws InterruptedException {
+  void spanCodeStackTrace() {
     doRequest(getUrl("/health"), okResponseBody("Alive!"));
 
     List<ExportTraceServiceRequest> traces = waitForTraces();
@@ -53,14 +68,55 @@ public void healthcheck() throws InterruptedException {
         .containsOnly(tuple("GET /health", Span.SpanKind.SPAN_KIND_SERVER));
 
     spans.forEach(
-        span -> {
-          assertThat(getAttributes(span.getAttributesList()))
-              // span breakdown feature disabled
-              .doesNotContainKeys(
-                  "elastic.span.self_time",
-                  "elastic.span.is_local_root",
-                  "elastic.span.local_root.id")
-              .containsKeys("code.stacktrace");
-        });
+        span ->
+            assertThat(getAttributes(span.getAttributesList())).containsKeys("code.stacktrace"));
+  }
+
+  @Test
+  void httpHeaderCapture() {
+    Map<String, String> headers = new HashMap<>();
+    headers.put("Hello", "World!");
+    doRequest(getUrl("/health"), headers, okResponseBody("Alive!"));
+
+    List<ExportTraceServiceRequest> traces = waitForTraces();
+    List<Span> spans = getSpans(traces).toList();
+    assertThat(spans)
+        .hasSize(1)
+        .extracting("name", "kind")
+        .containsOnly(tuple("GET /health", Span.SpanKind.SPAN_KIND_SERVER));
+
+    spans.forEach(
+        span ->
+            assertThat(getAttributes(span.getAttributesList()))
+                .containsEntry("http.request.header.hello", attributeArrayValue("World!"))
+                .containsEntry("http.response.header.content-length", attributeArrayValue("6"))
+                .containsKey("http.response.header.date"));
+  }
+
+  @Test
+  void messagingHeaderCapture() {
+    doRequest(
+        getUrl("/messages/send?headerName=My_Header&headerValue=my-header-value"), okResponse());
+    doRequest(getUrl("/messages/receive"), okResponse());
+
+    List<ExportTraceServiceRequest> traces = waitForTraces();
+    List<Span> spans = getSpans(traces).toList();
+    assertThat(spans)
+        .hasSize(3)
+        .extracting("name", "kind")
+        .containsOnly(
+            tuple("GET /messages/send", Span.SpanKind.SPAN_KIND_SERVER),
+            tuple("messages-destination publish", Span.SpanKind.SPAN_KIND_PRODUCER),
+            tuple("GET /messages/receive", Span.SpanKind.SPAN_KIND_SERVER));
+
+    spans.stream()
+        .filter(span -> span.getKind() == Span.SpanKind.SPAN_KIND_PRODUCER)
+        .forEach(
+            span ->
+                assertThat(getAttributes(span.getAttributesList()))
+                    .containsEntry(
+                        "messaging.destination.name", attributeValue("messages-destination"))
+                    .containsEntry(
+                        "messaging.header.My_Header", attributeArrayValue("my-header-value")));
   }
 }

smoke-tests/src/test/java/com/example/javaagent/smoketest/AgentFeaturesSmokeTest.java

@@ -70,22 +70,18 @@ public static boolean isDebugInCI() {
 
   private static boolean probeListeningDebugger(int port) {
     // the most straightforward way to probe for an active debugger listening on port is to start
-    // another JVM
-    // with the debug options and check the process exit status. Trying to probe for open network
-    // port messes with
-    // the debugger and makes IDEA stop it. The only downside of this is that the debugger will
-    // first attach to this
-    // probe JVM, then the one running in a docker container we are aiming to debug.
+    // another JVM with the debug options and check the process exit status. Trying to probe for
+    // open network port messes with the debugger and makes IDEA stop it. The only downside of this
+    // is that the debugger will first attach to this probe JVM, then the one running in a docker
+    // container we are aiming to debug.
     try {
       Process process =
           new ProcessBuilder()
               .command(
-                  JavaExecutable.getBinaryPath().toString(),
-                  jvmDebugArgument("localhost", port),
-                  "-version")
+                  JavaExecutable.getBinaryPath(), jvmDebugArgument("localhost", port), "-version")
               .start();
-      process.waitFor(5, TimeUnit.SECONDS);
-      return process.exitValue() == 0;
+      boolean processExit = process.waitFor(5, TimeUnit.SECONDS);
+      return processExit && process.exitValue() == 0;
     } catch (InterruptedException | IOException e) {
       return false;
     }

smoke-tests/src/test/java/com/example/javaagent/smoketest/JavaExecutable.java

@@ -26,6 +26,7 @@
 import com.google.protobuf.util.JsonFormat;
 import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest;
 import io.opentelemetry.proto.common.v1.AnyValue;
+import io.opentelemetry.proto.common.v1.ArrayValue;
 import io.opentelemetry.proto.common.v1.KeyValue;
 import io.opentelemetry.proto.resource.v1.Resource;
 import io.opentelemetry.proto.trace.v1.ResourceSpans;
@@ -34,6 +35,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -355,6 +357,14 @@ protected static AnyValue attributeValue(String value) {
     return AnyValue.newBuilder().setStringValue(value).build();
   }
 
+  protected static AnyValue attributeArrayValue(String... values) {
+    ArrayValue.Builder valueBuilder = ArrayValue.newBuilder();
+    for (String value : values) {
+      valueBuilder.addValues(AnyValue.newBuilder().setStringValue(value).build());
+    }
+    return AnyValue.newBuilder().setArrayValue(valueBuilder.build()).build();
+  }
+
   protected static Map<String, AnyValue> getAttributes(List<KeyValue> list) {
     Map<String, AnyValue> attributes = new HashMap<>();
     for (KeyValue kv : list) {
@@ -383,9 +393,15 @@ protected static String bytesToHex(byte[] bytes) {
   }
 
   protected void doRequest(String url, IOConsumer<Response> responseHandler) {
-    Request request = new Request.Builder().url(url).get().build();
+    doRequest(url, Collections.emptyMap(), responseHandler);
+  }
+
+  protected void doRequest(
+      String url, Map<String, String> headers, IOConsumer<Response> responseHandler) {
+    Request.Builder request = new Request.Builder().url(url).get();
+    headers.forEach(request::addHeader);
 
-    try (Response response = client.newCall(request).execute()) {
+    try (Response response = client.newCall(request.build()).execute()) {
       responseHandler.accept(response);
     } catch (IOException e) {
       throw new RuntimeException(e);

smoke-tests/src/test/java/com/example/javaagent/smoketest/SmokeTest.java

@@ -16,12 +16,14 @@ dependencies {
   testImplementation("org.springframework.boot:spring-boot-starter-test:${springBootVersion}")
   implementation("org.springframework.boot:spring-boot-starter-web:${springBootVersion}")
 
-
   implementation("io.opentelemetry:opentelemetry-api")
   implementation("io.opentelemetry.instrumentation:opentelemetry-instrumentation-annotations")
 
   implementation(project(":runtime-attach"))
 
+  implementation("org.springframework.boot:spring-boot-starter-artemis:${springBootVersion}")
+  // using a rather old version to keep java 8 compatibility
+  implementation("org.apache.activemq:artemis-jms-server:2.27.0")
 }
 
 java {

smoke-tests/test-app/build.gradle.kts

@@ -21,8 +21,10 @@
 import co.elastic.otel.agent.attach.RuntimeAttach;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.jms.annotation.EnableJms;
 
 @SpringBootApplication
+@EnableJms
 public class AppMain {
 
   public static void main(String[] args) {

smoke-tests/test-app/src/main/java/co/elastic/otel/test/AppMain.java

@@ -0,0 +1,87 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package co.elastic.otel.test;
+
+import java.util.Enumeration;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.TextMessage;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/messages")
+public class MessagingController {
+
+  private static final String DESTINATION = "messages-destination";
+
+  @Autowired
+  public MessagingController(JmsTemplate jmsTemplate) {
+    this.jmsTemplate = jmsTemplate;
+  }
+
+  private final JmsTemplate jmsTemplate;
+
+  @RequestMapping("/send")
+  public String send(
+      @RequestParam(name = "headerName", required = false) String headerName,
+      @RequestParam(name = "headerValue", required = false) String headerValue) {
+    jmsTemplate.send(
+        DESTINATION,
+        session -> {
+          TextMessage message = session.createTextMessage("Hello World");
+          if (headerName != null && headerValue != null) {
+            message.setStringProperty(headerName, headerValue);
+          }
+          return message;
+        });
+    return null;
+  }
+
+  @RequestMapping("/receive")
+  public String receive() throws JMSException {
+    Message received = jmsTemplate.receive(DESTINATION);
+    if (received instanceof TextMessage) {
+      TextMessage textMessage = (TextMessage) received;
+      StringBuilder sb = new StringBuilder();
+      sb.append("message: [").append(textMessage.getText()).append("]");
+
+      Enumeration<?> propertyNames = textMessage.getPropertyNames();
+      if (propertyNames.hasMoreElements()) {
+        sb.append(", headers: [");
+        int count = 0;
+        while (propertyNames.hasMoreElements()) {
+          String propertyName = (String) propertyNames.nextElement();
+          sb.append(count++ > 0 ? ", " : "")
+              .append(propertyName)
+              .append(" = ")
+              .append(textMessage.getStringProperty(propertyName));
+        }
+        sb.append("]");
+      }
+
+      return sb.toString();
+    } else {
+      return "nothing received";
+    }
+  }
+}

smoke-tests/test-app/src/main/java/co/elastic/otel/test/MessagingController.java

@@ -0,0 +1,2 @@
+# use an in-process activemq artemis instance
+spring.artemis.mode=embedded