Make OTEL Vertx Adapter extract forwarded headers for client ip if present

extensions/opentelemetry/opentelemetry/deployment/src/test/java/io/quarkus/opentelemetry/deployment/TestSpanExporter.java

@@ -0,0 +1,55 @@
+package io.quarkus.opentelemetry.deployment;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import javax.enterprise.context.ApplicationScoped;
+
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SpanExporter;
+
+@ApplicationScoped
+public class TestSpanExporter implements SpanExporter {
+    private final List<SpanData> finishedSpanItems = new ArrayList<>();
+    private boolean isStopped = false;
+
+    public List<SpanData> getFinishedSpanItems() {
+        synchronized (this) {
+            return Collections.unmodifiableList(new ArrayList<>(finishedSpanItems));
+        }
+    }
+
+    public void reset() {
+        synchronized (this) {
+            finishedSpanItems.clear();
+        }
+    }
+
+    @Override
+    public CompletableResultCode export(Collection<SpanData> spans) {
+        synchronized (this) {
+            if (isStopped) {
+                return CompletableResultCode.ofFailure();
+            }
+            finishedSpanItems.addAll(spans);
+        }
+        return CompletableResultCode.ofSuccess();
+    }
+
+    @Override
+    public CompletableResultCode flush() {
+        return CompletableResultCode.ofSuccess();
+    }
+
+    @Override
+    public CompletableResultCode shutdown() {
+        synchronized (this) {
+            finishedSpanItems.clear();
+            isStopped = true;
+        }
+        return CompletableResultCode.ofSuccess();
+    }
+}

extensions/opentelemetry/opentelemetry/deployment/src/test/java/io/quarkus/opentelemetry/deployment/TracerRouter.java

@@ -0,0 +1,26 @@
+package io.quarkus.opentelemetry.deployment;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.enterprise.event.Observes;
+import javax.inject.Inject;
+
+import io.opentelemetry.api.trace.Tracer;
+import io.quarkus.runtime.StartupEvent;
+import io.vertx.ext.web.Router;
+
+@ApplicationScoped
+public class TracerRouter {
+    @Inject
+    Router router;
+    @Inject
+    Tracer tracer;
+
+    public void register(@Observes StartupEvent ev) {
+        router.get("/tracer").handler(rc -> {
+            tracer.spanBuilder("io.quarkus.vertx.opentelemetry").startSpan()
+                    .setAttribute("test.message", "hello!")
+                    .end();
+            rc.response().end("Hello Tracer!");
+        });
+    }
+}

extensions/opentelemetry/opentelemetry/deployment/src/test/java/io/quarkus/opentelemetry/deployment/VertxOpenTelemetryForwardedTest.java

@@ -0,0 +1,41 @@
+package io.quarkus.opentelemetry.deployment;
+
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.HTTP_CLIENT_IP;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.List;
+
+import javax.inject.Inject;
+
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.quarkus.test.QuarkusUnitTest;
+import io.restassured.RestAssured;
+
+public class VertxOpenTelemetryForwardedTest {
+    @RegisterExtension
+    static final QuarkusUnitTest unitTest = new QuarkusUnitTest()
+            .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
+                    .addClass(TestSpanExporter.class)
+                    .addClass(TracerRouter.class));
+
+    @Inject
+    TestSpanExporter testSpanExporter;
+
+    @Test
+    void trace() {
+        RestAssured.given().header("Forwarded", "for=192.0.2.60;proto=http;by=203.0.113.43")
+                .when().get("/tracer").then()
+                .statusCode(200)
+                .body(is("Hello Tracer!"));
+
+        List<SpanData> spans = testSpanExporter.getFinishedSpanItems();
+
+        assertEquals("192.0.2.60", spans.get(1).getAttributes().get(HTTP_CLIENT_IP));
+    }
+}

extensions/opentelemetry/opentelemetry/deployment/src/test/java/io/quarkus/opentelemetry/deployment/VertxOpenTelemetryTest.java

@@ -12,44 +12,36 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
 import java.util.List;
 
-import javax.enterprise.context.ApplicationScoped;
-import javax.enterprise.event.Observes;
 import javax.inject.Inject;
 
 import org.jboss.shrinkwrap.api.ShrinkWrap;
 import org.jboss.shrinkwrap.api.spec.JavaArchive;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
-import io.opentelemetry.api.trace.Tracer;
-import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.trace.data.SpanData;
-import io.opentelemetry.sdk.trace.export.SpanExporter;
-import io.quarkus.runtime.StartupEvent;
 import io.quarkus.test.QuarkusUnitTest;
 import io.restassured.RestAssured;
-import io.vertx.ext.web.Router;
 
 public class VertxOpenTelemetryTest {
     @RegisterExtension
     static final QuarkusUnitTest unitTest = new QuarkusUnitTest()
-            .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class));
+            .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
+                    .addClass(TestSpanExporter.class)
+                    .addClass(TracerRouter.class));
 
     @Inject
-    MyExporter myExporter;
+    TestSpanExporter testSpanExporter;
 
     @Test
     void trace() {
         RestAssured.when().get("/tracer").then()
                 .statusCode(200)
                 .body(is("Hello Tracer!"));
 
-        List<SpanData> spans = myExporter.getFinishedSpanItems();
+        List<SpanData> spans = testSpanExporter.getFinishedSpanItems();
 
         assertEquals(2, spans.size());
         assertEquals("io.quarkus.vertx.opentelemetry", spans.get(0).getName());
@@ -62,64 +54,4 @@ void trace() {
         assertEquals("127.0.0.1", spans.get(1).getAttributes().get(HTTP_CLIENT_IP));
         assertNotNull(spans.get(1).getAttributes().get(HTTP_USER_AGENT));
     }
-
-    @ApplicationScoped
-    public static class TracerRouter {
-        @Inject
-        Router router;
-        @Inject
-        Tracer tracer;
-
-        public void register(@Observes StartupEvent ev) {
-            router.get("/tracer").handler(rc -> {
-                tracer.spanBuilder("io.quarkus.vertx.opentelemetry").startSpan()
-                        .setAttribute("test.message", "hello!")
-                        .end();
-                rc.response().end("Hello Tracer!");
-            });
-        }
-    }
-
-    @ApplicationScoped
-    public static class MyExporter implements SpanExporter {
-        private final List<SpanData> finishedSpanItems = new ArrayList<>();
-        private boolean isStopped = false;
-
-        public List<SpanData> getFinishedSpanItems() {
-            synchronized (this) {
-                return Collections.unmodifiableList(new ArrayList<>(finishedSpanItems));
-            }
-        }
-
-        public void reset() {
-            synchronized (this) {
-                finishedSpanItems.clear();
-            }
-        }
-
-        @Override
-        public CompletableResultCode export(Collection<SpanData> spans) {
-            synchronized (this) {
-                if (isStopped) {
-                    return CompletableResultCode.ofFailure();
-                }
-                finishedSpanItems.addAll(spans);
-            }
-            return CompletableResultCode.ofSuccess();
-        }
-
-        @Override
-        public CompletableResultCode flush() {
-            return CompletableResultCode.ofSuccess();
-        }
-
-        @Override
-        public CompletableResultCode shutdown() {
-            synchronized (this) {
-                finishedSpanItems.clear();
-                isStopped = true;
-            }
-            return CompletableResultCode.ofSuccess();
-        }
-    }
 }

extensions/opentelemetry/opentelemetry/deployment/src/test/java/io/quarkus/opentelemetry/deployment/VertxOpenTelemetryXForwardedTest.java

@@ -0,0 +1,41 @@
+package io.quarkus.opentelemetry.deployment;
+
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.HTTP_CLIENT_IP;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.List;
+
+import javax.inject.Inject;
+
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.quarkus.test.QuarkusUnitTest;
+import io.restassured.RestAssured;
+
+public class VertxOpenTelemetryXForwardedTest {
+    @RegisterExtension
+    static final QuarkusUnitTest unitTest = new QuarkusUnitTest()
+            .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
+                    .addClass(TestSpanExporter.class)
+                    .addClass(TracerRouter.class));
+
+    @Inject
+    TestSpanExporter testSpanExporter;
+
+    @Test
+    void trace() {
+        RestAssured.given().header("X-Forwarded-For", "203.0.113.195, 70.41.3.18, 150.172.238.178")
+                .when().get("/tracer").then()
+                .statusCode(200)
+                .body(is("Hello Tracer!"));
+
+        List<SpanData> spans = testSpanExporter.getFinishedSpanItems();
+
+        assertEquals("203.0.113.195", spans.get(1).getAttributes().get(HTTP_CLIENT_IP));
+    }
+}

extensions/opentelemetry/opentelemetry/runtime/src/main/java/io/quarkus/opentelemetry/runtime/tracing/vertx/VertxTracingAdapter.java

@@ -11,6 +11,7 @@
 import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.HTTP_TARGET;
 import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.HTTP_URL;
 import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.HTTP_USER_AGENT;
+import static io.quarkus.opentelemetry.runtime.tracing.vertx.VertxUtil.extractClientIP;
 import static io.vertx.core.http.HttpHeaders.CONTENT_LENGTH;
 import static io.vertx.core.http.HttpHeaders.USER_AGENT;
 
@@ -103,7 +104,7 @@ public <R> Span receiveRequest(
                 builder.setAttribute(HTTP_TARGET, httpServerRequest.path());
                 builder.setAttribute(HTTP_SCHEME, httpServerRequest.scheme());
                 builder.setAttribute(HTTP_HOST, httpServerRequest.host());
-                builder.setAttribute(HTTP_CLIENT_IP, httpServerRequest.remoteAddress().host());
+                builder.setAttribute(HTTP_CLIENT_IP, extractClientIP(httpServerRequest));
                 builder.setAttribute(HTTP_USER_AGENT, httpServerRequest.getHeader(USER_AGENT));
 
                 String contentLength = httpServerRequest.getHeader(CONTENT_LENGTH);

extensions/opentelemetry/opentelemetry/runtime/src/main/java/io/quarkus/opentelemetry/runtime/tracing/vertx/VertxUtil.java

@@ -0,0 +1,42 @@
+package io.quarkus.opentelemetry.runtime.tracing.vertx;
+
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import io.vertx.core.http.HttpServerRequest;
+
+public final class VertxUtil {
+    private static final String X_FORWARDED_FOR = "X-Forwarded-For";
+    private static final Pattern FORWARDED_FOR_PATTERN = Pattern.compile("for=\"?([^;,\"]+)\"?");
+    private static final String FORWARDED = "Forwarded";
+    private static final String COMMA_SPLITTER = ",";
+    private static final int SPLIT_LIMIT = -1;
+
+    private VertxUtil() {
+    }
+
+    private static Optional<String> getForwardedHeaderValue(HttpServerRequest httpServerRequest) {
+        return Optional.ofNullable(httpServerRequest.getHeader(FORWARDED))
+                .map(FORWARDED_FOR_PATTERN::matcher)
+                .filter(Matcher::find)
+                .map(matcher -> matcher.group(1).trim());
+    }
+
+    private static Optional<String> getXForwardedHeaderValue(HttpServerRequest httpServerRequest) {
+        return Optional.ofNullable(httpServerRequest.getHeader(X_FORWARDED_FOR))
+                .flatMap(o -> Stream.of(o.split(COMMA_SPLITTER, SPLIT_LIMIT))
+                        .findFirst());
+    }
+
+    public static String extractClientIP(HttpServerRequest httpServerRequest) {
+        // Tries to fetch Forwarded first since X-Forwarded can be lost by a proxy
+        // If Forwarded is not there tries to fetch the X-Forwarded-For header
+        // If none is found resorts to the remote address from the http request
+        return getForwardedHeaderValue(httpServerRequest)
+                .orElseGet(() -> getXForwardedHeaderValue(httpServerRequest)
+                        .orElseGet(() -> httpServerRequest.remoteAddress().host()));
+
+    }
+}