Zipkin settings add thrift encoding (#650)

* Zipkin settings add thrift encoding

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)

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<byte[]> 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<byte[]> 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;
+	}
 }

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

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 {
 

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