Fixes parse glitches and adds remote endpoint to reactor HttpClient (#1559)

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;

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)