Work in progress, updates to vizceral bridge and zipkin bridge so far

Author: kbahr

gradle/wrapper/gradle-wrapper.jar

@@ -1,6 +1,7 @@
 package io.netifi.proteus.tracing;
 
 import io.netifi.proteus.Proteus;
+import io.rsocket.RSocket;
 import java.util.Optional;
 import java.util.function.Function;
 import javax.inject.Inject;
@@ -10,11 +11,16 @@ public class ProteusTraceStreamsSupplier implements Function<TracesRequest, Flux
 
   private final ProteusTracingServiceClient client;
 
-  @Inject
   public ProteusTraceStreamsSupplier(Proteus proteus, Optional<String> tracingGroup) {
+    this(proteus::group, tracingGroup);
+  }
+
+  @Inject
+  public ProteusTraceStreamsSupplier(
+      Function<String, RSocket> rSocketFactory, Optional<String> tracingGroup) {
     client =
         new ProteusTracingServiceClient(
-            proteus.group(tracingGroup.orElse("com.netifi.proteus.tracing")));
+            rSocketFactory.apply(tracingGroup.orElse("com.netifi.proteus.tracing")));
   }
 
   @Override

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

@@ -13,6 +13,7 @@
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.ReplayProcessor;
 import reactor.ipc.netty.http.client.HttpClient;
 import reactor.ipc.netty.resources.PoolResources;
 import zipkin2.proto3.Span;
@@ -21,24 +22,30 @@ public class ProteusZipkinHttpBridge implements ProteusTracingService {
   private static final Logger logger = LoggerFactory.getLogger(ProteusZipkinHttpBridge.class);
 
   private static final String DEFAULT_ZIPKIN_SPANS_URL = "/api/v2/spans";
-  private static final String DEFAULT_ZIPKIN_TRACES_URL = "/api/v2/traces";
+  private static final Duration DEFAULT_TRACE_RETENTION = Duration.ofSeconds(60);
 
   private final String host;
   private final int port;
   private final String zipkinSpansUrl;
+  private final Duration retentionPeriod;
+  private final ReplayProcessor<Span> spansProcessor;
   private HttpClient httpClient;
-  private TracesStreamer tracesStreamer;
+
+  public ProteusZipkinHttpBridge(String host, int port, String zipkinSpansUrl) {
+    this(host, port, zipkinSpansUrl, DEFAULT_TRACE_RETENTION);
+  }
+
+  public ProteusZipkinHttpBridge(String host, int port) {
+    this(host, port, DEFAULT_ZIPKIN_SPANS_URL, DEFAULT_TRACE_RETENTION);
+  }
 
   public ProteusZipkinHttpBridge(
-      String host, int port, String zipkinSpansUrl, String zipkinTracesUrl) {
+      String host, int port, String zipkinSpansUrl, Duration retentionPeriod) {
     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_SPANS_URL, DEFAULT_ZIPKIN_TRACES_URL);
+    this.retentionPeriod = retentionPeriod;
+    this.spansProcessor = ReplayProcessor.createTimeout(this.retentionPeriod);
   }
 
   public static void main(String... args) {
@@ -51,8 +58,8 @@ public static void main(String... args) {
     int zipkinPort = Integer.getInteger("netifi.proteus.zipkinPort", 9411);
     String zipkinSpansUrl =
         System.getProperty("netifi.proteus.zipkinSpansUrl", DEFAULT_ZIPKIN_SPANS_URL);
-    String zipkinTracesUrl =
-        System.getProperty("netifi.proteus.zipkinTracesUrl", DEFAULT_ZIPKIN_TRACES_URL);
+    Duration traceRetentionPeriod =
+        Duration.ofSeconds(Integer.getInteger("netifi.proteus.traceRetentionPeriod", 60));
     long accessKey = Long.getLong("netifi.proteus.accessKey", 3855261330795754807L);
     String accessToken =
         System.getProperty("netifi.authentication.accessToken", "kTBDVtfRBO4tHOnZzSyY5ym2kfY");
@@ -63,7 +70,7 @@ public static void main(String... args) {
     logger.info("zipkin host - {}", zipkinHost);
     logger.info("zipkin port - {}", zipkinPort);
     logger.info("zipkin spans url - {}", zipkinSpansUrl);
-    logger.info("zipkin traces url - {}", zipkinTracesUrl);
+    logger.info("traces retention period - {}", traceRetentionPeriod);
     logger.info("access key - {}", accessKey);
 
     Proteus proteus =
@@ -78,7 +85,8 @@ public static void main(String... args) {
 
     proteus.addService(
         new ProteusTracingServiceServer(
-            new ProteusZipkinHttpBridge(zipkinHost, zipkinPort, zipkinSpansUrl, zipkinTracesUrl),
+            new ProteusZipkinHttpBridge(
+                zipkinHost, zipkinPort, zipkinSpansUrl, traceRetentionPeriod),
             Optional.empty(),
             Optional.empty()));
 
@@ -112,6 +120,9 @@ public Mono<Ack> streamSpans(Publisher<Span> messages, ByteBuf metadata) {
     return Flux.from(messages)
         .map(
             span -> {
+              // Retain span for replay
+              spansProcessor.onNext(span);
+
               try {
                 String json = JsonFormat.printer().print(span);
                 if (logger.isTraceEnabled()) {
@@ -151,6 +162,18 @@ public Mono<Ack> streamSpans(Publisher<Span> messages, ByteBuf metadata) {
 
   @Override
   public Flux<Trace> streamTraces(TracesRequest message, ByteBuf metadata) {
-    return tracesStreamer.streamTraces(message.getLookbackSeconds());
+    Duration skipDuration = retentionPeriod.minus(Duration.ofSeconds(message.getLookbackSeconds()));
+    // This isn't working for the moment - WIP
+    // if (skipDuration.isNegative()) {
+    //    } else {
+    //      traces = spansProcessor.skip(skipDuration);
+    //    }
+    return spansProcessor
+        .filter(p -> p != null)
+        .window(Duration.ofSeconds(1))
+        .flatMap(spans -> spans.collectList())
+        .filter(spans -> spans.size() > 0)
+        .map(spans -> Trace.newBuilder().addAllSpans(spans).build())
+        .onBackpressureDrop();
   }
 }

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

@@ -1,11 +1,11 @@
 package io.netifi.proteus.vizceral.service;
 
-import io.netifi.proteus.Proteus;
 import io.netifi.proteus.tracing.ProteusTraceStreamsSupplier;
 import io.netifi.proteus.tracing.Trace;
 import io.netifi.proteus.tracing.TracesRequest;
 import io.netifi.proteus.vizceral.*;
 import io.netty.buffer.ByteBuf;
+import io.rsocket.RSocket;
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
@@ -28,22 +28,22 @@ public class VizceralBridge implements VizceralService {
   private int tracesLookbackSeconds;
 
   public VizceralBridge(
-      Proteus proteus,
+      Function<String, RSocket> rSocketFactory,
       Optional<String> tracingGroup,
       int intervalSeconds,
       int tracesLookbackSeconds) {
     this(
-        new ProteusTraceStreamsSupplier(proteus, tracingGroup),
+        new ProteusTraceStreamsSupplier(rSocketFactory, tracingGroup),
         intervalSeconds,
         tracesLookbackSeconds);
   }
 
-  public VizceralBridge(Proteus proteus, Optional<String> tracingGroup) {
-    this(proteus, tracingGroup, 5, 10);
+  public VizceralBridge(Function<String, RSocket> rSocketFactory, Optional<String> tracingGroup) {
+    this(rSocketFactory, tracingGroup, 5, 10);
   }
 
-  public VizceralBridge(Proteus proteus) {
-    this(proteus, Optional.empty(), 5, 10);
+  public VizceralBridge(Function<String, RSocket> rSocketFactory) {
+    this(rSocketFactory, Optional.empty(), 5, 10);
   }
 
   VizceralBridge(