test header capture

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

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

@@ -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,49 @@ 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/JavaExecutable.java

@@ -70,22 +70,20 @@ 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(),
+                  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/SmokeTest.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) {
@@ -381,33 +391,37 @@ protected static String bytesToHex(byte[] bytes) {
     }
     return sb.toString();
   }
-
   protected void doRequest(String url, IOConsumer<Response> responseHandler) {
-    Request request = new Request.Builder().url(url).get().build();
+    doRequest(url, Collections.emptyMap(), responseHandler);
+  }
 
-    try (Response response = client.newCall(request).execute()) {
-      responseHandler.accept(response);
-    } catch (IOException e) {
-      throw new RuntimeException(e);
+  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.build()).execute()) {
+        responseHandler.accept(response);
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
     }
-  }
 
-  @FunctionalInterface
-  protected interface IOConsumer<T> {
-    void accept(T t) throws IOException;
-  }
+    @FunctionalInterface
+    protected interface IOConsumer<T> {
+      void accept(T t) throws IOException;
+    }
 
-  protected static IOConsumer<Response> okResponse() {
-    return r -> {
-      assertThat(r.code()).isEqualTo(200);
-    };
-  }
+    protected static IOConsumer<Response> okResponse() {
+      return r -> {
+        assertThat(r.code()).isEqualTo(200);
+      };
+    }
 
-  protected static IOConsumer<Response> okResponseBody(String body) {
-    return r -> {
-      assertThat(r.code()).isEqualTo(200);
-      assertThat(r.body()).isNotNull();
-      assertThat(r.body().string()).isEqualTo(body);
-    };
+    protected static IOConsumer<Response> okResponseBody(String body) {
+      return r -> {
+        assertThat(r.code()).isEqualTo(200);
+        assertThat(r.body()).isNotNull();
+        assertThat(r.body().string()).isEqualTo(body);
+      };
+    }
   }
-}

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

@@ -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/src/main/java/co/elastic/otel/test/AppMain.java

@@ -21,8 +21,11 @@
 import co.elastic.otel.agent.attach.RuntimeAttach;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Configuration;
+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/MessagingController.java

@@ -0,0 +1,71 @@
+package co.elastic.otel.test;
+
+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;
+import java.util.Enumeration;
+
+@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/resources/application.properties

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