Revert "Adds CollectedSpanHandler"

This reverts commit 3e9c457.

Author: Adrian Cole

zipkin-collector/activemq/src/main/java/zipkin2/collector/activemq/ActiveMQCollector.java

@@ -22,7 +22,6 @@
 import zipkin2.collector.CollectorComponent;
 import zipkin2.collector.CollectorMetrics;
 import zipkin2.collector.CollectorSampler;
-import zipkin2.collector.handler.CollectedSpanHandler;
 import zipkin2.storage.StorageComponent;
 
 /** This collector consumes encoded binary messages from a ActiveMQ queue. */
@@ -39,18 +38,13 @@ public static final class Builder extends CollectorComponent.Builder {
     String queue = "zipkin";
     int concurrency = 1;
 
-    @Override public Builder sampler(CollectorSampler sampler) {
-      this.delegate.sampler(sampler);
-      return this;
-    }
-
-    @Override public Builder addCollectedSpanHandler(CollectedSpanHandler collectedSpanHandler) {
-      this.delegate.addCollectedSpanHandler(collectedSpanHandler);
+    @Override public Builder storage(StorageComponent storage) {
+      this.delegate.storage(storage);
       return this;
     }
 
-    @Override public Builder storage(StorageComponent storage) {
-      this.delegate.storage(storage);
+    @Override public Builder sampler(CollectorSampler sampler) {
+      this.delegate.sampler(sampler);
       return this;
     }
 

zipkin-collector/core/src/main/java/zipkin2/collector/Collector.java

@@ -15,19 +15,16 @@
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.Executor;
 import java.util.function.Supplier;
 import java.util.logging.Logger;
-import zipkin2.Call;
 import zipkin2.Callback;
 import zipkin2.Span;
 import zipkin2.SpanBytesDecoderDetector;
 import zipkin2.codec.BytesDecoder;
 import zipkin2.codec.SpanBytesDecoder;
-import zipkin2.collector.handler.CollectedSpanHandler;
 import zipkin2.storage.StorageComponent;
 
 import static java.lang.String.format;
@@ -36,13 +33,21 @@
 
 /**
  * This component takes action on spans received from a transport. This includes deserializing,
- * sampling, invoking any handlers, and scheduling for storage.
+ * sampling and scheduling for storage.
  *
  * <p>Callbacks passed do not propagate to the storage layer. They only return success or failures
  * before storage is attempted. This ensures that calling threads are disconnected from storage
  * threads.
  */
 public class Collector { // not final for mock
+  static final Callback<Void> NOOP_CALLBACK = new Callback<Void>() {
+    @Override public void onSuccess(Void value) {
+    }
+
+    @Override public void onError(Throwable t) {
+    }
+  };
+
   /** Needed to scope this to the correct logging category */
   public static Builder newBuilder(Class<?> loggingClass) {
     if (loggingClass == null) throw new NullPointerException("loggingClass == null");
@@ -51,35 +56,14 @@ public static Builder newBuilder(Class<?> loggingClass) {
 
   public static final class Builder {
     final Logger logger;
-    final ArrayList<CollectedSpanHandler> collectedSpanHandlers = new ArrayList<>();
     StorageComponent storage;
+    CollectorSampler sampler;
     CollectorMetrics metrics;
 
     Builder(Logger logger) {
       this.logger = logger;
     }
 
-    /** @see {@link CollectorComponent.Builder#sampler(CollectorSampler)} */
-    public Builder sampler(CollectorSampler sampler) {
-      if (sampler == null) throw new NullPointerException("sampler == null");
-      this.collectedSpanHandlers.add(0, sampler); // sample first
-      return this;
-    }
-
-    /**
-     * @see {@link CollectorComponent.Builder#addCollectedSpanHandler(CollectedSpanHandler)}
-     * @since 2.17
-     */
-    public Builder addCollectedSpanHandler(CollectedSpanHandler collectedSpanHandler) {
-      if (collectedSpanHandler == null) {
-        throw new NullPointerException("collectedSpanHandler == null");
-      }
-      if (collectedSpanHandler != CollectedSpanHandler.NOOP) { // lenient on config bug
-        this.collectedSpanHandlers.add(collectedSpanHandler);
-      }
-      return this;
-    }
-
     /** @see {@link CollectorComponent.Builder#storage(StorageComponent)} */
     public Builder storage(StorageComponent storage) {
       if (storage == null) throw new NullPointerException("storage == null");
@@ -94,14 +78,21 @@ public Builder metrics(CollectorMetrics metrics) {
       return this;
     }
 
+    /** @see {@link CollectorComponent.Builder#sampler(CollectorSampler)} */
+    public Builder sampler(CollectorSampler sampler) {
+      if (sampler == null) throw new NullPointerException("sampler == null");
+      this.sampler = sampler;
+      return this;
+    }
+
     public Collector build() {
       return new Collector(this);
     }
   }
 
   final Logger logger;
   final CollectorMetrics metrics;
-  final CollectedSpanHandler handler;
+  final CollectorSampler sampler;
   final StorageComponent storage;
 
   Collector(Builder builder) {
@@ -110,7 +101,7 @@ public Collector build() {
     this.metrics = builder.metrics == null ? CollectorMetrics.NOOP_METRICS : builder.metrics;
     if (builder.storage == null) throw new NullPointerException("storage == null");
     this.storage = builder.storage;
-    this.handler = consolidate(builder.collectedSpanHandlers);
+    this.sampler = builder.sampler == null ? CollectorSampler.ALWAYS_SAMPLE : builder.sampler;
   }
 
   public void accept(List<Span> spans, Callback<Void> callback) {
@@ -128,25 +119,24 @@ public void accept(List<Span> spans, Callback<Void> callback, Executor executor)
       callback.onSuccess(null);
       return;
     }
+    metrics.incrementSpans(spans.size());
 
-    try {
-      metrics.incrementSpans(spans.size());
-
-      List<Span> handledSpans = handle(spans);
-      if (handledSpans.isEmpty()) {
-        callback.onSuccess(null);
-        return;
-      }
+    List<Span> sampledSpans = sample(spans);
+    if (sampledSpans.isEmpty()) {
+      callback.onSuccess(null);
+      return;
+    }
 
-      // In order to ensure callers are not blocked, we swap callbacks when we get to the storage
-      // phase of this process. Here, we create a callback whose sole purpose is classifying later
-      // errors on this bundle of spans in the same log category. This allows people to only turn on
-      // debug logging in one place.
-      executor.execute(new StoreSpans(handledSpans));
+    // In order to ensure callers are not blocked, we swap callbacks when we get to the storage
+    // phase of this process. Here, we create a callback whose sole purpose is classifying later
+    // errors on this bundle of spans in the same log category. This allows people to only turn on
+    // debug logging in one place.
+    try {
+      executor.execute(new StoreSpans(sampledSpans));
       callback.onSuccess(null);
     } catch (Throwable unexpected) { // ensure if a future is supplied we always set value or error
-      Call.propagateIfFatal(unexpected);
       callback.onError(unexpected);
+      throw unexpected;
     }
   }
 
@@ -214,15 +204,17 @@ String idString(Span span) {
     return span.traceId() + "/" + span.id();
   }
 
-  List<Span> handle(List<Span> input) {
-    List<Span> handled = new ArrayList<>(input.size());
+  List<Span> sample(List<Span> input) {
+    List<Span> sampled = new ArrayList<>(input.size());
     for (int i = 0, length = input.size(); i < length; i++) {
-      Span s = handler.handle(input.get(i));
-      if (s != null) handled.add(s);
+      Span s = input.get(i);
+      if (sampler.isSampled(s.traceId(), Boolean.TRUE.equals(s.debug()))) {
+        sampled.add(s);
+      }
     }
-    int dropped = input.size() - handled.size();
+    int dropped = input.size() - sampled.size();
     if (dropped > 0) metrics.incrementSpansDropped(dropped);
-    return handled;
+    return sampled;
   }
 
   class StoreSpans implements Callback<Void>, Runnable {
@@ -247,7 +239,7 @@ class StoreSpans implements Callback<Void>, Runnable {
     }
 
     @Override public void onError(Throwable t) {
-      handleStorageError(spans, t, NOOP_VOID);
+      handleStorageError(spans, t, NOOP_CALLBACK);
     }
 
     @Override public String toString() {
@@ -301,30 +293,4 @@ String appendSpanIds(List<Span> spans, StringBuilder message) {
 
     return message.append("]").toString();
   }
-
-  static CollectedSpanHandler consolidate(List<CollectedSpanHandler> handlers) {
-    if (handlers.isEmpty()) return CollectedSpanHandler.NOOP;
-    if (handlers.size() == 1) return handlers.get(0);
-    return new MultipleHandler(handlers);
-  }
-
-  static final class MultipleHandler implements CollectedSpanHandler {
-    final CollectedSpanHandler[] handlers; // Array ensures no iterators are created at runtime
-
-    MultipleHandler(List<CollectedSpanHandler> handlers) {
-      this.handlers = handlers.toArray(new CollectedSpanHandler[0]);
-    }
-
-    @Override public Span handle(Span span) {
-      for (CollectedSpanHandler handler : handlers) {
-        span = handler.handle(span);
-        if (span == null) return null;
-      }
-      return span;
-    }
-
-    @Override public String toString() {
-      return Arrays.toString(handlers);
-    }
-  }
 }

zipkin-collector/core/src/main/java/zipkin2/collector/CollectorComponent.java

@@ -15,7 +15,6 @@
 
 import java.util.List;
 import zipkin2.Component;
-import zipkin2.collector.handler.CollectedSpanHandler;
 import zipkin2.storage.SpanConsumer;
 import zipkin2.storage.StorageComponent;
 
@@ -36,31 +35,7 @@ public abstract class CollectorComponent extends Component {
 
   public abstract static class Builder {
     /**
-     * {@link CollectorSampler#isSampled(String, boolean) samples spans} to reduce load on the
-     * storage system. Defaults to always sample.
-     *
-     * <p>Sampling happens before {@link #addCollectedSpanHandler(CollectedSpanHandler) handlers}.
-     *
-     * @deprecated since 2.17, use {@link #addCollectedSpanHandler(CollectedSpanHandler)}
-     */
-    @Deprecated public abstract Builder sampler(CollectorSampler sampler);
-
-    /**
-     * Triggered on each collected span, before storage. This allows the ability to mutate or drop
-     * spans for reasons including remapping tags.
-     *
-     * <p>Handlers execute after {@link #sampler(CollectorSampler) sampling} and before {@link
-     * #storage(StorageComponent) storage}.
-     *
-     * @since 2.17
-     */
-    // empty implementation as this method was added late
-    public Builder addCollectedSpanHandler(CollectedSpanHandler collectedSpanHandler) {
-      return this;
-    }
-
-    /**
-     * Once spans are handled, they are {@link SpanConsumer#accept(List)} queued for storage} using
+     * Once spans are sampled, they are {@link SpanConsumer#accept(List)} queued for storage} using
      * this component.
      */
     public abstract Builder storage(StorageComponent storage);
@@ -71,6 +46,12 @@ public Builder addCollectedSpanHandler(CollectedSpanHandler collectedSpanHandler
      */
     public abstract Builder metrics(CollectorMetrics metrics);
 
+    /**
+     * {@link CollectorSampler#isSampled(String, boolean) samples spans} to reduce load on the
+     * storage system. Defaults to always sample.
+     */
+    public abstract Builder sampler(CollectorSampler sampler);
+
     public abstract CollectorComponent build();
   }
 }

zipkin-collector/core/src/main/java/zipkin2/collector/CollectorSampler.java

@@ -14,7 +14,6 @@
 package zipkin2.collector;
 
 import zipkin2.Span;
-import zipkin2.collector.handler.CollectedSpanHandler;
 import zipkin2.internal.HexCodec;
 
 /**
@@ -27,21 +26,21 @@
  * <p>Accepts a percentage of trace ids by comparing their absolute value against a potentially
  * dynamic boundary. eg {@code isSampled == abs(traceId) <= boundary}
  *
- * <p>While idempotent, this implementation's sample rate won't exactly match the input rate
- * because trace ids are not perfectly distributed across 64bits. For example, tests have shown an
- * error rate of 3% when 100K trace ids are {@link java.util.Random#nextLong random}.
+ * <p>While idempotent, this implementation's sample rate won't exactly match the input rate because
+ * trace ids are not perfectly distributed across 64bits. For example, tests have shown an error
+ * rate of 3% when 100K trace ids are {@link java.util.Random#nextLong random}.
  */
-public abstract class CollectorSampler implements CollectedSpanHandler {
+public abstract class CollectorSampler {
   public static final CollectorSampler ALWAYS_SAMPLE = CollectorSampler.create(1.0f);
 
   /** @param rate minimum sample rate is 0.0001, or 0.01% of traces */
   public static CollectorSampler create(float rate) {
-    if (rate < 0 || rate > 1) {
+    if (rate < 0 || rate > 1)
       throw new IllegalArgumentException("rate should be between 0 and 1: was " + rate);
-    }
     final long boundary = (long) (Long.MAX_VALUE * rate); // safe cast as less <= 1
     return new CollectorSampler() {
-      @Override protected long boundary() {
+      @Override
+      protected long boundary() {
         return boundary;
       }
     };
@@ -60,29 +59,20 @@ public static CollectorSampler create(float rate) {
    *
    * @param hexTraceId the lower 64 bits of the span's trace ID are checked against the boundary
    * @param debug when true, always passes sampling
-   * @deprecated since 2.17, use {@link #handle(Span)}
    */
-  @Deprecated public boolean isSampled(String hexTraceId, boolean debug) {
-    if (debug) return true;
+  public boolean isSampled(String hexTraceId, boolean debug) {
+    if (Boolean.TRUE.equals(debug)) return true;
     long traceId = HexCodec.lowerHexToUnsignedLong(hexTraceId);
     // The absolute value of Long.MIN_VALUE is larger than a long, so Math.abs returns identity.
     // This converts to MAX_VALUE to avoid always dropping when traceId == Long.MIN_VALUE
     long t = traceId == Long.MIN_VALUE ? Long.MAX_VALUE : Math.abs(traceId);
     return t <= boundary();
   }
 
-  /** Returns the input when the input is {@link Span#debug() debug} or its trace ID was sampled. */
-  @Override public Span handle(Span span) {
-    if (isSampled(span.traceId(), Boolean.TRUE.equals(span.debug()))) {
-      return span;
-    }
-    return null;
-  }
-
-  @Override public String toString() {
+  @Override
+  public String toString() {
     return "CollectorSampler(" + boundary() + ")";
   }
 
-  protected CollectorSampler() {
-  }
+  protected CollectorSampler() {}
 }