diff --git a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java index 13ce175e6b7..c996336697b 100644 --- a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java +++ b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java @@ -49,6 +49,7 @@ public final class BatchLogRecordProcessor implements LogRecordProcessor { BatchLogRecordProcessor.class.getSimpleName() + "_WorkerThread"; private final Worker worker; + private final Thread workerThread; /** * Returns a new Builder for {@link BatchLogRecordProcessor}. @@ -79,8 +80,8 @@ public static BatchLogRecordProcessorBuilder builder(LogRecordExporter logRecord exporterTimeoutNanos, new ArrayBlockingQueue<>(maxQueueSize), maxQueueSize); // TODO: use JcTools.newFixedSizeQueue(..) - Thread workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker); - workerThread.start(); + this.workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker); + this.workerThread.start(); } @Override @@ -115,6 +116,11 @@ List getBatch() { return worker.batch; } + // Visible for testing + Thread getWorkerThread() { + return workerThread; + } + @Override public String toString() { return "BatchLogRecordProcessor{" @@ -199,51 +205,80 @@ private void addLog(ReadWriteLogRecord logData) { @Override public void run() { - updateNextExportTime(); + boolean normalExit = false; + try { + updateNextExportTime(); - while (continueWork) { - if (flushRequested.get() != null) { - flush(); - } - while (!queue.isEmpty() && batch.size() < maxExportBatchSize) { - batch.add(queue.poll().toLogRecordData()); + while (!Thread.currentThread().isInterrupted() && continueWork) { + while (!Thread.currentThread().isInterrupted() + && continueWork + && flushRequested.get() != null) { + flush(); + } + if (!continueWork || Thread.currentThread().isInterrupted()) { + break; + } + while (!queue.isEmpty() && batch.size() < maxExportBatchSize) { + batch.add(queue.poll().toLogRecordData()); + } + if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) { + exportCurrentBatch(); + updateNextExportTime(); + } + if (queue.isEmpty()) { + try { + long pollWaitTime = nextExportTime - System.nanoTime(); + if (pollWaitTime > 0) { + logsNeeded.set(maxExportBatchSize - batch.size()); + signal.poll(pollWaitTime, TimeUnit.NANOSECONDS); + logsNeeded.set(Integer.MAX_VALUE); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } } - if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) { - exportCurrentBatch(); - updateNextExportTime(); + if (!continueWork && !Thread.currentThread().isInterrupted()) { + normalExit = true; } - if (queue.isEmpty()) { - try { - long pollWaitTime = nextExportTime - System.nanoTime(); - if (pollWaitTime > 0) { - logsNeeded.set(maxExportBatchSize - batch.size()); - signal.poll(pollWaitTime, TimeUnit.NANOSECONDS); - logsNeeded.set(Integer.MAX_VALUE); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; + } finally { + continueWork = false; + CompletableResultCode sentinel = + normalExit ? CompletableResultCode.ofSuccess() : CompletableResultCode.ofFailure(); + CompletableResultCode flushResult = flushRequested.getAndSet(sentinel); + if (flushResult != null) { + if (normalExit) { + flushResult.succeed(); + } else { + flushResult.fail(); } } } } private void flush() { - int logsToFlush = queue.size(); - while (logsToFlush > 0) { - ReadWriteLogRecord logRecord = queue.poll(); - assert logRecord != null; - batch.add(logRecord.toLogRecordData()); - logsToFlush--; - if (batch.size() >= maxExportBatchSize) { - exportCurrentBatch(); + CompletableResultCode flushResult = flushRequested.getAndSet(null); + try { + int logsToFlush = queue.size(); + while (logsToFlush > 0) { + ReadWriteLogRecord logRecord = queue.poll(); + assert logRecord != null; + batch.add(logRecord.toLogRecordData()); + logsToFlush--; + if (batch.size() >= maxExportBatchSize) { + exportCurrentBatch(); + } } - } - exportCurrentBatch(); - CompletableResultCode flushResult = flushRequested.get(); - if (flushResult != null) { - flushResult.succeed(); - flushRequested.set(null); + exportCurrentBatch(); + if (flushResult != null) { + flushResult.succeed(); + } + } catch (Throwable t) { + if (flushResult != null) { + flushResult.fail(); + } + throw t; } } @@ -274,16 +309,22 @@ private CompletableResultCode shutdown() { } private CompletableResultCode forceFlush() { - CompletableResultCode flushResult = new CompletableResultCode(); - // we set the atomic here to trigger the worker loop to do a flush of the entire queue. - if (flushRequested.compareAndSet(null, flushResult)) { - signal.offer(true); + while (true) { + CompletableResultCode current = flushRequested.get(); + if (current != null) { + return current; + } + if (!continueWork) { + return isShutdown.get() + ? CompletableResultCode.ofSuccess() + : CompletableResultCode.ofFailure(); + } + CompletableResultCode flushResult = new CompletableResultCode(); + if (flushRequested.compareAndSet(null, flushResult)) { + signal.offer(true); + return flushResult; + } } - CompletableResultCode possibleResult = flushRequested.get(); - // there's a race here where the flush happening in the worker loop could complete before we - // get what's in the atomic. In that case, just return success, since we know it succeeded in - // the interim. - return possibleResult == null ? CompletableResultCode.ofSuccess() : possibleResult; } private void exportCurrentBatch() { diff --git a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java index 1b86ba63dee..a77077666c8 100644 --- a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java +++ b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java @@ -14,14 +14,17 @@ import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import io.opentelemetry.api.internal.GuardedBy; +import io.opentelemetry.context.Context; import io.opentelemetry.internal.testing.slf4j.SuppressLogger; import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.logs.ReadWriteLogRecord; import io.opentelemetry.sdk.logs.SdkLoggerProvider; import io.opentelemetry.sdk.logs.data.LogRecordData; import java.time.Duration; @@ -29,10 +32,12 @@ import java.util.Arrays; import java.util.Collection; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import javax.annotation.Nullable; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -511,6 +516,410 @@ void toString_Valid() { + "}"); } + @Test + void shutdown_whileForceFlushInFlight_exportsLogsEmittedBeforeShutdown() { + CompletableLogRecordExporter exporter = new CompletableLogRecordExporter(); + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + CompletableResultCode firstFlush = processor.forceFlush(); + + // Wait until the worker enters exportCurrentBatch() for LOG_MESSAGE_1 during flush(). + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .satisfiesExactly(log -> assertThat(log).hasBody(LOG_MESSAGE_1))); + + // Log 2 is emitted while the first forceFlush() is still in flight, before shutdown() is + // called. + emitLog(sdkLoggerProvider, LOG_MESSAGE_2); + + // Trigger shutdown while the first forceFlush() is in flight. + CompletableResultCode shutdownResult = processor.shutdown(); + + // Unblock the in-flight export. + exporter.succeed(); + + firstFlush.join(10, TimeUnit.SECONDS); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + assertThat(shutdownResult.isSuccess()).isTrue(); + assertThat(exporter.getExported()) + .satisfiesExactly( + log -> assertThat(log).hasBody(LOG_MESSAGE_1), + log -> assertThat(log).hasBody(LOG_MESSAGE_2)); + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_whileForceFlushInFlight_queuesBehindIt() { + CompletableLogRecordExporter exporter = new CompletableLogRecordExporter(); + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + CompletableResultCode firstFlush = processor.forceFlush(); + + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .satisfiesExactly(log -> assertThat(log).hasBody(LOG_MESSAGE_1))); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_2); + CompletableResultCode secondFlush = processor.forceFlush(); + CompletableResultCode thirdFlush = processor.forceFlush(); + + // Complete only the first export; secondFlush and thirdFlush must wait for the second export. + exporter.completeExport(0); + firstFlush.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .satisfiesExactly( + log -> assertThat(log).hasBody(LOG_MESSAGE_1), + log -> assertThat(log).hasBody(LOG_MESSAGE_2))); + assertThat(secondFlush.isDone()).isFalse(); + assertThat(thirdFlush.isDone()).isFalse(); + + exporter.succeed(); + secondFlush.join(10, TimeUnit.SECONDS); + thirdFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isSuccess()).isTrue(); + assertThat(thirdFlush.isSuccess()).isTrue(); + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_fromWhenCompleteCallback_exportsNewlyQueuedLogs() { + CompletableLogRecordExporter exporter = new CompletableLogRecordExporter(); + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + CompletableResultCode firstFlush = processor.forceFlush(); + AtomicReference secondFlushRef = new AtomicReference<>(); + + firstFlush.whenComplete(() -> { + emitLog(sdkLoggerProvider, LOG_MESSAGE_2); + secondFlushRef.set(processor.forceFlush()); + }); + + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .satisfiesExactly(log -> assertThat(log).hasBody(LOG_MESSAGE_1))); + + exporter.succeed(); + + firstFlush.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + CompletableResultCode secondFlush = secondFlushRef.get(); + assertThat(secondFlush).isNotNull(); + secondFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isSuccess()).isTrue(); + assertThat(exporter.getExported()) + .satisfiesExactly( + log -> assertThat(log).hasBody(LOG_MESSAGE_1), + log -> assertThat(log).hasBody(LOG_MESSAGE_2)); + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_afterShutdown_completes() { + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter).build(); + CompletableResultCode shutdownResult = processor.shutdown(); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(shutdownResult.isSuccess()).isTrue(); + + CompletableResultCode secondShutdownResult = processor.shutdown(); + secondShutdownResult.join(1, TimeUnit.SECONDS); + assertThat(secondShutdownResult.isDone()).isTrue(); + assertThat(secondShutdownResult.isSuccess()).isTrue(); + + CompletableResultCode flushAfterShutdown = processor.forceFlush(); + flushAfterShutdown.join(1, TimeUnit.SECONDS); + assertThat(flushAfterShutdown.isDone()).isTrue(); + assertThat(flushAfterShutdown.isSuccess()).isTrue(); + } + + @Test + void forceFlush_whenWorkerInterrupted_failsPendingAndSubsequentFlushes() throws Exception { + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + await().until(() -> processor.getWorkerThread().getState() == Thread.State.TIMED_WAITING); + processor.getWorkerThread().interrupt(); + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + CompletableResultCode flushAfterInterruption = processor.forceFlush(); + flushAfterInterruption.join(1, TimeUnit.SECONDS); + assertThat(flushAfterInterruption.isDone()).isTrue(); + assertThat(flushAfterInterruption.isSuccess()).isFalse(); + + CompletableResultCode shutdownResult = processor.shutdown(); + shutdownResult.join(1, TimeUnit.SECONDS); + assertThat(shutdownResult.isDone()).isTrue(); + assertThat(shutdownResult.isSuccess()).isFalse(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + @SuppressLogger(BatchLogRecordProcessor.class) + void forceFlush_whenExporterThrows_completesFlushSuccessfully() { + when(mockLogRecordExporter.export(anyList())).thenThrow(new RuntimeException("export failed")); + + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + + CompletableResultCode flushResult = processor.forceFlush(); + flushResult.join(10, TimeUnit.SECONDS); + + assertThat(flushResult.isDone()).isTrue(); + assertThat(flushResult.isSuccess()).isTrue(); + assertThat(processor.getWorkerThread().isAlive()).isTrue(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_whenFlushThrows_failsPendingFlushAndTerminatesWorker() throws Exception { + ReadWriteLogRecord mockRecord = mock(ReadWriteLogRecord.class); + when(mockRecord.toLogRecordData()).thenThrow(new RuntimeException("toLogRecordData failed")); + + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + processor.onEmit(Context.root(), mockRecord); + + CompletableResultCode flushResult = processor.forceFlush(); + flushResult.join(10, TimeUnit.SECONDS); + + assertThat(flushResult.isDone()).isTrue(); + assertThat(flushResult.isSuccess()).isFalse(); + + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + CompletableResultCode subsequentFlush = processor.forceFlush(); + assertThat(subsequentFlush.isDone()).isTrue(); + assertThat(subsequentFlush.isSuccess()).isFalse(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_whileShutdownExportInFlight_completesPendingFlushOnNormalExit() { + CompletableLogRecordExporter exporter = new CompletableLogRecordExporter(); + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + CompletableResultCode shutdownResult = processor.shutdown(); + + // Wait until the worker enters exportCurrentBatch() for log1 during shutdown flush. + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .satisfiesExactly(log -> assertThat(log).hasBody(LOG_MESSAGE_1))); + + // Queue a trailing forceFlush() while the shutdown export is in flight + CompletableResultCode trailingFlush = processor.forceFlush(); + assertThat(trailingFlush.isDone()).isFalse(); + + // Unblock the export + exporter.succeed(); + + shutdownResult.join(10, TimeUnit.SECONDS); + trailingFlush.join(10, TimeUnit.SECONDS); + + assertThat(shutdownResult.isSuccess()).isTrue(); + assertThat(trailingFlush.isSuccess()).isTrue(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_whileWorkerInterruptedWithPendingFlush_failsPendingFlush() throws Exception { + CountDownLatch exportStarted = new CountDownLatch(1); + CountDownLatch exportBlocker = new CountDownLatch(1); + when(mockLogRecordExporter.export(anyList())) + .thenAnswer(invocation -> { + exportStarted.countDown(); + try { + exportBlocker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return CompletableResultCode.ofFailure(); + }); + + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + processor.forceFlush(); + + exportStarted.await(); + + // Queue a second flush while worker is inside export + CompletableResultCode secondFlush = processor.forceFlush(); + assertThat(secondFlush.isDone()).isFalse(); + + // Interrupt worker thread while second flush is pending + processor.getWorkerThread().interrupt(); + exportBlocker.countDown(); + + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + secondFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isDone()).isTrue(); + assertThat(secondFlush.isSuccess()).isFalse(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_duringExporterShutdown_returnsSuccess() throws Exception { + CountDownLatch shutdownStarted = new CountDownLatch(1); + CountDownLatch shutdownBlocker = new CountDownLatch(1); + when(mockLogRecordExporter.shutdown()) + .thenAnswer(invocation -> { + shutdownStarted.countDown(); + try { + shutdownBlocker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return CompletableResultCode.ofSuccess(); + }); + + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + CompletableResultCode shutdownResult = processor.shutdown(); + + // Wait until exporter.shutdown() is in flight + shutdownStarted.await(); + + // While exporter.shutdown() is in flight, continueWork is false and flushRequested is null + CompletableResultCode flushDuringShutdown = processor.forceFlush(); + assertThat(flushDuringShutdown.isDone()).isTrue(); + assertThat(flushDuringShutdown.isSuccess()).isTrue(); + + shutdownBlocker.countDown(); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(shutdownResult.isSuccess()).isTrue(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_withMultipleBatches_exportsAllBatches() { + WaitingLogRecordExporter exporter = + new WaitingLogRecordExporter(3, CompletableResultCode.ofSuccess()); + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(exporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .setMaxExportBatchSize(2) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + emitLog(sdkLoggerProvider, LOG_MESSAGE_2); + emitLog(sdkLoggerProvider, "log 3"); + + CompletableResultCode flushResult = processor.forceFlush(); + flushResult.join(10, TimeUnit.SECONDS); + + assertThat(flushResult.isSuccess()).isTrue(); + assertThat(exporter.waitForExport()).hasSize(3); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_concurrentCalls_allSucceed() throws Exception { + when(mockLogRecordExporter.export(anyList())).thenReturn(CompletableResultCode.ofSuccess()); + + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + + int threadCount = 8; + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(threadCount); + List results = new CopyOnWriteArrayList<>(); + + for (int i = 0; i < threadCount; i++) { + new Thread( + () -> { + try { + startLatch.await(); + results.add(processor.forceFlush()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + doneLatch.countDown(); + } + }) + .start(); + } + + startLatch.countDown(); + doneLatch.await(5, TimeUnit.SECONDS); + + assertThat(results).hasSize(threadCount); + for (CompletableResultCode result : results) { + result.join(10, TimeUnit.SECONDS); + assertThat(result.isSuccess()).isTrue(); + } + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + private static final class BlockingLogRecordExporter implements LogRecordExporter { final Object monitor = new Object(); @@ -574,9 +983,11 @@ private void unblock() { private static class CompletableLogRecordExporter implements LogRecordExporter { - private final List results = new ArrayList<>(); + private final Object lock = new Object(); + + private final List results = new CopyOnWriteArrayList<>(); - private final List exported = new ArrayList<>(); + private final List exported = new CopyOnWriteArrayList<>(); private volatile boolean succeeded; @@ -584,20 +995,31 @@ List getExported() { return exported; } + void completeExport(int index) { + results.get(index).succeed(); + } + void succeed() { - succeeded = true; - results.forEach(CompletableResultCode::succeed); + List toComplete; + synchronized (lock) { + succeeded = true; + toComplete = new ArrayList<>(results); + } + toComplete.forEach(CompletableResultCode::succeed); } @Override public CompletableResultCode export(Collection logs) { - exported.addAll(logs); - if (succeeded) { - return CompletableResultCode.ofSuccess(); + synchronized (lock) { + if (succeeded) { + exported.addAll(logs); + return CompletableResultCode.ofSuccess(); + } + CompletableResultCode result = new CompletableResultCode(); + results.add(result); + exported.addAll(logs); + return result; } - CompletableResultCode result = new CompletableResultCode(); - results.add(result); - return result; } @Override 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 422c9a576c1..0aa041e1bf0 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,6 +54,7 @@ public final class BatchSpanProcessor implements SpanProcessor { private final boolean exportUnsampledSpans; private final Worker worker; + private final Thread workerThread; /** * Returns a new {@link BatchSpanProcessor} with default configuration which batches spans @@ -103,8 +104,8 @@ public static BatchSpanProcessorBuilder builder(SpanExporter spanExporter) { exporterTimeoutNanos, JcTools.newFixedSizeQueue(maxQueueSize), maxQueueSize); - Thread workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker); - workerThread.start(); + this.workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker); + this.workerThread.start(); } @Override @@ -156,6 +157,11 @@ Queue getQueue() { return worker.queue; } + // Visible for testing + Thread getWorkerThread() { + return workerThread; + } + @Override public String toString() { return "BatchSpanProcessor{" @@ -243,29 +249,52 @@ private void addSpan(ReadableSpan span) { @Override public void run() { - updateNextExportTime(); + boolean normalExit = false; + try { + updateNextExportTime(); - while (continueWork) { - if (flushRequested.get() != null) { - flush(); - } - drain(maxExportBatchSize - batch.size()); + while (!Thread.currentThread().isInterrupted() && continueWork) { + while (!Thread.currentThread().isInterrupted() + && continueWork + && flushRequested.get() != null) { + flush(); + } + if (!continueWork || Thread.currentThread().isInterrupted()) { + break; + } + drain(maxExportBatchSize - batch.size()); - if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) { - exportCurrentBatch(); - updateNextExportTime(); - } - if (queue.isEmpty()) { - try { - long pollWaitTime = nextExportTime - System.nanoTime(); - if (pollWaitTime > 0) { - spansNeeded.set(maxExportBatchSize - batch.size()); - signal.poll(pollWaitTime, TimeUnit.NANOSECONDS); - spansNeeded.set(Integer.MAX_VALUE); + if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) { + exportCurrentBatch(); + updateNextExportTime(); + } + if (queue.isEmpty()) { + try { + long pollWaitTime = nextExportTime - System.nanoTime(); + if (pollWaitTime > 0) { + spansNeeded.set(maxExportBatchSize - batch.size()); + signal.poll(pollWaitTime, TimeUnit.NANOSECONDS); + spansNeeded.set(Integer.MAX_VALUE); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; + } + } + if (!continueWork && !Thread.currentThread().isInterrupted()) { + normalExit = true; + } + } finally { + continueWork = false; + CompletableResultCode sentinel = + normalExit ? CompletableResultCode.ofSuccess() : CompletableResultCode.ofFailure(); + CompletableResultCode flushResult = flushRequested.getAndSet(sentinel); + if (flushResult != null) { + if (normalExit) { + flushResult.succeed(); + } else { + flushResult.fail(); } } } @@ -278,19 +307,25 @@ private int drain(int limit) { } private void flush() { - int spansToFlush = queueSize.get(); - while (spansToFlush > 0) { - int drained = drain(maxExportBatchSize - batch.size()); - spansToFlush -= drained; - if (batch.size() >= maxExportBatchSize) { - exportCurrentBatch(); + CompletableResultCode flushResult = flushRequested.getAndSet(null); + try { + int spansToFlush = queueSize.get(); + while (spansToFlush > 0) { + int drained = drain(maxExportBatchSize - batch.size()); + spansToFlush -= drained; + if (batch.size() >= maxExportBatchSize) { + exportCurrentBatch(); + } } - } - exportCurrentBatch(); - CompletableResultCode flushResult = flushRequested.get(); - if (flushResult != null) { - flushResult.succeed(); - flushRequested.set(null); + exportCurrentBatch(); + if (flushResult != null) { + flushResult.succeed(); + } + } catch (Throwable t) { + if (flushResult != null) { + flushResult.fail(); + } + throw t; } } @@ -321,16 +356,22 @@ private CompletableResultCode shutdown() { } private CompletableResultCode forceFlush() { - CompletableResultCode flushResult = new CompletableResultCode(); - // we set the atomic here to trigger the worker loop to do a flush of the entire queue. - if (flushRequested.compareAndSet(null, flushResult)) { - signal.offer(true); + while (true) { + CompletableResultCode current = flushRequested.get(); + if (current != null) { + return current; + } + if (!continueWork) { + return isShutdown.get() + ? CompletableResultCode.ofSuccess() + : CompletableResultCode.ofFailure(); + } + CompletableResultCode flushResult = new CompletableResultCode(); + if (flushRequested.compareAndSet(null, flushResult)) { + signal.offer(true); + return flushResult; + } } - CompletableResultCode possibleResult = flushRequested.get(); - // there's a race here where the flush happening in the worker loop could complete before we - // get what's in the atomic. In that case, just return success, since we know it succeeded in - // the interim. - return possibleResult == null ? CompletableResultCode.ofSuccess() : possibleResult; } private void exportCurrentBatch() { diff --git a/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java b/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java index 2f1b3d005e4..fab9bd4ade9 100644 --- a/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java +++ b/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java @@ -13,6 +13,7 @@ import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -21,6 +22,9 @@ import io.github.netmikey.logunit.api.LogCapturer; import io.opentelemetry.api.internal.GuardedBy; import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.internal.testing.slf4j.SuppressLogger; import io.opentelemetry.sdk.common.CompletableResultCode; @@ -34,9 +38,11 @@ import java.util.Arrays; import java.util.Collection; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import javax.annotation.Nullable; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -730,6 +736,340 @@ void exporterThrowsNonRuntimeException() { await().untilAsserted(() -> assertThat(batchSpanProcessor.getQueue()).isEmpty()); } + @Test + void shutdown_whileForceFlushInFlight_exportsSpansEndedBeforeShutdown() { + CompletableSpanExporter exporter = new CompletableSpanExporter(); + BatchSpanProcessor processor = + BatchSpanProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + ReadableSpan span1 = createEndedSpan(SPAN_NAME_1); + CompletableResultCode firstFlush = processor.forceFlush(); + + // Wait until the worker enters exportCurrentBatch() for span1 during flush(). + await() + .untilAsserted( + () -> assertThat(exporter.getExported()).containsExactly(span1.toSpanData())); + + // Span 2 ends while the first forceFlush() is still in flight, before shutdown() is called. + ReadableSpan span2 = createEndedSpan(SPAN_NAME_2); + assertThat(processor.getQueue()).hasSize(1); + + // Trigger shutdown while the first forceFlush() is in flight. + CompletableResultCode shutdownResult = processor.shutdown(); + + // Unblock the in-flight export. + exporter.succeed(); + + firstFlush.join(10, TimeUnit.SECONDS); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + assertThat(shutdownResult.isSuccess()).isTrue(); + assertThat(exporter.getExported()).containsExactly(span1.toSpanData(), span2.toSpanData()); + } + + @Test + void forceFlush_whileForceFlushInFlight_queuesBehindIt() { + CompletableSpanExporter exporter = new CompletableSpanExporter(); + BatchSpanProcessor processor = + BatchSpanProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + ReadableSpan span1 = createEndedSpan(SPAN_NAME_1); + CompletableResultCode firstFlush = processor.forceFlush(); + + await() + .untilAsserted( + () -> assertThat(exporter.getExported()).containsExactly(span1.toSpanData())); + + ReadableSpan span2 = createEndedSpan(SPAN_NAME_2); + CompletableResultCode secondFlush = processor.forceFlush(); + CompletableResultCode thirdFlush = processor.forceFlush(); + + // Complete only the first export; secondFlush and thirdFlush must wait for the second export. + exporter.completeExport(0); + firstFlush.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .containsExactly(span1.toSpanData(), span2.toSpanData())); + assertThat(secondFlush.isDone()).isFalse(); + assertThat(thirdFlush.isDone()).isFalse(); + + exporter.succeed(); + secondFlush.join(10, TimeUnit.SECONDS); + thirdFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isSuccess()).isTrue(); + assertThat(thirdFlush.isSuccess()).isTrue(); + } + + @Test + void forceFlush_fromWhenCompleteCallback_exportsNewlyQueuedSpans() { + CompletableSpanExporter exporter = new CompletableSpanExporter(); + BatchSpanProcessor processor = + BatchSpanProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + ReadableSpan span1 = createEndedSpan(SPAN_NAME_1); + CompletableResultCode firstFlush = processor.forceFlush(); + AtomicReference span2Ref = new AtomicReference<>(); + AtomicReference secondFlushRef = new AtomicReference<>(); + + firstFlush.whenComplete(() -> { + span2Ref.set(createEndedSpan(SPAN_NAME_2)); + secondFlushRef.set(processor.forceFlush()); + }); + + await() + .untilAsserted( + () -> assertThat(exporter.getExported()).containsExactly(span1.toSpanData())); + + exporter.succeed(); + + firstFlush.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + CompletableResultCode secondFlush = secondFlushRef.get(); + assertThat(secondFlush).isNotNull(); + secondFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isSuccess()).isTrue(); + assertThat(exporter.getExported()) + .containsExactly(span1.toSpanData(), span2Ref.get().toSpanData()); + } + + @Test + void forceFlush_afterShutdown_completes() { + BatchSpanProcessor processor = BatchSpanProcessor.builder(mockSpanExporter).build(); + CompletableResultCode shutdownResult = processor.shutdown(); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(shutdownResult.isSuccess()).isTrue(); + + CompletableResultCode secondShutdownResult = processor.shutdown(); + secondShutdownResult.join(1, TimeUnit.SECONDS); + assertThat(secondShutdownResult.isDone()).isTrue(); + assertThat(secondShutdownResult.isSuccess()).isTrue(); + + CompletableResultCode flushAfterShutdown = processor.forceFlush(); + flushAfterShutdown.join(1, TimeUnit.SECONDS); + assertThat(flushAfterShutdown.isDone()).isTrue(); + assertThat(flushAfterShutdown.isSuccess()).isTrue(); + } + + @Test + void forceFlush_whenWorkerInterrupted_failsPendingAndSubsequentFlushes() throws Exception { + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + await().until(() -> processor.getWorkerThread().getState() == Thread.State.TIMED_WAITING); + processor.getWorkerThread().interrupt(); + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + CompletableResultCode flushAfterInterruption = processor.forceFlush(); + flushAfterInterruption.join(1, TimeUnit.SECONDS); + assertThat(flushAfterInterruption.isDone()).isTrue(); + assertThat(flushAfterInterruption.isSuccess()).isFalse(); + + CompletableResultCode shutdownResult = processor.shutdown(); + shutdownResult.join(1, TimeUnit.SECONDS); + assertThat(shutdownResult.isDone()).isTrue(); + assertThat(shutdownResult.isSuccess()).isFalse(); + } + + @Test + @SuppressLogger(BatchSpanProcessor.class) + void forceFlush_whenExporterThrows_completesFlushSuccessfully() { + when(mockSpanExporter.export(anyList())).thenThrow(new RuntimeException("export failed")); + + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + createEndedSpan(SPAN_NAME_1); + + CompletableResultCode flushResult = processor.forceFlush(); + flushResult.join(10, TimeUnit.SECONDS); + + assertThat(flushResult.isDone()).isTrue(); + assertThat(flushResult.isSuccess()).isTrue(); + assertThat(processor.getWorkerThread().isAlive()).isTrue(); + } + + @Test + void forceFlush_whenFlushThrows_failsPendingFlushAndTerminatesWorker() throws Exception { + ReadableSpan mockSpan = mock(ReadableSpan.class); + SpanContext sampledContext = + SpanContext.create( + "00000000000000000000000000000001", + "0000000000000001", + TraceFlags.getSampled(), + TraceState.getDefault()); + when(mockSpan.getSpanContext()).thenReturn(sampledContext); + when(mockSpan.toSpanData()).thenThrow(new RuntimeException("toSpanData failed")); + + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + processor.onEnd(mockSpan); + + CompletableResultCode flushResult = processor.forceFlush(); + flushResult.join(10, TimeUnit.SECONDS); + + assertThat(flushResult.isDone()).isTrue(); + assertThat(flushResult.isSuccess()).isFalse(); + + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + CompletableResultCode subsequentFlush = processor.forceFlush(); + assertThat(subsequentFlush.isDone()).isTrue(); + assertThat(subsequentFlush.isSuccess()).isFalse(); + } + + @Test + void forceFlush_whileShutdownExportInFlight_completesPendingFlushOnNormalExit() { + CompletableSpanExporter exporter = new CompletableSpanExporter(); + BatchSpanProcessor processor = + BatchSpanProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + ReadableSpan span1 = createEndedSpan(SPAN_NAME_1); + CompletableResultCode shutdownResult = processor.shutdown(); + + // Wait until the worker enters exportCurrentBatch() for span1 during shutdown flush. + await() + .untilAsserted( + () -> assertThat(exporter.getExported()).containsExactly(span1.toSpanData())); + + // Queue a trailing forceFlush() while the shutdown export is in flight + CompletableResultCode trailingFlush = processor.forceFlush(); + assertThat(trailingFlush.isDone()).isFalse(); + + // Unblock the export + exporter.succeed(); + + shutdownResult.join(10, TimeUnit.SECONDS); + trailingFlush.join(10, TimeUnit.SECONDS); + + assertThat(shutdownResult.isSuccess()).isTrue(); + assertThat(trailingFlush.isSuccess()).isTrue(); + } + + @Test + void forceFlush_whileWorkerInterruptedWithPendingFlush_failsPendingFlush() throws Exception { + CountDownLatch exportStarted = new CountDownLatch(1); + CountDownLatch exportBlocker = new CountDownLatch(1); + when(mockSpanExporter.export(anyList())) + .thenAnswer(invocation -> { + exportStarted.countDown(); + try { + exportBlocker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return CompletableResultCode.ofFailure(); + }); + + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + createEndedSpan(SPAN_NAME_1); + processor.forceFlush(); + + exportStarted.await(); + + // Queue a second flush while worker is inside export + CompletableResultCode secondFlush = processor.forceFlush(); + assertThat(secondFlush.isDone()).isFalse(); + + // Interrupt worker thread while second flush is pending + processor.getWorkerThread().interrupt(); + exportBlocker.countDown(); + + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + secondFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isDone()).isTrue(); + assertThat(secondFlush.isSuccess()).isFalse(); + } + + @Test + void forceFlush_duringExporterShutdown_returnsSuccess() throws Exception { + CountDownLatch shutdownStarted = new CountDownLatch(1); + CountDownLatch shutdownBlocker = new CountDownLatch(1); + when(mockSpanExporter.shutdown()) + .thenAnswer(invocation -> { + shutdownStarted.countDown(); + try { + shutdownBlocker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return CompletableResultCode.ofSuccess(); + }); + + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + CompletableResultCode shutdownResult = processor.shutdown(); + + // Wait until exporter.shutdown() is in flight + shutdownStarted.await(); + + // While exporter.shutdown() is in flight, continueWork is false and flushRequested is null + CompletableResultCode flushDuringShutdown = processor.forceFlush(); + assertThat(flushDuringShutdown.isDone()).isTrue(); + assertThat(flushDuringShutdown.isSuccess()).isTrue(); + + shutdownBlocker.countDown(); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(shutdownResult.isSuccess()).isTrue(); + } + + @Test + void forceFlush_concurrentCalls_allSucceed() throws Exception { + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + createEndedSpan(SPAN_NAME_1); + + int threadCount = 8; + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(threadCount); + List results = new CopyOnWriteArrayList<>(); + + for (int i = 0; i < threadCount; i++) { + new Thread( + () -> { + try { + startLatch.await(); + results.add(processor.forceFlush()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + doneLatch.countDown(); + } + }) + .start(); + } + + startLatch.countDown(); + doneLatch.await(5, TimeUnit.SECONDS); + + assertThat(results).hasSize(threadCount); + for (CompletableResultCode result : results) { + result.join(10, TimeUnit.SECONDS); + assertThat(result.isSuccess()).isTrue(); + } + } + private static final class BlockingSpanExporter implements SpanExporter { final Object monitor = new Object(); @@ -793,9 +1133,11 @@ private void unblock() { private static class CompletableSpanExporter implements SpanExporter { - private final List results = new ArrayList<>(); + private final Object lock = new Object(); - private final List exported = new ArrayList<>(); + private final List results = new CopyOnWriteArrayList<>(); + + private final List exported = new CopyOnWriteArrayList<>(); private volatile boolean succeeded; @@ -803,20 +1145,31 @@ List getExported() { return exported; } + void completeExport(int index) { + results.get(index).succeed(); + } + void succeed() { - succeeded = true; - results.forEach(CompletableResultCode::succeed); + List toComplete; + synchronized (lock) { + succeeded = true; + toComplete = new ArrayList<>(results); + } + toComplete.forEach(CompletableResultCode::succeed); } @Override public CompletableResultCode export(Collection spans) { - exported.addAll(spans); - if (succeeded) { - return CompletableResultCode.ofSuccess(); + synchronized (lock) { + if (succeeded) { + exported.addAll(spans); + return CompletableResultCode.ofSuccess(); + } + CompletableResultCode result = new CompletableResultCode(); + results.add(result); + exported.addAll(spans); + return result; } - CompletableResultCode result = new CompletableResultCode(); - results.add(result); - return result; } @Override