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

Introduces HttpZipkinSpanReporter

Browse Source 
Brave's span collector, which turned out to not be a great tool for
direct use. Brave internally creates spans before sending to its
collector, so validation is implicit. The flip side of this is using the
collector directly does not validate spans. This means it is easy to
send invalid ones, for example missing span names. The problem is more
difficult as the data is in binary (thrift).

This introduces HttpZipkinSpanReporter, which validates via zipkin-java
classes before sending on the wire. Moreover, this sends in json to make
debugging problems easier.

This does not fully remove the Brave dependency, as further work is
needed. Particularly, Brave is indirectly referenced in other code.

See https://github.com/openzipkin/zipkin-java/issues/68
See #98 (Reporter is an OpenTracing term)
This commit is contained in:
Adrian Cole
2016-01-14 19:18:40 +08:00
committed by Marcin Grzejszczak
parent f7fad008bc
commit 779b32b363
16 changed files with 333 additions and 96 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>

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

@@ -79,13 +79,10 @@ public class MessagingApplicationTests extends AbstractIntegrationTest {
}
private void thenThereIsAtLeastOneBinaryAnnotationWithKey(String binaryAnnotationKey) {
then(this.integrationTestSpanCollector.hashedSpans.stream()
.filter(Span::isSetBinary_annotations)
.map(Span::getBinary_annotations)
then(reporter.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) {

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>

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;
}
}}
}
}