diff --git a/docs/apidiffs/current_vs_latest/opentelemetry-sdk-logs.txt b/docs/apidiffs/current_vs_latest/opentelemetry-sdk-logs.txt index 403941132b2..4a013cfcd23 100644 --- a/docs/apidiffs/current_vs_latest/opentelemetry-sdk-logs.txt +++ b/docs/apidiffs/current_vs_latest/opentelemetry-sdk-logs.txt @@ -1,2 +1,4 @@ Comparing source compatibility of opentelemetry-sdk-logs-1.68.0-SNAPSHOT.jar against opentelemetry-sdk-logs-1.67.0.jar -No changes. \ No newline at end of file +*** MODIFIED INTERFACE: PUBLIC ABSTRACT io.opentelemetry.sdk.logs.export.LogRecordExporter (not serializable) + === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 + +++ NEW METHOD: PUBLIC(+) int getConcurrencyLimit() diff --git a/docs/apidiffs/current_vs_latest/opentelemetry-sdk-metrics.txt b/docs/apidiffs/current_vs_latest/opentelemetry-sdk-metrics.txt index c215c35f65f..f4bdfe261d8 100644 --- a/docs/apidiffs/current_vs_latest/opentelemetry-sdk-metrics.txt +++ b/docs/apidiffs/current_vs_latest/opentelemetry-sdk-metrics.txt @@ -1,2 +1,4 @@ Comparing source compatibility of opentelemetry-sdk-metrics-1.68.0-SNAPSHOT.jar against opentelemetry-sdk-metrics-1.67.0.jar -No changes. \ No newline at end of file +*** MODIFIED INTERFACE: PUBLIC ABSTRACT io.opentelemetry.sdk.metrics.export.MetricExporter (not serializable) + === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 + +++ NEW METHOD: PUBLIC(+) int getConcurrencyLimit() diff --git a/docs/apidiffs/current_vs_latest/opentelemetry-sdk-trace.txt b/docs/apidiffs/current_vs_latest/opentelemetry-sdk-trace.txt index a3a4e8b6251..62986cd6edf 100644 --- a/docs/apidiffs/current_vs_latest/opentelemetry-sdk-trace.txt +++ b/docs/apidiffs/current_vs_latest/opentelemetry-sdk-trace.txt @@ -1,2 +1,4 @@ Comparing source compatibility of opentelemetry-sdk-trace-1.68.0-SNAPSHOT.jar against opentelemetry-sdk-trace-1.67.0.jar -No changes. \ No newline at end of file +*** MODIFIED INTERFACE: PUBLIC ABSTRACT io.opentelemetry.sdk.trace.export.SpanExporter (not serializable) + === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 + +++ NEW METHOD: PUBLIC(+) int getConcurrencyLimit() diff --git a/sdk-extensions/autoconfigure/src/test/java/io/opentelemetry/sdk/autoconfigure/TracerProviderConfigurationTest.java b/sdk-extensions/autoconfigure/src/test/java/io/opentelemetry/sdk/autoconfigure/TracerProviderConfigurationTest.java index 450c98cacad..2a244def263 100644 --- a/sdk-extensions/autoconfigure/src/test/java/io/opentelemetry/sdk/autoconfigure/TracerProviderConfigurationTest.java +++ b/sdk-extensions/autoconfigure/src/test/java/io/opentelemetry/sdk/autoconfigure/TracerProviderConfigurationTest.java @@ -60,6 +60,7 @@ class TracerProviderConfigurationTest { @BeforeEach void setUp() { when(mockSpanExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess()); + when(mockSpanExporter.getConcurrencyLimit()).thenReturn(1); } @Test diff --git a/sdk/all/src/test/java/io/opentelemetry/sdk/OpenTelemetrySdkTest.java b/sdk/all/src/test/java/io/opentelemetry/sdk/OpenTelemetrySdkTest.java index e934ebe2802..e0ad7a74275 100644 --- a/sdk/all/src/test/java/io/opentelemetry/sdk/OpenTelemetrySdkTest.java +++ b/sdk/all/src/test/java/io/opentelemetry/sdk/OpenTelemetrySdkTest.java @@ -215,6 +215,7 @@ void getMeter() { @Test void meterBuilder() { + when(metricExporter.getConcurrencyLimit()).thenReturn(1); when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod(); OpenTelemetrySdk openTelemetry = OpenTelemetrySdk.builder() @@ -229,6 +230,7 @@ void meterBuilder() { @Test void meterBuilder_ViaProvider() { + when(metricExporter.getConcurrencyLimit()).thenReturn(1); when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod(); OpenTelemetrySdk openTelemetry = OpenTelemetrySdk.builder() @@ -257,6 +259,7 @@ void getMeterProvider() { // Demonstrates how clear or confusing is SDK configuration @Test void fullOpenTelemetrySdkConfigurationDemo() { + when(metricExporter.getConcurrencyLimit()).thenReturn(1); when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod(); OpenTelemetrySdk.builder() .setMeterProvider( @@ -294,6 +297,7 @@ void fullOpenTelemetrySdkConfigurationDemo() { // Demonstrates how clear or confusing is SDK configuration @Test void trivialOpenTelemetrySdkConfigurationDemo() { + when(metricExporter.getConcurrencyLimit()).thenReturn(1); when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod(); OpenTelemetrySdk.builder() .setMeterProvider( @@ -312,6 +316,7 @@ void trivialOpenTelemetrySdkConfigurationDemo() { // Demonstrates how clear or confusing is SDK configuration @Test void minimalOpenTelemetrySdkConfigurationDemo() { + when(metricExporter.getConcurrencyLimit()).thenReturn(1); when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod(); OpenTelemetrySdk.builder() .setMeterProvider( @@ -384,6 +389,7 @@ void shutdown() { void stringRepresentation() { SpanExporter spanExporter = mock(SpanExporter.class); when(spanExporter.toString()).thenReturn("MockSpanExporter{}"); + when(metricExporter.getConcurrencyLimit()).thenReturn(1); when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod(); when(metricExporter.toString()).thenReturn("MockMetricExporter{}"); LogRecordExporter logRecordExporter = mock(LogRecordExporter.class); 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 306ca41f66d..3b54070e648 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 @@ -19,8 +19,13 @@ import java.util.Collections; import java.util.List; import java.util.Queue; +import java.util.Set; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -28,6 +33,7 @@ import java.util.function.Supplier; import java.util.logging.Level; import java.util.logging.Logger; +import javax.annotation.Nullable; /** * Implementation of the {@link LogRecordProcessor} that batches logs exported by the SDK then @@ -159,6 +165,9 @@ private static final class Worker implements Runnable { private volatile boolean continueWork = true; private final ArrayList batch; private final long maxQueueSize; + @Nullable private final ExecutorService exportExecutor; + private final Set pendingExports = + Collections.newSetFromMap(new ConcurrentHashMap()); private Worker( LogRecordExporter logRecordExporter, @@ -180,6 +189,22 @@ private Worker( this.maxQueueSize = maxQueueSize; this.batch = new ArrayList<>(this.maxExportBatchSize); + int concurrencyLimit = logRecordExporter.getConcurrencyLimit(); + if (concurrencyLimit < 1) { + throw new IllegalArgumentException("log exporter concurrency limit must be positive"); + } + exportExecutor = + concurrencyLimit == 1 + ? null + : new ThreadPoolExecutor( + concurrencyLimit - 1, + concurrencyLimit - 1, + 0, + TimeUnit.NANOSECONDS, + new SynchronousQueue<>(), + new DaemonThreadFactory( + BatchLogRecordProcessor.class.getSimpleName() + "_Exporter"), + new ThreadPoolExecutor.CallerRunsPolicy()); } private void addLog(ReadWriteLogRecord logData) { @@ -240,6 +265,7 @@ private void flush() { } } exportCurrentBatch(); + waitForExports(); CompletableResultCode flushResult = flushRequested.get(); if (flushResult != null) { flushResult.succeed(); @@ -264,6 +290,9 @@ private CompletableResultCode shutdown() { CompletableResultCode shutdownResult = logRecordExporter.shutdown(); shutdownResult.whenComplete( () -> { + if (exportExecutor != null) { + exportExecutor.shutdownNow(); + } if (!flushResult.isSuccess() || !shutdownResult.isSuccess()) { result.fail(); } else { @@ -293,12 +322,47 @@ private void exportCurrentBatch() { return; } + if (exportExecutor != null) { + List exportBatch = new ArrayList<>(batch); + batch.clear(); + CompletableResultCode taskResult = new CompletableResultCode(); + pendingExports.add(taskResult); + try { + exportExecutor.execute( + () -> { + try { + exportBatch(exportBatch); + taskResult.succeed(); + } catch (Throwable t) { + ThrowableUtil.propagateIfFatal(t); + taskResult.fail(); + } finally { + pendingExports.remove(taskResult); + } + }); + } catch (Throwable t) { + pendingExports.remove(taskResult); + taskResult.fail(); + ThrowableUtil.propagateIfFatal(t); + logger.log(Level.WARNING, "Unable to submit export", t); + } + return; + } + + try { + exportBatch(batch); + } finally { + batch.clear(); + } + } + + private void exportBatch(List exportBatch) { try { // We always increment for every export invocation, so we increment before the export call // to make sure thrown errors don't affect it. - logProcessorInstrumentation.finishLogs(batch.size()); + logProcessorInstrumentation.finishLogs(exportBatch.size()); CompletableResultCode result = - logRecordExporter.export(Collections.unmodifiableList(batch)); + logRecordExporter.export(Collections.unmodifiableList(exportBatch)); result.join(exporterTimeoutNanos, TimeUnit.NANOSECONDS); if (!result.isSuccess()) { logger.log(Level.FINE, "Exporter failed"); @@ -306,8 +370,13 @@ private void exportCurrentBatch() { } catch (Throwable t) { ThrowableUtil.propagateIfFatal(t); logger.log(Level.WARNING, "Exporter threw an Exception", t); - } finally { - batch.clear(); + } + } + + private void waitForExports() { + while (!pendingExports.isEmpty()) { + CompletableResultCode.ofAll(new ArrayList<>(pendingExports)) + .join(exporterTimeoutNanos, TimeUnit.NANOSECONDS); } } } diff --git a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/LogRecordExporter.java b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/LogRecordExporter.java index cccdf326505..a077ffe6cd8 100644 --- a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/LogRecordExporter.java +++ b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/LogRecordExporter.java @@ -73,6 +73,20 @@ static LogRecordExporter noop() { */ CompletableResultCode export(Collection logs); + /** + * Returns the maximum number of export calls that this exporter can process concurrently. + * + *

The default value preserves the historical serialized behavior. Exporters that support + * concurrent calls may return a larger value and may block an export call when this capacity is + * exhausted to apply backpressure. + * + * @return the export concurrency limit, which must be positive + * @since 1.67.0 + */ + default int getConcurrencyLimit() { + return 1; + } + /** * Exports the collection of {@link LogRecordData} that have not yet been exported. * diff --git a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/SdkLoggerProviderMetricsTest.java b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/SdkLoggerProviderMetricsTest.java index e20b72a2cde..d7c731d0043 100644 --- a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/SdkLoggerProviderMetricsTest.java +++ b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/SdkLoggerProviderMetricsTest.java @@ -146,6 +146,7 @@ void batch() throws Exception { MeterProvider meterProvider = SdkMeterProvider.builder().registerMetricReader(metricReader).build(); + when(mockExporter.getConcurrencyLimit()).thenReturn(1); BatchLogRecordProcessor processor = BatchLogRecordProcessor.builder(mockExporter) .setMaxQueueSize(1) 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 579d6d2bc02..8f7c338ce6f 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 @@ -57,6 +57,7 @@ class BatchLogRecordProcessorTest { @BeforeEach void setUp() { when(mockLogRecordExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess()); + when(mockLogRecordExporter.getConcurrencyLimit()).thenReturn(1); } private void emitLog(SdkLoggerProvider sdkLoggerProvider, String message) { @@ -439,6 +440,7 @@ public void continuesIfExporterTimesOut() throws InterruptedException { CountDownLatch exportedAgain = new CountDownLatch(1); reset(mockLogRecordExporter); when(mockLogRecordExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess()); + when(mockLogRecordExporter.getConcurrencyLimit()).thenReturn(1); when(mockLogRecordExporter.export( argThat( logs -> { @@ -497,6 +499,46 @@ void shutdownPropagatesFailure() { assertThat(result.isSuccess()).isFalse(); } + @Test + @Timeout(5) + void exporterConcurrencyLimitAllowsConcurrentExports() throws InterruptedException { + ConcurrentLogRecordExporter exporter = new ConcurrentLogRecordExporter(); + SdkLoggerProvider loggerProvider = + SdkLoggerProvider.builder() + .addLogRecordProcessor( + BatchLogRecordProcessor.builder(exporter) + .setMaxExportBatchSize(1) + .setScheduleDelay(1, TimeUnit.HOURS) + .build()) + .build(); + + emitLog(loggerProvider, LOG_MESSAGE_1); + emitLog(loggerProvider, LOG_MESSAGE_2); + CompletableResultCode shutdown = loggerProvider.shutdown(); + + assertThat(exporter.entered.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(exporter.maxConcurrent.get()).isEqualTo(2); + exporter.release.countDown(); + shutdown.join(1, TimeUnit.SECONDS); + assertThat(shutdown.isSuccess()).isTrue(); + } + + @Test + void rejectsNegativeExporterConcurrencyLimit() { + when(mockLogRecordExporter.getConcurrencyLimit()).thenReturn(-1); + assertThatThrownBy(() -> BatchLogRecordProcessor.builder(mockLogRecordExporter).build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("log exporter concurrency limit must be positive"); + } + + @Test + void rejectsZeroExporterConcurrencyLimit() { + when(mockLogRecordExporter.getConcurrencyLimit()).thenReturn(0); + assertThatThrownBy(() -> BatchLogRecordProcessor.builder(mockLogRecordExporter).build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("log exporter concurrency limit must be positive"); + } + @Test void getLogRecordExporter() { assertThat( @@ -578,6 +620,43 @@ private void unblock() { } } + private static final class ConcurrentLogRecordExporter implements LogRecordExporter { + private final CountDownLatch entered = new CountDownLatch(2); + private final CountDownLatch release = new CountDownLatch(1); + private final AtomicInteger active = new AtomicInteger(); + private final AtomicInteger maxConcurrent = new AtomicInteger(); + + @Override + public int getConcurrencyLimit() { + return 2; + } + + @Override + public CompletableResultCode export(Collection logs) { + int current = active.incrementAndGet(); + maxConcurrent.accumulateAndGet(current, Math::max); + entered.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + active.decrementAndGet(); + } + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode flush() { + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode shutdown() { + return CompletableResultCode.ofSuccess(); + } + } + private static class CompletableLogRecordExporter implements LogRecordExporter { private final List results = new ArrayList<>(); diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/MetricExporter.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/MetricExporter.java index 685c5f09bf1..ad97ae10bef 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/MetricExporter.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/MetricExporter.java @@ -60,6 +60,20 @@ default MemoryMode getMemoryMode() { */ CompletableResultCode export(Collection metrics); + /** + * Returns the maximum number of export calls that this exporter can process concurrently. + * + *

The default value preserves the historical serialized behavior. Exporters that support + * concurrent calls may return a larger value and may block an export call when this capacity is + * exhausted to apply backpressure. + * + * @return the export concurrency limit, which must be positive + * @since 1.67.0 + */ + default int getConcurrencyLimit() { + return 1; + } + /** * A hint that any metrics previously {@link #export(Collection)}ed should be completed. * diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReader.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReader.java index 14e3751ba18..f3ea9887c17 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReader.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReader.java @@ -11,16 +11,22 @@ import io.opentelemetry.sdk.common.InternalTelemetryVersion; import io.opentelemetry.sdk.common.export.MemoryMode; import io.opentelemetry.sdk.common.internal.ComponentId; +import io.opentelemetry.sdk.common.internal.DaemonThreadFactory; import io.opentelemetry.sdk.metrics.Aggregation; import io.opentelemetry.sdk.metrics.InstrumentType; import io.opentelemetry.sdk.metrics.SdkMeterProvider; import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder; import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.metrics.data.MetricData; +import java.util.ArrayList; import java.util.Collection; import java.util.Iterator; +import java.util.List; +import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.logging.Level; @@ -55,6 +61,7 @@ public final class PeriodicMetricReader implements MetricReader { @Nullable private volatile ScheduledFuture scheduledFuture; private final int maxExportBatchSize; + @Nullable private final ExecutorService exportExecutor; /** * Returns a new {@link PeriodicMetricReader} which exports to the {@code exporter} once every @@ -81,6 +88,21 @@ public static PeriodicMetricReaderBuilder builder(MetricExporter exporter) { this.maxExportBatchSize = maxExportBatchSize; this.scheduled = new Scheduled(); this.internalTelemetryVersion = internalTelemetryVersion; + int concurrencyLimit = exporter.getConcurrencyLimit(); + if (concurrencyLimit < 1) { + throw new IllegalArgumentException("metric exporter concurrency limit must be positive"); + } + exportExecutor = + concurrencyLimit == 1 + ? null + : new ThreadPoolExecutor( + concurrencyLimit - 1, + concurrencyLimit - 1, + 0, + TimeUnit.NANOSECONDS, + new SynchronousQueue<>(), + new DaemonThreadFactory(PeriodicMetricReader.class.getSimpleName() + "_Exporter"), + new ThreadPoolExecutor.CallerRunsPolicy()); } @Override @@ -145,6 +167,9 @@ public CompletableResultCode shutdown() { // reset the interrupted status Thread.currentThread().interrupt(); } finally { + if (exportExecutor != null) { + exportExecutor.shutdownNow(); + } CompletableResultCode shutdownResult = scheduled.shutdown(); shutdownResult.whenComplete( () -> { @@ -217,6 +242,35 @@ private CompletableResultCode exportMetrics(Collection metricData) { } Collection> batches = MetricExportBatcher.batchMetrics(metricData, maxExportBatchSize); + if (exportExecutor != null) { + List results = new ArrayList<>(); + for (Collection batch : batches) { + CompletableResultCode batchResult = new CompletableResultCode(); + results.add(batchResult); + try { + exportExecutor.execute( + () -> { + try { + CompletableResultCode exportResult = exporter.export(batch); + exportResult.join(Long.MAX_VALUE, TimeUnit.NANOSECONDS); + if (exportResult.isSuccess()) { + batchResult.succeed(); + } else { + batchResult.fail(); + } + } catch (Throwable t) { + batchResult.fail(); + logger.log(Level.WARNING, "Exporter threw an Exception", t); + } + }); + } catch (Throwable t) { + batchResult.fail(); + logger.log(Level.WARNING, "Unable to submit export", t); + } + } + return CompletableResultCode.ofAll(results); + } + CompletableResultCode sequentialResult = new CompletableResultCode(); AtomicBoolean anyFailed = new AtomicBoolean(false); Iterator> batchIterator = batches.iterator(); diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReaderTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReaderTest.java index df00a7e14d4..38198ab6d8b 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReaderTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReaderTest.java @@ -44,6 +44,7 @@ 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; @@ -88,6 +89,7 @@ class PeriodicMetricReaderTest { void setup() { when(collectionRegistration.collectAllMetrics()) .thenReturn(Collections.singletonList(METRIC_DATA)); + when(metricExporter.getConcurrencyLimit()).thenReturn(1); } @Test @@ -295,6 +297,7 @@ void flush() throws Exception { @SuppressLogger(PeriodicMetricReader.class) void forceflush_callsFlush() { MetricExporter metricExporter = mock(MetricExporter.class); + when(metricExporter.getConcurrencyLimit()).thenReturn(1); when(metricExporter.export(any())) .thenReturn(CompletableResultCode.ofSuccess()) .thenReturn(CompletableResultCode.ofSuccess()) @@ -475,6 +478,7 @@ void invalidConfig() { @Test void periodicExport_SequentialBatches() throws Exception { MetricExporter mockExporter = mock(MetricExporter.class); + when(mockExporter.getConcurrencyLimit()).thenReturn(1); when(mockExporter.getAggregationTemporality(any())) .thenReturn(AggregationTemporality.CUMULATIVE); when(mockExporter.flush()).thenReturn(CompletableResultCode.ofSuccess()); @@ -518,6 +522,7 @@ void periodicExport_SequentialBatches() throws Exception { @SuppressLogger(PeriodicMetricReader.class) void periodicExport_SequentialBatches_PartialFailure() throws Exception { MetricExporter mockExporter = mock(MetricExporter.class); + when(mockExporter.getConcurrencyLimit()).thenReturn(1); when(mockExporter.getAggregationTemporality(any())) .thenReturn(AggregationTemporality.CUMULATIVE); when(mockExporter.flush()).thenReturn(CompletableResultCode.ofSuccess()); @@ -567,6 +572,7 @@ void periodicExport_SequentialBatches_PartialFailure() throws Exception { @Test void periodicExport_SequentialBatches_PurelySynchronous() throws Exception { MetricExporter mockExporter = mock(MetricExporter.class); + when(mockExporter.getConcurrencyLimit()).thenReturn(1); when(mockExporter.getAggregationTemporality(any())) .thenReturn(AggregationTemporality.CUMULATIVE); when(mockExporter.flush()).thenReturn(CompletableResultCode.ofSuccess()); @@ -600,6 +606,7 @@ void periodicExport_SequentialBatches_PurelySynchronous() throws Exception { @Test void periodicExport_SequentialBatches_PurelyAsynchronous() throws Exception { MetricExporter mockExporter = mock(MetricExporter.class); + when(mockExporter.getConcurrencyLimit()).thenReturn(1); when(mockExporter.getAggregationTemporality(any())) .thenReturn(AggregationTemporality.CUMULATIVE); when(mockExporter.flush()).thenReturn(CompletableResultCode.ofSuccess()); @@ -642,6 +649,84 @@ void periodicExport_SequentialBatches_PurelyAsynchronous() throws Exception { reader.shutdown(); } + @Test + @Timeout(5) + void exporterConcurrencyLimitAllowsConcurrentExports() throws Exception { + ConcurrentMetricExporter exporter = new ConcurrentMetricExporter(); + PeriodicMetricReader reader = + PeriodicMetricReader.builder(exporter) + .setInterval(Duration.ofSeconds(Integer.MAX_VALUE)) + .setMaxExportBatchSize(3) + .build(); + reader.register(collectionRegistration); + + AtomicReference flushResult = new AtomicReference<>(); + Thread flushThread = new Thread(() -> flushResult.set(reader.forceFlush())); + flushThread.start(); + assertThat(exporter.entered.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(exporter.maxConcurrent.get()).isEqualTo(2); + exporter.release.countDown(); + flushThread.join(1000); + assertThat(flushResult.get().join(1, TimeUnit.SECONDS).isSuccess()).isTrue(); + reader.shutdown(); + } + + @Test + @Timeout(5) + void exporterConcurrencyLimitWaitsForAsyncExports() throws Exception { + AsyncMetricExporter exporter = new AsyncMetricExporter(); + AtomicBoolean firstCollection = new AtomicBoolean(true); + when(collectionRegistration.collectAllMetrics()) + .thenAnswer( + invocation -> + firstCollection.getAndSet(false) + ? Collections.singletonList(METRIC_DATA) + : Collections.emptyList()); + PeriodicMetricReader reader = + PeriodicMetricReader.builder(exporter) + .setInterval(Duration.ofSeconds(Integer.MAX_VALUE)) + .setMaxExportBatchSize(6) + .build(); + reader.register(collectionRegistration); + + CompletableResultCode flushResult = reader.forceFlush(); + assertThat(exporter.entered.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(flushResult.isDone()).isFalse(); + exporter.results.get(0).succeed(); + assertThat(flushResult.join(1, TimeUnit.SECONDS).isSuccess()).isTrue(); + reader.shutdown(); + } + + @Test + void rejectsZeroExporterConcurrencyLimit() { + when(metricExporter.getConcurrencyLimit()).thenReturn(0); + assertThatThrownBy(() -> PeriodicMetricReader.builder(metricExporter).build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("metric exporter concurrency limit must be positive"); + } + + @Test + @SuppressLogger(PeriodicMetricReader.class) + void concurrentExporterFailureCompletesFlushAsFailure() throws Exception { + ConcurrentMetricExporter exporter = new ConcurrentMetricExporter(/* fail= */ true); + PeriodicMetricReader reader = + PeriodicMetricReader.builder(exporter) + .setInterval(Duration.ofSeconds(Integer.MAX_VALUE)) + .setMaxExportBatchSize(3) + .build(); + reader.register(collectionRegistration); + + AtomicReference flushResult = new AtomicReference<>(); + Thread flushThread = new Thread(() -> flushResult.set(reader.forceFlush())); + flushThread.start(); + assertThat(exporter.entered.await(1, TimeUnit.SECONDS)).isTrue(); + exporter.release.countDown(); + flushThread.join(1000); + assertThat(flushResult.get().join(1, TimeUnit.SECONDS).isSuccess()).isTrue(); + logCapturer.assertContains("Exporter failed"); + reader.shutdown(); + } + @Test @Timeout(10) @SuppressLogger(PeriodicMetricReader.class) @@ -700,6 +785,92 @@ public CompletableResultCode shutdown() { reader.shutdown(); } + private static final class ConcurrentMetricExporter implements MetricExporter { + private final CountDownLatch entered = new CountDownLatch(2); + private final CountDownLatch release = new CountDownLatch(1); + private final AtomicInteger active = new AtomicInteger(); + private final AtomicInteger maxConcurrent = new AtomicInteger(); + private final boolean fail; + + private ConcurrentMetricExporter() { + this(false); + } + + private ConcurrentMetricExporter(boolean fail) { + this.fail = fail; + } + + @Override + public int getConcurrencyLimit() { + return 2; + } + + @Override + public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) { + return AggregationTemporality.CUMULATIVE; + } + + @Override + public CompletableResultCode export(Collection metrics) { + int current = active.incrementAndGet(); + maxConcurrent.accumulateAndGet(current, Math::max); + entered.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + active.decrementAndGet(); + } + return fail ? CompletableResultCode.ofFailure() : CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode flush() { + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode shutdown() { + return CompletableResultCode.ofSuccess(); + } + } + + private static final class AsyncMetricExporter implements MetricExporter { + private final CountDownLatch entered = new CountDownLatch(1); + private final List results = new ArrayList<>(); + + @Override + public int getConcurrencyLimit() { + return 2; + } + + @Override + public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) { + return AggregationTemporality.CUMULATIVE; + } + + @Override + public CompletableResultCode export(Collection metrics) { + CompletableResultCode result = new CompletableResultCode(); + synchronized (results) { + results.add(result); + entered.countDown(); + } + return result; + } + + @Override + public CompletableResultCode flush() { + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode shutdown() { + return CompletableResultCode.ofSuccess(); + } + } + @Test void stringRepresentation() { when(metricExporter.toString()).thenReturn("MockMetricExporter{}"); 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 a471a53901f..c6750197255 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 @@ -23,8 +23,13 @@ import java.util.Collections; import java.util.List; import java.util.Queue; +import java.util.Set; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -32,6 +37,7 @@ import java.util.function.Supplier; import java.util.logging.Level; import java.util.logging.Logger; +import javax.annotation.Nullable; /** * Implementation of the {@link SpanProcessor} that batches spans exported by the SDK then pushes @@ -201,6 +207,9 @@ private static final class Worker implements Runnable { private final ArrayList batch; private final long maxQueueSize; private final AtomicInteger droppedSpanCount = new AtomicInteger(0); + @Nullable private final ExecutorService exportExecutor; + private final Set pendingExports = + Collections.newSetFromMap(new ConcurrentHashMap()); private Worker( SpanExporter spanExporter, @@ -223,6 +232,21 @@ private Worker( this.maxQueueSize = maxQueueSize; this.batch = new ArrayList<>(this.maxExportBatchSize); + int concurrencyLimit = spanExporter.getConcurrencyLimit(); + if (concurrencyLimit < 1) { + throw new IllegalArgumentException("span exporter concurrency limit must be positive"); + } + exportExecutor = + concurrencyLimit == 1 + ? null + : new ThreadPoolExecutor( + concurrencyLimit - 1, + concurrencyLimit - 1, + 0, + TimeUnit.NANOSECONDS, + new SynchronousQueue<>(), + new DaemonThreadFactory(BatchSpanProcessor.class.getSimpleName() + "_Exporter"), + new ThreadPoolExecutor.CallerRunsPolicy()); } private void addSpan(ReadableSpan span) { @@ -287,6 +311,7 @@ private void flush() { } } exportCurrentBatch(); + waitForExports(); CompletableResultCode flushResult = flushRequested.get(); if (flushResult != null) { flushResult.succeed(); @@ -311,6 +336,9 @@ private CompletableResultCode shutdown() { CompletableResultCode shutdownResult = spanExporter.shutdown(); shutdownResult.whenComplete( () -> { + if (exportExecutor != null) { + exportExecutor.shutdownNow(); + } if (!flushResult.isSuccess() || !shutdownResult.isSuccess()) { result.fail(); } else { @@ -352,11 +380,47 @@ private void exportCurrentBatch() { + ")"); } + if (exportExecutor != null) { + List exportBatch = new ArrayList<>(batch); + batch.clear(); + CompletableResultCode taskResult = new CompletableResultCode(); + pendingExports.add(taskResult); + try { + exportExecutor.execute( + () -> { + try { + exportBatch(exportBatch); + taskResult.succeed(); + } catch (Throwable t) { + ThrowableUtil.propagateIfFatal(t); + taskResult.fail(); + } finally { + pendingExports.remove(taskResult); + } + }); + } catch (Throwable t) { + pendingExports.remove(taskResult); + taskResult.fail(); + ThrowableUtil.propagateIfFatal(t); + logger.log(Level.WARNING, "Unable to submit export", t); + } + return; + } + + try { + exportBatch(batch); + } finally { + batch.clear(); + } + } + + private void exportBatch(List exportBatch) { 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)); + spanProcessorInstrumentation.finishSpans(exportBatch.size()); + CompletableResultCode result = + spanExporter.export(Collections.unmodifiableList(exportBatch)); result.join(exporterTimeoutNanos, TimeUnit.NANOSECONDS); if (!result.isSuccess()) { logger.log(Level.FINE, "Exporter failed"); @@ -364,8 +428,13 @@ private void exportCurrentBatch() { } catch (Throwable t) { ThrowableUtil.propagateIfFatal(t); logger.log(Level.WARNING, "Exporter threw an Exception", t); - } finally { - batch.clear(); + } + } + + private void waitForExports() { + while (!pendingExports.isEmpty()) { + CompletableResultCode.ofAll(new ArrayList<>(pendingExports)) + .join(exporterTimeoutNanos, TimeUnit.NANOSECONDS); } } } diff --git a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SpanExporter.java b/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SpanExporter.java index 1a1818bf6f2..d81892bf98f 100644 --- a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SpanExporter.java +++ b/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/SpanExporter.java @@ -74,6 +74,20 @@ static SpanExporter noop() { */ CompletableResultCode export(Collection spans); + /** + * Returns the maximum number of export calls that this exporter can process concurrently. + * + *

The default value preserves the historical behavior of {@link BatchSpanProcessor}, which + * serializes export calls. Exporters that support concurrent calls may return a larger value. An + * exporter may block an export call when this capacity is exhausted to apply backpressure. + * + * @return the export concurrency limit, which must be positive + * @since 1.67.0 + */ + default int getConcurrencyLimit() { + return 1; + } + /** * Exports the collection of sampled {@code Span}s that have not yet been exported. Note that * export operations can be performed simultaneously depending on the type of span processor being 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 504c54a4b19..515574d3361 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 @@ -810,6 +810,7 @@ void batch() throws Exception { MeterProvider meterProvider = SdkMeterProvider.builder().registerMetricReader(metricReader).build(); + when(mockExporter.getConcurrencyLimit()).thenReturn(1); BatchSpanProcessor processor = BatchSpanProcessor.builder(mockExporter) .setMaxQueueSize(1) 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 5e9bc38ff78..1d62edfe853 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 @@ -37,6 +37,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import javax.annotation.Nullable; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -72,6 +73,7 @@ class BatchSpanProcessorTest { void setUp() { when(mockSpanExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess()); when(mockSpanExporter.export(anyList())).thenReturn(CompletableResultCode.ofSuccess()); + when(mockSpanExporter.getConcurrencyLimit()).thenReturn(1); when(mockSampler.shutdown()).thenReturn(CompletableResultCode.ofSuccess()); } @@ -542,6 +544,7 @@ public void continuesIfExporterTimesOut() throws InterruptedException { CountDownLatch exportedAgain = new CountDownLatch(1); reset(mockSpanExporter); when(mockSpanExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess()); + when(mockSpanExporter.getConcurrencyLimit()).thenReturn(1); when(mockSpanExporter.export( argThat( spans -> { @@ -692,6 +695,46 @@ void shutdownPropagatesFailure() { assertThat(result.isSuccess()).isFalse(); } + @Test + @Timeout(5) + void exporterConcurrencyLimitAllowsConcurrentExports() throws InterruptedException { + ConcurrentSpanExporter exporter = new ConcurrentSpanExporter(); + sdkTracerProvider = + SdkTracerProvider.builder() + .addSpanProcessor( + BatchSpanProcessor.builder(exporter) + .setMaxExportBatchSize(1) + .setScheduleDelay(1, TimeUnit.HOURS) + .build()) + .build(); + + createEndedSpan(SPAN_NAME_1); + createEndedSpan(SPAN_NAME_2); + CompletableResultCode shutdown = sdkTracerProvider.shutdown(); + + assertThat(exporter.entered.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(exporter.maxConcurrent.get()).isEqualTo(2); + exporter.release.countDown(); + shutdown.join(1, TimeUnit.SECONDS); + assertThat(shutdown.isSuccess()).isTrue(); + } + + @Test + void rejectsNegativeExporterConcurrencyLimit() { + when(mockSpanExporter.getConcurrencyLimit()).thenReturn(-1); + assertThatThrownBy(() -> BatchSpanProcessor.builder(mockSpanExporter).build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("span exporter concurrency limit must be positive"); + } + + @Test + void rejectsZeroExporterConcurrencyLimit() { + when(mockSpanExporter.getConcurrencyLimit()).thenReturn(0); + assertThatThrownBy(() -> BatchSpanProcessor.builder(mockSpanExporter).build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("span exporter concurrency limit must be positive"); + } + @Test void getSpanExporter() { assertThat(BatchSpanProcessor.builder(mockSpanExporter).build().getSpanExporter()) @@ -797,6 +840,43 @@ private void unblock() { } } + private static final class ConcurrentSpanExporter implements SpanExporter { + private final CountDownLatch entered = new CountDownLatch(2); + private final CountDownLatch release = new CountDownLatch(1); + private final AtomicInteger active = new AtomicInteger(); + private final AtomicInteger maxConcurrent = new AtomicInteger(); + + @Override + public int getConcurrencyLimit() { + return 2; + } + + @Override + public CompletableResultCode export(Collection spans) { + int current = active.incrementAndGet(); + maxConcurrent.accumulateAndGet(current, Math::max); + entered.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + active.decrementAndGet(); + } + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode flush() { + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode shutdown() { + return CompletableResultCode.ofSuccess(); + } + } + private static class CompletableSpanExporter implements SpanExporter { private final List results = new ArrayList<>();