Add Zipkin stream support

@EnableZipkinStreamServer and bind to a Spring Cloud Stream message
broker. That's it. The default span store is in memory, but Zipkin
also supports MySQL (and Cassandra coming soon).

spring-cloud-sleuth-zipkin-stream/src/main/docker/Dockerfile

@@ -0,0 +1,5 @@
+FROM frolvlad/alpine-oraclejdk8
+VOLUME /tmp
+ADD spring-cloud-sleuth-zipkin-stream-1.0.0.BUILD-SNAPSHOT.jar app.jar
+RUN sh -c 'touch /app.jar'
+CMD ["java","-Djava.security.egd=file:/dev/./urandom","-jar","/app.jar"]

spring-cloud-sleuth-zipkin-stream/src/main/java/org/springframework/cloud/sleuth/zipkin/stream/EnableZipkinStreamServer.java

@@ -0,0 +1,36 @@
+/**
+ * Copyright 2015 The OpenZipkin Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.springframework.cloud.sleuth.zipkin.stream;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.cloud.sleuth.stream.SleuthSink;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.context.annotation.Import;
+
+import io.zipkin.server.EnableZipkinServer;
+
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@EnableBinding(SleuthSink.class)
+@EnableZipkinServer
+@Import(ZipkinMessageListener.class)
+public @interface EnableZipkinStreamServer {
+
+}

spring-cloud-sleuth-zipkin-stream/src/main/java/org/springframework/cloud/sleuth/zipkin/stream/ZipkinMessageListener.java

@@ -0,0 +1,223 @@
+package org.springframework.cloud.sleuth.zipkin.stream;
+
+import java.io.UnsupportedEncodingException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import javax.sql.DataSource;
+
+import org.apache.thrift.TException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionOutcome;
+import org.springframework.boot.autoconfigure.condition.SpringBootCondition;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.cloud.Cloud;
+import org.springframework.cloud.CloudFactory;
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.TimelineAnnotation;
+import org.springframework.cloud.sleuth.stream.Host;
+import org.springframework.cloud.sleuth.stream.SleuthSink;
+import org.springframework.cloud.sleuth.stream.Spans;
+import org.springframework.cloud.sleuth.zipkin.stream.ZipkinMessageListener.NotSleuthStreamClient;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+import org.springframework.core.env.ConfigurableEnvironment;
+import org.springframework.core.env.Environment;
+import org.springframework.core.env.MapPropertySource;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+import org.springframework.integration.annotation.MessageEndpoint;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.util.StringUtils;
+
+import io.zipkin.Annotation;
+import io.zipkin.BinaryAnnotation;
+import io.zipkin.BinaryAnnotation.Type;
+import io.zipkin.Endpoint;
+import io.zipkin.Span.Builder;
+import io.zipkin.SpanStore;
+import lombok.extern.apachecommons.CommonsLog;
+
+@MessageEndpoint
+@CommonsLog
+@Conditional(NotSleuthStreamClient.class)
+public class ZipkinMessageListener {
+
+	@Autowired
+	SpanStore spanStore;
+
+	@ServiceActivator(inputChannel = SleuthSink.INPUT)
+	public void sink(Spans input) throws TException {
+		List<io.zipkin.Span> spans = new ArrayList<>();
+		for (Span span : input.getSpans()) {
+			if (!span.getName().equals("message/" + SleuthSink.INPUT)) {
+				spans.add(convert(span, input.getHost()));
+			}
+			else {
+				log.warn("Message tracing cycle detected for: " + span);
+			}
+		}
+		if (!spans.isEmpty()) {
+			this.spanStore.accept(spans);
+		}
+	}
+
+	/**
+	 * Converts a given Sleuth span to a Zipkin Span.
+	 * <ul>
+	 * <li>Set ids, etc
+	 * <li>Create timeline annotations based on data from Span object.
+	 * <li>Create binary annotations based on data from Span object.
+	 * </ul>
+	 */
+	public io.zipkin.Span convert(Span span, Host host) {
+		Builder zipkinSpan = new io.zipkin.Span.Builder();
+
+		Endpoint ep = Endpoint.create(host.getServiceName(), host.getIpv4(),
+				host.getPort().shortValue());
+		List<Annotation> annotationList = createZipkinAnnotations(span, ep);
+		List<BinaryAnnotation> binaryAnnotationList = createZipkinBinaryAnnotations(span,
+				ep);
+		zipkinSpan.traceId(hash(span.getTraceId()));
+		if (span.getParents().size() > 0) {
+			if (span.getParents().size() > 1) {
+				log.error("zipkin doesn't support spans with multiple parents.  Omitting "
+						+ "other parents for " + span);
+			}
+			zipkinSpan.parentId(hash(span.getParents().get(0)));
+		}
+		zipkinSpan.id(hash(span.getSpanId()));
+		if (StringUtils.hasText(span.getName())) {
+			zipkinSpan.name(span.getName());
+		}
+		for (Annotation annotation : annotationList) {
+			zipkinSpan.addAnnotation(annotation);
+		}
+		for (BinaryAnnotation annotation : binaryAnnotationList) {
+			zipkinSpan.addBinaryAnnotation(annotation);
+		}
+		return zipkinSpan.build();
+	}
+
+	/**
+	 * Add annotations from the sleuth Span.
+	 */
+	private List<Annotation> createZipkinAnnotations(Span span, Endpoint endpoint) {
+		List<Annotation> annotationList = new ArrayList<>();
+		for (TimelineAnnotation ta : span.getTimelineAnnotations()) {
+			Annotation zipkinAnnotation = createZipkinAnnotation(ta.getMsg(),
+					ta.getTime(), endpoint, true);
+			annotationList.add(zipkinAnnotation);
+		}
+		return annotationList;
+	}
+
+	/**
+	 * Creates a list of Annotations that are present in sleuth Span object.
+	 *
+	 * @return list of Annotations that could be added to Zipkin Span.
+	 */
+	private List<BinaryAnnotation> createZipkinBinaryAnnotations(Span span,
+			Endpoint endpoint) {
+		List<BinaryAnnotation> l = new ArrayList<>();
+		for (Map.Entry<String, String> e : span.getAnnotations().entrySet()) {
+			BinaryAnnotation.Builder binaryAnn = new BinaryAnnotation.Builder();
+			binaryAnn.type(Type.STRING);
+			binaryAnn.key(e.getKey());
+			try {
+				binaryAnn.value(e.getValue().getBytes("UTF-8"));
+			}
+			catch (UnsupportedEncodingException ex) {
+				log.error("Error encoding string as UTF-8", ex);
+			}
+			binaryAnn.endpoint(endpoint);
+			l.add(binaryAnn.build());
+		}
+		return l;
+	}
+
+	/**
+	 * Create an annotation with the correct times and endpoint.
+	 *
+	 * @param value Annotation value
+	 * @param time timestamp will be extracted
+	 * @param endpoint the endpoint this annotation will be associated with.
+	 * @param sendRequest use the first or last timestamp.
+	 */
+	private static Annotation createZipkinAnnotation(String value, long time,
+			Endpoint endpoint, boolean sendRequest) {
+		Annotation.Builder annotation = new Annotation.Builder();
+		annotation.endpoint(endpoint);
+
+		// Zipkin is in microseconds
+		if (sendRequest) {
+			annotation.timestamp(time * 1000);
+		}
+		else {
+			annotation.timestamp(time * 1000);
+		}
+		annotation.value(value);
+		return annotation.build();
+	}
+
+	private static long hash(String string) {
+		long h = 1125899906842597L;
+		if (string == null) {
+			return h;
+		}
+		int len = string.length();
+
+		for (int i = 0; i < len; i++) {
+			h = 31 * h + string.charAt(i);
+		}
+		return h;
+	}
+
+	protected static class NotSleuthStreamClient extends SpringBootCondition {
+
+		@Override
+		public ConditionOutcome getMatchOutcome(ConditionContext context,
+				AnnotatedTypeMetadata metadata) {
+			Environment environment = context.getEnvironment();
+			if ("true".equals(environment
+					.resolvePlaceholders("${spring.sleuth.stream.enabled:}"))) {
+				return ConditionOutcome
+						.noMatch("Found spring.sleuth.stream.enabled=true");
+			}
+			if (environment instanceof ConfigurableEnvironment) {
+				ConfigurableEnvironment configurable = (ConfigurableEnvironment) environment;
+				configurable.getPropertySources()
+						.addLast(
+								new MapPropertySource("spring.sleuth.stream",
+										Collections.<String, Object>singletonMap(
+												"spring.sleuth.stream.enabled",
+												"false")));
+			}
+			return ConditionOutcome.match("Not found: spring.sleuth.stream.enabled");
+		}
+
+	}
+
+	@Configuration
+	@Profile("cloud")
+	protected static class CloudDataSourceConfiguration {
+
+		@Bean
+		public Cloud cloud() {
+			return new CloudFactory().getCloud();
+		}
+
+		@Bean
+		@ConfigurationProperties(DataSourceProperties.PREFIX)
+		public DataSource dataSource() {
+			return cloud().getSingletonServiceConnector(DataSource.class, null);
+		}
+
+	}
+
+}

spring-cloud-sleuth-zipkin-stream/src/main/java/org/springframework/cloud/sleuth/zipkin/stream/ZipkinQueryServerApplication.java

@@ -0,0 +1,15 @@
+package org.springframework.cloud.sleuth.zipkin.stream;
+
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+
+@SpringBootApplication
+@EnableZipkinStreamServer
+public class ZipkinQueryServerApplication {
+
+	public static void main(String[] args) throws Exception {
+		new SpringApplicationBuilder(ZipkinQueryServerApplication.class)
+				.properties("spring.config.name=zipkin-server").run(args);
+	}
+
+}

spring-cloud-sleuth-zipkin-stream/src/main/resources/zipkin-server.yml

@@ -0,0 +1,24 @@
+server:
+  port: 9411
+
+spring:
+  datasource:
+    schema: classpath:/mysql.sql
+    url: jdbc:mysql://${MYSQL_HOST:localhost}/test
+    username: root
+    password: root
+# Switch this on to create the schema on startup:
+    initialize: false
+    continueOnError: true
+  sleuth:
+    enabled: false
+zipkin:
+  store:
+    type: mysql # default is inMemory
+
+---
+spring:
+  profiles: test
+zipkin:
+  store:
+    type: mem # default is inMemory

spring-cloud-sleuth-zipkin-stream/src/test/java/org/springframework/cloud/sleuth/zipkin/stream/ZipkinServerApplicationTests.java

@@ -0,0 +1,31 @@
+package org.springframework.cloud.sleuth.zipkin.stream;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.IntegrationTest;
+import org.springframework.boot.test.SpringApplicationConfiguration;
+import org.springframework.cloud.sleuth.zipkin.stream.ZipkinQueryServerApplication;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringApplicationConfiguration(classes = ZipkinQueryServerApplication.class)
+@IntegrationTest({ "server.port=0", "spring.datasource.initialize=true" })
+@ActiveProfiles("test")
+public class ZipkinServerApplicationTests {
+
+	@Autowired
+	private JdbcTemplate jdbcTemplate;
+
+	@Test
+	public void contextLoads() {
+		int count = this.jdbcTemplate.queryForObject("SELECT COUNT(*) FROM zipkin_spans",
+				Integer.class);
+		assertEquals(0, count);
+	}
+
+}

spring-cloud-sleuth-zipkin-stream/src/test/resources/drop.sql

@@ -0,0 +1,5 @@
+drop table zipkin_spans;
+drop table zipkin_annotations;
+drop table zipkin_binary_annotations;
+drop table zipkin_dependencies;
+drop table zipkin_dependency_links;