Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/OpenTelemetry/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,9 @@ Notes](../../RELEASENOTES.md).

## Unreleased

* Fix resource leak in batch exporting processors for Blazor/WASM.
Comment thread
martincostello marked this conversation as resolved.
Outdated
([#7069](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7069))

## 1.15.2

Released 2026-Apr-08
Expand Down
10 changes: 5 additions & 5 deletions src/OpenTelemetry/Internal/BatchExportTaskWorker.cs
Original file line number Diff line number Diff line change
Expand Up @@ -225,14 +225,14 @@ private async Task ExporterProcAsync()
{
try
{
using var delayCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
using var waitAndDelayCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
await Task.WhenAny(
this.exportTrigger.WaitAsync(cancellationToken),
Task.Delay(this.ScheduledDelayMilliseconds, delayCts.Token)).ConfigureAwait(false);
this.exportTrigger.WaitAsync(waitAndDelayCts.Token),
Task.Delay(this.ScheduledDelayMilliseconds, waitAndDelayCts.Token)).ConfigureAwait(false);
#if NET8_0_OR_GREATER
await delayCts.CancelAsync().ConfigureAwait(false);
await waitAndDelayCts.CancelAsync().ConfigureAwait(false);
#else
delayCts.Cancel();
waitAndDelayCts.Cancel();
#endif
}
catch (OperationCanceledException)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -120,19 +120,20 @@ private async Task ExporterProcAsync()
{
var timeout = (int)(this.ExportIntervalMilliseconds - (sw.ElapsedMilliseconds % this.ExportIntervalMilliseconds));

var exportTriggerTask = this.exportTrigger.WaitAsync(cancellationToken);
Task? exportTriggerTask = null;
Task? triggeredTask = null;

try
{
using var delayCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
using var waitAndDelayCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
exportTriggerTask = this.exportTrigger.WaitAsync(waitAndDelayCts.Token);
triggeredTask = await Task.WhenAny(
exportTriggerTask,
Task.Delay(timeout, delayCts.Token)).ConfigureAwait(false);
Task.Delay(timeout, waitAndDelayCts.Token)).ConfigureAwait(false);
#if NET8_0_OR_GREATER
await delayCts.CancelAsync().ConfigureAwait(false);
await waitAndDelayCts.CancelAsync().ConfigureAwait(false);
#else
delayCts.Cancel();
waitAndDelayCts.Cancel();
#endif
}
catch (OperationCanceledException)
Expand All @@ -148,7 +149,7 @@ private async Task ExporterProcAsync()
}

// Check if the trigger was signaled by trying to acquire it with a timeout of 0
var exportWasTriggered = triggeredTask == exportTriggerTask;
var exportWasTriggered = triggeredTask != null && triggeredTask == exportTriggerTask;

if (exportWasTriggered)
{
Expand Down
126 changes: 126 additions & 0 deletions test/OpenTelemetry.Tests/Internal/TaskWorkerTests.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

using System.Diagnostics;
using OpenTelemetry.Metrics;
using Xunit;

namespace OpenTelemetry.Internal.Tests;

public class TaskWorkerTests
{
private const int BaselineCollectCount = 2;
private const int CircularBufferCapacity = 8;
private const int IdleCyclesBeforeTrigger = 2;
private const int MaxExportBatchSize = 4;
private const int PostBaselinePauseMilliseconds = 20;
private const int TriggerCompletionTimeoutMilliseconds = 100;
private const int WaitUntilPollingIntervalMilliseconds = 10;
private const int WorkerDelayMilliseconds = 250;
private const int WorkerTimeoutMilliseconds = 1000;
Comment thread
martincostello marked this conversation as resolved.

[Fact]
public async Task BatchExportTaskWorker_TriggerExportAfterIdleCycles_DoesNotWaitForScheduledDelay()
{
// Arrange
var circularBuffer = new CircularBuffer<Activity>(capacity: CircularBufferCapacity);
using var exporter = new TestActivityExporter();
using var worker = new BatchExportTaskWorker<Activity>(
circularBuffer,
exporter,
maxExportBatchSize: MaxExportBatchSize,
scheduledDelayMilliseconds: WorkerDelayMilliseconds,
exporterTimeoutMilliseconds: WorkerTimeoutMilliseconds);

worker.Start();

await Task.Delay(GetIdleWaitDuration());

using var activity = new Activity("test");

Assert.True(circularBuffer.Add(activity));
Assert.True(worker.TriggerExport());

// Act
await WaitUntilAsync(() => exporter.ExportCount >= 1, TriggerCompletionTimeoutMilliseconds);

// Assert
Assert.True(worker.Shutdown(WorkerTimeoutMilliseconds));
}

[Fact]
public async Task PeriodicExportingMetricReaderTaskWorker_TriggerExportAfterIdleCycles_DoesNotWaitForExportInterval()
{
// Arrange
using var reader = new TestMetricReader();
using var worker = new PeriodicExportingMetricReaderTaskWorker(
reader,
exportIntervalMilliseconds: WorkerDelayMilliseconds,
exportTimeoutMilliseconds: WorkerTimeoutMilliseconds);

worker.Start();

await WaitUntilAsync(() => reader.CollectCount >= BaselineCollectCount, WorkerTimeoutMilliseconds);

var baselineCollectCount = reader.CollectCount;

await Task.Delay(PostBaselinePauseMilliseconds);

Assert.True(worker.TriggerExport());

// Act
await WaitUntilAsync(() => reader.CollectCount >= baselineCollectCount + 1, TriggerCompletionTimeoutMilliseconds);

// Assert
Assert.True(worker.Shutdown(WorkerTimeoutMilliseconds));
}

private static async Task WaitUntilAsync(Func<bool> condition, int timeoutMilliseconds)
{
var sw = Stopwatch.StartNew();

while (sw.ElapsedMilliseconds < timeoutMilliseconds)
{
if (condition())
{
return;
}

await Task.Delay(WaitUntilPollingIntervalMilliseconds);
}

Assert.True(condition());
}

private static int GetIdleWaitDuration()
=> (IdleCyclesBeforeTrigger * WorkerDelayMilliseconds) + (WorkerDelayMilliseconds / 5);

private sealed class TestActivityExporter : BaseExporter<Activity>
{
public int ExportCount { get; private set; }

public override ExportResult Export(in Batch<Activity> batch)
{
this.ExportCount++;
return ExportResult.Success;
}
Comment thread
martincostello marked this conversation as resolved.
Outdated
}

#pragma warning disable CA2000 // BaseExportingMetricReader owns the exporter lifecycle
private sealed class TestMetricReader() : BaseExportingMetricReader(new TestMetricExporter())
#pragma warning restore CA2000
{
public int CollectCount { get; private set; }

protected override bool OnCollect(int timeoutMilliseconds)
{
this.CollectCount++;
Comment thread
martincostello marked this conversation as resolved.
Outdated
return true;
}
}

private sealed class TestMetricExporter : BaseExporter<Metric>
{
public override ExportResult Export(in Batch<Metric> batch) => ExportResult.Success;
}
}