From 59216c32f7848ec337fb68d1dbec8e87eeb6bf59 Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Wed, 31 Jul 2019 22:53:14 +0200 Subject: [PATCH] Calling asyncreporter#check upon bean registration (#1413) fixes gh-1411 --- .../test/java/integration/ZipkinTests.java | 20 ++++++--- .../zipkin2/ZipkinAutoConfiguration.java | 45 ++++++++++++++++++- ...ZipkinRestTemplateSenderConfiguration.java | 12 +++++ .../zipkin2/ZipkinAutoConfigurationTests.java | 13 +++++- 4 files changed, 82 insertions(+), 8 deletions(-) diff --git a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-zipkin/src/test/java/integration/ZipkinTests.java b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-zipkin/src/test/java/integration/ZipkinTests.java index d42e76674..6153c7447 100644 --- a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-zipkin/src/test/java/integration/ZipkinTests.java +++ b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-zipkin/src/test/java/integration/ZipkinTests.java @@ -29,6 +29,7 @@ import integration.ZipkinTests.WaitUntilZipkinIsUpConfig; import okhttp3.mockwebserver.MockResponse; import okhttp3.mockwebserver.MockWebServer; import okhttp3.mockwebserver.RecordedRequest; +import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; @@ -39,8 +40,8 @@ import zipkin2.Span; import zipkin2.codec.SpanBytesDecoder; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.web.server.LocalServerPort; import org.springframework.cloud.sleuth.zipkin2.ZipkinProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -66,13 +67,22 @@ public class ZipkinTests extends AbstractIntegrationTest { @Autowired ZipkinProperties zipkinProperties; - @Value("${local.server.port}") + @LocalServerPort private int port = 3380; private String sampleAppUrl = "http://localhost:" + this.port; + @BeforeClass + public static void setup() { + // enqueues a request for async reporter health check + zipkin.enqueue(new MockResponse()); + } + @Test public void should_propagate_spans_to_zipkin() throws Exception { + // takes the received request for async reporter health check + zipkin.takeRequest(); + // enqueues a request for spans zipkin.enqueue(new MockResponse()); long traceId = new Random().nextLong(); @@ -96,11 +106,11 @@ public class ZipkinTests extends AbstractIntegrationTest { List traceIdsNotFoundInZipkin = traceIdsNotFoundInZipkin(spans, traceId); List serviceNamesNotFoundInZipkin = serviceNamesNotFoundInZipkin(spans); List tagsNotFoundInZipkin = hasRequiredTag(spans); - log.info(String.format("The following trace IDs were not found in Zipkin [%s]", + log.info(String.format("The following trace IDs were not found in Zipkin %s", traceIdsNotFoundInZipkin)); - log.info(String.format("The following services were not found in Zipkin [%s]", + log.info(String.format("The following services were not found in Zipkin %s", serviceNamesNotFoundInZipkin)); - log.info(String.format("The following tags were not found in Zipkin [%s]", + log.info(String.format("The following tags were not found in Zipkin %s", tagsNotFoundInZipkin)); then(traceIdsNotFoundInZipkin).isEmpty(); then(serviceNamesNotFoundInZipkin).isEmpty(); diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin2/ZipkinAutoConfiguration.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin2/ZipkinAutoConfiguration.java index dc6ae7128..353fb17a5 100644 --- a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin2/ZipkinAutoConfiguration.java +++ b/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 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 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 asyncReporter) { + ExecutorService executor = Executors.newSingleThreadExecutor(); + Callable task = asyncReporter::check; + Future 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 diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin2/sender/ZipkinRestTemplateSenderConfiguration.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin2/sender/ZipkinRestTemplateSenderConfiguration.java index 05d5dba66..945e8706e 100644 --- a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin2/sender/ZipkinRestTemplateSenderConfiguration.java +++ b/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 diff --git a/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin2/ZipkinAutoConfigurationTests.java b/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin2/ZipkinAutoConfigurationTests.java index acae31d10..c04b75699 100644 --- a/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin2/ZipkinAutoConfigurationTests.java +++ b/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");