diff --git a/src/ServiceControl.Transports.ASBS.Tests/QueueLengthProviderBackoffTests.cs b/src/ServiceControl.Transports.ASBS.Tests/QueueLengthProviderBackoffTests.cs new file mode 100644 index 0000000000..c92ffbcaaa --- /dev/null +++ b/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); + } + } +} diff --git a/src/ServiceControl.Transports.ASBS/AuthenticationMethod.cs b/src/ServiceControl.Transports.ASBS/AuthenticationMethod.cs index 966b2f3c3d..50fd649115 100644 --- a/src/ServiceControl.Transports.ASBS/AuthenticationMethod.cs +++ b/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); } } \ No newline at end of file diff --git a/src/ServiceControl.Transports.ASBS/ManagementThrottleDetector.cs b/src/ServiceControl.Transports.ASBS/ManagementThrottleDetector.cs new file mode 100644 index 0000000000..e2bbc3366f --- /dev/null +++ b/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; + } +} diff --git a/src/ServiceControl.Transports.ASBS/QueueLengthProvider.cs b/src/ServiceControl.Transports.ASBS/QueueLengthProvider.cs index 8978c03fed..3cfdadeea1 100644 --- a/src/ServiceControl.Transports.ASBS/QueueLengthProvider.cs +++ b/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 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> GetQueueList(CancellationToken cancellationToken) { var queuePathToRuntimeInfo = new Dictionary(StringComparer.InvariantCultureIgnoreCase); @@ -109,7 +170,11 @@ void UpdateQueueLength(KeyValuePair monitoredEndpoint, IReadOnly readonly ConcurrentDictionary endpointQueueMappings = new ConcurrentDictionary(); readonly ServiceBusAdministrationClient managementClient; + readonly ManagementThrottleDetector throttleDetector = new(); readonly TimeSpan queryDelayInterval; readonly ILogger logger; + + // Upper bound for the reactive back-off when Azure Service Bus is throttling management operations. + static readonly TimeSpan MaxBackoffInterval = TimeSpan.FromMinutes(1); } } \ No newline at end of file diff --git a/src/ServiceControl.Transports.ASBS/ServiceControl.Transports.ASBS.csproj b/src/ServiceControl.Transports.ASBS/ServiceControl.Transports.ASBS.csproj index bd61cf9af1..916b33ee96 100644 --- a/src/ServiceControl.Transports.ASBS/ServiceControl.Transports.ASBS.csproj +++ b/src/ServiceControl.Transports.ASBS/ServiceControl.Transports.ASBS.csproj @@ -10,6 +10,10 @@ + + + + diff --git a/src/ServiceControl.Transports.ASBS/SharedAccessSignatureAuthentication.cs b/src/ServiceControl.Transports.ASBS/SharedAccessSignatureAuthentication.cs index 5368353371..3ad4a0f2a0 100644 --- a/src/ServiceControl.Transports.ASBS/SharedAccessSignatureAuthentication.cs +++ b/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); diff --git a/src/ServiceControl.Transports.ASBS/TokenCredentialAuthentication.cs b/src/ServiceControl.Transports.ASBS/TokenCredentialAuthentication.cs index d00e277891..27c5223cdc 100644 --- a/src/ServiceControl.Transports.ASBS/TokenCredentialAuthentication.cs +++ b/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) {