From 0290437330f785bde3271d4e4d93bd41cb2d1b17 Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Fri, 15 May 2020 20:26:19 +0800 Subject: [PATCH] Makes it more clear what the timeout code is doing and backfills tests (#1636) I was surprised to see an executor service created just to make a time limiter. This does it more simply and backfills the missing tests. --- .../zipkin2/ZipkinAutoConfiguration.java | 60 +++++++++------ .../zipkin2/ZipkinAutoConfigurationTests.java | 76 +++++++++++++++++++ 2 files changed, 111 insertions(+), 25 deletions(-) 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 df110882c..6d829b084 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,11 +16,8 @@ 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 java.util.concurrent.TimeoutException; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -92,43 +89,56 @@ public class ZipkinAutoConfiguration { @ConditionalOnMissingBean(name = REPORTER_BEAN_NAME) public Reporter reporter(ReporterMetrics reporterMetrics, ZipkinProperties zipkin, @Qualifier(SENDER_BEAN_NAME) Sender sender) { + CheckResult checkResult = checkResult(sender, 1_000L); + logCheckResult(sender, checkResult); + // historical constraint. Note: AsyncReporter supports memory bounds 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) { + private void logCheckResult(Sender sender, CheckResult checkResult) { if (log.isDebugEnabled() && checkResult != null && checkResult.ok()) { - log.debug("Check result of the [" + asyncReporter.toString() + "] is [" - + checkResult + "]"); + log.debug("Check result of the [" + sender.toString() + "] is [" + checkResult + + "]"); } else if (checkResult != null && !checkResult.ok()) { - log.warn("Check result of the [" + asyncReporter.toString() - + "] contains an error [" + checkResult + "]"); + log.warn("Check result of the [" + sender.toString() + "] contains an error [" + + checkResult + "]"); } } - private CheckResult checkResult(AsyncReporter asyncReporter) { - ExecutorService executor = Executors.newSingleThreadExecutor(); - Callable task = asyncReporter::check; - Future future = executor.submit(task); + /** Limits {@link Sender#check()} to {@code deadlineMillis}. */ + static CheckResult checkResult(Sender sender, long deadlineMillis) { + CheckResult[] outcome = new CheckResult[1]; + Thread thread = new Thread(sender + " check()") { + @Override + public void run() { + try { + outcome[0] = sender.check(); + } + catch (Throwable e) { + outcome[0] = CheckResult.failed(e); + } + } + }; + thread.start(); try { - return future.get(1, TimeUnit.SECONDS); + thread.join(deadlineMillis); + if (outcome[0] != null) { + return outcome[0]; + } + thread.interrupt(); + return CheckResult.failed(new TimeoutException( + thread.getName() + " timed out after " + deadlineMillis + "ms")); } - 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(); + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return CheckResult.failed(e); } } 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 3cb6423f7..851c1c6d2 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 @@ -17,6 +17,7 @@ package org.springframework.cloud.sleuth.zipkin2; import java.util.List; +import java.util.concurrent.TimeoutException; import brave.Span; import brave.Tracing; @@ -32,6 +33,7 @@ import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; import zipkin2.Call; +import zipkin2.CheckResult; import zipkin2.codec.Encoding; import zipkin2.reporter.AsyncReporter; import zipkin2.reporter.Reporter; @@ -51,7 +53,10 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.mock.env.MockEnvironment; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.BDDAssertions.then; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; /** * Not using {@linkplain SpringBootTest} as we need to change properties per test. @@ -315,6 +320,77 @@ public class ZipkinAutoConfigurationTests { Awaitility.await().untilAsserted(() -> then(sender.isSpanSent()).isTrue()); } + @Test + public void checkResult_onTime() { + Sender sender = mock(Sender.class); + when(sender.check()).thenReturn(CheckResult.OK); + + assertThat(ZipkinAutoConfiguration.checkResult(sender, 200).ok()).isTrue(); + } + + @Test + public void checkResult_onTime_notOk() { + Sender sender = mock(Sender.class); + RuntimeException exception = new RuntimeException("dead"); + when(sender.check()).thenReturn(CheckResult.failed(exception)); + + assertThat(ZipkinAutoConfiguration.checkResult(sender, 200).error()) + .isSameAs(exception); + } + + /** Bug in {@link Sender} as it shouldn't throw */ + @Test + public void checkResult_thrown() { + Sender sender = mock(Sender.class); + RuntimeException exception = new RuntimeException("dead"); + when(sender.check()).thenThrow(exception); + + assertThat(ZipkinAutoConfiguration.checkResult(sender, 200).error()) + .isSameAs(exception); + } + + @Test + public void checkResult_slow() { + assertThat(ZipkinAutoConfiguration.checkResult(new Sender() { + @Override + public CheckResult check() { + try { + Thread.sleep(500L); + } + catch (InterruptedException e) { + throw new AssertionError(e); + } + return CheckResult.OK; + } + + @Override + public Encoding encoding() { + return Encoding.JSON; + } + + @Override + public int messageMaxBytes() { + return 0; + } + + @Override + public int messageSizeInBytes(List list) { + return 0; + } + + @Override + public Call sendSpans(List list) { + return Call.create(null); + } + + @Override + public String toString() { + return "FakeSender{}"; + } + }, 200).error()).isInstanceOf(TimeoutException.class) + .hasMessage("FakeSender{} check() timed out after 200ms"); + } + @Configuration protected static class Config {