Migrate to custom OTel Exporter in dotnetup to prevent perf delay and data loss

src/Cli/Microsoft.DotNet.Cli.Telemetry/Implementation/BreezePartialContent.cs

@@ -51,7 +51,7 @@ internal static class BreezePartialContent
         var retriableIndices = new HashSet<int>();
         foreach (var error in errors)
         {
-            if (s_retriableStatusCodes.Contains(error.StatusCode))
+            if (IsRetriableStatusCode(error.StatusCode))
             {
                 retriableIndices.Add(error.Index);
             }
@@ -86,4 +86,7 @@ internal static class BreezePartialContent
 
         return stream.Length == 0 ? null : stream.ToArray();
     }
+
+    internal static bool IsRetriableStatusCode(int statusCode)
+        => s_retriableStatusCodes.Contains(statusCode);
 }

src/Cli/Microsoft.DotNet.Cli.Telemetry/Implementation/HttpTelemetryUploadTransport.cs

@@ -65,11 +65,36 @@ public async Task<TelemetryUploadResult> TryUploadAsync(byte[] payload, Cancella
             // blob is done. Otherwise re-persist just the retriable envelopes.
             return retriable is null
                 ? TelemetryUploadResult.Accepted
-                : TelemetryUploadResult.PartiallyAccepted(retriable);
+                : TelemetryUploadResult.PartiallyAccepted(retriable, GetRetryAfter(response));
         }
 
-        // Throttling, server errors, etc.: retain the blob and retry it later.
-        return TelemetryUploadResult.Rejected;
+        // Retain only statuses that Azure Monitor defines as retriable. Treat permanent
+        // whole-request failures as terminal so one poison blob cannot block later telemetry.
+        if (!BreezePartialContent.IsRetriableStatusCode((int)response.StatusCode))
+        {
+            return TelemetryUploadResult.PermanentlyRejected;
+        }
+
+        return GetRetryAfter(response) is { } retryAfter
+            ? TelemetryUploadResult.RejectedAfter(retryAfter)
+            : TelemetryUploadResult.Rejected;
+    }
+
+    private static TimeSpan? GetRetryAfter(HttpResponseMessage response)
+    {
+        var retryAfter = response.Headers.RetryAfter;
+        if (retryAfter?.Delta is { } delta)
+        {
+            return delta < TimeSpan.Zero ? TimeSpan.Zero : delta;
+        }
+
+        if (retryAfter?.Date is { } date)
+        {
+            var delay = date - DateTimeOffset.UtcNow;
+            return delay < TimeSpan.Zero ? TimeSpan.Zero : delay;
+        }
+
+        return null;
     }
 
     /// <summary>

src/Cli/Microsoft.DotNet.Cli.Telemetry/Implementation/PersistentStorageTelemetryBackgroundWorker.cs

@@ -0,0 +1,88 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System.Diagnostics;
+using System.Threading;
+
+namespace Microsoft.DotNet.Cli.Telemetry.Implementation;
+
+internal sealed class PersistentStorageTelemetryBackgroundWorker
+{
+    private readonly Func<CancellationToken, Task> _drainAsync;
+    private int _started;
+    private CancellationTokenSource? _cancellation;
+    private Task? _task;
+
+    public PersistentStorageTelemetryBackgroundWorker(
+        ITelemetryBlobStorage storage,
+        Uri ingestionTrackUri,
+        int leasePeriodMilliseconds,
+        int maxBlobsPerDrain)
+        : this(CreateDrainAsync(storage, ingestionTrackUri, leasePeriodMilliseconds, maxBlobsPerDrain))
+    {
+    }
+
+    internal PersistentStorageTelemetryBackgroundWorker(Func<CancellationToken, Task> drainAsync)
+    {
+        _drainAsync = drainAsync;
+    }
+
+    public void StartOnce()
+    {
+        if (Interlocked.Exchange(ref _started, 1) != 0)
+        {
+            return;
+        }
+
+        _cancellation = new CancellationTokenSource();
+        _task = Task.Run(() => DrainAsync(_cancellation.Token));
+    }
+
+    public bool Shutdown(int timeoutMilliseconds)
+    {
+        _cancellation?.Cancel();
+        if (_task is null)
+        {
+            return true;
+        }
+
+        try
+        {
+            return _task.Wait(timeoutMilliseconds);
+        }
+        catch (AggregateException)
+        {
+            return true;
+        }
+    }
+
+    private async Task DrainAsync(CancellationToken cancellationToken)
+    {
+        try
+        {
+            await _drainAsync(cancellationToken).ConfigureAwait(false);
+        }
+        catch (OperationCanceledException)
+        {
+        }
+        catch (Exception e)
+        {
+            Debug.Fail(e.ToString());
+        }
+    }
+
+    private static Func<CancellationToken, Task> CreateDrainAsync(
+        ITelemetryBlobStorage storage,
+        Uri ingestionTrackUri,
+        int leasePeriodMilliseconds,
+        int maxBlobsPerDrain)
+    {
+        var transport = new HttpTelemetryUploadTransport(ingestionTrackUri);
+        var uploader = new PersistentStorageTelemetryUploader(
+            storage,
+            transport,
+            leasePeriodMilliseconds,
+            maxBlobsPerDrain);
+        return cancellationToken => uploader.DrainAsync(cancellationToken);
+    }
+}

src/Cli/Microsoft.DotNet.Cli.Telemetry/Implementation/PersistentStorageTelemetryUploader.cs

@@ -45,9 +45,12 @@ public PersistentStorageTelemetryUploader(
     /// Leases, uploads, and deletes persisted blobs. Blobs that fail to upload are left for a
     /// later retry. This method never throws.
     /// </summary>
-    public async Task DrainAsync(CancellationToken cancellationToken)
+    public async Task<TelemetryDrainResult> DrainAsync(CancellationToken cancellationToken)
     {
         var processed = 0;
+        var forwardProgress = 0;
+        var shouldBackOff = false;
+        TimeSpan? retryAfter = null;
         // Retriable remainders from partially-accepted uploads. Persisted AFTER the enumeration
         // completes so we never mutate the storage collection while iterating it.
         List<byte[]>? retriableRemainders = null;
@@ -92,16 +95,26 @@ public async Task DrainAsync(CancellationToken cancellationToken)
                         {
                             (retriableRemainders ??= []).Add(remainder);
                         }
+                        shouldBackOff = true;
+                        retryAfter = result.RetryAfter;
                         deleted = blob.TryDelete();
                         break;
 
                     case TelemetryUploadOutcome.Accepted:
                         deleted = blob.TryDelete();
                         break;
 
+                    case TelemetryUploadOutcome.PermanentlyRejected:
+                        // Retrying cannot succeed. Delete this poison blob so later telemetry
+                        // in the storage directory can continue draining.
+                        deleted = blob.TryDelete();
+                        break;
+
                     case TelemetryUploadOutcome.Rejected:
                         // Leave the blob in place; its lease will expire and a later invocation
                         // will retry it.
+                        shouldBackOff = true;
+                        retryAfter = result.RetryAfter;
                         break;
                 }
             }
@@ -113,16 +126,31 @@ public async Task DrainAsync(CancellationToken cancellationToken)
             }
             catch (Exception e)
             {
-                // Swallow per-blob failures and keep going; the blob is retried later.
+                // Network and storage failures are normally transient. Stop this pass and let the
+                // caller back off before trying the blob again rather than moving immediately to
+                // the rest of the backlog.
                 Debug.Fail(e.ToString());
+                shouldBackOff = true;
             }
             finally
             {
-                if (leased && !deleted)
+                if (deleted)
+                {
+                    forwardProgress++;
+                }
+                else if (leased)
                 {
                     blob.TryRelease();
                 }
             }
+
+            if (shouldBackOff)
+            {
+                // A retryable response normally indicates service throttling or a transient
+                // failure. Stop this pass rather than submitting every remaining blob to a
+                // service that has already asked us to retry.
+                break;
+            }
         }
 
         if (retriableRemainders is not null)
@@ -132,5 +160,7 @@ public async Task DrainAsync(CancellationToken cancellationToken)
                 _storage.TryPersist(remainder);
             }
         }
+
+        return new TelemetryDrainResult(forwardProgress, shouldBackOff, retryAfter);
     }
 }

src/Cli/Microsoft.DotNet.Cli.Telemetry/Implementation/TelemetryUploadResult.cs

@@ -15,6 +15,10 @@ internal enum TelemetryUploadOutcome
     /// <see cref="TelemetryUploadResult.RetryPayload"/> and should be persisted for a later retry.</summary>
     PartiallyAccepted,
 
+    /// <summary>The server permanently rejected the payload. Retrying cannot succeed, so the
+    /// blob should be deleted to avoid blocking later telemetry.</summary>
+    PermanentlyRejected,
+
     /// <summary>The server did not accept the payload (throttling, server error, etc.). The
     /// blob should be retained and retried later.</summary>
     Rejected,
@@ -28,10 +32,11 @@ internal enum TelemetryUploadOutcome
 /// </summary>
 internal readonly struct TelemetryUploadResult
 {
-    private TelemetryUploadResult(TelemetryUploadOutcome outcome, byte[]? retryPayload)
+    private TelemetryUploadResult(TelemetryUploadOutcome outcome, byte[]? retryPayload, TimeSpan? retryAfter)
     {
         Outcome = outcome;
         RetryPayload = retryPayload;
+        RetryAfter = retryAfter;
     }
 
     public TelemetryUploadOutcome Outcome { get; }
@@ -42,10 +47,39 @@ private TelemetryUploadResult(TelemetryUploadOutcome outcome, byte[]? retryPaylo
     /// </summary>
     public byte[]? RetryPayload { get; }
 
-    public static TelemetryUploadResult Accepted { get; } = new(TelemetryUploadOutcome.Accepted, null);
+    /// <summary>
+    /// The server-provided delay before another upload attempt, when present.
+    /// </summary>
+    public TimeSpan? RetryAfter { get; }
+
+    public static TelemetryUploadResult Accepted { get; } = new(TelemetryUploadOutcome.Accepted, null, null);
+
+    public static TelemetryUploadResult Rejected { get; } = new(TelemetryUploadOutcome.Rejected, null, null);
+
+    public static TelemetryUploadResult PermanentlyRejected { get; } = new(TelemetryUploadOutcome.PermanentlyRejected, null, null);
+
+    public static TelemetryUploadResult RejectedAfter(TimeSpan retryAfter)
+        => new(TelemetryUploadOutcome.Rejected, null, retryAfter);
+
+    public static TelemetryUploadResult PartiallyAccepted(byte[] retryPayload, TimeSpan? retryAfter = null)
+        => new(TelemetryUploadOutcome.PartiallyAccepted, retryPayload, retryAfter);
+}
+
+/// <summary>
+/// The result of one storage drain pass.
+/// </summary>
+internal readonly struct TelemetryDrainResult
+{
+    public TelemetryDrainResult(int forwardProgress, bool shouldBackOff, TimeSpan? retryAfter)
+    {
+        ForwardProgress = forwardProgress;
+        ShouldBackOff = shouldBackOff;
+        RetryAfter = retryAfter;
+    }
+
+    public int ForwardProgress { get; }
 
-    public static TelemetryUploadResult Rejected { get; } = new(TelemetryUploadOutcome.Rejected, null);
+    public bool ShouldBackOff { get; }
 
-    public static TelemetryUploadResult PartiallyAccepted(byte[] retryPayload)
-        => new(TelemetryUploadOutcome.PartiallyAccepted, retryPayload);
+    public TimeSpan? RetryAfter { get; }
 }