From cbbb3c2a40c65a113a021a979e5a908323267b7e Mon Sep 17 00:00:00 2001 From: Pavol Loffay Date: Fri, 21 Jul 2017 11:34:30 +0200 Subject: [PATCH] Zipkin settings add thrift encoding (#650) * Zipkin settings add thrift encoding --- .../sleuth/zipkin/HttpZipkinSpanReporter.java | 15 ++++++++++- .../sleuth/zipkin/RestTemplateSender.java | 25 ++++++++++++++++--- .../zipkin/ZipkinAutoConfiguration.java | 4 +-- .../cloud/sleuth/zipkin/ZipkinProperties.java | 11 ++++++++ .../zipkin/HttpZipkinSpanReporterTest.java | 25 +++++++++++++++++++ 5 files changed, 73 insertions(+), 7 deletions(-) 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 72284aa76..c042c8981 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 @@ -9,6 +9,7 @@ import org.springframework.web.client.RestTemplate; import zipkin.Span; import zipkin.reporter.AsyncReporter; +import zipkin.reporter.Encoding; /** * Submits spans using Zipkin's {@code POST /spans} endpoint. @@ -28,7 +29,19 @@ public final class HttpZipkinSpanReporter implements ZipkinSpanReporter, Flushab */ public HttpZipkinSpanReporter(RestTemplate restTemplate, String baseUrl, int flushInterval, SpanMetricReporter spanMetricReporter) { - this.sender = new RestTemplateSender(restTemplate, baseUrl); + this(restTemplate, baseUrl, flushInterval, spanMetricReporter, Encoding.JSON); + } + + /** + * @param restTemplate {@link RestTemplate} used for sending requests to Zipkin + * @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. + * @param spanMetricReporter service to count number of accepted / dropped spans + * @param encoding span encoding. + */ + public HttpZipkinSpanReporter(RestTemplate restTemplate, String baseUrl, int flushInterval, + SpanMetricReporter spanMetricReporter, Encoding encoding) { + this.sender = new RestTemplateSender(restTemplate, baseUrl, encoding); this.delegate = AsyncReporter.builder(this.sender) .queuedMaxSpans(1000) // historical constraint. Note: AsyncReporter supports memory bounds .messageTimeout(flushInterval, TimeUnit.SECONDS) diff --git a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/RestTemplateSender.java b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/RestTemplateSender.java index 90c27c7d2..f852ef74d 100644 --- a/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/RestTemplateSender.java +++ b/spring-cloud-sleuth-zipkin/src/main/java/org/springframework/cloud/sleuth/zipkin/RestTemplateSender.java @@ -18,13 +18,18 @@ final class RestTemplateSender implements Sender { final RestTemplate restTemplate; final String url; - RestTemplateSender(RestTemplate restTemplate, String baseUrl) { + final Encoding encoding; + final MediaType mediaType; + + RestTemplateSender(RestTemplate restTemplate, String baseUrl, Encoding encoding) { this.restTemplate = restTemplate; this.url = baseUrl + (baseUrl.endsWith("/") ? "" : "/") + "api/v1/spans"; + this.encoding = encoding; + this.mediaType = mediaType(encoding); } @Override public Encoding encoding() { - return Encoding.JSON; + return this.encoding; } @Override public int messageMaxBytes() { @@ -42,7 +47,7 @@ final class RestTemplateSender implements Sender { @Override public void sendSpans(List encodedSpans, Callback callback) { if (this.closeCalled) throw new IllegalStateException("close"); try { - byte[] message = BytesMessageEncoder.JSON.encode(encodedSpans); + byte[] message = BytesMessageEncoder.forEncoding(this.encoding).encode(encodedSpans); post(message); callback.onComplete(); } catch (Throwable e) { @@ -67,9 +72,21 @@ final class RestTemplateSender implements Sender { void post(byte[] json) { HttpHeaders httpHeaders = new HttpHeaders(); - httpHeaders.setContentType(MediaType.APPLICATION_JSON); + httpHeaders.setContentType(this.mediaType); RequestEntity requestEntity = new RequestEntity<>(json, httpHeaders, HttpMethod.POST, URI.create(this.url)); this.restTemplate.exchange(requestEntity, String.class); } + + private MediaType mediaType(Encoding encoding) { + MediaType mediaType = null; + switch (this.encoding) { + case JSON: + mediaType = MediaType.APPLICATION_JSON; + break; + case THRIFT: + mediaType = new MediaType("application","x-thrift"); + } + return mediaType; + } } 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 ac99ca1b4..844236c2d 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 @@ -83,7 +83,7 @@ public class ZipkinAutoConfiguration { RestTemplate restTemplate = zipkinRestTemplate(zipkin); zipkinRestTemplateCustomizer.customize(restTemplate); return new HttpZipkinSpanReporter(restTemplate, zipkin.getBaseUrl(), zipkin.getFlushInterval(), - spanMetricReporter); + spanMetricReporter, zipkin.getEncoding()); } private RestTemplate zipkinRestTemplate(ZipkinProperties zipkinProperties) { @@ -259,4 +259,4 @@ class ZipkinRestTemplateWrapper extends RestTemplate { return originalUrl; } } -} \ No newline at end of file +} 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 037a9dc2f..df357fb94 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 @@ -18,6 +18,8 @@ package org.springframework.cloud.sleuth.zipkin; import org.springframework.boot.context.properties.ConfigurationProperties; +import zipkin.reporter.Encoding; + /** * Zipkin settings * @@ -33,6 +35,7 @@ public class ZipkinProperties { private String baseUrl = "http://localhost:9411/"; private boolean enabled = true; private int flushInterval = 1; + private Encoding encoding = Encoding.JSON; private Compression compression = new Compression(); private Service service = new Service(); @@ -87,6 +90,14 @@ public class ZipkinProperties { this.locator = locator; } + public Encoding getEncoding() { + return this.encoding; + } + + public void setEncoding(Encoding encoding) { + this.encoding = encoding; + } + /** When enabled, spans are gzipped before sent to the zipkin server */ public static class Compression { 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 index 14d808fd9..42dff1d69 100644 --- 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 @@ -15,17 +15,20 @@ import org.springframework.cloud.sleuth.sampler.AlwaysSampler; import org.springframework.cloud.sleuth.trace.DefaultTracer; import org.springframework.cloud.sleuth.util.ExceptionUtils; import org.springframework.web.client.RestTemplate; +import zipkin.reporter.Encoding; import zipkin.Span; import zipkin.junit.HttpFailure; import zipkin.junit.ZipkinRule; import java.util.ArrayList; +import java.util.Collections; import java.util.Random; import java.util.concurrent.atomic.AtomicReference; import static java.util.Arrays.asList; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.BDDAssertions.then; +import static org.awaitility.Awaitility.await; public class HttpZipkinSpanReporterTest { @@ -157,6 +160,28 @@ public class HttpZipkinSpanReporterTest { .contains("peer.service", "redisService"); } + @Test + public void testSenderThriftEncoding() { + ZipkinProperties zipkinProperties = new ZipkinProperties(); + zipkinProperties.setEncoding(Encoding.THRIFT); + zipkinProperties.setBaseUrl(zipkin.httpUrl()); + + HttpZipkinSpanReporter httpZipkinSpanReporter = new HttpZipkinSpanReporter(restTemplate(zipkinProperties) + , zipkinProperties.getBaseUrl(), 1, spanMetricReporter, zipkinProperties.getEncoding()); + + Tracer tracer = new DefaultTracer(new AlwaysSampler(), new Random(), new DefaultSpanNamer(), + new NoOpSpanLogger(),new ZipkinSpanListener(httpZipkinSpanReporter, + new ServerPropertiesEndpointLocator(new ServerProperties(), "foo", + zipkinProperties, new InetUtils(new InetUtilsProperties())), + null, Collections.emptyList()), new TraceKeys()); + + tracer.close(tracer.createSpan("foo")); + httpZipkinSpanReporter.flush(); + + await().until(() -> zipkin.getTraces().size() == 1); + assertThat(zipkin.getTraces().size()).isEqualTo(1); + } + static Span span(long traceId, String spanName) { return Span.builder().traceId(traceId).id(traceId).name(spanName).build(); }