From 0ecb23ee88acc5dc85975a5f2f32e333a668c57f Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Thu, 13 Feb 2020 15:28:31 -0800 Subject: [PATCH] Fixes parse glitches and adds remote endpoint to reactor HttpClient (#1559) --- .../client/HttpClientBeanPostProcessor.java | 19 ++++++++++++++++-- ...ReactorNettyHttpClientSpringBootTests.java | 20 +++++++++++++++++++ 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java index 7f90d40ed..052cf5467 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java @@ -16,6 +16,7 @@ package org.springframework.cloud.sleuth.instrument.web.client; +import java.net.InetSocketAddress; import java.util.List; import java.util.concurrent.atomic.AtomicReference; import java.util.function.BiConsumer; @@ -141,6 +142,7 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { WrappedHttpClientRequest request = new WrappedHttpClientRequest(req); Span clientSpan = parent != null ? handler().handleSend(request, parent) : handler().handleSend(request); + parseConnectionAddress(connection, clientSpan); // Swap the ref with the client span, so that other hooks can see it if (ref != null) { @@ -148,6 +150,14 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { } } + static void parseConnectionAddress(Connection connection, Span span) { + if (span.isNoop()) { + return; + } + InetSocketAddress socketAddress = connection.address(); + span.remoteIpAndPort(socketAddress.getHostString(), socketAddress.getPort()); + } + } private static class TracingDoOnResponse extends AbstractTracingDoOnHandler @@ -244,12 +254,12 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { @Override public String path() { - return delegate.path(); + return "/" + delegate.path(); // TODO: reactor/reactor-netty#999 } @Override public String url() { - return delegate.uri(); + return delegate.resourceUrl(); } @Override @@ -272,6 +282,11 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { this.delegate = delegate; } + @Override + public String method() { + return delegate.method().name(); + } + @Override public Object unwrap() { return delegate; diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientSpringBootTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientSpringBootTests.java index 518b61fb5..791146de6 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientSpringBootTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientSpringBootTests.java @@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit; import brave.propagation.B3SinglePropagation; import brave.propagation.Propagation; import brave.sampler.Sampler; +import io.netty.handler.codec.http.HttpResponseStatus; import org.junit.After; import org.junit.Test; import org.junit.runner.RunWith; @@ -30,6 +31,7 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.netty.DisposableServer; import reactor.netty.http.client.HttpClient; +import reactor.netty.http.client.HttpClientResponse; import reactor.netty.http.client.PrematureCloseException; import reactor.netty.http.server.HttpServer; import zipkin2.Span; @@ -76,6 +78,24 @@ public class ReactorNettyHttpClientSpringBootTests { this.spans.clear(); } + @Test + public void shouldRecordRemoteEndpoint() throws Exception { + disposableServer = HttpServer.create().port(0) + .handle((in, out) -> out.sendString(Flux.just("foo"))).bindNow(); + + HttpClientResponse response = httpClient.port(disposableServer.port()).get() + .uri("/").response().block(); + + assertThat(response.status()).isEqualTo(HttpResponseStatus.OK); + + Span clientSpan = takeClientSpan(); + + assertThat(clientSpan.remoteEndpoint()).satisfiesAnyOf( + ep -> assertThat(ep.ipv4()).isNotNull(), + ep -> assertThat(ep.ipv6()).isNotNull()); + assertThat(clientSpan.remoteEndpoint().portAsInt()).isNotZero(); + } + @Test public void shouldSendTraceContextToServer_rootSpan() throws Exception { disposableServer = HttpServer.create().port(0)