From 6a575ff3dd8de781256730c22d25e62963e0f7b6 Mon Sep 17 00:00:00 2001 From: Tomasz Masternak Date: Fri, 15 Feb 2019 14:56:59 +0100 Subject: [PATCH 1/2] dedicated class for all queue length specific variables --- .../QueueLengthProvider.cs | 63 +++++++++++-------- 1 file changed, 38 insertions(+), 25 deletions(-) diff --git a/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs b/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs index d11adf7..4598aab 100644 --- a/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs +++ b/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs @@ -16,9 +16,8 @@ public class QueueLengthProvider : IProvideQueueLength { - ConcurrentDictionary queues = new ConcurrentDictionary(); - ConcurrentDictionary sizes = new ConcurrentDictionary(); - ConcurrentDictionary problematicQueues = new ConcurrentDictionary(); + ConcurrentDictionary queueLengths = new ConcurrentDictionary(); + ConcurrentDictionary problematicQueuesNames = new ConcurrentDictionary(); string connectionString; QueueLengthStore store; @@ -38,19 +37,23 @@ public void Process(EndpointInstanceId endpointInstanceId, EndpointMetadataRepor var queueName = QueueNameSanitizer.Sanitize(metadataReport.LocalAddress); var queueClient = CloudStorageAccount.Parse(connectionString).CreateCloudQueueClient(); - var queue = queueClient.GetQueueReference(queueName); - queues.AddOrUpdate(endpointInputQueue, _ => queue, (_, currentQueue) => + var queueLength = new QueueLengthValue { - if (currentQueue.Name.Equals(queue.Name) == false) + QueueName = queueName, + Length = 0, + QueueReference = queueClient.GetQueueReference(queueName) + }; + + queueLengths.AddOrUpdate(endpointInputQueue, _ => queueLength, (_, currentLength) => + { + if (currentLength.QueueName.Equals(queueLength.QueueName)) { - sizes.TryRemove(currentQueue, out var _); + return currentLength; } - return queue; + return queueLength; }); - - sizes.TryAdd(queue, 0); } public void Process(EndpointInstanceId endpointInstanceId, TaggedLongValueOccurrence metricsReport) @@ -100,28 +103,31 @@ void UpdateQueueLengthStore() { var nowTicks = DateTime.UtcNow.Ticks; - foreach (var tableNamePair in queues) + foreach (var endpointQueueLengthPair in queueLengths) { - store.Store( - new[]{ new RawMessage.Entry - { - DateTicks = nowTicks, - Value = sizes.TryGetValue(tableNamePair.Value, out var size) ? size : 0 - }}, - tableNamePair.Key); + var queueLengthEntry = new RawMessage.Entry + { + DateTicks = nowTicks, + Value = endpointQueueLengthPair.Value.Length + }; + + store.Store(new[]{ queueLengthEntry }, endpointQueueLengthPair.Key); } } - Task FetchQueueSizes(CancellationToken token) => Task.WhenAll(sizes.Select(kvp => FetchLength(kvp.Key, token))); + Task FetchQueueSizes(CancellationToken token) => Task.WhenAll(queueLengths.Select(kvp => FetchLength(kvp.Value, token))); - async Task FetchLength(CloudQueue queue, CancellationToken token) + async Task FetchLength(QueueLengthValue queueLength, CancellationToken token) { try { - await queue.FetchAttributesAsync(token).ConfigureAwait(false); - sizes[queue] = queue.ApproximateMessageCount.GetValueOrDefault(); + var queueReference = queueLength.QueueReference; - problematicQueues.TryRemove(queue, out _); + await queueReference.FetchAttributesAsync(token).ConfigureAwait(false); + + queueLength.Length = queueReference.ApproximateMessageCount.GetValueOrDefault(); + + problematicQueuesNames.TryRemove(queueLength.QueueName, out _); } catch (OperationCanceledException) { @@ -130,13 +136,20 @@ async Task FetchLength(CloudQueue queue, CancellationToken token) catch (Exception ex) { // simple "log once" approach to do not flood logs - if (problematicQueues.TryAdd(queue, queue)) + if (problematicQueuesNames.TryAdd(queueLength.QueueName, queueLength.QueueName)) { - Logger.Error($"Obtaining Azure Storage Queue count failed for '{queue.Name}'", ex); + Logger.Error($"Obtaining Azure Storage Queue count failed for '{queueLength.QueueName}'", ex); } } } + class QueueLengthValue + { + public string QueueName; + public volatile int Length; + public CloudQueue QueueReference; + } + static TimeSpan QueryDelayInterval = TimeSpan.FromMilliseconds(200); static ILog Logger = LogManager.GetLogger(); } From 50cc8fa4af9b169c9d191f98556a3614ad7c41a4 Mon Sep 17 00:00:00 2001 From: Sean Feldman Date: Tue, 19 Feb 2019 13:36:31 -0700 Subject: [PATCH 2/2] Simplify the logic of adding/updating a queue Length Co-authored-by: Tomasz Masternak Co-authored-by: Sean Feldman --- .../QueueLengthProvider.cs | 12 ++---------- 1 file changed, 2 insertions(+), 10 deletions(-) diff --git a/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs b/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs index 4598aab..c8655ef 100644 --- a/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs +++ b/src/ServiceControl.Transports.AzureStorageQueues/QueueLengthProvider.cs @@ -38,22 +38,14 @@ public void Process(EndpointInstanceId endpointInstanceId, EndpointMetadataRepor var queueClient = CloudStorageAccount.Parse(connectionString).CreateCloudQueueClient(); - var queueLength = new QueueLengthValue + var emptyQueueLength = new QueueLengthValue { QueueName = queueName, Length = 0, QueueReference = queueClient.GetQueueReference(queueName) }; - queueLengths.AddOrUpdate(endpointInputQueue, _ => queueLength, (_, currentLength) => - { - if (currentLength.QueueName.Equals(queueLength.QueueName)) - { - return currentLength; - } - - return queueLength; - }); + queueLengths.AddOrUpdate(endpointInputQueue, _ => emptyQueueLength, (_, existingQueueLength) => existingQueueLength); } public void Process(EndpointInstanceId endpointInstanceId, TaggedLongValueOccurrence metricsReport)