Add structured abandonment details to Azure Storage logs

Test/DurableTask.AzureStorage.Tests/LoggingTests.cs

@@ -0,0 +1,219 @@
+//  ----------------------------------------------------------------------------------
+//  Copyright Microsoft Corporation
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//  http://www.apache.org/licenses/LICENSE-2.0
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//  ----------------------------------------------------------------------------------
+
+namespace DurableTask.AzureStorage.Tests
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Diagnostics.Tracing;
+    using System.Linq;
+    using System.Reflection;
+    using DurableTask.AzureStorage.Logging;
+    using DurableTask.Core.Logging;
+    using Microsoft.Extensions.Logging;
+    using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+    [TestClass]
+    public class LoggingTests
+    {
+        [TestMethod]
+        public void AbandoningMessage_HasExpectedStructuredFieldsAndPreservesMessage()
+        {
+            var logEvent = new LogEvents.AbandoningMessage(
+                "test-account",
+                "test-hub",
+                "TaskScheduled",
+                42,
+                "message-id",
+                "instance-id",
+                "execution-id",
+                "control-queue",
+                17,
+                "pop-receipt",
+                30,
+                "The activity work item could not be processed.");
+
+            var fields = (IReadOnlyDictionary<string, object>)logEvent;
+            Assert.AreEqual("test-account", fields["Account"]);
+            Assert.AreEqual("test-hub", fields["TaskHub"]);
+            Assert.AreEqual("TaskScheduled", fields["EventType"]);
+            Assert.AreEqual(42, fields["TaskEventId"]);
+            Assert.AreEqual("message-id", fields["MessageId"]);
+            Assert.AreEqual("instance-id", fields["InstanceId"]);
+            Assert.AreEqual("execution-id", fields["ExecutionId"]);
+            Assert.AreEqual("control-queue", fields["PartitionId"]);
+            Assert.AreEqual(17L, fields["SequenceNumber"]);
+            Assert.AreEqual("pop-receipt", fields["PopReceipt"]);
+            Assert.AreEqual(30, fields["VisibilityTimeoutSeconds"]);
+            Assert.AreEqual("The activity work item could not be processed.", fields["Details"]);
+            Assert.AreEqual(LogLevel.Warning, logEvent.Level);
+            Assert.AreEqual(EventIds.AbandoningMessage, logEvent.EventId.Id);
+            Assert.AreEqual(nameof(EventIds.AbandoningMessage), logEvent.EventId.Name);
+            Assert.AreEqual(
+                "instance-id: Abandoning [TaskScheduled#42] message back to control-queue and setting a visibility delay of 30ms",
+                ((ILogEvent)logEvent).FormattedMessage);
+        }
+
+        [TestMethod]
+        public void AbandoningMessage_EventSourceSchemaAppendsDetailsAndUsesVersionEight()
+        {
+            MethodInfo method = typeof(AnalyticsEventSource).GetMethod(nameof(AnalyticsEventSource.AbandoningMessage));
+            EventAttribute eventAttribute = method.GetCustomAttribute<EventAttribute>();
+            string[] parameterNames = method.GetParameters().Select(parameter => parameter.Name).ToArray();
+
+            Assert.AreEqual(8, eventAttribute.Version);
+            CollectionAssert.AreEqual(
+                new[]
+                {
+                    "Account",
+                    "TaskHub",
+                    "EventType",
+                    "TaskEventId",
+                    "MessageId",
+                    "InstanceId",
+                    "ExecutionId",
+                    "PartitionId",
+                    "SequenceNumber",
+                    "PopReceipt",
+                    "VisibilityTimeoutSeconds",
+                    "AppName",
+                    "ExtensionVersion",
+                    "Details",
+                },
+                parameterNames);
+        }
+
+        [TestMethod]
+        public void AbandoningMessage_WriteEventSourceWritesDetailsToFinalPayloadSlot()
+        {
+            const string details = "The dispatcher abandoned the work item.";
+            const string messageId = "event-source-test-message-id";
+            var logEvent = new LogEvents.AbandoningMessage(
+                "test-account",
+                "test-hub",
+                "TaskScheduled",
+                42,
+                messageId,
+                "instance-id",
+                "execution-id",
+                "control-queue",
+                17,
+                "pop-receipt",
+                30,
+                details);
+
+            using (var listener = new AbandoningMessageEventListener(messageId))
+            {
+                listener.Enable();
+                ((IEventSourceEvent)logEvent).WriteEventSource();
+
+                Assert.AreEqual(EventIds.AbandoningMessage, listener.EventId);
+                Assert.AreEqual("Details", listener.PayloadNames.Last());
+                Assert.AreEqual(Utils.ExtensionVersion, listener.Payload[listener.Payload.Count - 2]);
+                Assert.AreEqual(details, listener.Payload.Last());
+            }
+        }
+
+        [TestMethod]
+        public void AbandoningMessage_LogHelperPropagatesDetails()
+        {
+            var logger = new CapturingLogger();
+            var logHelper = new LogHelper(logger);
+
+            logHelper.AbandoningMessage(
+                "test-account",
+                "test-hub",
+                "TaskScheduled",
+                42,
+                "message-id",
+                "instance-id",
+                "execution-id",
+                "control-queue",
+                17,
+                "pop-receipt",
+                30,
+                "The orchestration work item could not be processed.");
+
+            var fields = (IReadOnlyDictionary<string, object>)logger.State;
+            Assert.AreEqual("The orchestration work item could not be processed.", fields["Details"]);
+        }
+
+        sealed class CapturingLogger : ILogger
+        {
+            public object State { get; private set; }
+
+            public IDisposable BeginScope<TState>(TState state) => NullScope.Instance;
+
+            public bool IsEnabled(LogLevel logLevel) => true;
+
+            public void Log<TState>(
+                LogLevel logLevel,
+                EventId eventId,
+                TState state,
+                Exception exception,
+                Func<TState, Exception, string> formatter)
+            {
+                this.State = state;
+            }
+        }
+
+        sealed class AbandoningMessageEventListener : EventListener
+        {
+            readonly string messageId;
+
+            public AbandoningMessageEventListener(string messageId)
+            {
+                this.messageId = messageId;
+            }
+
+            public int EventId { get; private set; } = -1;
+
+            public IReadOnlyList<string> PayloadNames { get; private set; } = Array.Empty<string>();
+
+            public IReadOnlyList<object> Payload { get; private set; } = Array.Empty<object>();
+
+            public void Enable()
+            {
+                this.EnableEvents(AnalyticsEventSource.Log, EventLevel.Verbose);
+            }
+
+            public override void Dispose()
+            {
+                this.DisableEvents(AnalyticsEventSource.Log);
+                base.Dispose();
+            }
+
+            protected override void OnEventWritten(EventWrittenEventArgs eventData)
+            {
+                if (eventData.EventSource == AnalyticsEventSource.Log &&
+                    eventData.EventId == EventIds.AbandoningMessage &&
+                    eventData.Payload.Count == 14 &&
+                    Equals(eventData.Payload[4], this.messageId))
+                {
+                    this.EventId = eventData.EventId;
+                    this.PayloadNames = eventData.PayloadNames.ToArray();
+                    this.Payload = eventData.Payload.ToArray();
+                }
+            }
+        }
+
+        sealed class NullScope : IDisposable
+        {
+            public static readonly NullScope Instance = new NullScope();
+
+            public void Dispose()
+            {
+            }
+        }
+    }
+}

src/DurableTask.AzureStorage/AnalyticsEventSource.cs

@@ -165,7 +165,7 @@ public void DeletingMessage(
                 ExtensionVersion);
         }
 
-        [Event(EventIds.AbandoningMessage, Level = EventLevel.Warning, Version = 7)]
+        [Event(EventIds.AbandoningMessage, Level = EventLevel.Warning, Version = 8)]
         public void AbandoningMessage(
             string Account,
             string TaskHub,
@@ -179,7 +179,8 @@ public void AbandoningMessage(
             string PopReceipt,
             int VisibilityTimeoutSeconds,
             string AppName,
-            string ExtensionVersion)
+            string ExtensionVersion,
+            string Details)
         {
             this.WriteEvent(
                 EventIds.AbandoningMessage,
@@ -195,7 +196,8 @@ public void AbandoningMessage(
                 PopReceipt ?? string.Empty,
                 VisibilityTimeoutSeconds,
                 AppName,
-                ExtensionVersion);
+                ExtensionVersion,
+                Details);
         }
 
         [Event(EventIds.AssertFailure, Level = EventLevel.Warning, Message = "An unexpected condition was detected: {2}", Version = 2)]

src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs

@@ -726,7 +726,9 @@ async Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAsync(boo
                     // Make sure we still own the partition. If not, abandon the session.
                     if (session.ControlQueue.IsReleased)
                     {
-                        await this.AbandonAndReleaseSessionAsync(session);
+                        await this.AbandonAndReleaseSessionAsync(
+                            session,
+                            "The control queue was released.");
                         return null;
                     }
 
@@ -771,13 +773,18 @@ async Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAsync(boo
                     if (outOfOrderMessages?.Count > 0)
                     {
                         // This will also remove the messages from the current batch.
-                        await this.AbandonMessagesAsync(session, outOfOrderMessages);
+                        await this.AbandonMessagesAsync(
+                            session,
+                            outOfOrderMessages,
+                            "Message was received out of order.");
                     }
 
                     if (session.CurrentMessageBatch.Count == 0)
                     {
                         // All messages were removed. Release the work item.
-                        await this.AbandonAndReleaseSessionAsync(session);
+                        await this.AbandonAndReleaseSessionAsync(
+                            session,
+                            "No processable messages remained in the session.");
                         return null;
                     }
 
@@ -870,7 +877,9 @@ async Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAsync(boo
                     if (session != null)
                     {
                         // host is shutting down - release any queued messages
-                        await this.AbandonAndReleaseSessionAsync(session);
+                        await this.AbandonAndReleaseSessionAsync(
+                            session,
+                            "Message processing was canceled during shutdown or listener cancellation.");
                     }
 
                     return null;
@@ -1125,11 +1134,11 @@ await this.trackingStore.UpdateInstanceStatusForCompletedOrchestrationAsync(
             return null;
         }
 
-        async Task AbandonAndReleaseSessionAsync(OrchestrationSession session)
+        async Task AbandonAndReleaseSessionAsync(OrchestrationSession session, string details)
         {
             try
             {
-                await this.AbandonSessionAsync(session);
+                await this.AbandonSessionAsync(session, details);
             }
             finally
             {
@@ -1488,20 +1497,25 @@ public Task AbandonTaskOrchestrationWorkItemAsync(TaskOrchestrationWorkItem work
                 return Utils.CompletedTask;
             }
 
-            return this.AbandonSessionAsync(session);
+            return this.AbandonSessionAsync(
+                session,
+                "The orchestration work item was abandoned by the dispatcher.");
         }
 
-        Task AbandonSessionAsync(OrchestrationSession session)
+        Task AbandonSessionAsync(OrchestrationSession session, string details)
         {
             session.StartNewLogicalTraceScope();
-            return this.AbandonMessagesAsync(session, session.CurrentMessageBatch.ToList());
+            return this.AbandonMessagesAsync(session, session.CurrentMessageBatch.ToList(), details);
         }
 
-        async Task AbandonMessagesAsync(OrchestrationSession session, IList<MessageData> messages)
+        async Task AbandonMessagesAsync(
+            OrchestrationSession session,
+            IList<MessageData> messages,
+            string details)
         {
             await messages.ParallelForEachAsync(
                 this.settings.MaxStorageOperationConcurrency,
-                message => session.ControlQueue.AbandonMessageAsync(message, session));
+                message => session.ControlQueue.AbandonMessageAsync(message, details, session));
 
             // Remove the messages from the current batch. The remaining messages
             // may still be able to be processed
@@ -1680,7 +1694,10 @@ public async Task AbandonTaskActivityWorkItemAsync(TaskActivityWorkItem workItem
 
             session.StartNewLogicalTraceScope();
 
-            await this.workItemQueue.AbandonMessageAsync(session.MessageData, session);
+            await this.workItemQueue.AbandonMessageAsync(
+                session.MessageData,
+                "The activity work item was abandoned by the dispatcher.",
+                session);
 
             if (this.activeActivitySessions.TryRemove(workItem.Id, out _))
             {

src/DurableTask.AzureStorage/Logging/LogEvents.cs

@@ -336,7 +336,8 @@ public AbandoningMessage(
                 string partitionId,
                 long sequenceNumber,
                 string popReceipt,
-                int visibilityTimeoutSeconds)
+                int visibilityTimeoutSeconds,
+                string details)
             {
                 this.Account = account;
                 this.TaskHub = taskHub;
@@ -349,6 +350,7 @@ public AbandoningMessage(
                 this.SequenceNumber = sequenceNumber;
                 this.PopReceipt = popReceipt;
                 this.VisibilityTimeoutSeconds = visibilityTimeoutSeconds;
+                this.Details = details;
             }
 
             [StructuredLogField]
@@ -384,6 +386,9 @@ public AbandoningMessage(
             [StructuredLogField]
             public int VisibilityTimeoutSeconds { get; }
 
+            [StructuredLogField]
+            public string Details { get; }
+
             public override EventId EventId => new EventId(
                 EventIds.AbandoningMessage,
                 nameof(EventIds.AbandoningMessage));
@@ -410,7 +415,8 @@ void IEventSourceEvent.WriteEventSource() => AnalyticsEventSource.Log.Abandoning
                 this.PopReceipt,
                 this.VisibilityTimeoutSeconds,
                 Utils.AppName,
-                Utils.ExtensionVersion);
+                Utils.ExtensionVersion,
+                this.Details);
         }
 
         internal class AssertFailure : StructuredLogEvent, IEventSourceEvent