Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,14 @@ Notes](../../RELEASENOTES.md).
blocks, which could stall concurrent scrapes being waited on.
([#7571](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7571))

* A scrape which is still collecting when the listener is disposed now returns
an HTTP 503 response.
([#TODO](https://github.com/open-telemetry/opentelemetry-dotnet/pull/TODO))
Comment thread
martincostello marked this conversation as resolved.
Resolved

* Shutting down the listener no longer waits indefinitely for its request
processing loop to stop, and no longer throws if the loop faulted.
([#TODO](https://github.com/open-telemetry/opentelemetry-dotnet/pull/TODO))
Comment thread
martincostello marked this conversation as resolved.
Resolved

## 1.17.0-beta.1

Released 2026-Jul-16
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,9 @@ namespace OpenTelemetry.Exporter;

internal sealed class PrometheusHttpListener : IDisposable
{
private static readonly TimeSpan RequestDrainTimeout = TimeSpan.FromSeconds(5);
private static readonly TimeSpan WorkerShutdownTimeout = TimeSpan.FromSeconds(5);

private readonly PrometheusExporter exporter;
private readonly HttpListener httpListener = new();
private readonly Lock syncObject = new();
Expand Down Expand Up @@ -114,7 +117,7 @@ public void Dispose()
// Wait for in-flight requests to finish (they will observe the
// cancelled token and return 503 quickly). Use a timeout to avoid
// blocking indefinitely if a request is unexpectedly stuck.
SpinWait.SpinUntil(() => Volatile.Read(ref this.activeRequestCount) == 0, TimeSpan.FromSeconds(5));
SpinWait.SpinUntil(() => Volatile.Read(ref this.activeRequestCount) == 0, RequestDrainTimeout);

try
{
Expand Down Expand Up @@ -157,7 +160,19 @@ private void Stop()
try
{
tokenSource.Cancel();
workerThread?.Wait();

// Bounded so that a processing loop which does not observe cancellation
// promptly cannot block Dispose and the caller's shutdown indefinitely.
if (workerThread != null &&
!workerThread.Wait(WorkerShutdownTimeout))
{
PrometheusExporterEventSource.Log.FailedShutdown(
new TimeoutException($"The {nameof(PrometheusHttpListener)} processing loop did not stop within {WorkerShutdownTimeout.TotalSeconds} seconds."));
}
}
catch (Exception ex)
{
PrometheusExporterEventSource.Log.FailedShutdown(ex);
}
finally
{
Expand Down Expand Up @@ -260,6 +275,13 @@ private async Task ProcessRequestAsync(HttpListenerContext context, Cancellation
{
requestCancelled.Token.ThrowIfCancellationRequested();

// Disposal can start while this scrape is collecting, which can take
// arbitrarily long. Bail out before composing a response.
if (this.disposed || cancellationToken.IsCancellationRequested)
{
throw new OperationCanceledException(cancellationToken);
}

context.Response.Headers.Add("Server", string.Empty);

if (!collectionResponse.Succeeded)
Expand Down Expand Up @@ -307,7 +329,7 @@ private async Task ProcessRequestAsync(HttpListenerContext context, Cancellation
this.exporter.CollectionManager.ExitCollect(protocol);
}
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
catch (OperationCanceledException) when (this.disposed || cancellationToken.IsCancellationRequested)
{
context.Response.StatusCode = 503;
context.Response.ContentLength64 = 0;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,9 @@ public sealed class PromToolFixture : PrometheusFixture
{
private const string DockerInternalHost = "host.docker.internal";

private static readonly TimeSpan CheckMetricsTimeout = TimeSpan.FromMinutes(1);
private static readonly TimeSpan ScrapeTimeout = TimeSpan.FromSeconds(30);

public async Task<ExecResult> CheckMetricsAsync(
Uri targetUri,
string accept,
Expand All @@ -31,14 +34,25 @@ public async Task<ExecResult> CheckMetricsAsync(
"-c",
$"set -eu;" +
$"tmp=/tmp/metrics.$$;" +
$"wget -qO \"$tmp\" --header=\"Accept: {accept}\" --header=\"Host: {targetUri.Host}\" \"{metricsUri}\"; " +
$"wget -qO \"$tmp\" -T {ScrapeTimeout.TotalSeconds:F0} --header=\"Accept: {accept}\" --header=\"Host: {targetUri.Host}\" \"{metricsUri}\"; " +
$"cat \"$tmp\"; " +
$"promtool check metrics --lint=all < \"$tmp\"",
];

return await this.Container
.ExecAsync(command, cancellationToken)
.ConfigureAwait(false);
using var timeout = new CancellationTokenSource(CheckMetricsTimeout);
using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeout.Token);

try
{
return await this.Container
.ExecAsync(command, linked.Token)
.ConfigureAwait(false);
}
catch (OperationCanceledException) when (timeout.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
{
throw new TimeoutException(
$"promtool did not finish checking the metrics from {targetUri} within {CheckMetricsTimeout.TotalSeconds} seconds.");
}
}

protected override IContainer CreateContainer() =>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -278,6 +278,8 @@ public void Start_ThrowsObjectDisposedException_AfterDisposal()
[Fact]
public async Task ProcessRequest_Returns503_AfterDisposal()
{
EnsureThreadPoolWorkerThreadsAvailable();

using var meter = new Meter(MeterName, MeterVersion);

var port = GetRandomPort();
Expand Down Expand Up @@ -324,10 +326,25 @@ public async Task ProcessRequest_Returns503_AfterDisposal()
// Wait until the request is actually inside the Collect delegate.
Assert.True(collectEntered.Wait(TimeSpan.FromSeconds(10)), "Request did not enter Collect in time.");

// Dispose the provider on a background thread. This sets disposed = true,
// Dispose the provider on a dedicated thread. This sets disposed = true,
// cancels the CancellationToken, and then blocks on SpinWait waiting for
// the in-flight request to drain.
var disposeTask = Task.Run(provider.Dispose);
// the in-flight request to drain. A dedicated thread is used rather than a
// thread pool work item because the held request already owns a pool worker:
// on a busy agent a queued work item can sit unstarted for longer than the
// window below, and the scrape would then be released before disposal had
// cancelled anything, producing a 200 instead of a 503.
using var disposeStarted = new ManualResetEventSlim(false);
var disposeTask = Task.Factory.StartNew(
() =>
{
disposeStarted.Set();
provider.Dispose();
},
CancellationToken.None,
TaskCreationOptions.LongRunning,
TaskScheduler.Default);

Assert.True(disposeStarted.Wait(TimeSpan.FromSeconds(10)), "Disposal did not start in time.");

// Confirm Dispose() is actually blocked (meaning it has cancelled the
// token and is now waiting for activeRequestCount to reach 0). If
Expand All @@ -337,9 +354,9 @@ public async Task ProcessRequest_Returns503_AfterDisposal()
var completed = await Task.WhenAny(disposeTask, Task.Delay(timeout, cts.Token));
Assert.NotSame(disposeTask, completed);

// Release the blocker so EnterCollect can finish.
// After EnterCollect completes, ProcessRequestAsync will hit
// cancellationToken.ThrowIfCancellationRequested() and return 503.
// Release the blocker so EnterCollect can finish. Once it has,
// ProcessRequestAsync observes that disposal has started and returns 503
// instead of sending the collected response.
collectBlocker.Set();

// Wait for both the scrape response and disposal to complete.
Expand Down