diff --git a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessor.java b/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessor.java index 2fe3a538950..a471a53901f 100644 --- a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessor.java +++ b/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 signal; private final AtomicReference flushRequested = new AtomicReference<>(); + private final AtomicBoolean isShutdown = new AtomicBoolean(false); private volatile boolean continueWork = true; private final ArrayList 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(); } } diff --git a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/LegacySpanProcessorInstrumentation.java b/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/LegacySpanProcessorInstrumentation.java index a13d53b0254..b8e81b6c250 100644 --- a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/LegacySpanProcessorInstrumentation.java +++ b/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 diff --git a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SemConvSpanProcessorInstrumentation.java b/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SemConvSpanProcessorInstrumentation.java index fdb308eeac7..87fbee7c1e6 100644 --- a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SemConvSpanProcessorInstrumentation.java +++ b/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; 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,7 +43,7 @@ final class SemConvSpanProcessorInstrumentation implements SpanProcessorInstrume componentId.getTypeName(), SemConvAttributes.OTEL_COMPONENT_NAME, componentId.getComponentName()); - droppedAttrs = + queueFullAttrs = Attributes.of( SemConvAttributes.OTEL_COMPONENT_TYPE, componentId.getTypeName(), @@ -50,23 +51,29 @@ final class SemConvSpanProcessorInstrumentation implements SpanProcessorInstrume 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 diff --git a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SimpleSpanProcessor.java b/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SimpleSpanProcessor.java index f369f872b41..aca04da2ff9 100644 --- a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SimpleSpanProcessor.java +++ b/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SimpleSpanProcessor.java @@ -99,11 +99,19 @@ 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 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); } @@ -111,16 +119,9 @@ public void onEnd(ReadableSpan span) { 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); diff --git a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SpanProcessorInstrumentation.java b/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SpanProcessorInstrumentation.java index ec730148e26..b2ed1ae9245 100644 --- a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SpanProcessorInstrumentation.java +++ b/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); diff --git a/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/SdkTracerProviderMetricsTest.java b/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/SdkTracerProviderMetricsTest.java index 1b3888bbd18..ffd93e3996d 100644 --- a/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/SdkTracerProviderMetricsTest.java +++ b/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/SdkTracerProviderMetricsTest.java @@ -58,10 +58,11 @@ void simple() { SdkMeterProvider.builder().registerMetricReader(metricReader).build(); InMemorySpanExporter exporter = InMemorySpanExporter.create(); - TracerProvider tracerProvider = + SimpleSpanProcessor processor = + SimpleSpanProcessor.builder(exporter).setMeterProvider(() -> meterProvider).build(); + SdkTracerProvider tracerProvider = SdkTracerProvider.builder() - .addSpanProcessor( - SimpleSpanProcessor.builder(exporter).setMeterProvider(() -> meterProvider).build()) + .addSpanProcessor(processor) .setMeterProvider(() -> meterProvider) .setSampler(sampler) .build(); @@ -706,6 +707,99 @@ void simple() { .hasAttributes( Attributes.of( OTEL_SPAN_SAMPLING_RESULT, "RECORD_ONLY"))))); + + // Spans rejected after the call to shutdown, regardless of completion so no join. + processor.shutdown(); + tracer.spanBuilder("span").startSpan().end(); + + assertThat(metricReader.collectAllMetrics()) + .satisfiesExactlyInAnyOrder( + m -> + assertThat(m) + .hasName("otel.sdk.processor.span.processed") + .hasLongSumSatisfying( + s -> + s.hasPointsSatisfying( + p -> + p.hasValue(2) + .hasAttributes( + Attributes.of( + OTEL_COMPONENT_NAME, + "simple_span_processor/0", + OTEL_COMPONENT_TYPE, + "simple_span_processor")), + p -> + p.hasValue(1) + .hasAttributes( + Attributes.of( + OTEL_COMPONENT_NAME, + "simple_span_processor/0", + OTEL_COMPONENT_TYPE, + "simple_span_processor", + ERROR_TYPE, + "already_shutdown")))), + m -> + assertThat(m) + .hasName("otel.sdk.span.started") + .hasLongSumSatisfying( + s -> + s.hasPointsSatisfying( + p -> + p.hasValue(2) + .hasAttributes( + Attributes.of( + OTEL_SPAN_PARENT_ORIGIN, + "none", + OTEL_SPAN_SAMPLING_RESULT, + "RECORD_AND_SAMPLE")), + p -> + p.hasValue(1) + .hasAttributes( + Attributes.of( + OTEL_SPAN_PARENT_ORIGIN, + "remote", + OTEL_SPAN_SAMPLING_RESULT, + "RECORD_AND_SAMPLE")), + p -> + p.hasValue(1) + .hasAttributes( + Attributes.of( + OTEL_SPAN_PARENT_ORIGIN, + "none", + OTEL_SPAN_SAMPLING_RESULT, + "RECORD_ONLY")), + p -> + p.hasValue(1) + .hasAttributes( + Attributes.of( + OTEL_SPAN_PARENT_ORIGIN, + "none", + OTEL_SPAN_SAMPLING_RESULT, + "DROP")), + p -> + p.hasValue(1) + .hasAttributes( + Attributes.of( + OTEL_SPAN_PARENT_ORIGIN, + "local", + OTEL_SPAN_SAMPLING_RESULT, + "DROP")))), + m -> + assertThat(m) + .hasName("otel.sdk.span.live") + .hasLongSumSatisfying( + s -> + s.hasPointsSatisfying( + p -> + p.hasValue(0) + .hasAttributes( + Attributes.of( + OTEL_SPAN_SAMPLING_RESULT, "RECORD_AND_SAMPLE")), + p -> + p.hasValue(0) + .hasAttributes( + Attributes.of( + OTEL_SPAN_SAMPLING_RESULT, "RECORD_ONLY"))))); } @Test @@ -745,6 +839,7 @@ void batch() throws Exception { // Queue is full, this span is dropped. tracer.spanBuilder("span").startSpan().end(); + // Export hasn't finished but processed metric is incremented assertThat(metricReader.collectAllMetrics()) .satisfiesExactlyInAnyOrder( m -> @@ -781,6 +876,14 @@ void batch() throws Exception { .hasLongSumSatisfying( s -> s.hasPointsSatisfying( + p -> + p.hasValue(1) + .hasAttributes( + Attributes.of( + OTEL_COMPONENT_NAME, + "batching_span_processor/0", + OTEL_COMPONENT_TYPE, + "batching_span_processor")), p -> p.hasValue(1) .hasAttributes( @@ -854,11 +957,96 @@ void batch() throws Exception { m -> assertThat(m) .hasName("otel.sdk.processor.span.processed") + .hasLongSumSatisfying( + s -> + s.hasPointsSatisfying( + p -> + p.hasValue(2) + .hasAttributes( + Attributes.of( + OTEL_COMPONENT_NAME, + "batching_span_processor/0", + OTEL_COMPONENT_TYPE, + "batching_span_processor")), + p -> + p.hasValue(1) + .hasAttributes( + Attributes.of( + OTEL_COMPONENT_NAME, + "batching_span_processor/0", + OTEL_COMPONENT_TYPE, + "batching_span_processor", + ERROR_TYPE, + "queue_full")))), + m -> + assertThat(m) + .hasName("otel.sdk.span.started") + .hasLongSumSatisfying( + s -> + s.hasPointsSatisfying( + p -> + p.hasValue(3) + .hasAttributes( + Attributes.of( + OTEL_SPAN_PARENT_ORIGIN, + "none", + OTEL_SPAN_SAMPLING_RESULT, + "RECORD_AND_SAMPLE")))), + m -> + assertThat(m) + .hasName("otel.sdk.span.live") + .hasLongSumSatisfying( + s -> + s.hasPointsSatisfying( + p -> + p.hasValue(0) + .hasAttributes( + Attributes.of( + OTEL_SPAN_SAMPLING_RESULT, "RECORD_AND_SAMPLE"))))); + + lenient().when(mockExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess()); + // Spans rejected after the call to shutdown, regardless of completion so no join. + processor.shutdown(); + + tracer.spanBuilder("span").startSpan().end(); + assertThat(metricReader.collectAllMetrics()) + .satisfiesExactlyInAnyOrder( + m -> + assertThat(m) + .hasName("otel.sdk.processor.span.queue.capacity") .hasLongSumSatisfying( s -> s.hasPointsSatisfying( p -> p.hasValue(1) + .hasAttributes( + Attributes.of( + OTEL_COMPONENT_NAME, + "batching_span_processor/0", + OTEL_COMPONENT_TYPE, + "batching_span_processor")))), + m -> + assertThat(m) + .hasName("otel.sdk.processor.span.queue.size") + .hasLongSumSatisfying( + s -> + s.hasPointsSatisfying( + p -> + p.hasValue(0) + .hasAttributes( + Attributes.of( + OTEL_COMPONENT_NAME, + "batching_span_processor/0", + OTEL_COMPONENT_TYPE, + "batching_span_processor")))), + m -> + assertThat(m) + .hasName("otel.sdk.processor.span.processed") + .hasLongSumSatisfying( + s -> + s.hasPointsSatisfying( + p -> + p.hasValue(2) .hasAttributes( Attributes.of( OTEL_COMPONENT_NAME, @@ -874,7 +1062,7 @@ void batch() throws Exception { OTEL_COMPONENT_TYPE, "batching_span_processor", ERROR_TYPE, - "export_failed")), + "already_shutdown")), p -> p.hasValue(1) .hasAttributes( @@ -892,7 +1080,7 @@ void batch() throws Exception { s -> s.hasPointsSatisfying( p -> - p.hasValue(3) + p.hasValue(4) .hasAttributes( Attributes.of( OTEL_SPAN_PARENT_ORIGIN, @@ -910,9 +1098,6 @@ void batch() throws Exception { .hasAttributes( Attributes.of( OTEL_SPAN_SAMPLING_RESULT, "RECORD_AND_SAMPLE"))))); - - lenient().when(mockExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess()); - processor.shutdown(); } @Test @@ -952,9 +1137,7 @@ void simpleExportError() { OTEL_COMPONENT_NAME, "simple_span_processor/0", OTEL_COMPONENT_TYPE, - "simple_span_processor", - ERROR_TYPE, - "export_failed")))), + "simple_span_processor")))), m -> assertThat(m) .hasName("otel.sdk.span.started")