Start dropping spans when the span exporter thread holds more than 10K spans references. (#1893)

* Start dropping spans when the span exporter thread holds more than 10K spans references.

* Fix checker framework.

* Fix findbugs by extracting inlined class definitions.

* Fix tests threads race condition.

* Address review feedback.

Author: bogdandrutu

exporters/metrics/ocagent/src/test/java/io/opencensus/exporter/metrics/ocagent/OcAgentMetricsExporterIntegrationTest.java

@@ -263,7 +263,7 @@ public void testExportMetrics() throws InterruptedException, IOException {
     for (Metric metricProto : metricProtos) {
       actualMetrics.add(metricProto.getMetricDescriptor().getName());
     }
-    assertThat(actualMetrics).containsExactlyElementsIn(expectedMetrics);
+    assertThat(actualMetrics).containsAllIn(expectedMetrics);
   }
 
   private static void registerAllViews() {

impl_core/src/main/java/io/opencensus/implcore/trace/export/SpanExporterImpl.java

@@ -18,8 +18,14 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import io.opencensus.common.Duration;
+import io.opencensus.common.ToLongFunction;
 import io.opencensus.implcore.internal.DaemonThreadFactory;
 import io.opencensus.implcore.trace.RecordEventsSpanImpl;
+import io.opencensus.metrics.DerivedLongCumulative;
+import io.opencensus.metrics.DerivedLongGauge;
+import io.opencensus.metrics.LabelValue;
+import io.opencensus.metrics.MetricOptions;
+import io.opencensus.metrics.Metrics;
 import io.opencensus.trace.export.ExportComponent;
 import io.opencensus.trace.export.SpanData;
 import io.opencensus.trace.export.SpanExporter;
@@ -32,9 +38,37 @@
 import java.util.logging.Logger;
 import javax.annotation.concurrent.GuardedBy;
 
+/*>>>
+import org.checkerframework.checker.nullness.qual.Nullable;
+*/
+
 /** Implementation of the {@link SpanExporter}. */
 public final class SpanExporterImpl extends SpanExporter {
   private static final Logger logger = Logger.getLogger(ExportComponent.class.getName());
+  private static final DerivedLongCumulative droppedSpans =
+      Metrics.getMetricRegistry()
+          .addDerivedLongCumulative(
+              "oc_worker_spans_dropped",
+              MetricOptions.builder()
+                  .setDescription("Number of spans dropped by the exporter thread.")
+                  .setUnit("1")
+                  .build());
+  private static final DerivedLongCumulative pushedSpans =
+      Metrics.getMetricRegistry()
+          .addDerivedLongCumulative(
+              "oc_worker_spans_pushed",
+              MetricOptions.builder()
+                  .setDescription("Number of spans pushed by the exporter thread to the exporter.")
+                  .setUnit("1")
+                  .build());
+  private static final DerivedLongGauge referencedSpans =
+      Metrics.getMetricRegistry()
+          .addDerivedLongGauge(
+              "oc_worker_spans_referenced",
+              MetricOptions.builder()
+                  .setDescription("Current number of spans referenced by the exporter thread.")
+                  .setUnit("1")
+                  .build());
 
   private final Worker worker;
   private final Thread workerThread;
@@ -88,13 +122,64 @@ private SpanExporterImpl(Worker worker) {
         new DaemonThreadFactory("ExportComponent.ServiceExporterThread").newThread(worker);
     this.workerThread.start();
     this.worker = worker;
+    droppedSpans.createTimeSeries(
+        Collections.<LabelValue>emptyList(), this.worker, new ReportDroppedSpans());
+    referencedSpans.createTimeSeries(
+        Collections.<LabelValue>emptyList(), this.worker, new ReportReferencedSpans());
+    pushedSpans.createTimeSeries(
+        Collections.<LabelValue>emptyList(), this.worker, new ReportPushedSpans());
+  }
+
+  private static class ReportDroppedSpans implements ToLongFunction</*@Nullable*/ Worker> {
+    @Override
+    public long applyAsLong(/*@Nullable*/ Worker worker) {
+      if (worker == null) {
+        return 0;
+      }
+      return worker.getDroppedSpans();
+    }
+  }
+
+  private static class ReportReferencedSpans implements ToLongFunction</*@Nullable*/ Worker> {
+    @Override
+    public long applyAsLong(/*@Nullable*/ Worker worker) {
+      if (worker == null) {
+        return 0;
+      }
+      return worker.getReferencedSpans();
+    }
+  }
+
+  private static class ReportPushedSpans implements ToLongFunction</*@Nullable*/ Worker> {
+    @Override
+    public long applyAsLong(/*@Nullable*/ Worker worker) {
+      if (worker == null) {
+        return 0;
+      }
+      return worker.getPushedSpans();
+    }
   }
 
   @VisibleForTesting
   Thread getServiceExporterThread() {
     return workerThread;
   }
 
+  @VisibleForTesting
+  long getDroppedSpans() {
+    return worker.getDroppedSpans();
+  }
+
+  @VisibleForTesting
+  long getReferencedSpans() {
+    return worker.getReferencedSpans();
+  }
+
+  @VisibleForTesting
+  long getPushedSpans() {
+    return worker.getPushedSpans();
+  }
+
   // Worker in a thread that batches multiple span data and calls the registered services to export
   // that data.
   //
@@ -110,14 +195,29 @@ private static final class Worker implements Runnable {
     @GuardedBy("monitor")
     private final List<RecordEventsSpanImpl> spans;
 
-    private final Map<String, Handler> serviceHandlers = new ConcurrentHashMap<String, Handler>();
+    @GuardedBy("monitor")
+    private long referencedSpans = 0;
+
+    @GuardedBy("monitor")
+    private long droppedSpans = 0;
+
+    @GuardedBy("monitor")
+    private long pushedSpans = 0;
+
+    private final Map<String, Handler> serviceHandlers = new ConcurrentHashMap<>();
     private final int bufferSize;
+    private final long maxReferencedSpans;
     private final long scheduleDelayMillis;
 
     // See SpanExporterImpl#addSpan.
     private void addSpan(RecordEventsSpanImpl span) {
       synchronized (monitor) {
+        if (referencedSpans == maxReferencedSpans) {
+          droppedSpans++;
+          return;
+        }
         this.spans.add(span);
+        referencedSpans++;
         if (spans.size() >= bufferSize) {
           monitor.notifyAll();
         }
@@ -154,6 +254,11 @@ private void onBatchExport(List<SpanData> spanDataList) {
     private Worker(int bufferSize, Duration scheduleDelay) {
       spans = new ArrayList<>(bufferSize);
       this.bufferSize = bufferSize;
+      // We notify the worker thread when bufferSize elements in the queue, so we will most likely
+      // have to process more than bufferSize elements but less than 2 * bufferSize in that cycle.
+      // During the processing time we want to allow the same amount of elements to be queued.
+      // So we need to have 4 * bufferSize maximum elements referenced as an estimate.
+      this.maxReferencedSpans = 4L * bufferSize;
       this.scheduleDelayMillis = scheduleDelay.toMillis();
     }
 
@@ -195,6 +300,24 @@ private void flush() {
       exportBatches(spansCopy);
     }
 
+    private long getDroppedSpans() {
+      synchronized (monitor) {
+        return droppedSpans;
+      }
+    }
+
+    private long getReferencedSpans() {
+      synchronized (monitor) {
+        return referencedSpans;
+      }
+    }
+
+    private long getPushedSpans() {
+      synchronized (monitor) {
+        return pushedSpans;
+      }
+    }
+
     @SuppressWarnings("argument.type.incompatible")
     private void exportBatches(ArrayList<RecordEventsSpanImpl> spanList) {
       ArrayList<SpanData> spanDataList = new ArrayList<>(bufferSize);
@@ -209,12 +332,23 @@ private void exportBatches(ArrayList<RecordEventsSpanImpl> spanList) {
           // Cannot clear because the exporter may still have a reference to this list (e.g. async
           // scheduled work), so just create a new list.
           spanDataList = new ArrayList<>(bufferSize);
+          // We removed reference for bufferSize Spans.
+          synchronized (monitor) {
+            referencedSpans -= bufferSize;
+            pushedSpans += bufferSize;
+          }
         }
       }
       // Last incomplete batch, send this as well.
       if (!spanDataList.isEmpty()) {
         // Wrap the list with unmodifiableList to ensure exporter does not change the list.
         onBatchExport(Collections.unmodifiableList(spanDataList));
+        // We removed reference for spanDataList.size() Spans.
+        synchronized (monitor) {
+          referencedSpans -= spanDataList.size();
+          pushedSpans += spanDataList.size();
+        }
+        spanDataList.clear();
       }
     }
   }

impl_core/src/test/java/io/opencensus/implcore/trace/export/SpanExporterImplTest.java

@@ -31,11 +31,15 @@
 import io.opencensus.trace.SpanId;
 import io.opencensus.trace.TraceId;
 import io.opencensus.trace.TraceOptions;
+import io.opencensus.trace.Tracestate;
 import io.opencensus.trace.config.TraceParams;
 import io.opencensus.trace.export.SpanData;
 import io.opencensus.trace.export.SpanExporter.Handler;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.Random;
+import javax.annotation.concurrent.GuardedBy;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -53,10 +57,14 @@ public class SpanExporterImplTest {
       SpanContext.create(
           TraceId.generateRandomId(random),
           SpanId.generateRandomId(random),
-          TraceOptions.builder().setIsSampled(true).build());
+          TraceOptions.builder().setIsSampled(true).build(),
+          Tracestate.builder().build());
   private final SpanContext notSampledSpanContext =
       SpanContext.create(
-          TraceId.generateRandomId(random), SpanId.generateRandomId(random), TraceOptions.DEFAULT);
+          TraceId.generateRandomId(random),
+          SpanId.generateRandomId(random),
+          TraceOptions.DEFAULT,
+          Tracestate.builder().build());
   private final InProcessRunningSpanStore runningSpanStore = new InProcessRunningSpanStore();
   private final SampledSpanStoreImpl sampledSpanStore =
       SampledSpanStoreImpl.getNoopSampledSpanStoreImpl();
@@ -143,6 +151,86 @@ public void exportMoreSpansThanTheBufferSize() {
             span6.toSpanData());
   }
 
+  private static class BlockingExporter extends Handler {
+    final Object monitor = new Object();
+
+    @GuardedBy("monitor")
+    Boolean condition = Boolean.FALSE;
+
+    @Override
+    public void export(Collection<SpanData> spanDataList) {
+      synchronized (monitor) {
+        while (!condition) {
+          try {
+            monitor.wait();
+          } catch (InterruptedException e) {
+            // Do nothing
+          }
+        }
+      }
+    }
+
+    private void unblock() {
+      synchronized (monitor) {
+        condition = Boolean.TRUE;
+        monitor.notifyAll();
+      }
+    }
+  }
+
+  @Test
+  public void exportMoreSpansThanTheMaximumLimit() {
+    final int bufferSize = 4;
+    final int maxReferencedSpans = bufferSize * 4;
+    SpanExporterImpl spanExporter = SpanExporterImpl.create(bufferSize, Duration.create(1, 0));
+    StartEndHandler startEndHandler =
+        new StartEndHandlerImpl(
+            spanExporter, runningSpanStore, sampledSpanStore, new SimpleEventQueue());
+    BlockingExporter blockingExporter = new BlockingExporter();
+
+    spanExporter.registerHandler("test.service", serviceHandler);
+    spanExporter.registerHandler("test.blocking", blockingExporter);
+
+    List<SpanData> spansToExport = new ArrayList<>(maxReferencedSpans);
+    for (int i = 0; i < maxReferencedSpans; i++) {
+      spansToExport.add(createSampledEndedSpan(startEndHandler, "span_1_" + i).toSpanData());
+    }
+
+    assertThat(spanExporter.getReferencedSpans()).isEqualTo(maxReferencedSpans);
+
+    // Now we should start dropping.
+    for (int i = 0; i < 7; i++) {
+      createSampledEndedSpan(startEndHandler, "span_2_" + i);
+      assertThat(spanExporter.getDroppedSpans()).isEqualTo(i + 1);
+    }
+
+    assertThat(spanExporter.getReferencedSpans()).isEqualTo(maxReferencedSpans);
+
+    // Release the blocking exporter
+    blockingExporter.unblock();
+
+    List<SpanData> exported = serviceHandler.waitForExport(maxReferencedSpans);
+    assertThat(exported).isNotNull();
+    assertThat(exported).containsExactlyElementsIn(spansToExport);
+    exported.clear();
+    spansToExport.clear();
+
+    // We cannot compare with maxReferencedSpans here because the worker thread may get
+    // unscheduled immediately after exporting, but before updating the pushed spans, if that is
+    // the case at most bufferSize spans will miss.
+    assertThat(spanExporter.getPushedSpans()).isAtLeast((long) maxReferencedSpans - bufferSize);
+
+    for (int i = 0; i < 7; i++) {
+      spansToExport.add(createSampledEndedSpan(startEndHandler, "span_3_" + i).toSpanData());
+      // No more dropped spans.
+      assertThat(spanExporter.getDroppedSpans()).isEqualTo(7);
+    }
+
+    exported = serviceHandler.waitForExport(7);
+    assertThat(exported).isNotNull();
+    assertThat(exported).containsExactlyElementsIn(spansToExport);
+  }
+
   @Test
   public void interruptWorkerThreadStops() throws InterruptedException {
     SpanExporterImpl spanExporter = SpanExporterImpl.create(4, Duration.create(1, 0));