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

Reroutes HttpZipkinSpanReporter to use zipkin.reporter.AsyncReporter (#409)

Browse Source 
AsyncReporter is a more robust version of what we were doing before.
Notably, it can give a memory threshold instead of span count for the
backlog. This change ports to use AsyncReporter internally.

See https://github.com/openzipkin/zipkin-reporter-java#asyncreporter
This commit is contained in:
Adrian Cole
2016-09-15 22:10:32 +08:00
committed by GitHub
parent 02473f7c18
commit e6d29898dd
7 changed files with 91 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
benchmarks/pom.xml
View File

@@ -163,6 +163,9 @@
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</repository>
<repository>
<id>spring-milestones</id>
@@ -172,14 +175,6 @@
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-staging</id>
<name>Spring Staging</name>
<url>https://repo.spring.io/libs-staging-local/</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-releases</id>
<name>Spring Releases</name>

11
pom.xml
View File

@@ -222,6 +222,9 @@
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</repository>
<repository>
<id>spring-milestones</id>
@@ -234,14 +237,6 @@
<enabled>false</enabled>
</releases>
</repository>
<repository>
<id>spring-staging</id>
<name>Spring Staging</name>
<url>https://repo.spring.io/libs-staging-local/</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-releases</id>
<name>Spring Releases</name>

13
spring-cloud-sleuth-dependencies/pom.xml
View File

@@ -16,7 +16,8 @@
<properties>
<spring-cloud-netflix.version>1.1.7.BUILD-SNAPSHOT</spring-cloud-netflix.version>
<aspectj.version>1.8.4</aspectj.version>
<zipkin.version>1.1.5</zipkin.version>
<zipkin.version>1.11.1</zipkin.version>
<zipkin-reporter.version>0.4.3</zipkin-reporter.version>
</properties>
<dependencyManagement>
<dependencies>
@@ -92,6 +93,16 @@
<artifactId>zipkin-junit</artifactId>
<version>${zipkin.version}</version>
</dependency>
<dependency>
<groupId>io.zipkin.reporter</groupId>
<artifactId>zipkin-reporter</artifactId>
<version>${zipkin-reporter.version}</version>
</dependency>
<dependency>
<groupId>io.zipkin.reporter</groupId>
<artifactId>zipkin-sender-urlconnection</artifactId>
<version>${zipkin-reporter.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
<profiles>

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

@@ -59,12 +59,12 @@
<dependency>
<groupId>io.zipkin.java</groupId>
<artifactId>zipkin</artifactId>
<version>1.1.5</version>
<version>1.11.1</version>
</dependency>
<dependency>
<groupId>io.zipkin.java</groupId>
<artifactId>zipkin-server</artifactId>
<version>1.1.5</version>
<version>1.11.1</version>
</dependency>
</dependencies>
</dependencyManagement>

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

@@ -61,6 +61,14 @@
<groupId>io.zipkin.java</groupId>
<artifactId>zipkin</artifactId>
</dependency>
<dependency>
<groupId>io.zipkin.reporter</groupId>
<artifactId>zipkin-reporter</artifactId>
</dependency>
<dependency>
<groupId>io.zipkin.reporter</groupId>
<artifactId>zipkin-sender-urlconnection</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-messaging</artifactId>

146
spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/HttpZipkinSpanReporter.java
View File

@@ -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<Span> 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<Span> 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<Span> 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();
}
}

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

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