From 779b32b3635d0830b49556e81a6ecc1863a18a48 Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Thu, 14 Jan 2016 19:18:40 +0800 Subject: [PATCH] Introduces HttpZipkinSpanReporter 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) --- .../pom.xml | 4 + .../MessagingApplicationTests.java | 9 +- .../spring-cloud-sleuth-sample-ribbon/pom.xml | 4 + spring-cloud-sleuth-zipkin/pom.xml | 10 +- .../DiscoveryClientEndpointLocator.java | 5 +- .../cloud/sleuth/zipkin/EndpointLocator.java | 2 +- .../zipkin/FallbackHavingEndpointLocator.java | 2 +- .../sleuth/zipkin/HttpZipkinSpanReporter.java | 139 ++++++++++++++++++ .../ServerPropertiesEndpointLocator.java | 6 +- .../zipkin/ZipkinAutoConfiguration.java | 17 +-- .../cloud/sleuth/zipkin/ZipkinProperties.java | 3 +- .../sleuth/zipkin/ZipkinSpanListener.java | 84 ++++++----- .../sleuth/zipkin/ZipkinSpanReporter.java | 11 ++ .../FallbackHavingEndpointLocatorTests.java | 4 +- .../zipkin/HttpZipkinSpanReporterTest.java | 96 ++++++++++++ .../zipkin/ZipkinSpanListenerTests.java | 33 ++--- 16 files changed, 333 insertions(+), 96 deletions(-) create mode 100644 spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/HttpZipkinSpanReporter.java create mode 100644 spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanReporter.java create mode 100644 spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/HttpZipkinSpanReporterTest.java diff --git a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/pom.xml b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/pom.xml index a63c5e8c8..584c7c8b9 100644 --- a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/pom.xml +++ b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/pom.xml @@ -80,6 +80,10 @@ org.projectlombok lombok + + com.github.kristofa + brave-core + org.springframework.boot spring-boot-starter-test diff --git a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/src/test/java/integration/MessagingApplicationTests.java b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/src/test/java/integration/MessagingApplicationTests.java index 1c593f7ce..fcce5dd2d 100644 --- a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/src/test/java/integration/MessagingApplicationTests.java +++ b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-messaging/src/test/java/integration/MessagingApplicationTests.java @@ -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) { diff --git a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-ribbon/pom.xml b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-ribbon/pom.xml index 822f69716..b440cf369 100644 --- a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-ribbon/pom.xml +++ b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-ribbon/pom.xml @@ -68,6 +68,10 @@ org.projectlombok lombok + + com.github.kristofa + brave-core + org.springframework.boot spring-boot-starter-test diff --git a/spring-cloud-sleuth-zipkin/pom.xml b/spring-cloud-sleuth-zipkin/pom.xml index bf26c2c46..670ba6568 100644 --- a/spring-cloud-sleuth-zipkin/pom.xml +++ b/spring-cloud-sleuth-zipkin/pom.xml @@ -42,8 +42,8 @@ true - com.github.kristofa - brave-core + io.zipkin + zipkin-java-core org.projectlombok @@ -61,6 +61,12 @@ assertj-core test + + com.squareup.okhttp3 + mockwebserver + 3.0.0 + test + diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/DiscoveryClientEndpointLocator.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/DiscoveryClientEndpointLocator.java index 11ced5f7e..a608c56f7 100644 --- a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/DiscoveryClientEndpointLocator.java +++ b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/DiscoveryClientEndpointLocator.java @@ -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) { diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/EndpointLocator.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/EndpointLocator.java index 8265d25a7..e51d4d1fb 100644 --- a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/EndpointLocator.java +++ b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/EndpointLocator.java @@ -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. diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/FallbackHavingEndpointLocator.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/FallbackHavingEndpointLocator.java index aad440560..1fde9c058 100644 --- a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/FallbackHavingEndpointLocator.java +++ b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/FallbackHavingEndpointLocator.java @@ -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; /** diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/HttpZipkinSpanReporter.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/HttpZipkinSpanReporter.java new file mode 100644 index 000000000..07556497b --- /dev/null +++ b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/HttpZipkinSpanReporter.java @@ -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 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 null. + */ + @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 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); + } +} diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ServerPropertiesEndpointLocator.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ServerPropertiesEndpointLocator.java index e81f49bee..de01c389f 100644 --- a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ServerPropertiesEndpointLocator.java +++ b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ServerPropertiesEndpointLocator.java @@ -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; } } } diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinAutoConfiguration.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinAutoConfiguration.java index 5cd7f5b85..6033d0205 100644 --- a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinAutoConfiguration.java +++ b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinAutoConfiguration.java @@ -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 diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinProperties.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinProperties.java index bac1b4a2b..efa70c9fd 100644 --- a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinProperties.java +++ b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinProperties.java @@ -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; } diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanListener.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanListener.java index 92177223a..b9d3bbfaf 100644 --- a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanListener.java +++ b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanListener.java @@ -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 { * */ // 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 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); } } diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanReporter.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanReporter.java new file mode 100644 index 000000000..5f1c44f41 --- /dev/null +++ b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanReporter.java @@ -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); +} diff --git a/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/FallbackHavingEndpointLocatorTests.java b/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/FallbackHavingEndpointLocatorTests.java index 2f7a18145..f6b1fc835 100644 --- a/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/FallbackHavingEndpointLocatorTests.java +++ b/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/FallbackHavingEndpointLocatorTests.java @@ -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() { diff --git a/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/HttpZipkinSpanReporterTest.java b/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/HttpZipkinSpanReporterTest.java new file mode 100644 index 000000000..272184331 --- /dev/null +++ b/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/HttpZipkinSpanReporterTest.java @@ -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 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(); + } +} diff --git a/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanListenerTests.java b/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanListenerTests.java index 4c0db6925..b81ae452f 100644 --- a/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanListenerTests.java +++ b/spring-cloud-sleuth-zipkin/src/test/java/org/springframework/cloud/sleuth/zipkin/ZipkinSpanListenerTests.java @@ -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 spans = new ArrayList<>(); + private List 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; } - }} + } +}