Record processed spans before export complete and reject on shutdown

sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessor.java

@@ -54,7 +54,6 @@ public final class BatchSpanProcessor implements SpanProcessor {
 
   private final boolean exportUnsampledSpans;
   private final Worker worker;
-  private final AtomicBoolean isShutdown = new AtomicBoolean(false);
 
   /**
    * Returns a new {@link BatchSpanProcessor} with default configuration which batches spans
@@ -130,9 +129,6 @@ public boolean isEndRequired() {
 
   @Override
   public CompletableResultCode shutdown() {
-    if (isShutdown.getAndSet(true)) {
-      return CompletableResultCode.ofSuccess();
-    }
     return worker.shutdown();
   }
 
@@ -200,6 +196,7 @@ private static final class Worker implements Runnable {
     private final AtomicInteger spansNeeded = new AtomicInteger(Integer.MAX_VALUE);
     private final BlockingQueue<Boolean> signal;
     private final AtomicReference<CompletableResultCode> flushRequested = new AtomicReference<>();
+    private final AtomicBoolean isShutdown = new AtomicBoolean(false);
     private volatile boolean continueWork = true;
     private final ArrayList<SpanData> batch;
     private final long maxQueueSize;
@@ -229,9 +226,13 @@ private Worker(
     }
 
     private void addSpan(ReadableSpan span) {
+      if (isShutdown.get()) {
+        spanProcessorInstrumentation.dropSpansAlreadyShutdown(1);
+        return;
+      }
       spanProcessorInstrumentation.buildQueueMetricsOnce(maxQueueSize, queue::size);
       if (!queue.offer(span)) {
-        spanProcessorInstrumentation.dropSpans(1);
+        spanProcessorInstrumentation.dropSpansQueueFull(1);
         droppedSpanCount.incrementAndGet();
       } else {
         if (queueSize.incrementAndGet() >= spansNeeded.get()) {
@@ -298,6 +299,9 @@ private void updateNextExportTime() {
     }
 
     private CompletableResultCode shutdown() {
+      if (isShutdown.getAndSet(true)) {
+        return CompletableResultCode.ofSuccess();
+      }
       CompletableResultCode result = new CompletableResultCode();
 
       CompletableResultCode flushResult = forceFlush();
@@ -348,24 +352,19 @@ private void exportCurrentBatch() {
                 + ")");
       }
 
-      String error = null;
       try {
+        // We always increment for every export invocation, so we increment before the export call
+        // to make sure thrown errors don't affect it.
+        spanProcessorInstrumentation.finishSpans(batch.size());
         CompletableResultCode result = spanExporter.export(Collections.unmodifiableList(batch));
         result.join(exporterTimeoutNanos, TimeUnit.NANOSECONDS);
         if (!result.isSuccess()) {
           logger.log(Level.FINE, "Exporter failed");
-          if (result.getFailureThrowable() != null) {
-            error = result.getFailureThrowable().getClass().getName();
-          } else {
-            error = "export_failed";
-          }
         }
       } catch (Throwable t) {
         ThrowableUtil.propagateIfFatal(t);
         logger.log(Level.WARNING, "Exporter threw an Exception", t);
-        error = t.getClass().getName();
       } finally {
-        spanProcessorInstrumentation.finishSpans(batch.size(), error);
         batch.clear();
       }
     }

sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/LegacySpanProcessorInstrumentation.java

@@ -51,16 +51,18 @@ final class LegacySpanProcessorInstrumentation implements SpanProcessorInstrumen
   }
 
   @Override
-  public void dropSpans(int count) {
+  public void dropSpansQueueFull(int count) {
     processedSpans().add(count, droppedAttrs);
   }
 
   @Override
-  public void finishSpans(int count, @Nullable String error) {
-    // Legacy metrics only record when no error.
-    if (error == null) {
-      processedSpans().add(count, standardAttrs);
-    }
+  public void dropSpansAlreadyShutdown(int count) {
+    // Legacy did not record this metric.
+  }
+
+  @Override
+  public void finishSpans(int count) {
+    processedSpans().add(count, standardAttrs);
   }
 
   @Override

sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SemConvSpanProcessorInstrumentation.java

@@ -27,7 +27,8 @@ final class SemConvSpanProcessorInstrumentation implements SpanProcessorInstrume
 
   private final Supplier<MeterProvider> meterProvider;
   private final Attributes standardAttrs;
-  private final Attributes droppedAttrs;
+  private final Attributes queueFullAttrs;
+  private final Attributes shutdownAttrs;
 
   @Nullable private Meter meter;
   @Nullable private volatile LongCounter processedSpans;
@@ -42,31 +43,37 @@ final class SemConvSpanProcessorInstrumentation implements SpanProcessorInstrume
             componentId.getTypeName(),
             SemConvAttributes.OTEL_COMPONENT_NAME,
             componentId.getComponentName());
-    droppedAttrs =
+    queueFullAttrs =
         Attributes.of(
             SemConvAttributes.OTEL_COMPONENT_TYPE,
             componentId.getTypeName(),
             SemConvAttributes.OTEL_COMPONENT_NAME,
             componentId.getComponentName(),
             SemConvAttributes.ERROR_TYPE,
             "queue_full");
+    shutdownAttrs =
+        Attributes.of(
+            SemConvAttributes.OTEL_COMPONENT_TYPE,
+            componentId.getTypeName(),
+            SemConvAttributes.OTEL_COMPONENT_NAME,
+            componentId.getComponentName(),
+            SemConvAttributes.ERROR_TYPE,
+            "already_shutdown");
   }
 
   @Override
-  public void dropSpans(int count) {
-    processedSpans().add(count, droppedAttrs);
+  public void dropSpansQueueFull(int count) {
+    processedSpans().add(count, queueFullAttrs);
   }
 
   @Override
-  public void finishSpans(int count, @Nullable String error) {
-    if (error == null) {
-      processedSpans().add(count, standardAttrs);
-      return;
-    }
+  public void dropSpansAlreadyShutdown(int count) {
+    processedSpans().add(count, shutdownAttrs);
+  }
 
-    Attributes attributes =
-        standardAttrs.toBuilder().put(SemConvAttributes.ERROR_TYPE, error).build();
-    processedSpans().add(count, attributes);
+  @Override
+  public void finishSpans(int count) {
+    processedSpans().add(count, standardAttrs);
   }
 
   @Override

sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SimpleSpanProcessor.java

@@ -99,28 +99,29 @@ public boolean isStartRequired() {
   @Override
   public void onEnd(ReadableSpan span) {
     if (span != null && (exportUnsampledSpans || span.getSpanContext().isSampled())) {
+      if (isShutdown.get()) {
+        spanProcessorInstrumentation.dropSpansAlreadyShutdown(1);
+        return;
+      }
+
       try {
         List<SpanData> spans = Collections.singletonList(span.toSpanData());
         CompletableResultCode result;
 
         synchronized (exporterLock) {
+          // We always increment for every export invocation, so we increment before the export
+          // call to make sure thrown errors don't affect it.
+          spanProcessorInstrumentation.finishSpans(1);
           result = spanExporter.export(spans);
         }
 
         pendingExports.add(result);
         result.whenComplete(
             () -> {
               pendingExports.remove(result);
-              String error = null;
               if (!result.isSuccess()) {
                 logger.log(Level.FINE, "Exporter failed");
-                if (result.getFailureThrowable() != null) {
-                  error = result.getFailureThrowable().getClass().getName();
-                } else {
-                  error = "export_failed";
-                }
               }
-              spanProcessorInstrumentation.finishSpans(1, error);
             });
       } catch (RuntimeException e) {
         logger.log(Level.WARNING, "Exporter threw an Exception", e);

sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SpanProcessorInstrumentation.java

@@ -9,7 +9,6 @@
 import io.opentelemetry.sdk.common.InternalTelemetryVersion;
 import io.opentelemetry.sdk.common.internal.ComponentId;
 import java.util.function.Supplier;
-import javax.annotation.Nullable;
 
 /** Metrics exported by span processors. */
 interface SpanProcessorInstrumentation {
@@ -27,10 +26,13 @@ static SpanProcessorInstrumentation get(
   }
 
   /** Records metrics for spans dropped because a queue is full. */
-  void dropSpans(int count);
+  void dropSpansQueueFull(int count);
 
-  /** Record metrics for spans processed, possibly with an error. */
-  void finishSpans(int count, @Nullable String error);
+  /** Record metrics for spans dropped since processor is shutdown. */
+  void dropSpansAlreadyShutdown(int count);
+
+  /** Record metrics for spans processed successfully. */
+  void finishSpans(int count);
 
   /** Registers metrics for processor queue capacity and size. */
   void buildQueueMetricsOnce(long capacity, LongCallable getSize);