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

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

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 21:03:28 +08:00
committed by GitHub
parent f00585a7d4
commit 91e506ce35
8 changed files with 153 additions and 124 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>

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

@@ -16,7 +16,8 @@
<properties>
<spring-cloud-netflix.version>1.2.0.BUILD-SNAPSHOT</spring-cloud-netflix.version>
<aspectj.version>1.8.4</aspectj.version>
<zipkin.version>1.8.4</zipkin.version>
<zipkin.version>1.11.1</zipkin.version>
<zipkin-reporter.version>0.4.3</zipkin-reporter.version>
</properties>
<dependencyManagement>
<dependencies>
@@ -92,6 +93,11 @@
<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>
</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.8.4</version>
<version>1.11.1</version>
</dependency>
<dependency>
<groupId>io.zipkin.java</groupId>
<artifactId>zipkin-server</artifactId>
<version>1.8.4</version>
<version>1.11.1</version>
</dependency>
</dependencies>
</dependencyManagement>

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

@@ -65,6 +65,10 @@
<groupId>io.zipkin.java</groupId>
<artifactId>zipkin</artifactId>
</dependency>
<dependency>
<groupId>io.zipkin.reporter</groupId>
<artifactId>zipkin-reporter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-messaging</artifactId>

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

@@ -2,30 +2,13 @@ package org.springframework.cloud.sleuth.zipkin;
import java.io.Closeable;
import java.io.Flushable;
import java.io.IOException;
import java.net.URI;
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 org.apache.commons.logging.Log;
import org.springframework.cloud.sleuth.metric.SpanMetricReporter;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.RequestEntity;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
import zipkin.Codec;
import zipkin.Span;
import static java.util.concurrent.TimeUnit.SECONDS;
import zipkin.reporter.AsyncReporter;
/**
* Submits spans using Zipkin's {@code POST /spans} endpoint.
@@ -33,17 +16,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 RestTemplate restTemplate;
private final String url;
private final BlockingQueue<Span> pending = new LinkedBlockingQueue<>(1000);
private final Flusher flusher; // Nullable for testing
private final SpanMetricReporter spanMetricReporter;
public final class HttpZipkinSpanReporter implements ZipkinSpanReporter, Flushable, Closeable {
private final RestTemplateSender sender;
private final AsyncReporter<Span> delegate;
/**
* @param restTemplate {@link RestTemplate} used for sending requests to Zipkin
@@ -53,10 +28,12 @@ public final class HttpZipkinSpanReporter
*/
public HttpZipkinSpanReporter(RestTemplate restTemplate, String baseUrl, int flushInterval,
SpanMetricReporter spanMetricReporter) {
this.restTemplate = restTemplate;
this.url = baseUrl + (baseUrl.endsWith("/") ? "" : "/") + "api/v1/spans";
this.flusher = flushInterval > 0 ? new Flusher(this, flushInterval) : null;
this.spanMetricReporter = spanMetricReporter;
this.sender = new RestTemplateSender(restTemplate, baseUrl);
this.delegate = AsyncReporter.builder(this.sender)
.queuedMaxSpans(1000) // historical constraint. Note: AsyncReporter supports memory bounds
.messageTimeout(flushInterval, TimeUnit.SECONDS)
.metrics(new ReporterMetricsAdapter(spanMetricReporter))
.build();
}
/**
@@ -66,10 +43,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);
}
/**
@@ -77,78 +51,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 (RestClientException 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) {
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.setContentType(MediaType.APPLICATION_JSON);
RequestEntity<byte[]> requestEntity = new RequestEntity<>(json, httpHeaders, HttpMethod.POST, URI.create(this.url));
this.restTemplate.exchange(requestEntity, String.class);
}
/**
* 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);
}
}

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

@@ -0,0 +1,75 @@
package org.springframework.cloud.sleuth.zipkin;
import java.net.URI;
import java.util.List;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.RequestEntity;
import org.springframework.web.client.RestTemplate;
import zipkin.reporter.BytesMessageEncoder;
import zipkin.reporter.Callback;
import zipkin.reporter.Encoding;
import zipkin.reporter.Sender;
final class RestTemplateSender implements Sender {
final RestTemplate restTemplate;
final String url;
RestTemplateSender(RestTemplate restTemplate, String baseUrl) {
this.restTemplate = restTemplate;
this.url = baseUrl + (baseUrl.endsWith("/") ? "" : "/") + "api/v1/spans";
}
@Override public Encoding encoding() {
return Encoding.JSON;
}
@Override public int messageMaxBytes() {
// This will drop a span larger than 5MiB. Note: values like 512KiB benchmark better.
return 5 * 1024 * 1024;
}
@Override public int messageSizeInBytes(List<byte[]> spans) {
return encoding().listSizeInBytes(spans);
}
/** close is typically called from a different thread */
transient boolean closeCalled;
@Override public void sendSpans(List<byte[]> encodedSpans, Callback callback) {
if (this.closeCalled) throw new IllegalStateException("close");
try {
byte[] message = BytesMessageEncoder.JSON.encode(encodedSpans);
post(message);
callback.onComplete();
} catch (Throwable e) {
callback.onError(e);
if (e instanceof Error) throw (Error) e;
}
}
/** Sends an empty json message to the configured endpoint. */
@Override public CheckResult check() {
try {
post(new byte[] {'[', ']'});
return CheckResult.OK;
} catch (Exception e) {
return CheckResult.failed(e);
}
}
@Override public void close() {
this.closeCalled = true;
}
void post(byte[] json) {
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.setContentType(MediaType.APPLICATION_JSON);
RequestEntity<byte[]> requestEntity =
new RequestEntity<>(json, httpHeaders, HttpMethod.POST, URI.create(this.url));
this.restTemplate.exchange(requestEntity, String.class);
}
}