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