SpringCloud/spring-cloud-sleuth
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Premature async rest template (#479)

Browse Source 
without this change the async rest template provides wrong value of the span duration
with this change the span is closed via a callback

fixes #475
This commit is contained in:
Marcin Grzejszczak
2016-12-16 13:59:03 +01:00
parent 8ff34c28a9
commit 22a4c77219
3 changed files with 167 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceAsyncRestTemplate.java
View File

@@ -20,11 +20,13 @@ import java.net.URI;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.util.ExceptionUtils;
import org.springframework.core.task.AsyncListenableTaskExecutor;
import org.springframework.http.HttpMethod;
import org.springframework.http.client.AsyncClientHttpRequestFactory;
import org.springframework.http.client.ClientHttpRequestFactory;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.client.AsyncRequestCallback;
import org.springframework.web.client.AsyncRestTemplate;
import org.springframework.web.client.ResponseExtractor;
@@ -75,27 +77,62 @@ public class TraceAsyncRestTemplate extends AsyncRestTemplate {
protected <T> ListenableFuture<T> doExecute(URI url, HttpMethod method,
AsyncRequestCallback requestCallback, ResponseExtractor<T> responseExtractor)
throws RestClientException {
try {
return super.doExecute(url, method, requestCallback, responseExtractor);
} finally {
ListenableFuture<T> future = super.doExecute(url, method, requestCallback, responseExtractor);
Span span = this.tracer.getCurrentSpan();
future.addCallback(new TraceListenableFutureCallback<>(this.tracer, span));
// potential race can happen here
if (span != null && span.equals(this.tracer.getCurrentSpan())) {
this.tracer.detach(span);
}
return future;
}
private static class TraceListenableFutureCallback<T> implements ListenableFutureCallback<T> {
private final Tracer tracer;
private final Span parent;
private TraceListenableFutureCallback(Tracer tracer, Span parent) {
this.tracer = tracer;
this.parent = parent;
}
@Override
public void onFailure(Throwable ex) {
continueSpan();
this.tracer.addTag(Span.SPAN_ERROR_TAG_NAME, ExceptionUtils.getExceptionMessage(ex));
finish();
}
}
private void finish() {
if (!isTracing()) {
return;
@Override
public void onSuccess(T result) {
continueSpan();
finish();
}
private void continueSpan() {
this.tracer.continueSpan(this.parent);
}
private void finish() {
if (!isTracing()) {
return;
}
currentSpan().logEvent(Span.CLIENT_RECV);
this.tracer.close(currentSpan());
}
private Span currentSpan() {
return this.tracer.getCurrentSpan();
}
private boolean isTracing() {
return this.tracer.isTracing();
}
currentSpan().logEvent(Span.CLIENT_RECV);
this.tracer.close(this.currentSpan());
}
private Span currentSpan() {
return this.tracer.getCurrentSpan();
}
private boolean isTracing() {
return this.tracer.isTracing();
}
}

17
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/RestTemplateTraceAspectIntegrationTests.java
View File

@@ -3,6 +3,7 @@ package org.springframework.cloud.sleuth.instrument.web;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -49,11 +50,9 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
@DirtiesContext
public class RestTemplateTraceAspectIntegrationTests {
@Autowired
private WebApplicationContext context;
@Autowired
private AspectTestingController controller;
@Autowired WebApplicationContext context;
@Autowired AspectTestingController controller;
@Autowired Tracer tracer;
private MockMvc mockMvc;
@@ -61,6 +60,14 @@ public class RestTemplateTraceAspectIntegrationTests {
public void init() {
this.mockMvc = MockMvcBuilders.webAppContextSetup(this.context).build();
this.controller.reset();
ExceptionUtils.setFail(true);
}
@Before
@After
public void verify() {
then(this.tracer.getCurrentSpan()).isNull();
then(ExceptionUtils.getLastException()).isNull();
}
@Test

104
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebAsyncClientAutoConfigurationTests.java
View File

@@ -18,7 +18,10 @@ package org.springframework.cloud.sleuth.instrument.web.client;
import java.io.IOException;
import java.net.URI;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.runners.Enclosed;
import org.junit.runner.RunWith;
@@ -26,18 +29,28 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.boot.test.WebIntegrationTest;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.sampler.AlwaysSampler;
import org.springframework.cloud.sleuth.util.ArrayListSpanAccumulator;
import org.springframework.cloud.sleuth.util.ExceptionUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.AsyncClientHttpRequest;
import org.springframework.http.client.AsyncClientHttpRequestFactory;
import org.springframework.http.client.ClientHttpRequest;
import org.springframework.http.client.ClientHttpRequestFactory;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.AsyncRestTemplate;
import static org.assertj.core.api.BDDAssertions.then;
import static org.springframework.cloud.sleuth.assertions.SleuthAssertions.then;
/**
* @author Marcin Grzejszczak
@@ -147,4 +160,93 @@ public class TraceWebAsyncClientAutoConfigurationTests {
}
}
@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = { DurationChecking.TestConfiguration.class })
@WebIntegrationTest(randomPort = true)
public static class DurationChecking {
@Autowired AsyncRestTemplate asyncRestTemplate;
@Autowired Environment environment;
@Autowired ArrayListSpanAccumulator accumulator;
@Autowired Tracer tracer;
@Before
public void setup() {
ExceptionUtils.setFail(true);
}
@Test
public void should_close_span_upon_success_callback()
throws ExecutionException, InterruptedException {
ListenableFuture<ResponseEntity<String>> future = this.asyncRestTemplate
.getForEntity("http://localhost:" + port() + "/foo", String.class);
String result = future.get().getBody();
then(result).isEqualTo("foo");
then(this.accumulator.getSpans().stream().filter(
span -> span.logs().stream().filter(log -> Span.CLIENT_RECV.equals(log.getEvent())).findFirst().isPresent()
).findFirst().get()).matches(span -> span.getAccumulatedMicros() >= TimeUnit.MILLISECONDS.toMicros(100));
then(this.tracer.getCurrentSpan()).isNull();
then(ExceptionUtils.getLastException()).isNull();
}
@Test
public void should_close_span_upon_failure_callback()
throws ExecutionException, InterruptedException {
ListenableFuture<ResponseEntity<String>> future;
try {
future = this.asyncRestTemplate
.getForEntity("http://localhost:" + port() + "/blowsup", String.class);
future.get();
} catch (Exception e) {
then(e.getMessage()).contains("Internal Server Error");
}
then(this.accumulator.getSpans().stream().filter(
span -> span.logs().stream().filter(log -> Span.CLIENT_RECV.equals(log.getEvent())).findFirst().isPresent()
).findFirst().get()).matches(span -> span.getAccumulatedMicros() >= TimeUnit.MILLISECONDS.toMicros(100))
.hasATag(Span.SPAN_ERROR_TAG_NAME, "500 Internal Server Error");
then(this.tracer.getCurrentSpan()).isNull();
then(ExceptionUtils.getLastException()).isNull();
}
int port() {
return this.environment.getProperty("local.server.port", Integer.class);
}
@EnableAutoConfiguration
@Configuration
public static class TestConfiguration {
@Bean ArrayListSpanAccumulator accumulator() {
return new ArrayListSpanAccumulator();
}
@Bean
MyController myController() {
return new MyController();
}
@Bean AlwaysSampler sampler() {
return new AlwaysSampler();
}
}
@RestController
public static class MyController {
@RequestMapping("/foo")
String foo() throws Exception {
Thread.sleep(100);
return "foo";
}
@RequestMapping("/blowsup")
String blowsup() throws Exception {
Thread.sleep(100);
throw new RuntimeException("boom");
}
}
}
}