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

Merge pull request #108 from spring-cloud/remove-brave-dep-merged

Browse Source 
Introduces HttpZipkinSpanReporter (rebased)
This commit is contained in:
Marcin Grzejszczak
2016-01-14 15:19:42 +01:00
parent f7fad008bc bf191d7cba
commit 2965d5053a
21 changed files with 376 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/pom.xml
View File

@@ -80,6 +80,10 @@
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.github.kristofa</groupId>
<artifactId>brave-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>

14
spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/src/test/java/integration/IntegrationTestSpanCollector.java → spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/src/test/java/integration/IntegrationTestZipkinSpanReporter.java
View File

@@ -15,8 +15,9 @@
*/
package integration;
import com.github.kristofa.brave.LoggingSpanCollector;
import com.twitter.zipkin.gen.Span;
import io.zipkin.Span;
import lombok.extern.apachecommons.CommonsLog;
import org.springframework.cloud.sleuth.zipkin.ZipkinSpanReporter;
import java.util.Collections;
import java.util.LinkedList;
@@ -27,13 +28,14 @@ import java.util.List;
*
* @author Marcin Grzejszczak
*/
public class IntegrationTestSpanCollector extends LoggingSpanCollector {
@CommonsLog
public class IntegrationTestZipkinSpanReporter implements ZipkinSpanReporter {
public List<Span> hashedSpans = Collections.<Span>synchronizedList(new LinkedList<Span>());
public List<Span> hashedSpans = Collections.synchronizedList(new LinkedList<>());
@Override
public void collect(Span span) {
super.collect(span);
public void report(Span span) {
log.info(span);
hashedSpans.add(span);
}

34
spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/src/test/java/integration/MessagingApplicationTests.java
View File

@@ -15,31 +15,26 @@
*/
package integration;
import static org.assertj.core.api.BDDAssertions.then;
import java.util.Collection;
import integration.MessagingApplicationTests.IntegrationSpanCollectorConfig;
import org.junit.After;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.boot.test.WebIntegrationTest;
import org.springframework.cloud.sleuth.zipkin.ZipkinSpanReporter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.JdkIdGenerator;
import org.springframework.util.StringUtils;
import com.github.kristofa.brave.SpanCollector;
import com.twitter.zipkin.gen.BinaryAnnotation;
import com.twitter.zipkin.gen.Span;
import integration.MessagingApplicationTests.IntegrationSpanCollectorConfig;
import sample.SampleMessagingApplication;
import tools.AbstractIntegrationTest;
import java.util.Collection;
import static org.assertj.core.api.BDDAssertions.then;
@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = { IntegrationSpanCollectorConfig.class, SampleMessagingApplication.class })
@WebIntegrationTest
@@ -48,7 +43,7 @@ public class MessagingApplicationTests extends AbstractIntegrationTest {
private static int port = 3381;
private static String sampleAppUrl = "http://localhost:" + port;
@Autowired IntegrationTestSpanCollector integrationTestSpanCollector;
@Autowired IntegrationTestZipkinSpanReporter integrationTestSpanCollector;
@After
public void cleanup() {
@@ -79,24 +74,21 @@ public class MessagingApplicationTests extends AbstractIntegrationTest {
}
private void thenThereIsAtLeastOneBinaryAnnotationWithKey(String binaryAnnotationKey) {
then(this.integrationTestSpanCollector.hashedSpans.stream()
.filter(Span::isSetBinary_annotations)
.map(Span::getBinary_annotations)
then(integrationTestSpanCollector.hashedSpans.stream()
.map(s -> s.binaryAnnotations)
.flatMap(Collection::stream)
.filter(binaryAnnotation -> StringUtils.hasText(binaryAnnotation.getKey()))
.map(BinaryAnnotation::getKey)
.anyMatch(binaryAnnotationKey::equals)).isTrue();
.anyMatch(b -> b.key.equals(binaryAnnotationKey))).isTrue();
}
private void thenAllSpansHaveTraceIdEqualTo(String traceId) {
then(this.integrationTestSpanCollector.hashedSpans.stream().allMatch(span -> span.getTrace_id() == zipkinHashedTraceId(traceId))).isTrue();
then(integrationTestSpanCollector.hashedSpans.stream().allMatch(span -> span.traceId == zipkinHashedTraceId(traceId))).isTrue();
}
@Configuration
public static class IntegrationSpanCollectorConfig {
@Bean
SpanCollector integrationTestSpanCollector() {
return new IntegrationTestSpanCollector();
ZipkinSpanReporter integrationTestZipkinSpanReporter() {
return new IntegrationTestZipkinSpanReporter();
}
}

4
spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-ribbon/pom.xml
View File

@@ -68,6 +68,10 @@
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.github.kristofa</groupId>
<artifactId>brave-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>

5
spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-zipkin/pom.xml
View File

@@ -86,6 +86,11 @@
<groupId>io.zipkin</groupId>
<artifactId>zipkin-java-server</artifactId>
</dependency>
<dependency>
<groupId>com.github.kristofa</groupId>
<artifactId>brave-core</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

16
spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-zipkin/src/main/java/sample/SampleZipkinApplication.java
View File

@@ -16,24 +16,25 @@
package sample;
import io.zipkin.Span;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.cloud.sleuth.Sampler;
import org.springframework.cloud.sleuth.sampler.AlwaysSampler;
import org.springframework.cloud.sleuth.zipkin.ZipkinSpanReporter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.scheduling.annotation.EnableAsync;
import com.github.kristofa.brave.LoggingSpanCollector;
import com.github.kristofa.brave.SpanCollector;
/**
* @author Spencer Gibb
*/
@SpringBootApplication
@EnableAspectJAutoProxy(proxyTargetClass = true)
@EnableAsync
@Slf4j
public class SampleZipkinApplication {
public static final String CLIENT_NAME = "testApp";
@@ -50,8 +51,13 @@ public class SampleZipkinApplication {
// Use this for debugging (or if there is no Zipkin server running on port 9411)
@Bean
@ConditionalOnProperty(value="sample.zipkin.enabled", havingValue="false")
public SpanCollector spanCollector() {
return new LoggingSpanCollector();
public ZipkinSpanReporter spanCollector() {
return new ZipkinSpanReporter() {
@Override
public void report(Span span) {
log.info("Reporting span [{}]", span);
}
};
}
}

40
spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-zipkin/src/test/java/integration/IntegrationTestSpanCollector.java
View File

@@ -1,40 +0,0 @@
/*
* Copyright 2013-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 integration;
import com.github.kristofa.brave.LoggingSpanCollector;
import com.twitter.zipkin.gen.Span;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
/**
* Span Collector that logs spans and adds Spans to a list
*
* @author Marcin Grzejszczak
*/
public class IntegrationTestSpanCollector extends LoggingSpanCollector {
public List<Span> hashedSpans = Collections.<Span>synchronizedList(new LinkedList<Span>());
@Override
public void collect(Span span) {
super.collect(span);
hashedSpans.add(span);
}
}

24
spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-zipkin/src/test/java/integration/ZipkinTests.java
View File

@@ -15,27 +15,23 @@
*/
package integration;
import integration.ZipkinTests.WaitUntilZipkinIsUpConfig;
import io.zipkin.server.ZipkinServer;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.boot.test.WebIntegrationTest;
import org.springframework.cloud.sleuth.zipkin.HttpZipkinSpanReporter;
import org.springframework.cloud.sleuth.zipkin.ZipkinProperties;
import org.springframework.cloud.sleuth.zipkin.ZipkinSpanReporter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.JdkIdGenerator;
import com.github.kristofa.brave.EmptySpanCollectorMetricsHandler;
import com.github.kristofa.brave.HttpSpanCollector;
import com.github.kristofa.brave.SpanCollector;
import com.github.kristofa.brave.SpanCollectorMetricsHandler;
import integration.ZipkinTests.WaitUntilZipkinIsUpConfig;
import io.zipkin.server.ZipkinServer;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import sample.SampleZipkinApplication;
import tools.AbstractIntegrationTest;
@@ -77,7 +73,7 @@ public class ZipkinTests extends AbstractIntegrationTest {
public static class WaitUntilZipkinIsUpConfig {
@Bean
@SneakyThrows
public SpanCollector spanCollector(final ZipkinProperties zipkin) {
public ZipkinSpanReporter spanCollector(final ZipkinProperties zipkin) {
await().until(new Runnable() {
@Override
public void run() {
@@ -94,11 +90,9 @@ public class ZipkinTests extends AbstractIntegrationTest {
return getSpanCollector(zipkin);
}
private SpanCollector getSpanCollector(ZipkinProperties zipkin) {
private ZipkinSpanReporter getSpanCollector(ZipkinProperties zipkin) {
String url = "http://localhost:" + zipkin.getPort();
// TODO: parameterize this
SpanCollectorMetricsHandler metrics = new EmptySpanCollectorMetricsHandler();
return HttpSpanCollector.create(url, zipkin.getHttpConfig(), metrics);
return new HttpZipkinSpanReporter(url, zipkin.getFlushInterval());
}
}
}

10
spring-cloud-sleuth-zipkin/pom.xml
View File

@@ -42,8 +42,8 @@
<optional>true</optional>
</dependency>
<dependency>
<groupId>com.github.kristofa</groupId>
<artifactId>brave-core</artifactId>
<groupId>io.zipkin</groupId>
<artifactId>zipkin-java-core</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
@@ -61,6 +61,12 @@
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>mockwebserver</artifactId>
<version>3.0.0</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

5
spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/DiscoveryClientEndpointLocator.java
View File

@@ -20,7 +20,7 @@ import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.cloud.util.InetUtils;
import com.twitter.zipkin.gen.Endpoint;
import io.zipkin.Endpoint;
/**
* An {@link EndpointLocator} that tries to find local service information from a
@@ -43,8 +43,7 @@ public class DiscoveryClientEndpointLocator implements EndpointLocator {
if (instance == null) {
throw new NoServiceInstanceAvailableException();
}
return new Endpoint(getIpAddress(instance),
new Integer(instance.getPort()).shortValue(), instance.getServiceId());
return Endpoint.create(instance.getServiceId(), getIpAddress(instance), instance.getPort());
}
private int getIpAddress(ServiceInstance instance) {

2
spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/EndpointLocator.java
View File

@@ -16,7 +16,7 @@
package org.springframework.cloud.sleuth.zipkin;
import com.twitter.zipkin.gen.Endpoint;
import io.zipkin.Endpoint;
/**
* Strategy for locating a zipkin {@linkplain Endpoint} for the current process.

2
spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/FallbackHavingEndpointLocator.java
View File

@@ -1,6 +1,6 @@
package org.springframework.cloud.sleuth.zipkin;
import com.twitter.zipkin.gen.Endpoint;
import io.zipkin.Endpoint;
import lombok.extern.slf4j.Slf4j;
/**

139
spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/HttpZipkinSpanReporter.java Normal file
View File

@@ -0,0 +1,139 @@
package org.springframework.cloud.sleuth.zipkin;
import io.zipkin.Codec;
import io.zipkin.Span;
import java.io.Closeable;
import java.io.Flushable;
import java.io.IOException;
import java.io.InputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import lombok.extern.apachecommons.CommonsLog;
import static java.util.concurrent.TimeUnit.SECONDS;
/**
* Submits spans using Zipkin's {@code POST /spans} endpoint.
*/
@CommonsLog
public final class HttpZipkinSpanReporter implements ZipkinSpanReporter, Flushable, Closeable {
private static final Charset UTF_8 = Charset.forName("UTF-8");
private final String url;
private final BlockingQueue<Span> pending = new LinkedBlockingQueue<>(1000);
private final Flusher flusher; // Nullable for testing
/**
* @param baseUrl URL of the zipkin query server instance. Like: http://localhost:9411/
* @param flushInterval in seconds. 0 implies spans are {@link #flush() flushed} externally.
*/
public HttpZipkinSpanReporter(String baseUrl, int flushInterval) {
this.url = baseUrl + (baseUrl.endsWith("/") ? "" : "/") + "api/v1/spans";
this.flusher = flushInterval > 0 ? new Flusher(this, flushInterval) : null;
}
/**
* Queues the span for collection, or drops it if the queue is full.
*
* @param span Span, should not be <code>null</code>.
*/
@Override
public void report(Span span) {
// TODO: metrics.incrementAcceptedSpans(1);
if (!pending.offer(span)) {
// TODO: metrics.incrementDroppedSpans(1);
}
}
/**
* Calling this will flush any pending spans to the http transport on the current thread.
*/
@Override
public void flush() {
if (pending.isEmpty()) return;
List<Span> drained = new ArrayList<>(pending.size());
pending.drainTo(drained);
if (drained.isEmpty()) return;
// json-encode the spans for transport
byte[] json = Codec.JSON.writeSpans(drained);
// NOTE: https://github.com/openzipkin/zipkin-java/issues/66 will throw instead of return null.
if (json == null) {
log.debug("failed to encode spans, dropping them: " + drained);
// TODO: metrics.incrementDroppedSpans(spanCount);
return;
}
// Send the json to the zipkin endpoint
try {
postSpans(json);
} catch (IOException e) {
if (log.isDebugEnabled()) { // don't pollute logs unless debug is on.
// TODO: logger test
log.debug("error POSTing spans to " + url + ": as json: " + new String(json, UTF_8), e);
}
// TODO: metrics.incrementDroppedSpans(spanCount);
return;
}
}
/** Calls flush on a fixed interval */
static final class Flusher implements Runnable {
final Flushable flushable;
final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
Flusher(Flushable flushable, int flushInterval) {
this.flushable = flushable;
this.scheduler.scheduleWithFixedDelay(this, 0, flushInterval, SECONDS);
}
@Override
public void run() {
try {
flushable.flush();
} catch (IOException ignored) {
}
}
}
void postSpans(byte[] json) throws IOException {
// intentionally not closing the connection, so as to use keep-alives
HttpURLConnection connection = (HttpURLConnection) new URL(url).openConnection();
connection.setRequestMethod("POST");
connection.addRequestProperty("Content-Type", "application/json");
connection.setDoOutput(true);
connection.setFixedLengthStreamingMode(json.length);
connection.getOutputStream().write(json);
try (InputStream in = connection.getInputStream()) {
while (in.read() != -1) ; // skip
} catch (IOException e) {
try (InputStream err = connection.getErrorStream()) {
if (err != null) { // possible, if the connection was dropped
while (err.read() != -1) ; // skip
}
}
throw e;
}
}
/**
* Requests a cease of delivery. There will be at most one in-flight request processing after this
* call returns.
*/
@Override
public void close() {
if (flusher != null) flusher.scheduler.shutdown();
// throw any outstanding spans on the floor
int dropped = pending.drainTo(new LinkedList<>());
// TODO: metrics.incrementDroppedSpans(dropped);
}
}

6
spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ServerPropertiesEndpointLocator.java
View File

@@ -21,7 +21,7 @@ import org.springframework.boot.context.embedded.EmbeddedServletContainerInitial
import org.springframework.cloud.util.InetUtils;
import org.springframework.context.event.EventListener;
import com.twitter.zipkin.gen.Endpoint;
import io.zipkin.Endpoint;
/**
* @author Dave Syer
@@ -43,7 +43,7 @@ public class ServerPropertiesEndpointLocator implements EndpointLocator {
public Endpoint local() {
int address = getAddress();
Integer port = getPort();
Endpoint ep = new Endpoint(address, port.shortValue(), this.appName);
Endpoint ep = Endpoint.create(this.appName, address, port);
return ep;
}
@@ -71,7 +71,7 @@ public class ServerPropertiesEndpointLocator implements EndpointLocator {
return InetUtils.getIpAddressAsInt(this.serverProperties.getAddress().getHostAddress());
}
else {
return 127 <<24|1;
return 127 << 24 | 1;
}
}
}

17
spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinAutoConfiguration.java
View File

@@ -16,9 +16,6 @@
package org.springframework.cloud.sleuth.zipkin;
import com.github.kristofa.brave.EmptySpanCollectorMetricsHandler;
import com.github.kristofa.brave.HttpSpanCollector;
import com.github.kristofa.brave.SpanCollectorMetricsHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -31,25 +28,21 @@ import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.github.kristofa.brave.SpanCollector;
/**
* @author Spencer Gibb
*/
@Configuration
@EnableConfigurationProperties
@ConditionalOnClass(SpanCollector.class)
@ConditionalOnProperty(value = "spring.zipkin.enabled", matchIfMissing = true)
public class ZipkinAutoConfiguration {
@Bean
@ConditionalOnMissingBean(SpanCollector.class)
public SpanCollector spanCollector() {
@ConditionalOnMissingBean(ZipkinSpanReporter.class)
public ZipkinSpanReporter reporter() {
ZipkinProperties zipkin = zipkinProperties();
String url = "http://" + zipkin.getHost() + ":" + zipkin.getPort();
// TODO: parameterize this
SpanCollectorMetricsHandler metrics = new EmptySpanCollectorMetricsHandler();
return HttpSpanCollector.create(url, zipkin.getHttpConfig(), metrics);
return new HttpZipkinSpanReporter(url, zipkin.getFlushInterval());
}
@Bean
@@ -58,8 +51,8 @@ public class ZipkinAutoConfiguration {
}
@Bean
public ZipkinSpanListener sleuthTracer(SpanCollector spanCollector, EndpointLocator endpointLocator) {
return new ZipkinSpanListener(spanCollector, endpointLocator.local());
public ZipkinSpanListener sleuthTracer(ZipkinSpanReporter reporter, EndpointLocator endpointLocator) {
return new ZipkinSpanListener(reporter, endpointLocator.local());
}
@Configuration

3
spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinProperties.java
View File

@@ -16,7 +16,6 @@
package org.springframework.cloud.sleuth.zipkin;
import com.github.kristofa.brave.HttpSpanCollector;
import org.springframework.boot.context.properties.ConfigurationProperties;
import lombok.Data;
@@ -32,5 +31,5 @@ public class ZipkinProperties {
private String host = "localhost";
private int port = 9411;
private boolean enabled = true;
private HttpSpanCollector.Config httpConfig = HttpSpanCollector.Config.builder().build();
private int flushInterval = 1;
}

84
spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanListener.java
View File

@@ -31,12 +31,10 @@ import org.springframework.context.event.EventListener;
import org.springframework.core.annotation.Order;
import org.springframework.util.StringUtils;
import com.github.kristofa.brave.SpanCollector;
import com.twitter.zipkin.gen.Annotation;
import com.twitter.zipkin.gen.AnnotationType;
import com.twitter.zipkin.gen.BinaryAnnotation;
import com.twitter.zipkin.gen.Endpoint;
import com.twitter.zipkin.gen.zipkinCoreConstants;
import io.zipkin.Annotation;
import io.zipkin.BinaryAnnotation;
import io.zipkin.Constants;
import io.zipkin.Endpoint;
import lombok.extern.apachecommons.CommonsLog;
@@ -48,7 +46,7 @@ public class ZipkinSpanListener {
private static final Charset UTF_8 = Charset.forName("UTF-8");
private static final byte[] UNKNOWN_BYTES = "unknown".getBytes(UTF_8);
private SpanCollector spanCollector;
private ZipkinSpanReporter reporter;
/**
* Endpoint is the visible IP address of this service, the port it is listening on and
* the service name from discovery.
@@ -56,8 +54,8 @@ public class ZipkinSpanListener {
// Visible for testing
Endpoint localEndpoint;
public ZipkinSpanListener(SpanCollector spanCollector, Endpoint localEndpoint) {
this.spanCollector = spanCollector;
public ZipkinSpanListener(ZipkinSpanReporter reporter, Endpoint localEndpoint) {
this.reporter = reporter;
this.localEndpoint = localEndpoint;
}
@@ -75,7 +73,7 @@ public class ZipkinSpanListener {
// If an inbound RPC call, it should log a "sr" annotation.
// If possible, it should log a binary annotation of "ca", indicating the
// caller's address (ex X-Forwarded-For header)
event.getParent().log(zipkinCoreConstants.SERVER_RECV);
event.getParent().log(Constants.SERVER_RECV);
}
}
@@ -85,21 +83,21 @@ public class ZipkinSpanListener {
// For an outbound RPC call, it should log a "cs" annotation.
// If possible, it should log a binary annotation of "sa", indicating the
// destination address.
event.getSpan().log(zipkinCoreConstants.CLIENT_SEND);
event.getSpan().log(Constants.CLIENT_SEND);
}
@EventListener
@Order(0)
public void clientReceive(ClientReceivedEvent event) {
event.getSpan().log(zipkinCoreConstants.CLIENT_RECV);
event.getSpan().log(Constants.CLIENT_RECV);
}
@EventListener
@Order(0)
public void serverSend(ServerSentEvent event) {
if (event.getParent() != null && event.getParent().isRemote()) {
event.getParent().log(zipkinCoreConstants.SERVER_SEND);
this.spanCollector.collect(convert(event.getParent()));
event.getParent().log(Constants.SERVER_SEND);
this.reporter.report(convert(event.getParent()));
}
}
@@ -110,7 +108,7 @@ public class ZipkinSpanListener {
// Zipkin Span.duration corresponds with Sleuth's Span.begin and end
assert event.getSpan().getEnd() != 0;
if (event.getSpan().isExportable()) {
this.spanCollector.collect(convert(event.getSpan()));
this.reporter.report(convert(event.getSpan()));
}
}
@@ -123,8 +121,8 @@ public class ZipkinSpanListener {
* </ul>
*/
// Visible for testing
com.twitter.zipkin.gen.Span convert(Span span) {
com.twitter.zipkin.gen.Span zipkinSpan = new com.twitter.zipkin.gen.Span();
io.zipkin.Span convert(Span span) {
io.zipkin.Span.Builder zipkinSpan = new io.zipkin.Span.Builder();
// A zipkin span without any annotations cannot be queried, add special "lc" to avoid that.
if (span.logs().isEmpty() && span.tags().isEmpty()) {
@@ -132,45 +130,45 @@ public class ZipkinSpanListener {
byte[] processId = span.getProcessId() != null
? span.getProcessId().toLowerCase().getBytes(UTF_8)
: UNKNOWN_BYTES;
BinaryAnnotation component = new BinaryAnnotation()
.setAnnotation_type(AnnotationType.STRING)
.setKey("lc") // LOCAL_COMPONENT
.setValue(processId)
.setHost(this.localEndpoint);
zipkinSpan.addToBinary_annotations(component);
BinaryAnnotation component = new BinaryAnnotation.Builder()
.type(BinaryAnnotation.Type.STRING)
.key("lc") // LOCAL_COMPONENT
.value(processId)
.endpoint(this.localEndpoint).build();
zipkinSpan.addBinaryAnnotation(component);
} else {
addZipkinAnnotations(zipkinSpan, span, this.localEndpoint);
addZipkinBinaryAnnotations(zipkinSpan, span, this.localEndpoint);
}
zipkinSpan.setTimestamp(span.getBegin() * 1000L);
zipkinSpan.setDuration((span.getEnd() - span.getBegin()) * 1000L);
zipkinSpan.setTrace_id(hash(span.getTraceId()));
zipkinSpan.timestamp(span.getBegin() * 1000L);
zipkinSpan.duration((span.getEnd() - span.getBegin()) * 1000L);
zipkinSpan.traceId(hash(span.getTraceId()));
if (span.getParents().size() > 0) {
if (span.getParents().size() > 1) {
log.error("Zipkin doesn't support spans with multiple parents. Omitting "
+ "other parents for " + span);
}
zipkinSpan.setParent_id(hash(span.getParents().get(0)));
zipkinSpan.parentId(hash(span.getParents().get(0)));
}
zipkinSpan.setId(hash(span.getSpanId()));
zipkinSpan.id(hash(span.getSpanId()));
if (StringUtils.hasText(span.getName())) {
zipkinSpan.setName(span.getName());
zipkinSpan.name(span.getName());
}
return zipkinSpan;
return zipkinSpan.build();
}
/**
* Add annotations from the sleuth Span.
*/
private void addZipkinAnnotations(com.twitter.zipkin.gen.Span zipkinSpan,
private void addZipkinAnnotations(io.zipkin.Span.Builder zipkinSpan,
Span span, Endpoint endpoint) {
for (Log ta : span.logs()) {
Annotation zipkinAnnotation = new Annotation()
.setHost(endpoint)
.setTimestamp(ta.getTime() * 1000) // Zipkin is in microseconds
.setValue(ta.getMsg());
zipkinSpan.addToAnnotations(zipkinAnnotation);
Annotation zipkinAnnotation = new Annotation.Builder()
.endpoint(endpoint)
.timestamp(ta.getTime() * 1000) // Zipkin is in microseconds
.value(ta.getMsg()).build();
zipkinSpan.addAnnotation(zipkinAnnotation);
}
}
@@ -179,15 +177,15 @@ public class ZipkinSpanListener {
*
* @return list of Annotations that could be added to Zipkin Span.
*/
private void addZipkinBinaryAnnotations(com.twitter.zipkin.gen.Span zipkinSpan,
private void addZipkinBinaryAnnotations(io.zipkin.Span.Builder zipkinSpan,
Span span, Endpoint endpoint) {
for (Map.Entry<String, String> e : span.tags().entrySet()) {
BinaryAnnotation binaryAnn = new BinaryAnnotation()
.setAnnotation_type(AnnotationType.STRING)
.setKey(e.getKey())
.setValue(e.getValue().getBytes(UTF_8))
.setHost(endpoint);
zipkinSpan.addToBinary_annotations(binaryAnn);
BinaryAnnotation binaryAnn = new BinaryAnnotation.Builder()
.type(BinaryAnnotation.Type.STRING)
.key(e.getKey())
.value(e.getValue().getBytes(UTF_8))
.endpoint(endpoint).build();
zipkinSpan.addBinaryAnnotation(binaryAnn);
}
}

11
spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanReporter.java Normal file
View File

@@ -0,0 +1,11 @@
package org.springframework.cloud.sleuth.zipkin;
import io.zipkin.Span;
public interface ZipkinSpanReporter {
/**
* Receives completed spans from {@link ZipkinSpanListener} and submits them to a Zipkin
* collector.
*/
void report(Span span);
}

4
spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/FallbackHavingEndpointLocatorTests.java
View File

@@ -1,6 +1,6 @@
package org.springframework.cloud.sleuth.zipkin;
import com.twitter.zipkin.gen.Endpoint;
import io.zipkin.Endpoint;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.BDDMockito;
@@ -14,7 +14,7 @@ public class FallbackHavingEndpointLocatorTests {
@Mock DiscoveryClientEndpointLocator discoveryClientEndpointLocator;
@Mock ServerPropertiesEndpointLocator serverPropertiesEndpointLocator;
Endpoint expectedEndpoint = new Endpoint();
Endpoint expectedEndpoint = Endpoint.create("my-tomcat", 127 << 24 | 1, 8080);
@Test
public void should_use_system_property_locator_if_discovery_client_locator_is_not_present() {

96
spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/HttpZipkinSpanReporterTest.java Normal file
View File

@@ -0,0 +1,96 @@
package org.springframework.cloud.sleuth.zipkin;
import okhttp3.mockwebserver.MockResponse;
import okhttp3.mockwebserver.MockWebServer;
import okhttp3.mockwebserver.RecordedRequest;
import okhttp3.mockwebserver.SocketPolicy;
import io.zipkin.Codec;
import io.zipkin.Span;
import java.util.List;
import org.junit.Rule;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
public class HttpZipkinSpanReporterTest {
@Rule
public final MockWebServer server = new MockWebServer();
// set flush interval to 0 so that tests can drive flushing explicitly
HttpZipkinSpanReporter reporter = new HttpZipkinSpanReporter(server.url("").toString(), 0);
@Test
public void reportDoesntDoIO() throws Exception {
reporter.report(span(1L, "foo"));
assertThat(server.getRequestCount()).isZero();
}
@Test
public void reportIncrementsAcceptedMetrics() throws Exception {
reporter.report(span(1L, "foo"));
// TODO: assertThat(metrics.acceptedSpans.get()).isEqualTo(1);
// TODO: assertThat(metrics.droppedSpans.get()).isZero();
}
@Test
public void dropsWhenQueueIsFull() throws Exception {
for (int i = 0; i < 1001; i++)
reporter.report(span(1L, "foo"));
// TODO: assertThat(metrics.acceptedSpans.get()).isEqualTo(1001);
// TODO: assertThat(metrics.droppedSpans.get()).isEqualTo(1);
}
@Test
public void postsSpans() throws Exception {
server.enqueue(new MockResponse());
reporter.report(span(1L, "foo"));
reporter.report(span(2L, "bar"));
reporter.flush(); // manually flush the spans
// Ensure a proper request was sent
RecordedRequest request = server.takeRequest();
assertThat(request.getRequestLine()).isEqualTo("POST /api/v1/spans HTTP/1.1");
assertThat(request.getHeader("Content-Type")).isEqualTo("application/json");
// Now, let's read back the spans we sent!
List<io.zipkin.Span> zipkinSpans = Codec.JSON.readSpans(request.getBody().readByteArray());
assertThat(zipkinSpans).containsExactly(
span(1L, "foo"),
span(2L, "bar")
);
}
@Test
public void incrementsDroppedSpansWhenServerErrors() throws Exception {
server.enqueue(new MockResponse().setResponseCode(500));
reporter.report(span(1L, "foo"));
reporter.report(span(2L, "bar"));
reporter.flush(); // manually flush the spans
// TODO: assertThat(metrics.droppedSpans.get()).isEqualTo(2);
}
@Test
public void incrementsDroppedSpansWhenServerDisconnects() throws Exception {
server.enqueue(new MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AFTER_REQUEST));
reporter.report(span(1L, "foo"));
reporter.report(span(2L, "bar"));
reporter.flush(); // manually flush the spans
// TODO: assertThat(metrics.droppedSpans.get()).isEqualTo(2);
}
static Span span(long traceId, String spanName) {
return new io.zipkin.Span.Builder().traceId(traceId).id(traceId).name(spanName).build();
}
}

33
spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanListenerTests.java
View File

@@ -19,9 +19,7 @@ package org.springframework.cloud.sleuth.zipkin;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import com.twitter.zipkin.gen.Endpoint;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import javax.annotation.PostConstruct;
@@ -49,9 +47,6 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.github.kristofa.brave.EmptySpanCollector;
import com.github.kristofa.brave.SpanCollector;
/**
* @author Dave Syer
*
@@ -85,7 +80,7 @@ public class ZipkinSpanListenerTests {
long start = System.currentTimeMillis();
parent.log("http/request/retry"); // System.currentTimeMillis
com.twitter.zipkin.gen.Span result = listener.convert(parent);
io.zipkin.Span result = listener.convert(parent);
assertThat(result.timestamp)
.isEqualTo(parent.getBegin() * 1000);
@@ -102,18 +97,18 @@ public class ZipkinSpanListenerTests {
parent.log("http/request/retry");
parent.tag("spring-boot/version", "1.3.1.RELEASE");
com.twitter.zipkin.gen.Span result = listener.convert(parent);
io.zipkin.Span result = listener.convert(parent);
assertThat(result.annotations.get(0).host)
assertThat(result.annotations.get(0).endpoint)
.isEqualTo(listener.localEndpoint);
assertThat(result.binary_annotations.get(0).host)
.isEqualTo(result.annotations.get(0).host);
assertThat(result.binaryAnnotations.get(0).endpoint)
.isEqualTo(result.annotations.get(0).endpoint);
}
/** zipkin's Endpoint.serviceName should never be null. */
@Test
public void localEndpointIncludesServiceName() {
assertThat(listener.localEndpoint.service_name)
assertThat(listener.localEndpoint.serviceName)
.isNotEmpty();
}
@@ -127,7 +122,7 @@ public class ZipkinSpanListenerTests {
Trace context = this.traceManager.startSpan("foo");
this.traceManager.close(context);
assertEquals(1, this.test.spans.size());
assertThat(this.test.spans.get(0).getBinary_annotations().get(0).getHost().getService_name())
assertThat(this.test.spans.get(0).binaryAnnotations.get(0).endpoint.serviceName)
.isEqualTo("unknown"); // TODO: "unknown" bc process id, documented as not nullable, is null.
}
@@ -151,7 +146,7 @@ public class ZipkinSpanListenerTests {
@Configuration
protected static class ZipkinTestConfiguration {
private List<com.twitter.zipkin.gen.Span> spans = new ArrayList<>();
private List<io.zipkin.Span> spans = new ArrayList<>();
@Bean
public Sampler<?> defaultSampler() {
@@ -159,13 +154,9 @@ public class ZipkinSpanListenerTests {
}
@Bean
public SpanCollector collector() {
return new EmptySpanCollector() {
@Override
public void collect(com.twitter.zipkin.gen.Span span) {
ZipkinTestConfiguration.this.spans.add(span);
}
};
public ZipkinSpanReporter reporter() {
return this.spans::add;
}
}}
}
}