Add Netty client payload and headers data capture

README.md

@@ -18,13 +18,13 @@ List of supported frameworks with additional capabilities:
 | [Apache HttpClient](https://hc.apache.org/index.html)                                                  | 4.0+            |
 | [gRPC](https://github.com/grpc/grpc-java)                                                              | 1.5+            |
 | [JAX-RS Client](https://javaee.github.io/javaee-spec/javadocs/javax/ws/rs/client/package-summary.html) | 2.0+            |
-| [Micronaut](https://micronaut.io/)  (via Netty and only server)                                        | 1.0+            |
-| [Netty](https://github.com/netty/netty)  (only server)                                                 | 4.0+            |
+| [Micronaut](https://micronaut.io/)  (basic support via Netty)                                          | 1.0+            |
+| [Netty](https://github.com/netty/netty)                                                                | 4.0+            |
 | [OkHttp](https://github.com/square/okhttp/)                                                            | 3.0+            |
 | [Servlet](https://javaee.github.io/javaee-spec/javadocs/javax/servlet/package-summary.html)            | 2.3+            |
 | [Spark Web Framework](https://github.com/perwendel/spark)                                              | 2.3+            |
-| [Spring Webflux](https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/web/reactive/package-summary.html) (only server)       | 5.0+            |
-| [Vert.x](https://vertx.io) (only server)                                                               | 3.0+            |
+| [Spring Webflux](https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/web/reactive/package-summary.html)       | 5.0+            |
+| [Vert.x](https://vertx.io)                                                                             | 3.0+            |
 
 ### Adding custom filter implementation
 

instrumentation/micronaut-1.0/build.gradle.kts

@@ -17,6 +17,7 @@ dependencies {
     testImplementation("io.micronaut:micronaut-http-server-netty:${micronautVersion}")
     testImplementation("io.micronaut:micronaut-runtime:${micronautVersion}")
     testImplementation("io.micronaut:micronaut-inject:${micronautVersion}")
+    testImplementation("io.micronaut:micronaut-http-client:${micronautVersion}")
     testAnnotationProcessor("io.micronaut:micronaut-inject-java:${micronautVersion}")
 }
 

instrumentation/micronaut-1.0/src/test/java/io/opentelemetry/javaagent/instrumentation/hypertrace/micronaut/MicronautClientInstrumentationTest.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright The Hypertrace Authors
+ *
+ * Licensed 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 io.opentelemetry.javaagent.instrumentation.hypertrace.micronaut;
+
+import io.micronaut.http.HttpRequest;
+import io.micronaut.http.HttpResponse;
+import io.micronaut.http.client.HttpClient;
+import io.micronaut.http.client.annotation.Client;
+import io.micronaut.test.annotation.MicronautTest;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import javax.inject.Inject;
+import org.hypertrace.agent.core.instrumentation.HypertraceSemanticAttributes;
+import org.hypertrace.agent.testing.AbstractInstrumenterTest;
+import org.hypertrace.agent.testing.TestHttpServer;
+import org.hypertrace.agent.testing.TestHttpServer.GetJsonHandler;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+@MicronautTest
+public class MicronautClientInstrumentationTest extends AbstractInstrumenterTest {
+
+  private static final String REQUEST_BODY = "hello_foo_bar";
+  private static final String REQUEST_HEADER_NAME = "reqheadername";
+  private static final String REQUEST_HEADER_VALUE = "reqheadervalue";
+
+  private static final TestHttpServer testHttpServer = new TestHttpServer();
+
+  @BeforeAll
+  public static void startServer() throws Exception {
+    testHttpServer.start();
+  }
+
+  @AfterAll
+  public static void closeServer() throws Exception {
+    testHttpServer.close();
+  }
+
+  @Inject
+  @Client("/")
+  private HttpClient client;
+
+  @Test
+  public void getJson() throws InterruptedException, TimeoutException {
+    String retrieve =
+        client
+            .toBlocking()
+            .retrieve(
+                HttpRequest.GET(
+                        String.format("http://localhost:%d/get_json", testHttpServer.port()))
+                    .header(REQUEST_HEADER_NAME, REQUEST_HEADER_VALUE));
+    Assertions.assertEquals(GetJsonHandler.RESPONSE_BODY, retrieve);
+
+    TEST_WRITER.waitForTraces(1);
+    List<List<SpanData>> traces = TEST_WRITER.getTraces();
+    Assertions.assertEquals(1, traces.size());
+    Assertions.assertEquals(1, traces.get(0).size());
+    SpanData clientSpan = traces.get(0).get(0);
+    Assertions.assertEquals(
+        REQUEST_HEADER_VALUE,
+        clientSpan
+            .getAttributes()
+            .get(HypertraceSemanticAttributes.httpRequestHeader(REQUEST_HEADER_NAME)));
+    Assertions.assertEquals(
+        TestHttpServer.RESPONSE_HEADER_VALUE,
+        clientSpan
+            .getAttributes()
+            .get(
+                HypertraceSemanticAttributes.httpResponseHeader(
+                    TestHttpServer.RESPONSE_HEADER_NAME)));
+    Assertions.assertNull(
+        clientSpan.getAttributes().get(HypertraceSemanticAttributes.HTTP_REQUEST_BODY));
+    Assertions.assertEquals(
+        GetJsonHandler.RESPONSE_BODY,
+        clientSpan.getAttributes().get(HypertraceSemanticAttributes.HTTP_RESPONSE_BODY));
+  }
+
+  @Test
+  public void post() throws InterruptedException, TimeoutException {
+    HttpResponse<Object> response =
+        client
+            .toBlocking()
+            .exchange(
+                HttpRequest.POST(
+                        String.format("http://localhost:%d/post", testHttpServer.port()),
+                        REQUEST_BODY)
+                    .header(REQUEST_HEADER_NAME, REQUEST_HEADER_VALUE)
+                    .header("Content-Type", "application/json"));
+    Assertions.assertEquals(204, response.getStatus().getCode());
+
+    TEST_WRITER.waitForTraces(1);
+    List<List<SpanData>> traces = TEST_WRITER.getTraces();
+    Assertions.assertEquals(1, traces.size());
+    Assertions.assertEquals(1, traces.get(0).size());
+    SpanData clientSpan = traces.get(0).get(0);
+    Assertions.assertEquals(
+        REQUEST_HEADER_VALUE,
+        clientSpan
+            .getAttributes()
+            .get(HypertraceSemanticAttributes.httpRequestHeader(REQUEST_HEADER_NAME)));
+    Assertions.assertEquals(
+        TestHttpServer.RESPONSE_HEADER_VALUE,
+        clientSpan
+            .getAttributes()
+            .get(
+                HypertraceSemanticAttributes.httpResponseHeader(
+                    TestHttpServer.RESPONSE_HEADER_NAME)));
+    Assertions.assertEquals(
+        REQUEST_BODY,
+        clientSpan.getAttributes().get(HypertraceSemanticAttributes.HTTP_REQUEST_BODY));
+    Assertions.assertNull(
+        clientSpan.getAttributes().get(HypertraceSemanticAttributes.HTTP_RESPONSE_BODY));
+  }
+}

instrumentation/netty/netty-4.0/build.gradle.kts

@@ -47,5 +47,6 @@ dependencies {
     implementation("io.netty:netty-codec-http:4.0.0.Final")
 
     testImplementation(project(":testing-common"))
+    testImplementation("org.asynchttpclient:async-http-client:2.0.9")
 }
 

instrumentation/netty/netty-4.0/src/main/java/io/opentelemetry/javaagent/instrumentation/hypertrace/netty/v4_0/server/DataCaptureUtils.java → instrumentation/netty/netty-4.0/src/main/java/io/opentelemetry/javaagent/instrumentation/hypertrace/netty/v4_0/DataCaptureUtils.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.opentelemetry.javaagent.instrumentation.hypertrace.netty.v4_0.server;
+package io.opentelemetry.javaagent.instrumentation.hypertrace.netty.v4_0;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;

instrumentation/netty/netty-4.0/src/main/java/io/opentelemetry/javaagent/instrumentation/hypertrace/netty/v4_0/NettyChannelPipelineInstrumentation.java

@@ -25,10 +25,16 @@
 
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.http.HttpClientCodec;
 import io.netty.handler.codec.http.HttpRequestDecoder;
+import io.netty.handler.codec.http.HttpRequestEncoder;
+import io.netty.handler.codec.http.HttpResponseDecoder;
 import io.netty.handler.codec.http.HttpResponseEncoder;
 import io.netty.handler.codec.http.HttpServerCodec;
 import io.opentelemetry.javaagent.instrumentation.api.CallDepthThreadLocalMap;
+import io.opentelemetry.javaagent.instrumentation.hypertrace.netty.v4_0.client.HttpClientRequestTracingHandler;
+import io.opentelemetry.javaagent.instrumentation.hypertrace.netty.v4_0.client.HttpClientResponseTracingHandler;
+import io.opentelemetry.javaagent.instrumentation.hypertrace.netty.v4_0.client.HttpClientTracingHandler;
 import io.opentelemetry.javaagent.instrumentation.hypertrace.netty.v4_0.server.HttpServerBlockingRequestHandler;
 import io.opentelemetry.javaagent.instrumentation.hypertrace.netty.v4_0.server.HttpServerRequestTracingHandler;
 import io.opentelemetry.javaagent.instrumentation.hypertrace.netty.v4_0.server.HttpServerResponseTracingHandler;
@@ -123,22 +129,36 @@ public static void addHandler(
           pipeline.addLast(
               HttpServerBlockingRequestHandler.class.getName(),
               new HttpServerBlockingRequestHandler());
+        } else
+        //         Client pipeline handlers
+        if (handler instanceof HttpClientCodec) {
+          pipeline.replace(
+              io.opentelemetry.javaagent.instrumentation.netty.v4_0.client.HttpClientTracingHandler
+                  .class
+                  .getName(),
+              HttpClientTracingHandler.class.getName(),
+              new HttpClientTracingHandler());
+
+          // add OTEL request handler to start spans
+          pipeline.addAfter(
+              HttpClientTracingHandler.class.getName(),
+              io.opentelemetry.javaagent.instrumentation.netty.v4_0.client
+                  .HttpClientRequestTracingHandler.class
+                  .getName(),
+              new io.opentelemetry.javaagent.instrumentation.netty.v4_0.client
+                  .HttpClientRequestTracingHandler());
+        } else if (handler instanceof HttpRequestEncoder) {
+          pipeline.addLast(
+              HttpClientRequestTracingHandler.class.getName(),
+              new HttpClientRequestTracingHandler());
+        } else if (handler instanceof HttpResponseDecoder) {
+          pipeline.replace(
+              io.opentelemetry.javaagent.instrumentation.netty.v4_0.client
+                  .HttpClientResponseTracingHandler.class
+                  .getName(),
+              HttpClientResponseTracingHandler.class.getName(),
+              new HttpClientResponseTracingHandler());
         }
-        // TODO add client instrumentation
-        //        else
-        // Client pipeline handlers
-        //          if (handler instanceof HttpClientCodec) {
-        //            pipeline.addLast(
-        //                HttpClientTracingHandler.class.getName(), new HttpClientTracingHandler());
-        //          } else if (handler instanceof HttpRequestEncoder) {
-        //            pipeline.addLast(
-        //                HttpClientRequestTracingHandler.class.getName(),
-        //                new HttpClientRequestTracingHandler());
-        //          } else if (handler instanceof HttpResponseDecoder) {
-        //            pipeline.addLast(
-        //                HttpClientResponseTracingHandler.class.getName(),
-        //                new HttpClientResponseTracingHandler());
-        //          }
       } catch (IllegalArgumentException e) {
         // Prevented adding duplicate handlers.
       }

instrumentation/netty/netty-4.0/src/main/java/io/opentelemetry/javaagent/instrumentation/hypertrace/netty/v4_0/client/HttpClientRequestTracingHandler.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright The Hypertrace Authors
+ *
+ * Licensed 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 io.opentelemetry.javaagent.instrumentation.hypertrace.netty.v4_0.client;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.http.HttpContent;
+import io.netty.handler.codec.http.HttpMessage;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.util.Attribute;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.javaagent.instrumentation.hypertrace.netty.v4_0.AttributeKeys;
+import io.opentelemetry.javaagent.instrumentation.hypertrace.netty.v4_0.DataCaptureUtils;
+import java.nio.charset.Charset;
+import java.util.HashMap;
+import java.util.Map;
+import org.hypertrace.agent.config.Config.AgentConfig;
+import org.hypertrace.agent.core.config.HypertraceConfig;
+import org.hypertrace.agent.core.instrumentation.HypertraceSemanticAttributes;
+import org.hypertrace.agent.core.instrumentation.buffer.BoundedBuffersFactory;
+import org.hypertrace.agent.core.instrumentation.buffer.BoundedByteArrayOutputStream;
+import org.hypertrace.agent.core.instrumentation.utils.ContentLengthUtils;
+import org.hypertrace.agent.core.instrumentation.utils.ContentTypeCharsetUtils;
+import org.hypertrace.agent.core.instrumentation.utils.ContentTypeUtils;
+
+public class HttpClientRequestTracingHandler extends ChannelOutboundHandlerAdapter {
+
+  private final AgentConfig agentConfig = HypertraceConfig.get();
+
+  @Override
+  public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise prm) {
+    Channel channel = ctx.channel();
+    Context context =
+        channel
+            .attr(
+                io.opentelemetry.javaagent.instrumentation.netty.v4_0.AttributeKeys.CLIENT_CONTEXT)
+            .get();
+    if (context == null) {
+      ctx.write(msg, prm);
+      return;
+    }
+    Span span = Span.fromContext(context);
+
+    if (msg instanceof HttpRequest) {
+      HttpRequest httpRequest = (HttpRequest) msg;
+
+      Map<String, String> headersMap = headersToMap(httpRequest);
+      if (agentConfig.getDataCapture().getHttpHeaders().getRequest().getValue()) {
+        headersMap.forEach((key, value) -> span.setAttribute(key, value));
+      }
+
+      CharSequence contentType = DataCaptureUtils.getContentType(httpRequest);
+      if (agentConfig.getDataCapture().getHttpBody().getRequest().getValue()
+          && contentType != null
+          && ContentTypeUtils.shouldCapture(contentType.toString())) {
+
+        CharSequence contentLengthHeader = DataCaptureUtils.getContentLength(httpRequest);
+        int contentLength = ContentLengthUtils.parseLength(contentLengthHeader);
+
+        String charsetString = ContentTypeUtils.parseCharset(contentType.toString());
+        Charset charset = ContentTypeCharsetUtils.toCharset(charsetString);
+
+        // set the buffer to capture response body
+        // the buffer is used byt captureBody method
+        Attribute<BoundedByteArrayOutputStream> bufferAttr =
+            ctx.channel().attr(AttributeKeys.REQUEST_BODY_BUFFER);
+        bufferAttr.set(BoundedBuffersFactory.createStream(contentLength, charset));
+      }
+    }
+
+    if ((msg instanceof HttpContent || msg instanceof ByteBuf)
+        && agentConfig.getDataCapture().getHttpBody().getRequest().getValue()) {
+      DataCaptureUtils.captureBody(span, channel, AttributeKeys.REQUEST_BODY_BUFFER, msg);
+    }
+
+    ctx.write(msg, prm);
+  }
+
+  private static Map<String, String> headersToMap(HttpMessage httpMessage) {
+    Map<String, String> map = new HashMap<>();
+    for (Map.Entry<String, String> entry : httpMessage.headers().entries()) {
+      AttributeKey<String> key = HypertraceSemanticAttributes.httpRequestHeader(entry.getKey());
+      map.put(key.getKey(), entry.getValue());
+    }
+    return map;
+  }
+}