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

Providing support to enable tracing in Executors

Browse Source 
User can opt to wrap java.unit.concurrent package usage with a
traceable version of the same thing.

Fixes gh-60, fixes gh-58
This commit is contained in:
Gaurav Rai Mazra
2015-11-24 22:23:30 +05:30
committed by Dave Syer
parent 2f253e035c
commit c588d73552
3 changed files with 361 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/concurrent/executor/TraceableExecutorService.java Normal file
View File

@@ -0,0 +1,115 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.sleuth.instrument.concurrent.executor;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.cloud.sleuth.Trace;
import org.springframework.cloud.sleuth.instrument.TraceCallable;
import org.springframework.cloud.sleuth.instrument.TraceRunnable;
/**
* A decorator class for {@link ExecutorService} to support tracing in Executors
* @author Gaurav Rai Mazra
*
*/
public class TraceableExecutorService implements ExecutorService {
final ExecutorService delegate;
final Trace trace;
public TraceableExecutorService(final ExecutorService delegate, final Trace trace) {
this.delegate = delegate;
this.trace = trace;
}
@Override
public void execute(Runnable command) {
final Runnable r = new TraceRunnable(trace, command);
this.delegate.execute(r);
}
@Override
public void shutdown() {
this.delegate.shutdown();
}
@Override
public List<Runnable> shutdownNow() {
return this.delegate.shutdownNow();
}
@Override
public boolean isShutdown() {
return this.delegate.isShutdown();
}
@Override
public boolean isTerminated() {
return this.delegate.isTerminated();
}
@Override
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
return this.delegate.awaitTermination(timeout, unit);
}
@Override
public <T> Future<T> submit(Callable<T> task) {
Callable<T> c = new TraceCallable<>(this.trace, task);
return this.delegate.submit(c);
}
@Override
public <T> Future<T> submit(Runnable task, T result) {
Runnable r = new TraceRunnable(trace, task);
return this.delegate.submit(r, result);
}
@Override
public Future<?> submit(Runnable task) {
Runnable r = new TraceRunnable(trace, task);
return this.delegate.submit(r);
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
return this.delegate.invokeAll(tasks);
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException {
return this.delegate.invokeAll(tasks, timeout, unit);
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
return this.delegate.invokeAny(tasks);
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return this.delegate.invokeAny(tasks, timeout, unit);
}
}

69
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/concurrent/executor/TraceableScheduledExecutorService.java Normal file
View File

@@ -0,0 +1,69 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.sleuth.instrument.concurrent.executor;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import org.springframework.cloud.sleuth.Trace;
import org.springframework.cloud.sleuth.instrument.TraceCallable;
import org.springframework.cloud.sleuth.instrument.TraceRunnable;
/**
* A decorator class for {@link ScheduledExecutorService} to support tracing in Executors
* @author Gaurav Rai Mazra
*
*/
public class TraceableScheduledExecutorService extends TraceableExecutorService implements ScheduledExecutorService {
public TraceableScheduledExecutorService(final ScheduledExecutorService delegate, final Trace trace) {
super(delegate, trace);
}
private ScheduledExecutorService getScheduledExecutorService() {
return (ScheduledExecutorService) this.delegate;
}
@Override
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
Runnable r = new TraceRunnable(this.trace, command);
return getScheduledExecutorService().schedule(r, delay, unit);
}
@Override
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
Callable<V> c = new TraceCallable<>(this.trace,callable);
return getScheduledExecutorService().schedule(c, delay, unit);
}
@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
Runnable r = new TraceRunnable(this.trace, command);
return getScheduledExecutorService().scheduleAtFixedRate(r, initialDelay, period, unit);
}
@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
Runnable r = new TraceRunnable(this.trace, command);
return getScheduledExecutorService().scheduleWithFixedDelay(r, initialDelay, delay, unit);
}
}

177
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/concurrent/executor/TraceableExecutorServiceTests.java Normal file
View File

@@ -0,0 +1,177 @@
package org.springframework.cloud.sleuth.instrument.concurrent.executor;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
import static org.mockito.Matchers.isA;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.springframework.cloud.sleuth.RandomUuidGenerator;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.Trace;
import org.springframework.cloud.sleuth.TraceContextHolder;
import org.springframework.cloud.sleuth.TraceScope;
import org.springframework.cloud.sleuth.event.SpanAcquiredEvent;
import org.springframework.cloud.sleuth.event.SpanReleasedEvent;
import org.springframework.cloud.sleuth.instrument.TraceRunnable;
import org.springframework.cloud.sleuth.sampler.AlwaysSampler;
import org.springframework.cloud.sleuth.trace.DefaultTrace;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
public class TraceableExecutorServiceTests {
private ApplicationEventPublisher publisher;
private ExecutorService traceableExecutorService;
private Trace trace;
private ExecutorService executorService;
private int NUM_SPANS = 11;
private int TOTAL_THREADS = 10;
@Before
public void setUp() throws Exception {
this.publisher = Mockito.mock(ApplicationEventPublisher.class);
this.trace = new DefaultTrace(new AlwaysSampler(), new RandomUuidGenerator(), this.publisher);
ExecutorService es = Executors.newFixedThreadPool(3);
this.traceableExecutorService = new TraceableExecutorService(es, this.trace);
this.executorService = Executors.newFixedThreadPool(3);
}
@After
public void tearDown() throws Exception {
this.trace = null;
this.traceableExecutorService.shutdown();
this.executorService.shutdown();
}
@Test
public void test_whenTraceContextOfWorkerThreadIsNotClosed_thenException() {
//THis test case ideally should fail but it is not failing because of the
// https://github.com/spring-cloud/spring-cloud-sleuth/issues/60 comment two
final AtomicInteger counter = new AtomicInteger(0);
final CountDownLatch latch = new CountDownLatch(TOTAL_THREADS);
TraceScope scope = this.trace.startSpan("PARENT");
for (int i = 0; i < TOTAL_THREADS; i++) {
traceableExecutorService.execute(new MyRunnable(counter, latch));
}
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
scope.close();
verify(this.publisher, times(NUM_SPANS)).publishEvent(isA(SpanAcquiredEvent.class));
verify(publisher, times(NUM_SPANS)).publishEvent(isA(SpanReleasedEvent.class));
ArgumentCaptor<ApplicationEvent> captor = ArgumentCaptor
.forClass(ApplicationEvent.class);
verify(publisher, atLeast(NUM_SPANS)).publishEvent(captor.capture());
List<Span> spans = new ArrayList<>();
for (ApplicationEvent event : captor.getAllValues()) {
if (event instanceof SpanReleasedEvent) {
spans.add(((SpanReleasedEvent) event).getSpan());
}
}
assertThat("spans was wrong size", spans.size(), is(NUM_SPANS));
}
@Test
public void test_whenTraceContextOfWorkerThreadIsClosed_thenNoException() {
final AtomicInteger counter = new AtomicInteger(0);
final CountDownLatch latch = new CountDownLatch(TOTAL_THREADS);
TraceScope scope = this.trace.startSpan("PARENT");
for (int i = 0; i < TOTAL_THREADS; i++) {
final TraceRunnableAdapter command = new TraceRunnableAdapter(new TraceRunnable(this.trace, new MyRunnable(counter, latch)));
executorService.execute(command);
}
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
scope.close();
verify(this.publisher, times(NUM_SPANS)).publishEvent(isA(SpanAcquiredEvent.class));
verify(publisher, times(NUM_SPANS)).publishEvent(isA(SpanReleasedEvent.class));
ArgumentCaptor<ApplicationEvent> captor = ArgumentCaptor
.forClass(ApplicationEvent.class);
verify(publisher, atLeast(NUM_SPANS)).publishEvent(captor.capture());
List<Span> spans = new ArrayList<>();
for (ApplicationEvent event : captor.getAllValues()) {
if (event instanceof SpanReleasedEvent) {
spans.add(((SpanReleasedEvent) event).getSpan());
}
}
assertThat("spans was wrong size", spans.size(), is(NUM_SPANS));
}
class TraceRunnableAdapter implements Runnable {
private final Runnable delegate;
public TraceRunnableAdapter(final Runnable delegate) {
this.delegate = delegate;
}
@Override
public void run() {
try {
this.delegate.run();
}
finally {
TraceContextHolder.removeCurrentSpan();
}
}
}
class MyRunnable implements Runnable {
private final AtomicInteger counter;
private final CountDownLatch latch;
MyRunnable(final AtomicInteger counter, final CountDownLatch latch) {
this.counter = counter;
this.latch = latch;
}
@Override
public void run() {
try {
try {
TimeUnit.MILLISECONDS.sleep(100l);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
finally {
counter.incrementAndGet();
latch.countDown();
}
}
}
}