diff --git a/benchmarks/pom.xml b/benchmarks/pom.xml index e50b4d101..546f6a989 100644 --- a/benchmarks/pom.xml +++ b/benchmarks/pom.xml @@ -163,6 +163,9 @@ true + + false + spring-milestones @@ -172,14 +175,6 @@ false - - spring-staging - Spring Staging - https://repo.spring.io/libs-staging-local/ - - false - - spring-releases Spring Releases diff --git a/pom.xml b/pom.xml index ebcac002a..dc4b2e792 100644 --- a/pom.xml +++ b/pom.xml @@ -222,6 +222,9 @@ true + + false + spring-milestones @@ -234,14 +237,6 @@ false - - spring-staging - Spring Staging - https://repo.spring.io/libs-staging-local/ - - false - - spring-releases Spring Releases diff --git a/spring-cloud-sleuth-dependencies/pom.xml b/spring-cloud-sleuth-dependencies/pom.xml index 0f62d7ca8..3f2e86c29 100644 --- a/spring-cloud-sleuth-dependencies/pom.xml +++ b/spring-cloud-sleuth-dependencies/pom.xml @@ -16,7 +16,8 @@ 1.1.7.BUILD-SNAPSHOT 1.8.4 - 1.1.5 + 1.11.1 + 0.4.3 @@ -92,6 +93,16 @@ zipkin-junit ${zipkin.version} + + io.zipkin.reporter + zipkin-reporter + ${zipkin-reporter.version} + + + io.zipkin.reporter + zipkin-sender-urlconnection + ${zipkin-reporter.version} + diff --git a/spring-cloud-sleuth-samples/pom.xml b/spring-cloud-sleuth-samples/pom.xml index 4c5f15eaa..f0f558db9 100644 --- a/spring-cloud-sleuth-samples/pom.xml +++ b/spring-cloud-sleuth-samples/pom.xml @@ -59,12 +59,12 @@ io.zipkin.java zipkin - 1.1.5 + 1.11.1 io.zipkin.java zipkin-server - 1.1.5 + 1.11.1 diff --git a/spring-cloud-sleuth-zipkin/pom.xml b/spring-cloud-sleuth-zipkin/pom.xml index 2d8c740c8..f7e7484a5 100644 --- a/spring-cloud-sleuth-zipkin/pom.xml +++ b/spring-cloud-sleuth-zipkin/pom.xml @@ -61,6 +61,14 @@ io.zipkin.java zipkin + + io.zipkin.reporter + zipkin-reporter + + + io.zipkin.reporter + zipkin-sender-urlconnection + org.springframework spring-messaging 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 index 917109248..98fb93bba 100644 --- 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 @@ -1,29 +1,15 @@ package org.springframework.cloud.sleuth.zipkin; -import java.io.ByteArrayOutputStream; 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 java.util.concurrent.TimeUnit; -import java.util.zip.GZIPOutputStream; -import org.apache.commons.logging.Log; import org.springframework.cloud.sleuth.metric.SpanMetricReporter; -import zipkin.Codec; import zipkin.Span; +import zipkin.reporter.AsyncReporter; +import zipkin.reporter.urlconnection.URLConnectionSender; -import static java.util.concurrent.TimeUnit.SECONDS; /** * Submits spans using Zipkin's {@code POST /spans} endpoint. @@ -31,17 +17,9 @@ import static java.util.concurrent.TimeUnit.SECONDS; * @author Adrian Cole * @since 1.0.0 */ -public final class HttpZipkinSpanReporter - implements ZipkinSpanReporter, Flushable, Closeable { - private static final Log log = org.apache.commons.logging.LogFactory - .getLog(HttpZipkinSpanReporter.class); - 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 - private final boolean compressionEnabled; - private final SpanMetricReporter spanMetricReporter; +public final class HttpZipkinSpanReporter implements ZipkinSpanReporter, Flushable, Closeable { + private final URLConnectionSender sender; + private final AsyncReporter delegate; /** * @param baseUrl URL of the zipkin query server instance. Like: http://localhost:9411/ @@ -51,10 +29,15 @@ public final class HttpZipkinSpanReporter */ public HttpZipkinSpanReporter(String baseUrl, int flushInterval, boolean compressionEnabled, SpanMetricReporter spanMetricReporter) { - this.url = baseUrl + (baseUrl.endsWith("/") ? "" : "/") + "api/v1/spans"; - this.flusher = flushInterval > 0 ? new Flusher(this, flushInterval) : null; - this.compressionEnabled = compressionEnabled; - this.spanMetricReporter = spanMetricReporter; + this.sender = URLConnectionSender.builder() + .endpoint(baseUrl + (baseUrl.endsWith("/") ? "" : "/") + "api/v1/spans") + .compressionEnabled(compressionEnabled) + .build(); + this.delegate = AsyncReporter.builder(this.sender) + .queuedMaxSpans(1000) // historical constraint. Note: AsyncReporter supports memory bounds + .messageTimeout(flushInterval, TimeUnit.SECONDS) + .metrics(new ReporterMetricsAdapter(spanMetricReporter)) + .build(); } /** @@ -64,10 +47,7 @@ public final class HttpZipkinSpanReporter */ @Override public void report(Span span) { - this.spanMetricReporter.incrementAcceptedSpans(1); - if (!this.pending.offer(span)) { - this.spanMetricReporter.incrementDroppedSpans(1); - } + this.delegate.report(span); } /** @@ -75,101 +55,15 @@ public final class HttpZipkinSpanReporter */ @Override public void flush() { - if (this.pending.isEmpty()) - return; - List drained = new ArrayList<>(this.pending.size()); - this.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) { - if (log.isDebugEnabled()) { - log.debug("failed to encode spans, dropping them: " + drained); - } - this.spanMetricReporter.incrementDroppedSpans(drained.size()); - 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 " + this.url + ": as json: " + new String(json, - UTF_8), e); - } - this.spanMetricReporter.incrementDroppedSpans(drained.size()); - } + this.delegate.flush(); } /** - * 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 { - this.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(this.url).openConnection(); - connection.setRequestMethod("POST"); - connection.addRequestProperty("Content-Type", "application/json"); - if (this.compressionEnabled) { - connection.addRequestProperty("Content-Encoding", "gzip"); - ByteArrayOutputStream gzipped = new ByteArrayOutputStream(); - try (GZIPOutputStream compressor = new GZIPOutputStream(gzipped)) { - compressor.write(json); - } - json = gzipped.toByteArray(); - } - 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. + * Blocks until in-flight spans are sent and drops any that are left pending. */ @Override public void close() { - if (this.flusher != null) - this.flusher.scheduler.shutdown(); - // throw any outstanding spans on the floor - int dropped = this.pending.drainTo(new LinkedList<>()); - this.spanMetricReporter.incrementDroppedSpans(dropped); + this.delegate.close(); + this.sender.close(); } } diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ReporterMetricsAdapter.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ReporterMetricsAdapter.java new file mode 100644 index 000000000..b1772ff64 --- /dev/null +++ b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/ReporterMetricsAdapter.java @@ -0,0 +1,43 @@ +package org.springframework.cloud.sleuth.zipkin; + +import org.springframework.cloud.sleuth.metric.SpanMetricReporter; + +import zipkin.reporter.ReporterMetrics; + +final class ReporterMetricsAdapter implements ReporterMetrics { + private final SpanMetricReporter spanMetricReporter; + + public ReporterMetricsAdapter(SpanMetricReporter spanMetricReporter) { + this.spanMetricReporter = spanMetricReporter; + } + + @Override + public ReporterMetrics forTransport(String transport) { + return this; + } + + @Override + public void incrementMessages() { + } + + @Override + public void incrementMessagesDropped() { + } + + @Override + public void incrementSpans(int i) { + this.spanMetricReporter.incrementAcceptedSpans(i); + } + + @Override + public void incrementSpanBytes(int i) { + } + + @Override + public void incrementMessageBytes(int i) { + } + + @Override public void incrementSpansDropped(int i) { + this.spanMetricReporter.incrementDroppedSpans(i); + } +}