diff --git a/AzureBatchQueue.Tests/MessageQueueTests.cs b/AzureBatchQueue.Tests/MessageQueueTests.cs index 90a2739..785fba8 100644 --- a/AzureBatchQueue.Tests/MessageQueueTests.cs +++ b/AzureBatchQueue.Tests/MessageQueueTests.cs @@ -1,6 +1,12 @@ +using System.Text.Json; +using Azure; +using Azure.Storage.Blobs; +using Azure.Storage.Queues; +using Azure.Storage.Queues.Models; using AzureBatchQueue.JsonNet; using AzureBatchQueue.Tests.Serializers; using FluentAssertions; +using Microsoft.Extensions.Logging; using NUnit.Framework; namespace AzureBatchQueue.Tests; @@ -145,6 +151,8 @@ public async Task When_quarantine_small_message() public async Task When_quarantine_large_message() { using var queueTest = await Queue(maxDequeueCount: 0); // quarantine after first read + var logger = new RecordingLogger(); + queueTest.Queue.WithLogger(logger); var largeMessage = new string('*', 65 * 1024); await queueTest.Queue.Send(largeMessage); @@ -154,6 +162,9 @@ public async Task When_quarantine_large_message() var msgFromQuarantine = (await queueTest.Queue.ReceiveFromQuarantine()).Single(); msgFromQuarantine.Item.Should().Be(largeMessage); msgFromQuarantine.MessageId.BlobName.Should().NotBeEmpty(); + var copy = logger.Entries.Single(x => x.Name == "BlobCopiedForQuarantine"); + copy.Fields["DestBlob"].Should().Be(msgFromQuarantine.MessageId.BlobName); + copy.Fields["SourceBlob"].Should().Be(logger.Entries.Single(x => x.Name == "BlobDeleted").Fields["BlobName"]); } [Test] @@ -168,6 +179,172 @@ public async Task When_exception_on_deserialize() Assert.DoesNotThrowAsync(async () => await gZipQueue.Queue.Receive()); } + [Test] + public async Task When_receiving_a_missing_blob_returns_healthy_messages() + { + using var queueTest = await Queue(); + await queueTest.Queue.Send(new string('*', 65 * 1024)); + await queueTest.DeleteMessageBlob(); + var largeMessage = new string('h', 65 * 1024); + await queueTest.Queue.Send("healthy-inline"); + await queueTest.Queue.Send(largeMessage); + + var messages = await queueTest.Queue.Receive(3); + + messages.Select(x => x.Item).Should().BeEquivalentTo(new[] { "healthy-inline", largeMessage }); + (await queueTest.MessageCount()).Should().Be(2); + (await queueTest.MessageCount(fromQuarantine: true)).Should().Be(0); + } + + [Test] + public async Task When_quarantining_a_missing_blob_keeps_healthy_messages() + { + using var queueTest = await Queue(maxDequeueCount: 0); + await queueTest.Queue.Send(new string('*', 65 * 1024)); + await queueTest.DeleteMessageBlob(); + var largeMessage = new string('h', 65 * 1024); + await queueTest.Queue.Send("healthy-inline"); + await queueTest.Queue.Send(largeMessage); + + (await queueTest.Queue.Receive(3)).Should().BeEmpty(); + + (await queueTest.MessageCount()).Should().Be(0); + (await queueTest.MessageCount(fromQuarantine: true)).Should().Be(2); + var quarantined = await queueTest.Queue.ReceiveFromQuarantine(2); + quarantined.Select(x => x.Item).Should().BeEquivalentTo(new[] { "healthy-inline", largeMessage }); + } + + [TestCase(0)] + [TestCase(5)] + public async Task When_container_is_missing_keeps_message_for_retry(int maxDequeueCount) + { + using var queueTest = await Queue(maxDequeueCount); + await queueTest.Queue.Send(new string('*', 65 * 1024)); + await queueTest.Blobs.DeleteAsync(); + + if (maxDequeueCount == 0) + (await queueTest.Queue.Receive()).Should().BeEmpty(); + else + { + var error = Assert.ThrowsAsync(async () => await queueTest.Queue.Receive()); + error!.ErrorCode.Should().Be("ContainerNotFound"); + } + + (await queueTest.MessageCount()).Should().Be(1); + } + + [Test] + public async Task When_receiving_a_missing_blob_from_quarantine_keeps_message() + { + using var queueTest = await Queue(); + await queueTest.Queue.QuarantineData(new string('*', 65 * 1024), CancellationToken.None); + await queueTest.DeleteMessageBlob(fromQuarantine: true); + + var error = Assert.ThrowsAsync(async () => await queueTest.Queue.ReceiveFromQuarantine()); + + error!.ErrorCode.Should().Be("BlobNotFound"); + (await queueTest.MessageCount(fromQuarantine: true)).Should().Be(1); + (await queueTest.MessageCount()).Should().Be(0); + } + + [Test] + public async Task When_processing_a_blob_logs_its_message_and_lifecycle() + { + using var queueTest = await Queue(); + var logger = new RecordingLogger(); + queueTest.Queue.WithLogger(logger); + + await queueTest.Queue.Send(new string('*', 65 * 1024)); + var message = (await queueTest.Queue.Receive()).Single(); + await queueTest.Queue.DeleteMessage(message.MessageId); + + var events = logger.Entries.Where(x => Equals(x.Fields.GetValueOrDefault("BlobName"), message.MessageId.BlobName)).ToArray(); + events.Select(x => x.Name).Should().ContainInOrder("UploadingBlob", "BlobUploaded", "SendingMessage", "MessageSent", + "MessageReceived", "DeletingMessage", "MessageDeleted", "DeletingBlob", "BlobDeleted"); + events.Single(x => x.Name == "MessageSent").Fields["MessageId"].Should().Be(message.MessageId.Id); + events.Single(x => x.Name == "BlobDeleted").Fields["Deleted"].Should().Be(true); + } + + [TestCase(1024)] + [TestCase(65 * 1024)] + public async Task When_updating_a_blob_logs_the_payload_reference(int size) + { + using var queueTest = await Queue(); + var logger = new RecordingLogger(); + queueTest.Queue.WithLogger(logger); + await queueTest.Queue.Send(new string('*', 65 * 1024)); + var original = (await queueTest.Queue.Receive()).Single(); + + await queueTest.Queue.UpdateMessage(original.MessageId, new string('+', size)); + var updated = (await queueTest.Queue.Receive()).Single(); + + var change = logger.Entries.Single(x => x.Name == "MessageUpdated"); + change.Fields["MessageId"].Should().Be(original.MessageId.Id); + change.Fields["BlobName"].Should().Be(updated.MessageId.BlobName); + logger.Entries.Count(x => x.Name == "BlobDeleted").Should().Be(size == 1024 ? 1 : 0); + } + + [Test] + public async Task When_sending_fails_logs_the_blob_and_operation() + { + using var queueTest = await Queue(); + var logger = new RecordingLogger(); + queueTest.Queue.WithLogger(logger); + await queueTest.RawQueue.DeleteAsync(); + try + { + var error = Assert.ThrowsAsync(async () => await queueTest.Queue.Send(new string('*', 65 * 1024))); + + var failed = logger.Entries.Single(x => x.Name == "MessageSendFailed"); + failed.Fields["OperationId"].Should().Be(logger.Entries.Single(x => x.Name == "SendingMessage").Fields["OperationId"]); + failed.Fields["BlobName"].Should().Be(logger.Entries.Single(x => x.Name == "BlobUploaded").Fields["BlobName"]); + failed.Exception.Should().BeSameAs(error); + } + finally + { + await queueTest.Init(); + } + } + + [Test] + public async Task When_messages_share_a_blob_logs_both_ids_and_who_deletes_it() + { + using var queueTest = await Queue(); + var logger = new RecordingLogger(); + queueTest.Queue.WithLogger(logger); + await queueTest.Queue.Send(new string('*', 65 * 1024)); + var original = (await queueTest.RawQueue.PeekMessagesAsync(1)).Value.Single(); + await queueTest.RawQueue.SendMessageAsync(original.Body); + + var messages = await queueTest.Queue.Receive(2); + await queueTest.Queue.DeleteMessage(messages[0].MessageId); + + var received = logger.Entries.Where(x => x.Name == "MessageReceived").ToArray(); + received.Select(x => x.Fields["MessageId"]).Should().BeEquivalentTo(messages.Select(x => x.MessageId.Id)); + received.Select(x => x.Fields["BlobName"]).Should().OnlyContain(x => Equals(x, messages[0].MessageId.BlobName)); + logger.Entries.Single(x => x.Name == "BlobDeleted").Fields["MessageId"].Should().Be(messages[0].MessageId.Id); + + await queueTest.Queue.UpdateMessage(messages[1].MessageId, null); + (await queueTest.Queue.Receive()).Should().BeEmpty(); + logger.Entries.Single(x => x.Name == "MissingBlobMessageDeleted").Fields["MessageId"].Should().Be(messages[1].MessageId.Id); + } + + sealed record LogEntry(string Name, Dictionary Fields, Exception? Exception); + + sealed class RecordingLogger : ILogger + { + public System.Collections.Concurrent.ConcurrentQueue Entries { get; } = new(); + public bool IsEnabled(LogLevel logLevel) => true; + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + var fields = ((IEnumerable>)state!).ToDictionary(x => x.Key, x => x.Value); + Entries.Enqueue(new LogEntry(eventId.Name ?? "", fields, exception)); + } + } + static async Task> Queue(int maxDequeueCount = 5, IMessageQueueSerializer? serializer = null) { var queue = new QueueTest(maxDequeueCount, serializer); @@ -177,16 +354,40 @@ static async Task> Queue(int maxDequeueCount = 5, IMessageQueueS class QueueTest : IDisposable { + readonly QueueClient quarantineQueue; + public QueueTest(int maxDequeueCount = 5, IMessageQueueSerializer? serializer = null) { - Queue = new MessageQueue("UseDevelopmentStorage=true", "test", maxDequeueCount: maxDequeueCount, serializer: serializer); + const string connection = "UseDevelopmentStorage=true"; + Queue = new MessageQueue(connection, "test", maxDequeueCount: maxDequeueCount, serializer: serializer); + var options = new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 }; + RawQueue = new QueueClient(connection, "test", options); + quarantineQueue = new QueueClient(connection, "test-quarantine", options); + Blobs = new BlobContainerClient(connection, "overflow-test"); } public async Task Init() => await Queue.Init(); public MessageQueue Queue { get; } + public QueueClient RawQueue { get; } + public BlobContainerClient Blobs { get; } + + public async Task DeleteMessageBlob(bool fromQuarantine = false) + { + var queue = fromQuarantine ? quarantineQueue : RawQueue; + var message = (await queue.PeekMessagesAsync(1)).Value.Single(); + using var envelope = JsonDocument.Parse(message.Body.ToString()); + await Blobs.DeleteBlobAsync(envelope.RootElement.GetProperty("__MSQ_QUEUE_BLOBNAME__").GetString()!); + } + + public async Task MessageCount(bool fromQuarantine = false) + { + var queue = fromQuarantine ? quarantineQueue : RawQueue; + return (await queue.GetPropertiesAsync()).Value.ApproximateMessagesCount; + } + public void Dispose() { - Queue.Delete().GetAwaiter().GetResult(); + Task.WhenAll(Queue.Delete(), Blobs.DeleteIfExistsAsync()).GetAwaiter().GetResult(); } } diff --git a/AzureBatchQueue/MessageQueue.cs b/AzureBatchQueue/MessageQueue.cs index d7e64b1..17f96e4 100644 --- a/AzureBatchQueue/MessageQueue.cs +++ b/AzureBatchQueue/MessageQueue.cs @@ -1,4 +1,5 @@ using System.Text.Json.Serialization; +using Azure; using Azure.Storage.Blobs; using Azure.Storage.Queues; using Azure.Storage.Queues.Models; @@ -37,29 +38,36 @@ public MessageQueue(string connectionString, string queueName, public async Task Send(T item, TimeSpan? visibilityTimeout = null, CancellationToken ct = default) { - var (data, _) = await SerializeAndOffloadIfBig(item, ct: ct); - await queue.SendMessageAsync(data, visibilityTimeout ?? TimeSpan.Zero, null, ct); + var payload = await SerializeAndOffloadIfBig(item, ct: ct); + await SendPayload(queue, payload.Data, payload.BlobName, ct, visibilityTimeout ?? TimeSpan.Zero); } public async Task DeleteMessage(MessageId id, CancellationToken ct = default) { - await queue.DeleteMessageAsync(id.Id, id.PopReceipt, ct); + await DeleteQueueMessage(queue, id, ct); if (id.BlobName is not null) - await container.DeleteBlobIfExistsAsync(id.BlobName, cancellationToken: ct); + await DeleteBlob(id, ct); } public async Task UpdateMessage(MessageId id, T? item, TimeSpan visibilityTimeout = default, CancellationToken ct = default) { + logger.UpdatingMessage(id.Id, queue.Name, id.BlobName); if (item == null) { await queue.UpdateMessageAsync(id.Id, id.PopReceipt, (BinaryData?)null, visibilityTimeout, ct); + logger.MessageUpdated(id.Id, queue.Name, id.BlobName, id.BlobName); return; } - var (data, offloaded) = await SerializeAndOffloadIfBig(item, id.BlobName, ct); - await queue.UpdateMessageAsync(id.Id, id.PopReceipt, data, visibilityTimeout, ct); - if (!offloaded && id.BlobName != null) - await container.DeleteBlobAsync(id.BlobName, cancellationToken: ct); + var payload = await SerializeAndOffloadIfBig(item, id.BlobName, ct); + await queue.UpdateMessageAsync(id.Id, id.PopReceipt, payload.Data, visibilityTimeout, ct); + logger.MessageUpdated(id.Id, queue.Name, payload.BlobName, id.BlobName); + if (!payload.Offloaded && id.BlobName != null) + { + logger.DeletingBlob(id.BlobName, id.Id, queue.Name); + var deleted = await container.DeleteBlobAsync(id.BlobName, cancellationToken: ct); + logger.BlobDeleted(id.BlobName, id.Id, queue.Name, true, deleted.Headers.RequestId); + } } async Task SerializeAndOffloadIfBig(T item, string? blobName = null, CancellationToken ct = default) @@ -73,7 +81,9 @@ async Task SerializeAndOffloadIfBig(T item, string? blobName = null, Ca return Payload.FitsInMessage(stream); var blobRef = blobName != null ? BlobRef.Get(blobName) : BlobRef.Create(); - await container.GetBlobClient(blobRef.BlobName).UploadAsync(stream, overwrite: true, ct); + logger.UploadingBlob(blobRef.BlobName, queue.Name, stream.Length); + var uploaded = await container.GetBlobClient(blobRef.BlobName).UploadAsync(stream, overwrite: true, ct); + logger.BlobUploaded(blobRef.BlobName, queue.Name, uploaded.Value.ETag.ToString(), uploaded.GetRawResponse().Headers.RequestId); return Payload.OffloadedToBlob(blobRef); } @@ -85,6 +95,7 @@ public async Task[]> Receive(int maxMessages = MaxMessagesReceiv if (!r.HasValue) return Array.Empty>(); + LogReceivedMessages(queue, r.Value); var quarantineMessages = r.Value.Where(x => x.DequeueCount > maxDequeueCount).ToList(); var validMessages = r.Value.Except(quarantineMessages); @@ -103,6 +114,7 @@ public async Task[]> ReceiveFromQuarantine(int maxMessages = Max if (!r.HasValue) return Array.Empty>(); + LogReceivedMessages(quarantineQueue, r.Value); var queueMessages = await Task.WhenAll(r.Value.Select(x => ToQueueMessage(x, ct, fromQuarantine: true))); return queueMessages.Where(x => x != null).ToArray()!; @@ -114,29 +126,32 @@ async Task QuarantineMessage(QueueMessage queueMessage, CancellationToken ct = d try { BinaryData messageBody; + string? quarantineBlobName = null; if (IsBlobRef(queueMessage.Body, out blobRef) && blobRef != null) { - var newBlobRef = await CopyBlobForQuarantine(blobRef, queueMessage.MessageId, ct); + var newBlobRef = await TryCopyBlobForQuarantine(blobRef, queueMessage, ct); + if (newBlobRef == null) + return; + messageBody = Payload.OffloadedToBlob(newBlobRef).Data; + quarantineBlobName = newBlobRef.BlobName; } else { messageBody = queueMessage.Body; } - await quarantineQueue.SendMessageAsync(messageBody, cancellationToken: ct); + await SendPayload(quarantineQueue, messageBody, quarantineBlobName, ct); var messageId = new MessageId(queueMessage.MessageId, queueMessage.PopReceipt, blobRef?.BlobName); await DeleteMessage(messageId, ct); - logger.LogInformation("QueueMessage {msgId} with {popReceipt} was quarantined after {dequeueCount} unsuccessful attempts.", - queueMessage.MessageId, queueMessage.PopReceipt, queueMessage.DequeueCount); + logger.MessageQuarantined(queueMessage.MessageId, queueMessage.PopReceipt, queueMessage.DequeueCount); } catch (Exception ex) { - logger.LogError(ex, "Failed to quarantine queue message with {messageId}, {popReceipt}, and {blobName}.", - queueMessage.MessageId, queueMessage.PopReceipt, blobRef?.BlobName); + logger.QuarantineFailed(queueMessage.MessageId, queueMessage.PopReceipt, blobRef?.BlobName, ex); } } @@ -145,17 +160,30 @@ async Task QuarantineMessage(QueueMessage queueMessage, CancellationToken ct = d // In this case, later we will insert a duplicate message to quarantine, with the same ref to blob. // When one message is processed, we remove the blob. The other message will have a BlobNotFound error. // So it's better to do some more heavy-lifting and copy blob content on quarantine. - async Task CopyBlobForQuarantine(BlobRef sourceBlobRef, string messageId, CancellationToken ct) + async Task TryCopyBlobForQuarantine(BlobRef sourceBlobRef, QueueMessage message, CancellationToken ct) { var sourceBlobClient = container.GetBlobClient(sourceBlobRef.BlobName); var newBlobRef = BlobRef.Create(); var destinationBlobClient = container.GetBlobClient(newBlobRef.BlobName); - var content = await sourceBlobClient.DownloadContentAsync(ct); - await destinationBlobClient.UploadAsync(content.Value.Content, ct); + BinaryData content; + try + { + var downloaded = await sourceBlobClient.DownloadContentAsync(ct); + content = downloaded.Value.Content; + logger.BlobDownloaded(sourceBlobRef.BlobName, message.MessageId, queue.Name, + downloaded.Value.Details.ETag.ToString(), downloaded.GetRawResponse().Headers.RequestId); + } + catch (RequestFailedException ex) when (ex is { Status: 404, ErrorCode: "BlobNotFound" }) + { + await DeleteMissingBlobMessage(message, sourceBlobRef, ex, ct); + return null; + } + logger.UploadingBlob(newBlobRef.BlobName, queue.Name, content.ToMemory().Length); + var uploaded = await destinationBlobClient.UploadAsync(content, ct); + logger.BlobUploaded(newBlobRef.BlobName, queue.Name, uploaded.Value.ETag.ToString(), uploaded.GetRawResponse().Headers.RequestId); - logger.LogInformation("Copied blob {SourceBlob} to {DestBlob} for quarantine message {MsgId}", - sourceBlobRef.BlobName, newBlobRef.BlobName, messageId); + logger.BlobCopiedForQuarantine(sourceBlobRef.BlobName, newBlobRef.BlobName, message.MessageId); return newBlobRef; } @@ -169,10 +197,12 @@ public async Task Dequarantine(CancellationToken ct = default) if (!response.HasValue || response.Value.Length == 0) return; + LogReceivedMessages(quarantineQueue, response.Value); foreach (var msg in response.Value) { - await queue.SendMessageAsync(msg.Body, TimeSpan.Zero, cancellationToken: ct); - await quarantineQueue.DeleteMessageAsync(msg.MessageId, msg.PopReceipt, ct); + IsBlobRef(msg.Body, out var blobRef); + await SendPayload(queue, msg.Body, blobRef?.BlobName, ct, TimeSpan.Zero); + await DeleteQueueMessage(quarantineQueue, new MessageId(msg.MessageId, msg.PopReceipt, blobRef?.BlobName), ct); } } while (!ct.IsCancellationRequested); } @@ -180,7 +210,7 @@ public async Task Dequarantine(CancellationToken ct = default) public async Task QuarantineData(T item, CancellationToken ct) { var payload = await SerializeAndOffloadIfBig(item, ct: ct); - await quarantineQueue.SendMessageAsync(payload.Data, cancellationToken: ct); + await SendPayload(quarantineQueue, payload.Data, payload.BlobName, ct); } async Task?> ToQueueMessage(QueueMessage m, CancellationToken ct, bool fromQuarantine = false) @@ -192,10 +222,17 @@ public async Task QuarantineData(T item, CancellationToken ct) { var blobData = await container.GetBlobClient(blobRef!.BlobName).DownloadContentAsync(ct); payload = blobData.Value.Content; + logger.BlobDownloaded(blobRef.BlobName, m.MessageId, fromQuarantine ? quarantineQueue.Name : queue.Name, + blobData.Value.Details.ETag.ToString(), blobData.GetRawResponse().Headers.RequestId); + } + catch (RequestFailedException ex) when (!fromQuarantine && ex is { Status: 404, ErrorCode: "BlobNotFound" }) + { + await DeleteMissingBlobMessage(m, blobRef!, ex, ct); + return null; } catch (Exception ex) { - logger.LogError(ex, "Exception when loading blob {BlobName} for {MessageId}", blobRef!.BlobName, m.MessageId); + logger.BlobDownloadFailed(blobRef!.BlobName, m.MessageId, ex); throw; } } @@ -207,7 +244,7 @@ public async Task QuarantineData(T item, CancellationToken ct) } catch (Exception ex) { - logger.LogError(ex, "Exception when deserializing a message {MessageId}", m.MessageId); + logger.MessageDeserializationFailed(m.MessageId, ex); if (!fromQuarantine) await QuarantineMessage(m, ct); @@ -221,6 +258,55 @@ public async Task QuarantineData(T item, CancellationToken ct) return msg; } + async Task DeleteMissingBlobMessage(QueueMessage message, BlobRef blobRef, RequestFailedException error, CancellationToken ct) + { + await DeleteQueueMessage(queue, new MessageId(message.MessageId, message.PopReceipt, blobRef.BlobName), ct); + + logger.MissingBlobMessageDeleted(message.MessageId, queue.Name, blobRef.BlobName, error); + } + + async Task SendPayload(QueueClient target, BinaryData data, string? blobName, CancellationToken ct, TimeSpan? visibilityTimeout = null) + { + var operationId = Guid.NewGuid(); + logger.SendingMessage(target.Name, blobName, operationId); + try + { + var sent = await target.SendMessageAsync(data, visibilityTimeout, cancellationToken: ct); + logger.MessageSent(sent.Value.MessageId, target.Name, blobName, operationId, sent.GetRawResponse().Headers.RequestId); + } + catch (RequestFailedException ex) + { + logger.MessageSendFailed(ex, target.Name, blobName, operationId, ex.Status, ex.ErrorCode, ex.GetRawResponse()?.Headers.RequestId); + throw; + } + } + + async Task DeleteQueueMessage(QueueClient source, MessageId id, CancellationToken ct) + { + logger.DeletingMessage(id.Id, source.Name, id.BlobName); + await source.DeleteMessageAsync(id.Id, id.PopReceipt, ct); + logger.MessageDeleted(id.Id, source.Name, id.BlobName); + } + + async Task DeleteBlob(MessageId id, CancellationToken ct) + { + logger.DeletingBlob(id.BlobName!, id.Id, queue.Name); + var deleted = await container.DeleteBlobIfExistsAsync(id.BlobName!, cancellationToken: ct); + logger.BlobDeleted(id.BlobName!, id.Id, queue.Name, deleted.Value, deleted.GetRawResponse().Headers.RequestId); + } + + void LogReceivedMessages(QueueClient source, QueueMessage[] messages) + { + if (!logger.IsEnabled(LogLevel.Information)) + return; + + foreach (var message in messages) + { + IsBlobRef(message.Body, out var blobRef); + logger.MessageReceived(message.MessageId, source.Name, blobRef?.BlobName, message.DequeueCount); + } + } + static bool IsBlobRef(BinaryData data, out BlobRef? o) { try @@ -239,12 +325,14 @@ static bool IsBlobRef(BinaryData data, out BlobRef? o) public async Task Delete(bool deleteBlobs = false) { + logger.DeletingQueueStorage(queue.Name, deleteBlobs); var tasks = new List { queue.DeleteAsync(), quarantineQueue.DeleteAsync() }; if (deleteBlobs) tasks.Add(container.DeleteAsync()); await Task.WhenAll(tasks); + logger.QueueStorageDeleted(queue.Name, deleteBlobs); } public async Task ClearMessages() => await queue.ClearMessagesAsync(); @@ -262,13 +350,13 @@ record BlobRef([property: JsonPropertyName("__MSQ_QUEUE_BLOBNAME__")] string Blo static string FileName() => $"{DateTime.UtcNow:yyyy-MM-dd}/{DateTime.UtcNow:s}{Guid.NewGuid():N}.json.gzip"; } - record Payload(BinaryData Data, bool Offloaded) + record Payload(BinaryData Data, bool Offloaded, string? BlobName = null) { public static Payload FitsInMessage(MemoryStream stream) => new(BinaryData.FromStream(stream), false); public static Payload OffloadedToBlob(BlobRef blobRef) => - new(BinaryData.FromObjectAsJson(blobRef), true); + new(BinaryData.FromObjectAsJson(blobRef), true, blobRef.BlobName); } } diff --git a/AzureBatchQueue/QueueLog.cs b/AzureBatchQueue/QueueLog.cs new file mode 100644 index 0000000..9a3b836 --- /dev/null +++ b/AzureBatchQueue/QueueLog.cs @@ -0,0 +1,97 @@ +using Microsoft.Extensions.Logging; + +namespace AzureBatchQueue; + +internal static partial class QueueLog +{ + [LoggerMessage(1001, LogLevel.Warning, "Deleted message {MessageId} from queue {QueueName} because blob {BlobName} was not found.")] + public static partial void MissingBlobMessageDeleted(this ILogger logger, string messageId, string queueName, string blobName, Exception exception); + + [LoggerMessage(1002, LogLevel.Information, "QueueMessage {msgId} with {popReceipt} was quarantined after {dequeueCount} unsuccessful attempts.")] + public static partial void MessageQuarantined(this ILogger logger, string msgId, string popReceipt, long dequeueCount); + + [LoggerMessage(1003, LogLevel.Error, "Failed to quarantine queue message with {messageId}, {popReceipt}, and {blobName}.")] + public static partial void QuarantineFailed(this ILogger logger, string messageId, string popReceipt, string? blobName, Exception exception); + + [LoggerMessage(1004, LogLevel.Information, "Copied blob {SourceBlob} to {DestBlob} for quarantine message {MsgId}")] + public static partial void BlobCopiedForQuarantine(this ILogger logger, string sourceBlob, string destBlob, string msgId); + + [LoggerMessage(1005, LogLevel.Error, "Exception when loading blob {BlobName} for {MessageId}")] + public static partial void BlobDownloadFailed(this ILogger logger, string blobName, string messageId, Exception exception); + + [LoggerMessage(1006, LogLevel.Error, "Exception when deserializing a message {MessageId}")] + public static partial void MessageDeserializationFailed(this ILogger logger, string messageId, Exception exception); + + [LoggerMessage(1007, LogLevel.Warning, "Queue {queueName} was not found when flushing {messageId} with {itemsCount} items left.")] + public static partial void QueueNotFoundOnFlush(this ILogger logger, Exception exception, string queueName, MessageId messageId, int itemsCount); + + [LoggerMessage(1008, LogLevel.Error, "Accessing already flushed message with {messageId}.")] + public static partial void FlushedMessageAccessed(this ILogger logger, Exception exception, MessageId messageId); + + [LoggerMessage(1009, LogLevel.Error, "Unhandled exception when flushing message {messageId}.")] + public static partial void FlushFailed(this ILogger logger, Exception exception, MessageId messageId); + + [LoggerMessage(1010, LogLevel.Warning, "Message {MsgId} was not fully processed within a timeout ({FlushPeriod}) sec in queue {QueueName}." + + " {RemainingCount} items were not completed ({NotProcessed} not processed on time and {FailedCount} failed) from {TotalCount} total")] + public static partial void BatchIncomplete(this ILogger logger, MessageId msgId, double flushPeriod, string queueName, + int remainingCount, int notProcessed, int failedCount, int totalCount); + + [LoggerMessage(1011, LogLevel.Debug, "Message {msgId} was fully processed with {totalCount} items")] + public static partial void BatchProcessed(this ILogger logger, MessageId msgId, int totalCount); + + [LoggerMessage(1012, LogLevel.Information, "Message {MsgId} was quarantined after {DequeueCount} unsuccessful attempts in queue {QueueName}." + + " {RemainingCount} items were not completed ({NotProcessed} not processed on time and {FailedCount} failed) from {TotalCount} total")] + public static partial void BatchQuarantined(this ILogger logger, MessageId msgId, long dequeueCount, string queueName, + int remainingCount, int notProcessed, int failedCount, int totalCount); + + [LoggerMessage(1013, LogLevel.Error, "VisibilityTimeout {VisibilityTimeout} cannot be less that 1 sec." + + " UtcNow: {UtcNow}, MessageVisibleTime: {MessageVisibleTime}, InsertedOn: {InsertedOn}")] + public static partial void InvalidVisibilityTimeout(this ILogger logger, TimeSpan visibilityTimeout, string utcNow, + string messageVisibleTime, string insertedOn); + + [LoggerMessage(1101, LogLevel.Information, "Uploading blob {BlobName} for queue {QueueName} with {ContentLength} bytes")] + public static partial void UploadingBlob(this ILogger logger, string blobName, string queueName, long contentLength); + + [LoggerMessage(1102, LogLevel.Information, "Uploaded blob {BlobName} for queue {QueueName}; ETag {ETag}, Azure request {RequestId}")] + public static partial void BlobUploaded(this ILogger logger, string blobName, string queueName, string eTag, string? requestId); + + [LoggerMessage(1103, LogLevel.Information, "Sending message to queue {QueueName} referencing blob {BlobName}; operation {OperationId}")] + public static partial void SendingMessage(this ILogger logger, string queueName, string? blobName, Guid operationId); + + [LoggerMessage(1104, LogLevel.Information, "Sent message {MessageId} to queue {QueueName} referencing blob {BlobName}; operation {OperationId}, Azure request {RequestId}")] + public static partial void MessageSent(this ILogger logger, string messageId, string queueName, string? blobName, Guid operationId, string? requestId); + + [LoggerMessage(1105, LogLevel.Information, "Received message {MessageId} from queue {QueueName} referencing blob {BlobName}; dequeue count {DequeueCount}")] + public static partial void MessageReceived(this ILogger logger, string messageId, string queueName, string? blobName, long dequeueCount); + + [LoggerMessage(1106, LogLevel.Information, "Deleting message {MessageId} from queue {QueueName} referencing blob {BlobName}")] + public static partial void DeletingMessage(this ILogger logger, string messageId, string queueName, string? blobName); + + [LoggerMessage(1107, LogLevel.Information, "Deleted message {MessageId} from queue {QueueName} referencing blob {BlobName}")] + public static partial void MessageDeleted(this ILogger logger, string messageId, string queueName, string? blobName); + + [LoggerMessage(1108, LogLevel.Information, "Deleting blob {BlobName} for message {MessageId} in queue {QueueName}")] + public static partial void DeletingBlob(this ILogger logger, string blobName, string messageId, string queueName); + + [LoggerMessage(1109, LogLevel.Information, "Deleted blob {BlobName} for message {MessageId} in queue {QueueName}; deleted {Deleted}, Azure request {RequestId}")] + public static partial void BlobDeleted(this ILogger logger, string blobName, string messageId, string queueName, bool deleted, string? requestId); + + [LoggerMessage(1110, LogLevel.Information, "Updating message {MessageId} in queue {QueueName} referencing blob {BlobName}")] + public static partial void UpdatingMessage(this ILogger logger, string messageId, string queueName, string? blobName); + + [LoggerMessage(1111, LogLevel.Information, "Updated message {MessageId} in queue {QueueName} referencing blob {BlobName}; previous blob {PreviousBlobName}")] + public static partial void MessageUpdated(this ILogger logger, string messageId, string queueName, string? blobName, string? previousBlobName); + + [LoggerMessage(1112, LogLevel.Error, "Failed sending message to queue {QueueName} referencing blob {BlobName}; operation {OperationId}, status {Status}, code {ErrorCode}, Azure request {RequestId}")] + public static partial void MessageSendFailed(this ILogger logger, Exception exception, string queueName, string? blobName, + Guid operationId, int status, string? errorCode, string? requestId); + + [LoggerMessage(1113, LogLevel.Information, "Deleting storage for queue {QueueName}; delete blob container {DeleteBlobs}")] + public static partial void DeletingQueueStorage(this ILogger logger, string queueName, bool deleteBlobs); + + [LoggerMessage(1114, LogLevel.Information, "Deleted storage for queue {QueueName}; delete blob container {DeleteBlobs}")] + public static partial void QueueStorageDeleted(this ILogger logger, string queueName, bool deleteBlobs); + + [LoggerMessage(1115, LogLevel.Information, "Downloaded blob {BlobName} for message {MessageId} from queue {QueueName}; ETag {ETag}, Azure request {RequestId}")] + public static partial void BlobDownloaded(this ILogger logger, string blobName, string messageId, string queueName, string eTag, string? requestId); +} diff --git a/AzureBatchQueue/TimerBatch.cs b/AzureBatchQueue/TimerBatch.cs index a1edddb..36710a0 100644 --- a/AzureBatchQueue/TimerBatch.cs +++ b/AzureBatchQueue/TimerBatch.cs @@ -58,16 +58,15 @@ async Task Flush() } catch (Azure.RequestFailedException ex) when (ex.ErrorCode == "QueueNotFound") { - logger.LogWarning(ex, "Queue {queueName} was not found when flushing {messageId} with {itemsCount} items left.", - batchQueue.Name, msg.MessageId, items.NotProcessedCount()); + logger.QueueNotFoundOnFlush(ex, batchQueue.Name, msg.MessageId, items.NotProcessedCount()); } catch (Azure.RequestFailedException ex) when (ex.ErrorCode == "MessageNotFound") { - logger.LogError(ex, "Accessing already flushed message with {messageId}.", msg.MessageId); + logger.FlushedMessageAccessed(ex, msg.MessageId); } catch (Exception ex) { - logger.LogError(ex, "Unhandled exception when flushing message {messageId}.", msg.MessageId); + logger.FlushFailed(ex, msg.MessageId); throw; } @@ -95,8 +94,7 @@ async Task Update() var remaining = Remaining(); await batchQueue.UpdateMessage(msg.MessageId, remaining); - logger.LogWarning("Message {MsgId} was not fully processed within a timeout ({FlushPeriod}) sec in queue {QueueName}." + - " {RemainingCount} items were not completed ({NotProcessed} not processed on time and {FailedCount} failed) from {TotalCount} total", + logger.BatchIncomplete( msg.MessageId, FlushPeriod.TotalSeconds, batchQueue.Name, @@ -110,7 +108,7 @@ async Task Delete() { await batchQueue.DeleteMessage(msg.MessageId); - logger.LogDebug("Message {msgId} was fully processed with {totalCount} items", msg.MessageId, items.Items().Length); + logger.BatchProcessed(msg.MessageId, items.Items().Length); } async Task Quarantine() @@ -118,8 +116,7 @@ async Task Quarantine() var remaining = Remaining(); await batchQueue.QuarantineData(msg.MessageId, remaining); - logger.LogInformation("Message {MsgId} was quarantined after {DequeueCount} unsuccessful attempts in queue {QueueName}." + - " {RemainingCount} items were not completed ({NotProcessed} not processed on time and {FailedCount} failed) from {TotalCount} total", + logger.BatchQuarantined( msg.MessageId, msg.Metadata.DequeueCount, batchQueue.Name, @@ -197,9 +194,8 @@ TimeSpan CalculateFlushPeriod(TimeSpan visibilityTimeout) if (visibilityTimeout >= TimeSpan.FromSeconds(1)) return visibilityTimeout.Subtract(TimeSpan.FromMilliseconds(100)); - logger.LogError("VisibilityTimeout {VisibilityTimeout} cannot be less that 1 sec." + - " UtcNow: {UtcNow}, MessageVisibleTime: {MessageVisibleTime}, InsertedOn: {InsertedOn}", - visibilityTimeout, DateTimeOffset.UtcNow.ToString("O"), Metadata.VisibilityTime.ToString("O"), Metadata.InsertedOn.ToString("O")); + logger.InvalidVisibilityTimeout(visibilityTimeout, DateTimeOffset.UtcNow.ToString("O"), + Metadata.VisibilityTime.ToString("O"), Metadata.InsertedOn.ToString("O")); throw new ArgumentOutOfRangeException(nameof(visibilityTimeout)); }