From 2cb83adc8d56172146ee0c7ee706dc564be47b73 Mon Sep 17 00:00:00 2001 From: danielmarbach Date: Wed, 19 Dec 2018 17:09:27 +0100 Subject: [PATCH 1/7] Graceful shutdown SQL provider --- .../QueueLengthProvider.cs | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/src/ServiceControl.Transports.SQLServer/QueueLengthProvider.cs b/src/ServiceControl.Transports.SQLServer/QueueLengthProvider.cs index bce6134..e0e0cf5 100644 --- a/src/ServiceControl.Transports.SQLServer/QueueLengthProvider.cs +++ b/src/ServiceControl.Transports.SQLServer/QueueLengthProvider.cs @@ -66,16 +66,22 @@ public Task Start() poller = Task.Run(async () => { - while (!stop.Token.IsCancellationRequested) + var token = stop.Token; + + while (!token.IsCancellationRequested) { try { - await Task.Delay(QueryDelayInterval); + await Task.Delay(QueryDelayInterval, token).ConfigureAwait(false); - await QueryTableSizes(); + await QueryTableSizes(token).ConfigureAwait(false); UpdateQueueLengthStore(); } + catch (OperationCanceledException) + { + // no-op + } catch (Exception e) { Logger.Error("Error querying sql queue sizes.", e); @@ -109,7 +115,7 @@ void UpdateQueueLengthStore() } } - async Task QueryTableSizes() + async Task QueryTableSizes(CancellationToken token) { var chunks = tableSizes .Select((i, index) => new @@ -125,9 +131,9 @@ async Task QueryTableSizes() { using (var connection = new SqlConnection(connectionString)) { - await connection.OpenAsync(); + await connection.OpenAsync(token).ConfigureAwait(false); - await UpdateChunk(connection, chunk, stop.Token); + await UpdateChunk(connection, chunk, token).ConfigureAwait(false); } } } @@ -138,11 +144,11 @@ async Task UpdateChunk(SqlConnection connection, KeyValuePair[] c using (var command = new SqlCommand(query, connection)) { - var reader = await command.ExecuteReaderAsync(token); + var reader = await command.ExecuteReaderAsync(token).ConfigureAwait(false); foreach (var chunkPair in chunk) { - await reader.ReadAsync(token); + await reader.ReadAsync(token).ConfigureAwait(false); var queueLength = reader.GetInt32(0); @@ -155,7 +161,7 @@ async Task UpdateChunk(SqlConnection connection, KeyValuePair[] c tableSizes.TryUpdate(chunkPair.Key, queueLength, chunkPair.Value); } - await reader.NextResultAsync(token); + await reader.NextResultAsync(token).ConfigureAwait(false); } } } From e29ad577cd9bd10b6f10168f550aba7c7c0ea6be Mon Sep 17 00:00:00 2001 From: danielmarbach Date: Wed, 19 Dec 2018 17:12:48 +0100 Subject: [PATCH 2/7] AzureServiceBus Queue length provider --- .../QueueLengthProvider.cs | 26 ++++++++++++------- 1 file changed, 16 insertions(+), 10 deletions(-) diff --git a/src/ServiceControl.Transports.AzureServiceBus/QueueLengthProvider.cs b/src/ServiceControl.Transports.AzureServiceBus/QueueLengthProvider.cs index 74969c2..82bd1ca 100644 --- a/src/ServiceControl.Transports.AzureServiceBus/QueueLengthProvider.cs +++ b/src/ServiceControl.Transports.AzureServiceBus/QueueLengthProvider.cs @@ -21,7 +21,7 @@ public class QueueLengthProvider : IProvideQueueLength ManagementClient managementClient; CancellationTokenSource stop = new CancellationTokenSource(); - Task pooler; + Task poller; public void Initialize(string connectionString, QueueLengthStore store) { @@ -46,23 +46,29 @@ public Task Start() { stop = new CancellationTokenSource(); - pooler = Task.Run(async () => + poller = Task.Run(async () => { - while (!stop.Token.IsCancellationRequested) + var token = stop.Token; + + while (!token.IsCancellationRequested) { try { Logger.Debug("Waiting for next interval"); - await Task.Delay(QueryDelayInterval).ConfigureAwait(false); + await Task.Delay(QueryDelayInterval, token).ConfigureAwait(false); Logger.DebugFormat("Querying management client."); - var queues = await managementClient.GetQueuesAsync().ConfigureAwait(false); + var queues = await managementClient.GetQueuesAsync(cancellationToken: token).ConfigureAwait(false); var lookup = queues.ToLookup(x => x.Path, StringComparer.InvariantCultureIgnoreCase); Logger.DebugFormat("Retrieved details of {0} queues", lookup.Count); - await UpdateQueueLengthStore(lookup); + await UpdateQueueLengthStore(lookup, token); + } + catch (OperationCanceledException) + { + // no-op } catch (Exception e) { @@ -74,7 +80,7 @@ public Task Start() return TaskEx.Completed; } - private async Task UpdateQueueLengthStore(ILookup queueData) + async Task UpdateQueueLengthStore(ILookup queueData, CancellationToken token) { var timestamp = DateTime.UtcNow.Ticks; foreach (var mapping in endpointQueueMappings) @@ -87,7 +93,7 @@ private async Task UpdateQueueLengthStore(ILookup queu new RawMessage.Entry { DateTicks = timestamp, - Value = (await managementClient.GetQueueRuntimeInfoAsync( queue.Path )).MessageCountDetails.ActiveMessageCount + Value = (await managementClient.GetQueueRuntimeInfoAsync(queue.Path, token).ConfigureAwait(false)).MessageCountDetails.ActiveMessageCount } }; queueLengthStore.Store(entries, new EndpointInputQueue(mapping.Key.EndpointName, queue.Path)); @@ -102,8 +108,8 @@ private async Task UpdateQueueLengthStore(ILookup queu public async Task Stop() { stop.Cancel(); - await managementClient.CloseAsync(); - await pooler; + await poller.ConfigureAwait(false); + await managementClient.CloseAsync().ConfigureAwait(false); } static TimeSpan QueryDelayInterval = TimeSpan.FromMilliseconds(200); From fc2e244d6a272e101d661ee1883e22a0dd7c9cf6 Mon Sep 17 00:00:00 2001 From: danielmarbach Date: Wed, 19 Dec 2018 17:17:31 +0100 Subject: [PATCH 3/7] SQS QueueLengthProvider graceful shutdown --- .../QueueAttributesRequestCache.cs | 9 +++--- .../QueueLengthProvider.cs | 29 ++++++++++++------- 2 files changed, 24 insertions(+), 14 deletions(-) diff --git a/src/ServiceControl.Transports.AmazonSQS/QueueAttributesRequestCache.cs b/src/ServiceControl.Transports.AmazonSQS/QueueAttributesRequestCache.cs index ba84c97..ed3d7f5 100644 --- a/src/ServiceControl.Transports.AmazonSQS/QueueAttributesRequestCache.cs +++ b/src/ServiceControl.Transports.AmazonSQS/QueueAttributesRequestCache.cs @@ -1,6 +1,7 @@ namespace ServiceControl.Transports.AmazonSQS { using System.Collections.Concurrent; + using System.Threading; using System.Threading.Tasks; using Amazon.SQS; using Amazon.SQS.Model; @@ -13,14 +14,14 @@ public QueueAttributesRequestCache(IAmazonSQS sqsClient) this.sqsClient = sqsClient; } - public async Task GetQueueAttributesRequest(string queueName) + public async Task GetQueueAttributesRequest(string queueName, CancellationToken token) { if (cache.TryGetValue(queueName, out var attReq)) { return attReq; } - var queueUrl = await GetQueueUrl(queueName).ConfigureAwait(false); + var queueUrl = await GetQueueUrl(queueName, token).ConfigureAwait(false); attReq = new GetQueueAttributesRequest {QueueUrl = queueUrl}; attReq.AttributeNames.Add("ApproximateNumberOfMessages"); @@ -30,9 +31,9 @@ public async Task GetQueueAttributesRequest(string qu return attReq; } - async Task GetQueueUrl(string queueName) + async Task GetQueueUrl(string queueName, CancellationToken token) { - var response = await sqsClient.GetQueueUrlAsync(queueName) + var response = await sqsClient.GetQueueUrlAsync(queueName, token) .ConfigureAwait(false); return response.QueueUrl; } diff --git a/src/ServiceControl.Transports.AmazonSQS/QueueLengthProvider.cs b/src/ServiceControl.Transports.AmazonSQS/QueueLengthProvider.cs index 9f3b4cf..c0b7c85 100644 --- a/src/ServiceControl.Transports.AmazonSQS/QueueLengthProvider.cs +++ b/src/ServiceControl.Transports.AmazonSQS/QueueLengthProvider.cs @@ -24,7 +24,7 @@ public class QueueLengthProvider : IProvideQueueLength string queueNamePrefix; CancellationTokenSource stop = new CancellationTokenSource(); - Task pooler; + Task poller; Func clientFactory = () => new AmazonSQSClient(); @@ -77,21 +77,26 @@ public Task Start() { stop = new CancellationTokenSource(); - pooler = Task.Run(async () => + poller = Task.Run(async () => { using (var client = clientFactory()) { var cache = new QueueAttributesRequestCache(client); + var token = stop.Token; - while (!stop.Token.IsCancellationRequested) + while (!token.IsCancellationRequested) { try { - await FetchQueueSizes(cache, client).ConfigureAwait(false); + await FetchQueueSizes(cache, client, token).ConfigureAwait(false); UpdateQueueLengthStore(); - await Task.Delay(QueryDelayInterval).ConfigureAwait(false); + await Task.Delay(QueryDelayInterval, token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // no-op } catch (Exception e) { @@ -108,7 +113,7 @@ public Task Stop() { stop.Cancel(); - return pooler; + return poller; } void UpdateQueueLengthStore() @@ -127,17 +132,21 @@ void UpdateQueueLengthStore() } } - Task FetchQueueSizes(QueueAttributesRequestCache cache, IAmazonSQS client) => Task.WhenAll(sizes.Select(kvp => FetchLength(kvp.Key, client, cache))); + Task FetchQueueSizes(QueueAttributesRequestCache cache, IAmazonSQS client, CancellationToken token) => Task.WhenAll(sizes.Select(kvp => FetchLength(kvp.Key, client, cache, token))); - async Task FetchLength(string queue, IAmazonSQS client, QueueAttributesRequestCache cache) + async Task FetchLength(string queue, IAmazonSQS client, QueueAttributesRequestCache cache, CancellationToken token) { try { - var attReq = await cache.GetQueueAttributesRequest(queue).ConfigureAwait(false); - var response = await client.GetQueueAttributesAsync(attReq).ConfigureAwait(false); + var attReq = await cache.GetQueueAttributesRequest(queue, token).ConfigureAwait(false); + var response = await client.GetQueueAttributesAsync(attReq, token).ConfigureAwait(false); sizes[queue] = response.ApproximateNumberOfMessages; } + catch (OperationCanceledException) + { + // no-op + } catch (Exception ex) { Logger.Error($"Obtaining an approximate number of messages failed for '{queue}'", ex); From 5a96d5df2aaaf44702cf0dc1efa9fd7753ebbfe0 Mon Sep 17 00:00:00 2001 From: danielmarbach Date: Wed, 19 Dec 2018 17:20:46 +0100 Subject: [PATCH 4/7] ASQ QueueLengthProvider graceful shutdown --- .../QueueLengthProvider.cs | 27 ++++++++++++------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs b/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs index 09a4103..d11adf7 100644 --- a/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs +++ b/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs @@ -24,7 +24,7 @@ public class QueueLengthProvider : IProvideQueueLength QueueLengthStore store; CancellationTokenSource stop = new CancellationTokenSource(); - Task pooler; + Task poller; public void Initialize(string connectionString, QueueLengthStore store) { @@ -62,17 +62,22 @@ public Task Start() { stop = new CancellationTokenSource(); - pooler = Task.Run(async () => + poller = Task.Run(async () => { - while (!stop.Token.IsCancellationRequested) + var token = stop.Token; + while (!token.IsCancellationRequested) { try { - await FetchQueueSizes().ConfigureAwait(false); + await FetchQueueSizes(token).ConfigureAwait(false); UpdateQueueLengthStore(); - await Task.Delay(QueryDelayInterval); + await Task.Delay(QueryDelayInterval, token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // no-op } catch (Exception e) { @@ -88,7 +93,7 @@ public Task Stop() { stop.Cancel(); - return pooler; + return poller; } void UpdateQueueLengthStore() @@ -107,17 +112,21 @@ void UpdateQueueLengthStore() } } - Task FetchQueueSizes() => Task.WhenAll(sizes.Select(kvp => FetchLength(kvp.Key))); + Task FetchQueueSizes(CancellationToken token) => Task.WhenAll(sizes.Select(kvp => FetchLength(kvp.Key, token))); - async Task FetchLength(CloudQueue queue) + async Task FetchLength(CloudQueue queue, CancellationToken token) { try { - await queue.FetchAttributesAsync(stop.Token).ConfigureAwait(false); + await queue.FetchAttributesAsync(token).ConfigureAwait(false); sizes[queue] = queue.ApproximateMessageCount.GetValueOrDefault(); problematicQueues.TryRemove(queue, out _); } + catch (OperationCanceledException) + { + // no-op + } catch (Exception ex) { // simple "log once" approach to do not flood logs From 73153a8c650b7feeec138da09d2672c705b7fd4d Mon Sep 17 00:00:00 2001 From: danielmarbach Date: Wed, 19 Dec 2018 17:23:28 +0100 Subject: [PATCH 5/7] AzureServiceBus QueueLengthProvider graceful shutdown --- .../QueueLengthProvider.cs | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/src/ServiceControl.Transports.LegacyAzureServiceBus/QueueLengthProvider.cs b/src/ServiceControl.Transports.LegacyAzureServiceBus/QueueLengthProvider.cs index b630a8e..512a69d 100644 --- a/src/ServiceControl.Transports.LegacyAzureServiceBus/QueueLengthProvider.cs +++ b/src/ServiceControl.Transports.LegacyAzureServiceBus/QueueLengthProvider.cs @@ -22,7 +22,7 @@ public class QueueLengthProvider : IProvideQueueLength NamespaceManager namespaceManager; CancellationTokenSource stop = new CancellationTokenSource(); - Task pooler; + Task poller; public void Initialize(string connectionString, QueueLengthStore store) { @@ -47,10 +47,10 @@ public Task Start() { stop = new CancellationTokenSource(); - pooler = Task.Run(async () => + poller = Task.Run(async () => { - - while (!stop.Token.IsCancellationRequested) + var token = stop.Token; + while (!token.IsCancellationRequested) { try { @@ -64,7 +64,11 @@ public Task Start() UpdateQueueLengthStore(lookup); Logger.Debug("Waiting for next interval"); - await Task.Delay(QueryDelayInterval).ConfigureAwait(false); + await Task.Delay(QueryDelayInterval, token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // no-op } catch (Exception e) { @@ -105,7 +109,7 @@ public Task Stop() { stop.Cancel(); - return pooler; + return poller; } static TimeSpan QueryDelayInterval = TimeSpan.FromMilliseconds(200); From 37e8636e64202fce23a311fe5f3e25142485e8ab Mon Sep 17 00:00:00 2001 From: danielmarbach Date: Wed, 19 Dec 2018 17:25:48 +0100 Subject: [PATCH 6/7] RabbitMQ QueueLengthProvider graceful shutdown --- .../QueueLengthProvider.cs | 29 ++++++++++++------- 1 file changed, 19 insertions(+), 10 deletions(-) diff --git a/src/ServiceControl.Transports.RabbitMQ/QueueLengthProvider.cs b/src/ServiceControl.Transports.RabbitMQ/QueueLengthProvider.cs index 18b43d7..b6eec8c 100644 --- a/src/ServiceControl.Transports.RabbitMQ/QueueLengthProvider.cs +++ b/src/ServiceControl.Transports.RabbitMQ/QueueLengthProvider.cs @@ -43,17 +43,22 @@ public Task Start() { queryExecutor.Initialize(); - queryLoop = Task.Run(async () => + poller = Task.Run(async () => { - while (!stoppedTokenSource.Token.IsCancellationRequested) + var token = stoppedTokenSource.Token; + while (!token.IsCancellationRequested) { try { - await FetchQueueLengths(); + await FetchQueueLengths(token); UpdateQueueLengths(); - await Task.Delay(QueryDelayInterval).ConfigureAwait(false); + await Task.Delay(QueryDelayInterval, token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // no-op } catch (Exception e) { @@ -69,7 +74,7 @@ public async Task Stop() { stoppedTokenSource.Cancel(); - await queryLoop; + await poller; queryExecutor.Dispose(); } @@ -96,7 +101,7 @@ void UpdateQueueLengths() } } - async Task FetchQueueLengths() + async Task FetchQueueLengths(CancellationToken token) { foreach (var endpointQueuePair in endpointQueues) { @@ -114,7 +119,7 @@ await queryExecutor.Execute(m => { Logger.Warn($"Error querying queue length for {queueName}", e); } - }); + }, token); } } @@ -137,7 +142,7 @@ public void Initialize() connectionFactory = new ConnectionFactory(connectionConfiguration, null, false, false); } - public async Task Execute(Action action) + public async Task Execute(Action action, CancellationToken token) { try { @@ -149,7 +154,7 @@ public async Task Execute(Action action) //Connection implements reconnection logic while (connection.IsOpen == false) { - await Task.Delay(ReconnectionDelay); + await Task.Delay(ReconnectionDelay, token); } if (model == null || model.IsClosed) @@ -161,6 +166,10 @@ public async Task Execute(Action action) action(model); } + catch (OperationCanceledException) + { + // no-op + } catch (Exception e) { Logger.Warn("Error querying queue length.", e); @@ -179,7 +188,7 @@ public void Dispose() ConcurrentDictionary sizes = new ConcurrentDictionary(); CancellationTokenSource stoppedTokenSource = new CancellationTokenSource(); - Task queryLoop; + Task poller; QueueLengthStore store; QueryExecutor queryExecutor; From ede4858e5b7cb72022488bcbb0f7fdb3d51994d6 Mon Sep 17 00:00:00 2001 From: danielmarbach Date: Wed, 19 Dec 2018 17:27:22 +0100 Subject: [PATCH 7/7] Share connection between chunks --- .../QueueLengthProvider.cs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/ServiceControl.Transports.SQLServer/QueueLengthProvider.cs b/src/ServiceControl.Transports.SQLServer/QueueLengthProvider.cs index e0e0cf5..7864f7f 100644 --- a/src/ServiceControl.Transports.SQLServer/QueueLengthProvider.cs +++ b/src/ServiceControl.Transports.SQLServer/QueueLengthProvider.cs @@ -127,12 +127,12 @@ async Task QueryTableSizes(CancellationToken token) .Select(grp => grp.Select(g => g.i).ToArray()) .ToList(); - foreach (var chunk in chunks) + using (var connection = new SqlConnection(connectionString)) { - using (var connection = new SqlConnection(connectionString)) - { - await connection.OpenAsync(token).ConfigureAwait(false); + await connection.OpenAsync(token).ConfigureAwait(false); + foreach (var chunk in chunks) + { await UpdateChunk(connection, chunk, token).ConfigureAwait(false); } }