Calling asyncreporter#check upon bean registration (#1413)

fixes gh-1411

spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin2/ZipkinAutoConfiguration.java

@@ -16,8 +16,15 @@
 
 package org.springframework.cloud.sleuth.zipkin2;
 
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import zipkin2.CheckResult;
 import zipkin2.Span;
 import zipkin2.reporter.AsyncReporter;
 import zipkin2.reporter.Reporter;
@@ -70,6 +77,8 @@ import org.springframework.web.client.RestTemplate;
 @Import({ ZipkinSenderConfigurationImportSelector.class, SamplerAutoConfiguration.class })
 public class ZipkinAutoConfiguration {
 
+	private static final Log log = LogFactory.getLog(ZipkinAutoConfiguration.class);
+
 	/**
 	 * Zipkin reporter bean name. Name of the bean matters for supporting multiple tracing
 	 * systems.
@@ -87,9 +96,43 @@ public class ZipkinAutoConfiguration {
 	public Reporter<Span> reporter(ReporterMetrics reporterMetrics,
 			ZipkinProperties zipkin, @Qualifier(SENDER_BEAN_NAME) Sender sender) {
 		// historical constraint. Note: AsyncReporter supports memory bounds
-		return AsyncReporter.builder(sender).queuedMaxSpans(1000)
+		AsyncReporter<Span> asyncReporter = AsyncReporter.builder(sender)
+				.queuedMaxSpans(1000)
 				.messageTimeout(zipkin.getMessageTimeout(), TimeUnit.SECONDS)
 				.metrics(reporterMetrics).build(zipkin.getEncoder());
+		CheckResult checkResult = checkResult(asyncReporter);
+		logCheckResult(asyncReporter, checkResult);
+		return asyncReporter;
+	}
+
+	private void logCheckResult(AsyncReporter asyncReporter, CheckResult checkResult) {
+		if (log.isDebugEnabled() && checkResult != null && checkResult.ok()) {
+			log.debug("Check result of the [" + asyncReporter.toString() + "] is ["
+					+ checkResult + "]");
+		}
+		else if (checkResult != null && !checkResult.ok()) {
+			log.warn("Check result of the [" + asyncReporter.toString()
+					+ "] contains an error [" + checkResult + "]");
+		}
+	}
+
+	private CheckResult checkResult(AsyncReporter<Span> asyncReporter) {
+		ExecutorService executor = Executors.newSingleThreadExecutor();
+		Callable<CheckResult> task = asyncReporter::check;
+		Future<CheckResult> future = executor.submit(task);
+		try {
+			return future.get(1, TimeUnit.SECONDS);
+		}
+		catch (Exception ex) {
+			log.warn(
+					"An exception took place when trying to retrieve the check result. Will return null.",
+					ex);
+			return null;
+		}
+		finally {
+			future.cancel(true);
+			executor.shutdown();
+		}
 	}
 
 	@Bean

spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin2/sender/ZipkinRestTemplateSenderConfiguration.java

@@ -38,6 +38,8 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.HttpMethod;
+import org.springframework.http.client.ClientHttpRequestFactory;
+import org.springframework.http.client.SimpleClientHttpRequestFactory;
 import org.springframework.web.client.RequestCallback;
 import org.springframework.web.client.ResponseExtractor;
 import org.springframework.web.client.RestClientException;
@@ -145,6 +147,8 @@ class ZipkinRestTemplateWrapper extends RestTemplate {
 
 	private static final Log log = LogFactory.getLog(ZipkinRestTemplateWrapper.class);
 
+	private static final int DEFAULT_TIMEOUT = 500;
+
 	private final ZipkinProperties zipkinProperties;
 
 	private final ZipkinUrlExtractor extractor;
@@ -153,6 +157,14 @@ class ZipkinRestTemplateWrapper extends RestTemplate {
 			ZipkinUrlExtractor extractor) {
 		this.zipkinProperties = zipkinProperties;
 		this.extractor = extractor;
+		setRequestFactory(clientHttpRequestFactory());
+	}
+
+	private ClientHttpRequestFactory clientHttpRequestFactory() {
+		SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
+		factory.setReadTimeout(DEFAULT_TIMEOUT);
+		factory.setConnectTimeout(DEFAULT_TIMEOUT);
+		return factory;
 	}
 
 	@Override

spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin2/ZipkinAutoConfigurationTests.java

@@ -92,7 +92,10 @@ public class ZipkinAutoConfigurationTests {
 		span.finish();
 
 		Awaitility.await().untilAsserted(
-				() -> then(this.server.getRequestCount()).isGreaterThan(0));
+				() -> then(this.server.getRequestCount()).isGreaterThan(1));
+		// first request is for health check
+		this.server.takeRequest();
+		// second request is the span one
 		RecordedRequest request = this.server.takeRequest();
 		then(request.getPath()).isEqualTo("/api/v2/spans");
 		then(request.getBody().readUtf8()).contains("localEndpoint");
@@ -120,6 +123,9 @@ public class ZipkinAutoConfigurationTests {
 
 		Awaitility.await().untilAsserted(
 				() -> then(this.server.getRequestCount()).isGreaterThan(0));
+		// first request is for health check
+		this.server.takeRequest();
+		// second request is the span one
 		RecordedRequest request = this.server.takeRequest();
 		then(request.getPath()).isEqualTo("/api/v1/spans");
 		then(request.getBody().readUtf8()).contains("binaryAnnotations");
@@ -246,7 +252,10 @@ public class ZipkinAutoConfigurationTests {
 		span.finish();
 
 		Awaitility.await().untilAsserted(
-				() -> then(this.server.getRequestCount()).isGreaterThan(0));
+				() -> then(this.server.getRequestCount()).isGreaterThan(1));
+		// first request is for health check
+		this.server.takeRequest();
+		// second request is the span one
 		RecordedRequest request = this.server.takeRequest();
 		then(request.getPath()).isEqualTo("/api/v2/spans");
 		then(request.getBody().readUtf8()).contains("localEndpoint");