Merge branch 'develop' into release/0.8.10

Author: rdegnan

build.gradle

@@ -88,3 +88,17 @@ project(':tracing-idl') {
 
     apply from: file('../gradle/java.gradle')
 }
+
+project(':viz-idl') {
+    description = 'Netifi Proteus Vizceral IDL'
+    ext.artifactName = 'viz-idl'
+
+    apply from: file('../gradle/java.gradle')
+}
+
+project(':vizceral') {
+    description = 'Netifi Proteus Vizceral Service'
+    ext.artifactName = 'vizceral'
+
+    apply from: file('../gradle/java.gradle')
+}

client/src/main/java/io/netifi/proteus/presence/BrokerInfoPresenceNotifier.java

@@ -55,6 +55,7 @@ public void watch(String group) {
                         }
                       }
                     })
+                .onErrorResume(err -> Mono.delay(Duration.ofMillis(500)).then(Mono.error(err)))
                 .retry()
                 .subscribe(this::joinEvent));
   }

settings.gradle

@@ -9,3 +9,5 @@ include 'metrics-micrometer'
 include 'metrics-prometheus'
 include 'tracing-openzipkin'
 include 'tracing-idl'
+include 'viz-idl'
+include 'vizceral'

tracing-idl/src/main/proto/proteus/tracing.proto

@@ -3,13 +3,29 @@ syntax = "proto3";
 package io.netifi.proteus.tracing;
 
 import "zipkin/proto3/zipkin.proto";
+import "google/protobuf/empty.proto";
 
 option java_package = "io.netifi.proteus.tracing";
 option java_outer_classname = "ProteusTracingServices";
 option java_multiple_files = true;
 
 message Ack {}
 
+message Trace {
+    repeated zipkin.proto3.Span spans = 1;
+}
+
+message Traces {
+    repeated Trace traces = 1;
+}
+
+message TracesRequest {
+    int32 lookbackSeconds = 1;
+}
+
 service ProteusTracingService {
+
     rpc StreamSpans (stream zipkin.proto3.Span) returns (Ack) {}
+
+    rpc StreamTraces (TracesRequest) returns (stream Trace) {}
 }

tracing-openzipkin/build.gradle

@@ -15,6 +15,8 @@ dependencies {
 
     compile 'io.opentracing:opentracing-api:0.31.0'
     compile 'io.opentracing.brave:brave-opentracing:0.31.2'
+    compile group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.9.6'
+    compile group: 'com.hubspot.jackson', name: 'jackson-datatype-protobuf', version: '0.9.10-jackson2.9-proto3'
 
     testCompile 'org.apache.logging.log4j:log4j-api:2.9.0'
     testCompile 'org.apache.logging.log4j:log4j-core:2.9.0'

tracing-openzipkin/src/main/java/io/netifi/proteus/tracing/ProteusTraceStreamsSupplier.java

@@ -0,0 +1,23 @@
+package io.netifi.proteus.tracing;
+
+import io.netifi.proteus.Proteus;
+import reactor.core.publisher.Flux;
+
+import javax.inject.Inject;
+import java.util.Optional;
+import java.util.function.Function;
+
+public class ProteusTraceStreamsSupplier implements Function<TracesRequest, Flux<Trace>> {
+
+  private final ProteusTracingServiceClient client;
+
+  @Inject
+  public ProteusTraceStreamsSupplier(Proteus proteus, Optional<String> tracingGroup) {
+    client = new ProteusTracingServiceClient(proteus.group(tracingGroup.orElse("com.netifi.proteus.tracing")));
+  }
+
+  @Override
+  public Flux<Trace> apply(TracesRequest message) {
+    return client.streamTraces(message);
+  }
+}

tracing-openzipkin/src/main/java/io/netifi/proteus/tracing/ProteusZipkinHttpBridge.java

@@ -1,6 +1,22 @@
 package io.netifi.proteus.tracing;
 
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.core.TreeNode;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+import com.fasterxml.jackson.databind.module.SimpleDeserializers;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.databind.type.CollectionType;
+import com.google.protobuf.Empty;
 import com.google.protobuf.InvalidProtocolBufferException;
+import com.hubspot.jackson.datatype.protobuf.ProtobufModule;
+import com.hubspot.jackson.datatype.protobuf.builtin.deserializers.ListValueDeserializer;
 import io.netifi.proteus.Proteus;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelOption;
@@ -11,33 +27,41 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.ipc.netty.http.client.HttpClient;
+import reactor.ipc.netty.http.client.HttpClientResponse;
 import reactor.ipc.netty.resources.PoolResources;
 import zipkin2.proto3.Span;
 
+import java.io.IOException;
 import java.time.Duration;
+import java.util.List;
 import java.util.Optional;
 import java.util.StringJoiner;
+import java.util.function.Function;
 
 public class ProteusZipkinHttpBridge implements ProteusTracingService {
   private static final Logger logger = LoggerFactory.getLogger(ProteusZipkinHttpBridge.class);
 
-  private static final String DEFAULT_ZIPKIN_URL = "/api/v2/spans";
+  private static final String DEFAULT_ZIPKIN_SPANS_URL = "/api/v2/spans";
+  private static final String DEFAULT_ZIPKIN_TRACES_URL = "/api/v2/traces";
 
   private final String host;
-
   private final int port;
-
-  private final String zipkinUrl;
+  private final String zipkinSpansUrl;
   private HttpClient httpClient;
+  private TracesStreamer tracesStreamer;
 
-  public ProteusZipkinHttpBridge(String host, int port, String zipkinUrl) {
-    this.zipkinUrl = zipkinUrl;
+  public ProteusZipkinHttpBridge(String host,
+                                 int port,
+                                 String zipkinSpansUrl,
+                                 String zipkinTracesUrl) {
     this.host = host;
     this.port = port;
+    this.zipkinSpansUrl = zipkinSpansUrl;
+    this.tracesStreamer = new TracesStreamer(zipkinTracesUrl, Mono.fromCallable(this::getClient));
   }
 
   public ProteusZipkinHttpBridge(String host, int port) {
-    this(host, port, DEFAULT_ZIPKIN_URL);
+    this(host, port, DEFAULT_ZIPKIN_SPANS_URL, DEFAULT_ZIPKIN_TRACES_URL);
   }
 
   public static void main(String... args) {
@@ -48,7 +72,8 @@ public static void main(String... args) {
     int brokerPort = Integer.getInteger("netifi.proteus.port", 8001);
     String zipkinHost = System.getProperty("netifi.proteus.zipkinHost", "localhost");
     int zipkinPort = Integer.getInteger("netifi.proteus.zipkinPort", 9411);
-    String zipkinUrl = System.getProperty("netifi.proteus.zipkinUrl", DEFAULT_ZIPKIN_URL);
+    String zipkinSpansUrl = System.getProperty("netifi.proteus.zipkinSpansUrl", DEFAULT_ZIPKIN_SPANS_URL);
+    String zipkinTracesUrl = System.getProperty("netifi.proteus.zipkinTracesUrl", DEFAULT_ZIPKIN_TRACES_URL);
     long accessKey = Long.getLong("netifi.proteus.accessKey", 3855261330795754807L);
     String accessToken =
         System.getProperty("netifi.authentication.accessToken", "kTBDVtfRBO4tHOnZzSyY5ym2kfY");
@@ -58,7 +83,8 @@ public static void main(String... args) {
     logger.info("broker port - {}", brokerPort);
     logger.info("zipkin host - {}", zipkinHost);
     logger.info("zipkin port - {}", zipkinPort);
-    logger.info("zipkin url - {}", zipkinUrl);
+    logger.info("zipkin spans url - {}", zipkinSpansUrl);
+    logger.info("zipkin traces url - {}", zipkinTracesUrl);
     logger.info("access key - {}", accessKey);
 
     Proteus proteus =
@@ -73,7 +99,11 @@ public static void main(String... args) {
 
     proteus.addService(
         new ProteusTracingServiceServer(
-            new ProteusZipkinHttpBridge(zipkinHost, zipkinPort, zipkinUrl),
+            new ProteusZipkinHttpBridge(
+                zipkinHost,
+                zipkinPort,
+                zipkinSpansUrl,
+                zipkinTracesUrl),
             Optional.empty(),
             Optional.empty()));
 
@@ -124,12 +154,12 @@ public Mono<Ack> streamSpans(Publisher<Span> messages, ByteBuf metadata) {
                     .reduce(new StringJoiner(","), StringJoiner::add)
                     .map(stringJoiner -> "[" + stringJoiner.toString() + "]"))
         .onBackpressureBuffer(1 << 16)
-               .flatMap(stringMono -> stringMono)
+        .flatMap(stringMono -> stringMono)
         .concatMap(
             spans ->
                 getClient()
                     .post(
-                        zipkinUrl,
+                        zipkinSpansUrl,
                         request -> {
                           request.addHeader("Content-Type", "application/json");
                           return request.sendString(Mono.just(spans));
@@ -139,7 +169,12 @@ public Mono<Ack> streamSpans(Publisher<Span> messages, ByteBuf metadata) {
             8)
         .doOnError(
             throwable ->
-                logger.error("error sending data to tracing data to url " + zipkinUrl, throwable))
+                logger.error("error sending data to tracing data to url " + zipkinSpansUrl, throwable))
         .then(Mono.never());
   }
+
+  @Override
+  public Flux<Trace> streamTraces(TracesRequest message, ByteBuf metadata) {
+    return tracesStreamer.streamTraces(message.getLookbackSeconds());
+  }
 }

tracing-openzipkin/src/main/java/io/netifi/proteus/tracing/TracesStreamer.java

@@ -0,0 +1,121 @@
+package io.netifi.proteus.tracing;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+import com.fasterxml.jackson.databind.module.SimpleDeserializers;
+import com.hubspot.jackson.datatype.protobuf.ProtobufModule;
+import reactor.core.Exceptions;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.ipc.netty.http.client.HttpClient;
+import zipkin2.proto3.Span;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+public class TracesStreamer {
+
+  private final ObjectMapper objectMapper = protoMapper();
+  private Function<Integer, Mono<InputStream>> inputSource;
+
+  public TracesStreamer(String zipkinUrl,
+                        Mono<HttpClient> client) {
+    this.inputSource = zipkinServerStream(zipkinUrl, client);
+  }
+
+  public TracesStreamer(Mono<InputStream> inputSource) {
+    this.inputSource = v -> inputSource;
+  }
+
+  public Flux<Trace> streamTraces(int lookbackSeconds) {
+    return streamTraces(inputSource.apply(lookbackSeconds));
+  }
+
+  Flux<Trace> streamTraces(Mono<InputStream> input) {
+    return input.map(is -> {
+      try {
+        return objectMapper.<Traces>readValue(
+            is,
+            new TypeReference<Traces>() {
+            });
+      } catch (IOException e) {
+        throw Exceptions.propagate(e);
+      }
+    }).flatMapIterable(Traces::getTracesList)
+        ;
+  }
+
+  private static Function<Integer, Mono<InputStream>> zipkinServerStream(String zipkinUrl,
+                                                                         Mono<HttpClient> client) {
+    return lookbackSeconds -> client
+        .flatMap(c -> c
+            .get(zipkinQuery(zipkinUrl, lookbackSeconds))
+            .flatMap(resp ->
+                resp.receive()
+                    .aggregate()
+                    .asInputStream()));
+  }
+
+  private static String zipkinQuery(String zipkinUrl, int lookbackSeconds) {
+    long lookbackMicros = TimeUnit.SECONDS.toMillis(lookbackSeconds);
+    return zipkinUrl + "?lookback=" + lookbackMicros + "&limit=100000";
+  }
+
+  private ObjectMapper protoMapper() {
+    ObjectMapper mapper = new ObjectMapper();
+    ProtobufModule module = new CustomProtoModule();
+    mapper.registerModule(module);
+    return mapper;
+  }
+
+  public static class CustomProtoModule extends ProtobufModule {
+    @Override
+    public void setupModule(SetupContext context) {
+      super.setupModule(context);
+      SimpleDeserializers deser = new SimpleDeserializers();
+      deser.addDeserializer(Traces.class, new TracersDeserializer());
+      context.addDeserializers(deser);
+    }
+  }
+
+  public static class TracersDeserializer extends StdDeserializer<Traces> {
+
+    public TracersDeserializer() {
+      this(null);
+    }
+
+    protected TracersDeserializer(Class<?> vc) {
+      super(vc);
+    }
+
+    @Override
+    public Traces deserialize(JsonParser p,
+                              DeserializationContext ctxt) throws IOException {
+      p.nextToken();
+      Traces.Builder tracesBuilder = Traces.newBuilder();
+      while (p.currentToken() != JsonToken.END_ARRAY) {
+        tracesBuilder.addTraces(nextTrace(p, ctxt));
+      }
+      p.nextToken();
+      return tracesBuilder.build();
+    }
+
+    private Trace nextTrace(JsonParser p,
+                            DeserializationContext ctxt) throws IOException {
+      Trace.Builder traceBuilder = Trace.newBuilder();
+      p.nextToken();
+      while (p.currentToken() != JsonToken.END_ARRAY) {
+        traceBuilder.addSpans(ctxt.readValue(p, Span.class));
+        p.nextToken();
+      }
+      p.nextToken();
+      return traceBuilder.build();
+    }
+  }
+}

tracing-openzipkin/src/test/java/io/netifi/proteus/tracing/ZipkinTracesStreamerTest.java

@@ -0,0 +1,34 @@
+package io.netifi.proteus.tracing;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import reactor.core.publisher.Mono;
+
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+
+public class ZipkinTracesStreamerTest {
+
+  private TracesStreamer tracesStreamer;
+
+  @Before
+  public void setUp() {
+    tracesStreamer = new TracesStreamer(zipkinSource());
+  }
+
+  @Test
+  public void zipkinServerTracesStreaming() {
+    List<Trace> traces = tracesStreamer
+        .streamTraces(42)
+        .collectList()
+        .block(Duration.ofSeconds(10));
+    Assert.assertFalse(traces.isEmpty());
+  }
+
+  private Mono<InputStream> zipkinSource() {
+    return Mono.fromCallable(() ->
+        getClass().getClassLoader().getResourceAsStream("zipkin.json"));
+  }
+}