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

Added PollerMetadata for StreamSpanReporter (#343)

Browse Source 
Added configurable PollerMetadata bean to be used with the Inbound
Channel Adapter in the StreamSpanReporter. Fixed delay and max messages
per poll are configurable via configuration properties. Implementors can
provide a bean of type PollerMetadata and name
StreamSpanReporter.POLLER to take full control of the poller.

Fixes: gh-338
This commit is contained in:
Steve Oakey
2016-07-29 09:40:46 -04:00
committed by Marcin Grzejszczak
parent c8992a65b3
commit cf9e379ca1
5 changed files with 221 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/SleuthStreamAutoConfiguration.java
View File

@@ -38,20 +38,23 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.Ordered;
import org.springframework.integration.config.GlobalChannelInterceptor;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.scheduling.support.PeriodicTrigger;
/**
* {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration Auto-configuration}
* for sending spans over Spring Cloud Stream. This is for the producer
* (via {@link SleuthSource}). A consumer can enable binding to {@link SleuthSink} and
* receive the messages coming from the source (they have the same channel name so there
* is no additional configuration to do by default).
* {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration
* Auto-configuration} for sending spans over Spring Cloud Stream. This is for
* the producer (via {@link SleuthSource}). A consumer can enable binding to
* {@link SleuthSink} and receive the messages coming from the source (they have
* the same channel name so there is no additional configuration to do by
* default).
*
* @author Dave Syer
* @since 1.0.0
*/
@Configuration
@EnableConfigurationProperties({SleuthStreamProperties.class, SamplerProperties.class})
@EnableConfigurationProperties({ SleuthStreamProperties.class, SamplerProperties.class })
@AutoConfigureAfter(TraceMetricsAutoConfiguration.class)
@AutoConfigureBefore(ChannelBindingAutoConfiguration.class)
@EnableBinding(SleuthSource.class)
@@ -77,6 +80,15 @@ public class SleuthStreamAutoConfiguration {
return new StreamSpanReporter(endpointLocator, spanMetricReporter);
}
@Bean(name = StreamSpanReporter.POLLER)
@ConditionalOnMissingBean(name = StreamSpanReporter.POLLER)
public PollerMetadata defaultStreamSpanReporterPoller(SleuthStreamProperties sleuth) {
PollerMetadata poller = new PollerMetadata();
poller.setTrigger(new PeriodicTrigger(sleuth.getPoller().getFixedDelay()));
poller.setMaxMessagesPerPoll(sleuth.getPoller().getMaxMessagesPerPoll());
return poller;
}
@Configuration
@ConditionalOnMissingClass("org.springframework.cloud.client.discovery.DiscoveryClient")
protected static class DefaultEndpointLocatorConfiguration {

33
spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/SleuthStreamProperties.java
View File

@@ -28,6 +28,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
public class SleuthStreamProperties {
private boolean enabled = true;
private String group = SleuthSink.INPUT;
private Poller poller = new Poller();
public boolean isEnabled() {
return this.enabled;
@@ -44,4 +45,36 @@ public class SleuthStreamProperties {
public void setGroup(String group) {
this.group = group;
}
public Poller getPoller() {
return this.poller;
}
public static class Poller {
/**
* Fixed delay (ms). Default: 1000
*/
private long fixedDelay = 1000L;
/**
* Max messages per poll. Default: -1 (unbounded)
*/
private int maxMessagesPerPoll = -1;
public long getFixedDelay() {
return this.fixedDelay;
}
public int getMaxMessagesPerPoll() {
return this.maxMessagesPerPoll;
}
public void setFixedDelay(long fixedDelay) {
this.fixedDelay = fixedDelay;
}
public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
}
}

11
spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/StreamSpanReporter.java
View File

@@ -27,7 +27,7 @@ import org.springframework.cloud.sleuth.SpanReporter;
import org.springframework.cloud.sleuth.metric.SpanMetricReporter;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Poller;
/**
* A message source for spans. Also handles RPC flavoured annotations.
@@ -38,6 +38,13 @@ import org.springframework.integration.annotation.MessageEndpoint;
@MessageEndpoint
public class StreamSpanReporter implements SpanReporter {
/**
* Bean name for the
* {@link org.springframework.integration.scheduling.PollerMetadata
* PollerMetadata}
*/
public static final String POLLER = "streamSpanReporterPoller";
private BlockingQueue<Span> queue = new LinkedBlockingQueue<>();
private final HostLocator endpointLocator;
private final SpanMetricReporter spanMetricReporter;
@@ -51,7 +58,7 @@ public class StreamSpanReporter implements SpanReporter {
this.queue = queue;
}
@InboundChannelAdapter(value = SleuthSource.OUTPUT)
@InboundChannelAdapter(value = SleuthSource.OUTPUT, poller = @Poller(POLLER))
public Spans poll() {
List<Span> result = new LinkedList<>();
this.queue.drainTo(result);

142
spring-cloud-sleuth-stream/src/test/java/org/springframework/cloud/sleuth/stream/SleuthStreamAutoConfigurationTest.java Normal file
View File

@@ -0,0 +1,142 @@
package org.springframework.cloud.sleuth.stream;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import org.junit.After;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.test.EnvironmentTestUtils;
import org.springframework.cloud.sleuth.autoconfig.TraceAutoConfiguration;
import org.springframework.cloud.sleuth.log.NoOpSpanLogger;
import org.springframework.cloud.sleuth.log.SpanLogger;
import org.springframework.cloud.sleuth.metric.TraceMetricsAutoConfiguration;
import org.springframework.cloud.stream.config.ChannelBindingAutoConfiguration;
import org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.scheduling.support.SimpleTriggerContext;
import static org.assertj.core.api.Assertions.assertThat;
public class SleuthStreamAutoConfigurationTest {
private AnnotationConfigApplicationContext ctx;
private static DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
private static final String TEST_SCHEDULED = "2016-01-01 12:00:00";
private static final String TEST_COMPLETION = "2016-01-01 12:00:02";
private static Date LAST_SCHEDULED_DATE;
private static Date LAST_EXECUTED_DATE;
private static Date LAST_COMPLETED_DATE;
@BeforeClass
public static void setupTests() throws ParseException {
LAST_SCHEDULED_DATE = dateFormat.parse(TEST_SCHEDULED);
LAST_EXECUTED_DATE = new Date(LAST_SCHEDULED_DATE.getTime());
LAST_COMPLETED_DATE = dateFormat.parse(TEST_COMPLETION);
}
@After
public void cleanup() {
if (ctx != null) {
ctx.close();
}
}
@Test
public void shouldUseDefaultPollerConfiguration() {
ctx = createContext();
ctx.refresh();
PollerMetadata poller = ctx.getBean(StreamSpanReporter.POLLER,
PollerMetadata.class);
assertThat(poller).isNotNull();
assertPollerConfigurationUsingConfigurationProperties(poller);
}
@Test
public void shouldUseCustomPollerConfiguration() {
ctx = createContext();
EnvironmentTestUtils.addEnvironment(ctx,
"spring.sleuth.stream.poller.fixed-delay=5000",
"spring.sleuth.stream.poller.max-messages-per-poll=100");
ctx.refresh();
PollerMetadata poller = ctx.getBean(StreamSpanReporter.POLLER,
PollerMetadata.class);
assertThat(poller).isNotNull();
assertPollerConfigurationUsingConfigurationProperties(poller);
}
@Test
public void shouldUseCustomPollerBean() {
ctx = createContext(CustomPollerConfiguration.class);
ctx.refresh();
PollerMetadata poller = ctx.getBean(StreamSpanReporter.POLLER,
PollerMetadata.class);
assertThat(poller).isNotNull();
assertPollerConfiguration(poller, 500L, 5000L);
}
private void assertPollerConfigurationUsingConfigurationProperties(
PollerMetadata poller) {
SleuthStreamProperties sleuth = ctx.getBean(SleuthStreamProperties.class);
assertPollerConfiguration(poller, sleuth.getPoller().getMaxMessagesPerPoll(),
sleuth.getPoller().getFixedDelay());
}
private void assertPollerConfiguration(PollerMetadata poller,
long expectedMaxMessages, long expectedFixedDelay) {
assertThat(poller.getMaxMessagesPerPoll()).isEqualTo(expectedMaxMessages);
Trigger trigger = poller.getTrigger();
assertThat(trigger).isInstanceOf(PeriodicTrigger.class);
TriggerContext triggerContext = new SimpleTriggerContext(LAST_SCHEDULED_DATE,
LAST_EXECUTED_DATE, LAST_COMPLETED_DATE);
Date nextExecution = trigger.nextExecutionTime(triggerContext);
assertThat(nextExecution.getTime())
.isEqualTo(LAST_COMPLETED_DATE.getTime() + expectedFixedDelay);
}
public AnnotationConfigApplicationContext createContext(Class<?>... classes) {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
if (classes != null && classes.length > 0) {
context.register(classes);
}
context.register(BaseConfiguration.class);
return context;
}
@Configuration
@Import({ SleuthStreamAutoConfiguration.class, TraceMetricsAutoConfiguration.class,
TestSupportBinderAutoConfiguration.class,
ChannelBindingAutoConfiguration.class, TraceAutoConfiguration.class,
PropertyPlaceholderAutoConfiguration.class })
public static class BaseConfiguration {
@Bean
SpanLogger spanLogger() {
return new NoOpSpanLogger();
}
}
@Configuration
public static class CustomPollerConfiguration {
@Bean(name = StreamSpanReporter.POLLER)
PollerMetadata customPoller() {
PollerMetadata poller = new PollerMetadata();
poller.setMaxMessagesPerPoll(500);
poller.setTrigger(new PeriodicTrigger(5000L));
return poller;
}
}
}

30
spring-cloud-sleuth-stream/src/test/java/org/springframework/cloud/sleuth/stream/StreamSpanListenerTests.java
View File

@@ -33,7 +33,6 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.actuate.metrics.CounterService;
import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.sleuth.Sampler;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.SpanReporter;
@@ -53,22 +52,29 @@ import org.springframework.context.annotation.Import;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Dave Syer
*
*/
@SpringApplicationConfiguration(classes = TestConfiguration.class)
@ContextConfiguration(classes = TestConfiguration.class)
@RunWith(SpringJUnit4ClassRunner.class)
public class StreamSpanListenerTests {
@Autowired Tracer tracer;
@Autowired ApplicationContext application;
@Autowired ZipkinTestConfiguration test;
@Autowired StreamSpanReporter listener;
@Autowired CounterService counterService;
@Autowired SpanReporter spanReporter;
@Autowired
Tracer tracer;
@Autowired
ApplicationContext application;
@Autowired
ZipkinTestConfiguration test;
@Autowired
StreamSpanReporter listener;
@Autowired
CounterService counterService;
@Autowired
SpanReporter spanReporter;
@PostConstruct
public void init() {
@@ -149,7 +155,8 @@ public class StreamSpanListenerTests {
TraceMetricsAutoConfiguration.class, TestSupportBinderAutoConfiguration.class,
ChannelBindingAutoConfiguration.class, TraceAutoConfiguration.class,
PropertyPlaceholderAutoConfiguration.class })
protected static class TestConfiguration {}
protected static class TestConfiguration {
}
@Configuration
@MessageEndpoint
@@ -160,7 +167,7 @@ public class StreamSpanListenerTests {
@Autowired
StreamSpanReporter listener;
@ServiceActivator(inputChannel=SleuthSource.OUTPUT)
@ServiceActivator(inputChannel = SleuthSource.OUTPUT)
public void handle(Message<?> msg) {
}
@@ -174,7 +181,8 @@ public class StreamSpanListenerTests {
return new AlwaysSampler();
}
@Bean CounterService counterService() {
@Bean
CounterService counterService() {
return Mockito.mock(CounterService.class);
}