Handle ASB queue-length management throttling (pipeline detection)

src/ServiceControl.Transports.ASBS.Tests/QueueLengthProviderBackoffTests.cs

@@ -0,0 +1,61 @@
+namespace ServiceControl.Transports.UnitTests.ASBS
+{
+    using System;
+    using NUnit.Framework;
+    using ServiceControl.Transports.ASBS;
+
+    [TestFixture]
+    class QueueLengthProviderBackoffTests
+    {
+        // Base is the configured QueueLengthQueryDelayInterval (default). Max is the internal reactive-backoff cap.
+        static readonly TimeSpan Base = TimeSpan.FromMilliseconds(500);
+        static readonly TimeSpan Max = TimeSpan.FromSeconds(60);
+
+        [Test]
+        public void Doubles_the_delay_when_throttled()
+        {
+            // A throttled cycle backs off so the provider stops making the throttling worse.
+            Assert.That(QueueLengthProvider.NextDelay(Base, Base, Max, throttled: true),
+                Is.EqualTo(TimeSpan.FromSeconds(1)));
+        }
+
+        [Test]
+        public void Caps_the_delay_at_max_when_throttled()
+        {
+            // 40s doubled would be 80s; the cap keeps it bounded.
+            Assert.That(QueueLengthProvider.NextDelay(TimeSpan.FromSeconds(40), Base, Max, throttled: true),
+                Is.EqualTo(Max));
+        }
+
+        [Test]
+        public void Halves_the_delay_on_success_while_still_backed_off()
+        {
+            // Success steps the cadence back down gradually rather than snapping to base, avoiding
+            // throttle -> reset -> throttle oscillation.
+            Assert.That(QueueLengthProvider.NextDelay(TimeSpan.FromSeconds(2), Base, Max, throttled: false),
+                Is.EqualTo(TimeSpan.FromSeconds(1)));
+        }
+
+        [Test]
+        public void Never_drops_below_base_on_success()
+        {
+            Assert.That(QueueLengthProvider.NextDelay(Base, Base, Max, throttled: false),
+                Is.EqualTo(Base));
+        }
+
+        [Test]
+        public void Http_429_is_classified_as_throttling()
+        {
+            Assert.That(ManagementThrottleDetector.IsThrottleResponse(429), Is.True);
+        }
+
+        [TestCase(200)]
+        [TestCase(404)]
+        [TestCase(500)]
+        [TestCase(503)]
+        public void Non_429_responses_are_not_classified_as_throttling(int statusCode)
+        {
+            Assert.That(ManagementThrottleDetector.IsThrottleResponse(statusCode), Is.False);
+        }
+    }
+}

src/ServiceControl.Transports.ASBS/AuthenticationMethod.cs

@@ -5,7 +5,7 @@
 
     public abstract class AuthenticationMethod
     {
-        public abstract ServiceBusAdministrationClient BuildManagementClient();
+        public abstract ServiceBusAdministrationClient BuildManagementClient(ServiceBusAdministrationClientOptions options = null);
         public abstract AzureServiceBusTransport CreateTransportDefinition(ConnectionSettings connectionSettings, TopicTopology topology);
     }
 }

src/ServiceControl.Transports.ASBS/ManagementThrottleDetector.cs

@@ -0,0 +1,27 @@
+namespace ServiceControl.Transports.ASBS
+{
+    using System.Threading;
+    using Azure.Core;
+    using Azure.Core.Pipeline;
+
+    // Counts the HTTP 429 throttling responses on the management client's pipeline, including the ones the SDK's
+    // retry absorbs so they never throw. Attach at PerRetry so every attempt is seen.
+    sealed class ManagementThrottleDetector : HttpPipelineSynchronousPolicy
+    {
+        // Interlocked: the pipeline runs on SDK I/O threads. The loop snapshots this around each query.
+        public long ThrottledResponseCount => Interlocked.Read(ref throttledResponseCount);
+
+        public override void OnReceivedResponse(HttpMessage message)
+        {
+            if (IsThrottleResponse(message.Response.Status))
+            {
+                Interlocked.Increment(ref throttledResponseCount);
+            }
+        }
+
+        // HTTP 429 (Too Many Requests) is how the Service Bus management endpoint signals throttling.
+        internal static bool IsThrottleResponse(int statusCode) => statusCode == 429;
+
+        long throttledResponseCount;
+    }
+}

src/ServiceControl.Transports.ASBS/QueueLengthProvider.cs

@@ -5,6 +5,8 @@ namespace ServiceControl.Transports.ASBS
     using System.Collections.Generic;
     using System.Threading;
     using System.Threading.Tasks;
+    using Azure.Core;
+    using Azure.Messaging.ServiceBus;
     using Azure.Messaging.ServiceBus.Administration;
     using Microsoft.Extensions.Logging;
 
@@ -16,7 +18,10 @@ public QueueLengthProvider(TransportSettings settings, Action<QueueLengthEntry[]
 
             queryDelayInterval = connectionSettings.QueryDelayInterval ?? TimeSpan.FromMilliseconds(500);
 
-            managementClient = connectionSettings.AuthenticationMethod.BuildManagementClient();
+            // PerRetry so the detector sees the 429s the SDK retries away (they never surface as exceptions).
+            var clientOptions = new ServiceBusAdministrationClientOptions();
+            clientOptions.AddPolicy(throttleDetector, HttpPipelinePosition.PerRetry);
+            managementClient = connectionSettings.AuthenticationMethod.BuildManagementClient(clientOptions);
             this.logger = logger;
         }
 
@@ -29,32 +34,88 @@ public override void TrackEndpointInputQueue(EndpointToQueueMapping queueToTrack
 
         protected override async Task ExecuteAsync(CancellationToken stoppingToken)
         {
+            // currentDelay backs off while throttled and recovers to the base once clear; throttled latches the log.
+            var currentDelay = queryDelayInterval;
+            var throttled = false;
+
             while (!stoppingToken.IsCancellationRequested)
             {
+                bool throttledThisCycle;
+
                 try
                 {
                     logger.LogDebug("Waiting for next interval");
-                    await Task.Delay(queryDelayInterval, stoppingToken);
+                    await Task.Delay(currentDelay, stoppingToken);
 
                     logger.LogDebug("Querying management client");
 
+                    // A query can succeed yet still have been throttled (429s retried away), so detect via the counter.
+                    var throttledResponsesBefore = throttleDetector.ThrottledResponseCount;
+
                     var queueRuntimeInfos = await GetQueueList(stoppingToken);
 
                     logger.LogDebug("Retrieved details of {QueueCount} queues", queueRuntimeInfos.Count);
 
                     UpdateAllQueueLengths(queueRuntimeInfos);
+
+                    throttledThisCycle = throttleDetector.ThrottledResponseCount > throttledResponsesBefore;
                 }
                 catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
                 {
-                    // no-op
+                    break;
+                }
+                catch (ServiceBusException e) when (e.Reason == ServiceBusFailureReason.ServiceBusy)
+                {
+                    // Fallback for when the SDK's retries were exhausted and it threw.
+                    throttledThisCycle = true;
                 }
                 catch (Exception e)
                 {
+                    // Unrelated error: log and keep the current cadence.
                     logger.LogError(e, "Error querying Azure Service Bus queue sizes");
+                    continue;
+                }
+
+                // Store nothing on a throttled cycle; the last values stand. Ideally we'd record an explicit
+                // "no value" (null/-1) so the graph shows a gap, but Value is a non-nullable long and the
+                // API/ServicePulse coerce gaps to 0 - that needs a new API contract + a ServicePulse using it.
+                currentDelay = NextDelay(currentDelay, queryDelayInterval, MaxBackoffInterval, throttledThisCycle);
+
+                if (throttledThisCycle)
+                {
+                    if (!throttled)
+                    {
+                        throttled = true;
+                        logger.LogWarning("Azure Service Bus is throttling the management operations used to read queue lengths. Backing off to a {CurrentDelay} query interval. This is expected on busy or large namespaces; increase the 'QueueLengthQueryDelayInterval' connection string setting to reduce it. Queue length metrics may be delayed until the throttling clears.", currentDelay);
+                    }
+                    else
+                    {
+                        logger.LogDebug("Still throttled by Azure Service Bus; backing off query interval to {CurrentDelay}", currentDelay);
+                    }
+                }
+                else if (throttled && currentDelay == queryDelayInterval)
+                {
+                    // Recover only once fully back at the base, so a slow ramp-down doesn't re-arm the warning.
+                    throttled = false;
+                    logger.LogInformation("Azure Service Bus management throttling has cleared; resumed querying queue lengths every {QueryDelayInterval}", queryDelayInterval);
                 }
             }
         }
 
+        // Pure back-off policy (unit-tested). Throttled -> exponential up to the cap; success -> halve back toward
+        // the base so it settles near the sustainable rate instead of oscillating.
+        internal static TimeSpan NextDelay(TimeSpan current, TimeSpan baseDelay, TimeSpan maxDelay, bool throttled)
+        {
+            if (throttled)
+            {
+                var doubled = TimeSpan.FromTicks(current.Ticks * 2);
+                return doubled > maxDelay ? maxDelay : doubled;
+            }
+
+            var halved = TimeSpan.FromTicks(current.Ticks / 2);
+            return halved < baseDelay ? baseDelay : halved;
+        }
+
         async Task<IReadOnlyDictionary<string, QueueRuntimeProperties>> GetQueueList(CancellationToken cancellationToken)
         {
             var queuePathToRuntimeInfo = new Dictionary<string, QueueRuntimeProperties>(StringComparer.InvariantCultureIgnoreCase);
@@ -109,7 +170,11 @@ void UpdateQueueLength(KeyValuePair<string, string> monitoredEndpoint, IReadOnly
 
         readonly ConcurrentDictionary<string, string> endpointQueueMappings = new ConcurrentDictionary<string, string>();
         readonly ServiceBusAdministrationClient managementClient;
+        readonly ManagementThrottleDetector throttleDetector = new();
         readonly TimeSpan queryDelayInterval;
         readonly ILogger<QueueLengthProvider> logger;
+
+        // Upper bound for the reactive back-off when Azure Service Bus is throttling management operations.
+        static readonly TimeSpan MaxBackoffInterval = TimeSpan.FromMinutes(1);
     }
 }

src/ServiceControl.Transports.ASBS/ServiceControl.Transports.ASBS.csproj

@@ -10,6 +10,10 @@
     <ProjectReference Include="..\ServiceControl.Transports\ServiceControl.Transports.csproj" Private="false" ExcludeAssets="runtime" />
   </ItemGroup>
 
+  <ItemGroup>
+    <InternalsVisibleTo Include="ServiceControl.Transports.ASBS.Tests" />
+  </ItemGroup>
+
   <ItemGroup>
     <PackageReference Include="Azure.Identity" />
     <PackageReference Include="Azure.ResourceManager.Monitor" />

src/ServiceControl.Transports.ASBS/SharedAccessSignatureAuthentication.cs

@@ -9,8 +9,10 @@ public class SharedAccessSignatureAuthentication : AuthenticationMethod
 
         public string ConnectionString { get; }
 
-        public override ServiceBusAdministrationClient BuildManagementClient()
-            => new ServiceBusAdministrationClient(ConnectionString);
+        public override ServiceBusAdministrationClient BuildManagementClient(ServiceBusAdministrationClientOptions options = null)
+            => options is null
+                ? new ServiceBusAdministrationClient(ConnectionString)
+                : new ServiceBusAdministrationClient(ConnectionString, options);
 
         public override AzureServiceBusTransport CreateTransportDefinition(ConnectionSettings connectionSettings, TopicTopology topology)
             => new AzureServiceBusTransport(ConnectionString, topology);

src/ServiceControl.Transports.ASBS/TokenCredentialAuthentication.cs

@@ -27,7 +27,10 @@ public TokenCredentialAuthentication(string fullyQualifiedNamespace, string? cli
 
         public string? ClientId { get; }
 
-        public override ServiceBusAdministrationClient BuildManagementClient() => new(FullyQualifiedNamespace, Credential);
+        public override ServiceBusAdministrationClient BuildManagementClient(ServiceBusAdministrationClientOptions? options = null)
+            => options is null
+                ? new(FullyQualifiedNamespace, Credential)
+                : new(FullyQualifiedNamespace, Credential, options);
 
         public override AzureServiceBusTransport CreateTransportDefinition(ConnectionSettings connectionSettings, TopicTopology topology)
         {