diff --git a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/SimpleLogRecordProcessor.java b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/SimpleLogRecordProcessor.java index 218018e2afd..6d81ba54ac2 100644 --- a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/SimpleLogRecordProcessor.java +++ b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/SimpleLogRecordProcessor.java @@ -16,6 +16,7 @@ import io.opentelemetry.sdk.logs.LogRecordProcessor; import io.opentelemetry.sdk.logs.ReadWriteLogRecord; import io.opentelemetry.sdk.logs.data.LogRecordData; +import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Set; @@ -142,7 +143,9 @@ public CompletableResultCode shutdown() { @Override public CompletableResultCode forceFlush() { - return CompletableResultCode.ofAll(pendingExports); + // Snapshot pendingExports: completed exports are removed concurrently, and ofAll requires a + // collection whose size matches the elements it iterates, or the result may never complete. + return CompletableResultCode.ofAll(new ArrayList<>(pendingExports)); } /** diff --git a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/SimpleLogRecordProcessorTest.java b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/SimpleLogRecordProcessorTest.java index bfbb14ce414..9c16af20eb6 100644 --- a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/SimpleLogRecordProcessorTest.java +++ b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/SimpleLogRecordProcessorTest.java @@ -11,6 +11,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyCollection; import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -23,7 +24,12 @@ import io.opentelemetry.sdk.logs.ReadWriteLogRecord; import io.opentelemetry.sdk.logs.data.LogRecordData; import io.opentelemetry.sdk.testing.logs.TestLogRecordData; +import java.util.ArrayList; import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -37,6 +43,8 @@ class SimpleLogRecordProcessorTest { private static final LogRecordData LOG_RECORD_DATA = TestLogRecordData.builder().build(); + private static final int CONCURRENT_ITERATIONS = 100; + private static final int CONCURRENT_EXPORTS = 16; @Mock private LogRecordExporter logRecordExporter; @Mock private ReadWriteLogRecord readWriteLogRecord; @@ -110,6 +118,89 @@ void forceFlush() { assertThat(flush.isSuccess()).isTrue(); } + @Test + void forceFlush_NoPendingExports() { + CompletableResultCode flush = logRecordProcessor.forceFlush(); + assertThat(flush.isDone()).isTrue(); + assertThat(flush.isSuccess()).isTrue(); + } + + @Test + void forceFlush_FailedExport() { + CompletableResultCode export1 = new CompletableResultCode(); + CompletableResultCode export2 = new CompletableResultCode(); + + when(logRecordExporter.export(any())).thenReturn(export1, export2); + + logRecordProcessor.onEmit(Context.current(), readWriteLogRecord); + logRecordProcessor.onEmit(Context.current(), readWriteLogRecord); + + CompletableResultCode flush = logRecordProcessor.forceFlush(); + RuntimeException error = new RuntimeException("export failed"); + export1.failExceptionally(error); + assertThat(flush.isDone()).isFalse(); + + export2.succeed(); + assertThat(flush.isDone()).isTrue(); + assertThat(flush.isSuccess()).isFalse(); + assertThat(flush.getFailureThrowable()).isSameAs(error); + } + + @Test + void forceFlush_RepeatedAfterExportsComplete() { + CompletableResultCode export = new CompletableResultCode(); + + when(logRecordExporter.export(any())).thenReturn(export); + + logRecordProcessor.onEmit(Context.current(), readWriteLogRecord); + + CompletableResultCode flush1 = logRecordProcessor.forceFlush(); + export.succeed(); + assertThat(flush1.isDone()).isTrue(); + assertThat(flush1.isSuccess()).isTrue(); + + for (int i = 0; i < 3; i++) { + CompletableResultCode flush = logRecordProcessor.forceFlush(); + assertThat(flush.isDone()).isTrue(); + assertThat(flush.isSuccess()).isTrue(); + } + assertThat(flush1.isSuccess()).isTrue(); + } + + // Regression test for https://github.com/open-telemetry/opentelemetry-java/issues/8995: exports + // completing while forceFlush() collects them must not leave the flush result incomplete. The + // exact interleaving cannot be forced without production test hooks, so race a bounded number of + // times. Once every export has completed, a correct flush result is expected to be done already, + // so the check does not depend on a timeout. + @Test + void forceFlush_ExportsCompleteConcurrently() throws InterruptedException { + for (int i = 0; i < CONCURRENT_ITERATIONS; i++) { + List exports = + emitAsyncExports(logRecordProcessor, CONCURRENT_EXPORTS); + + CompletableResultCode flush = callWhileCompleting(exports, logRecordProcessor::forceFlush); + + assertThat(exports).allMatch(CompletableResultCode::isDone); + assertThat(flush.isDone()).as("flush done in iteration %s", i).isTrue(); + assertThat(flush.isSuccess()).isTrue(); + } + } + + @Test + void shutdown_ExportsCompleteConcurrently() throws InterruptedException { + for (int i = 0; i < CONCURRENT_ITERATIONS; i++) { + LogRecordProcessor processor = SimpleLogRecordProcessor.create(logRecordExporter); + List exports = emitAsyncExports(processor, CONCURRENT_EXPORTS); + + CompletableResultCode shutdown = callWhileCompleting(exports, processor::shutdown); + + assertThat(exports).allMatch(CompletableResultCode::isDone); + assertThat(shutdown.isDone()).as("shutdown done in iteration %s", i).isTrue(); + assertThat(shutdown.isSuccess()).isTrue(); + } + verify(logRecordExporter, times(CONCURRENT_ITERATIONS)).shutdown(); + } + @Test void shutdown() { CompletableResultCode export1 = new CompletableResultCode(); @@ -149,4 +240,58 @@ void toString_Valid() { assertThat(logRecordProcessor.toString()) .isEqualTo("SimpleLogRecordProcessor{logRecordExporter=MockLogRecordExporter}"); } + + @SuppressWarnings("MockitoDoSetup") + private List emitAsyncExports(LogRecordProcessor processor, int count) { + List exports = new ArrayList<>(); + // doAnswer avoids invoking the previous answer while re-stubbing, which would record an extra + // result that was never exported. + doAnswer( + invocation -> { + CompletableResultCode export = new CompletableResultCode(); + exports.add(export); + return export; + }) + .when(logRecordExporter) + .export(any()); + for (int i = 0; i < count; i++) { + processor.onEmit(Context.current(), readWriteLogRecord); + } + return exports; + } + + /** Invokes {@code call} while another thread completes {@code exports}. */ + private static CompletableResultCode callWhileCompleting( + List exports, Supplier call) + throws InterruptedException { + CountDownLatch ready = new CountDownLatch(1); + CountDownLatch start = new CountDownLatch(1); + Thread completer = + new Thread( + () -> { + ready.countDown(); + try { + if (!start.await(10, TimeUnit.SECONDS)) { + return; + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + exports.forEach(CompletableResultCode::succeed); + }); + // Daemon, and every wait below is bounded, so a regression fails the test instead of hanging + // the build. The bounds are failure guards; on a passing run each wait is normally very short. + completer.setDaemon(true); + completer.start(); + try { + assertThat(ready.await(10, TimeUnit.SECONDS)).as("completer thread started").isTrue(); + start.countDown(); + return call.get(); + } finally { + start.countDown(); + completer.join(TimeUnit.SECONDS.toMillis(10)); + assertThat(completer.isAlive()).as("completer thread finished").isFalse(); + } + } } 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 c0aabdcd836..e9934f3ccab 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 @@ -17,6 +17,7 @@ import io.opentelemetry.sdk.trace.ReadableSpan; import io.opentelemetry.sdk.trace.SpanProcessor; import io.opentelemetry.sdk.trace.data.SpanData; +import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Set; @@ -162,7 +163,9 @@ public CompletableResultCode shutdown() { @Override public CompletableResultCode forceFlush() { - return CompletableResultCode.ofAll(pendingExports); + // Snapshot pendingExports: completed exports are removed concurrently, and ofAll requires a + // collection whose size matches the elements it iterates, or the result may never complete. + return CompletableResultCode.ofAll(new ArrayList<>(pendingExports)); } /** diff --git a/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/SimpleSpanProcessorTest.java b/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/SimpleSpanProcessorTest.java index fa8831d4d4c..037353a3887 100644 --- a/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/SimpleSpanProcessorTest.java +++ b/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/SimpleSpanProcessorTest.java @@ -11,6 +11,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyCollection; import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -36,8 +37,12 @@ import io.opentelemetry.sdk.trace.export.BatchSpanProcessorTest.WaitingSpanExporter; import io.opentelemetry.sdk.trace.samplers.Sampler; import io.opentelemetry.sdk.trace.samplers.SamplingResult; +import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -51,6 +56,8 @@ @MockitoSettings(strictness = Strictness.LENIENT) class SimpleSpanProcessorTest { private static final String SPAN_NAME = "MySpanName"; + private static final int CONCURRENT_ITERATIONS = 100; + private static final int CONCURRENT_EXPORTS = 16; @Mock private ReadableSpan readableSpan; @Mock private ReadWriteSpan readWriteSpan; @Mock private SpanExporter spanExporter; @@ -262,6 +269,96 @@ void forceFlush() { assertThat(flush.isSuccess()).isTrue(); } + @Test + void forceFlush_NoPendingExports() { + CompletableResultCode flush = simpleSampledSpansProcessor.forceFlush(); + assertThat(flush.isDone()).isTrue(); + assertThat(flush.isSuccess()).isTrue(); + } + + @Test + void forceFlush_FailedExport() { + CompletableResultCode export1 = new CompletableResultCode(); + CompletableResultCode export2 = new CompletableResultCode(); + + when(spanExporter.export(any())).thenReturn(export1, export2); + + when(readableSpan.getSpanContext()).thenReturn(SAMPLED_SPAN_CONTEXT); + when(readableSpan.toSpanData()).thenReturn(TestUtils.makeBasicSpan()); + + simpleSampledSpansProcessor.onEnd(readableSpan); + simpleSampledSpansProcessor.onEnd(readableSpan); + + CompletableResultCode flush = simpleSampledSpansProcessor.forceFlush(); + RuntimeException error = new RuntimeException("export failed"); + export1.failExceptionally(error); + assertThat(flush.isDone()).isFalse(); + + export2.succeed(); + assertThat(flush.isDone()).isTrue(); + assertThat(flush.isSuccess()).isFalse(); + assertThat(flush.getFailureThrowable()).isSameAs(error); + } + + @Test + void forceFlush_RepeatedAfterExportsComplete() { + CompletableResultCode export = new CompletableResultCode(); + + when(spanExporter.export(any())).thenReturn(export); + + when(readableSpan.getSpanContext()).thenReturn(SAMPLED_SPAN_CONTEXT); + when(readableSpan.toSpanData()).thenReturn(TestUtils.makeBasicSpan()); + + simpleSampledSpansProcessor.onEnd(readableSpan); + + CompletableResultCode flush1 = simpleSampledSpansProcessor.forceFlush(); + export.succeed(); + assertThat(flush1.isDone()).isTrue(); + assertThat(flush1.isSuccess()).isTrue(); + + for (int i = 0; i < 3; i++) { + CompletableResultCode flush = simpleSampledSpansProcessor.forceFlush(); + assertThat(flush.isDone()).isTrue(); + assertThat(flush.isSuccess()).isTrue(); + } + assertThat(flush1.isSuccess()).isTrue(); + } + + // Regression test for https://github.com/open-telemetry/opentelemetry-java/issues/8995: exports + // completing while forceFlush() collects them must not leave the flush result incomplete. The + // exact interleaving cannot be forced without production test hooks, so race a bounded number of + // times. Once every export has completed, a correct flush result is expected to be done already, + // so the check does not depend on a timeout. + @Test + void forceFlush_ExportsCompleteConcurrently() throws InterruptedException { + for (int i = 0; i < CONCURRENT_ITERATIONS; i++) { + List exports = + endAsyncExports(simpleSampledSpansProcessor, CONCURRENT_EXPORTS); + + CompletableResultCode flush = + callWhileCompleting(exports, simpleSampledSpansProcessor::forceFlush); + + assertThat(exports).allMatch(CompletableResultCode::isDone); + assertThat(flush.isDone()).as("flush done in iteration %s", i).isTrue(); + assertThat(flush.isSuccess()).isTrue(); + } + } + + @Test + void shutdown_ExportsCompleteConcurrently() throws InterruptedException { + for (int i = 0; i < CONCURRENT_ITERATIONS; i++) { + SpanProcessor processor = SimpleSpanProcessor.create(spanExporter); + List exports = endAsyncExports(processor, CONCURRENT_EXPORTS); + + CompletableResultCode shutdown = callWhileCompleting(exports, processor::shutdown); + + assertThat(exports).allMatch(CompletableResultCode::isDone); + assertThat(shutdown.isDone()).as("shutdown done in iteration %s", i).isTrue(); + assertThat(shutdown.isSuccess()).isTrue(); + } + verify(spanExporter, times(CONCURRENT_ITERATIONS)).shutdown(); + } + @Test void shutdown() { CompletableResultCode export1 = new CompletableResultCode(); @@ -302,4 +399,60 @@ void getSpanExporter() { assertThat(((SimpleSpanProcessor) SimpleSpanProcessor.create(spanExporter)).getSpanExporter()) .isSameAs(spanExporter); } + + @SuppressWarnings("MockitoDoSetup") + private List endAsyncExports(SpanProcessor processor, int count) { + List exports = new ArrayList<>(); + // doAnswer avoids invoking the previous answer while re-stubbing, which would record an extra + // result that was never exported. + doAnswer( + invocation -> { + CompletableResultCode export = new CompletableResultCode(); + exports.add(export); + return export; + }) + .when(spanExporter) + .export(any()); + when(readableSpan.getSpanContext()).thenReturn(SAMPLED_SPAN_CONTEXT); + when(readableSpan.toSpanData()).thenReturn(TestUtils.makeBasicSpan()); + for (int i = 0; i < count; i++) { + processor.onEnd(readableSpan); + } + return exports; + } + + /** Invokes {@code call} while another thread completes {@code exports}. */ + private static CompletableResultCode callWhileCompleting( + List exports, Supplier call) + throws InterruptedException { + CountDownLatch ready = new CountDownLatch(1); + CountDownLatch start = new CountDownLatch(1); + Thread completer = + new Thread( + () -> { + ready.countDown(); + try { + if (!start.await(10, TimeUnit.SECONDS)) { + return; + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + exports.forEach(CompletableResultCode::succeed); + }); + // Daemon, and every wait below is bounded, so a regression fails the test instead of hanging + // the build. The bounds are failure guards; on a passing run each wait is normally very short. + completer.setDaemon(true); + completer.start(); + try { + assertThat(ready.await(10, TimeUnit.SECONDS)).as("completer thread started").isTrue(); + start.countDown(); + return call.get(); + } finally { + start.countDown(); + completer.join(TimeUnit.SECONDS.toMillis(10)); + assertThat(completer.isAlive()).as("completer thread finished").isFalse(); + } + } }