Merge remote-tracking branch 'refs/remotes/Azure/master'

Author: solankisamir

Framework/BrokeredMessageExtensions.cs

@@ -0,0 +1,40 @@
+//  ----------------------------------------------------------------------------------
+//  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
+{
+    using System;
+    using System.Diagnostics;
+    using Microsoft.ServiceBus.Messaging;
+    using Tracing;
+
+    /// <summary>
+    /// Extension methods for BrokeredMessage
+    /// </summary>
+    public static class BrokeredMessageExtensions
+    {
+        /// <summary>
+        /// Returns delivery latency of the message
+        /// </summary>        
+        public static double DeliveryLatency(this BrokeredMessage message)
+        {
+            if (message == null)
+            {
+                return 0;
+            }
+
+            DateTime actualEnqueueTimeUtc = (message.ScheduledEnqueueTimeUtc == DateTime.MinValue) ? message.EnqueuedTimeUtc : message.ScheduledEnqueueTimeUtc;
+            return (DateTime.UtcNow - actualEnqueueTimeUtc).TotalMilliseconds;
+        }
+    }
+}

Framework/Framework.csproj

@@ -100,6 +100,7 @@
     <Reference Include="System.Xml" />
   </ItemGroup>
   <ItemGroup>
+    <Compile Include="BrokeredMessageExtensions.cs" />
     <Compile Include="SynchronousTaskScheduler.cs" />
     <Compile Include="History\GenericEvent.cs" />
     <Compile Include="CompressionSettings.cs" />

Framework/Properties/AssemblyInfo.cs

@@ -27,6 +27,6 @@
 [assembly: AssemblyCulture("")]
 [assembly: ComVisible(false)]
 [assembly: Guid("1d626d8b-330b-4c6a-b689-c0fefbeb99cd")]
-[assembly: AssemblyVersion("1.0.0.8")]
-[assembly: AssemblyFileVersion("1.0.0.8")]
-[assembly: InternalsVisibleTo("FrameworkUnitTests")]
+[assembly: AssemblyVersion("1.0.0.9")]
+[assembly: AssemblyFileVersion("1.0.0.9")]
+[assembly: InternalsVisibleTo("FrameworkUnitTests")]

Framework/TaskActivityDispatcher.cs

@@ -61,8 +61,7 @@ protected override async Task<BrokeredMessage> OnFetchWorkItem(TimeSpan receiveT
             {
                 TraceHelper.TraceSession(TraceEventType.Information,
                     receivedMessage.SessionId,
-                    GetFormattedLog("New message to process: " + receivedMessage.MessageId + " [" +
-                                    receivedMessage.SequenceNumber + "]"));
+                    GetFormattedLog($"New message to process: {receivedMessage.MessageId} [{receivedMessage.SequenceNumber}], latency: {receivedMessage.DeliveryLatency()}ms"));
             }
 
             return receivedMessage;

Framework/TaskOrchestrationContext.cs

@@ -15,8 +15,11 @@ namespace DurableTask
 {
     using System;
     using System.Collections.Generic;
+    using System.Diagnostics;
+    using System.Globalization;
     using System.Threading;
     using System.Threading.Tasks;
+    using DurableTask.Tracing;
     using Command;
     using History;
 
@@ -251,6 +254,10 @@ public void HandleTaskCompletedEvent(TaskCompletedEvent completedEvent)
 
                 openTasks.Remove(taskId);
             }
+            else
+            {
+                LogDuplicateEvent("TaskCompleted", completedEvent, taskId);
+            }
         }
 
         public void HandleTaskFailedEvent(TaskFailedEvent failedEvent)
@@ -269,6 +276,10 @@ public void HandleTaskFailedEvent(TaskFailedEvent failedEvent)
 
                 openTasks.Remove(taskId);
             }
+            else
+            {
+                LogDuplicateEvent("TaskFailed", failedEvent, taskId);
+            }
         }
 
         public void HandleSubOrchestrationInstanceCompletedEvent(SubOrchestrationInstanceCompletedEvent completedEvent)
@@ -281,6 +292,10 @@ public void HandleSubOrchestrationInstanceCompletedEvent(SubOrchestrationInstanc
 
                 openTasks.Remove(taskId);
             }
+            else
+            {
+                LogDuplicateEvent("SubOrchestrationInstanceCompleted", completedEvent, taskId);
+            }
         }
 
         public void HandleSubOrchestrationInstanceFailedEvent(SubOrchestrationInstanceFailedEvent failedEvent)
@@ -298,6 +313,10 @@ public void HandleSubOrchestrationInstanceFailedEvent(SubOrchestrationInstanceFa
 
                 openTasks.Remove(taskId);
             }
+            else
+            {
+                LogDuplicateEvent("SubOrchestrationInstanceFailed", failedEvent, taskId);
+            }
         }
 
         public void HandleTimerFiredEvent(TimerFiredEvent timerFiredEvent)
@@ -309,6 +328,22 @@ public void HandleTimerFiredEvent(TimerFiredEvent timerFiredEvent)
                 info.Result.SetResult(timerFiredEvent.TimerId.ToString());
                 openTasks.Remove(taskId);
             }
+            else
+            {
+                LogDuplicateEvent("TimerFired", timerFiredEvent, taskId);
+            }
+        }
+
+        private void LogDuplicateEvent(string source, HistoryEvent historyEvent, int taskId)
+        {
+            TraceHelper.TraceSession(
+                TraceEventType.Warning,
+                this.OrchestrationInstance.InstanceId,
+                "Duplicate {0} Event: {1}, type: {2}, ts: {3}",
+                source,
+                taskId.ToString(),
+                historyEvent.EventType,
+                historyEvent.Timestamp.ToString(CultureInfo.InvariantCulture));
         }
 
         public void HandleExecutionTerminatedEvent(ExecutionTerminatedEvent terminatedEvent)

Framework/TaskOrchestrationDispatcher.cs

@@ -124,18 +124,20 @@ await Utils.ExecuteWithRetries(() => session.ReceiveBatchAsync(PrefetchCount),
 
             TraceHelper.TraceSession(TraceEventType.Information, session.SessionId,
                 GetFormattedLog(
-                    string.Format("{0} new messages to process: {1}",
-                        newMessages.Count(),
-                        string.Join(",", newMessages.Select(m => m.MessageId)))));
+                    $@"{newMessages.Count()} new messages to process: {
+                        string.Join(",", newMessages.Select(m => m.MessageId))
+                        }, max latency: {
+                        newMessages.Max(message => message.DeliveryLatency())}ms"
+                        ));
 
             return new SessionWorkItem {Session = session, Messages = newMessages};
         }
 
         protected override async Task OnProcessWorkItem(SessionWorkItem sessionWorkItem)
         {
-            var messagesToSend = new List<BrokeredMessage>();
-            var timerMessages = new List<BrokeredMessage>();
-            var subOrchestrationMessages = new List<BrokeredMessage>();
+            var messagesToSend = new List<MessageContainer>();
+            var timerMessages = new List<MessageContainer>();
+            var subOrchestrationMessages = new List<MessageContainer>();
             bool isCompleted = false;
             bool continuedAsNew = false;
 
@@ -195,7 +197,7 @@ protected override async Task OnProcessWorkItem(SessionWorkItem sessionWorkItem)
                 foreach (OrchestratorAction decision in decisions)
                 {
                     TraceHelper.TraceInstance(TraceEventType.Information, runtimeState.OrchestrationInstance,
-                        "Processing orchestrator action of type {0}", decision.OrchestratorActionType);
+                        "Processing orchestrator action of type {0}, id {1}", decision.OrchestratorActionType, decision.Id.ToString());
                     switch (decision.OrchestratorActionType)
                     {
                         case OrchestratorActionType.ScheduleOrchestrator:
@@ -206,7 +208,7 @@ protected override async Task OnProcessWorkItem(SessionWorkItem sessionWorkItem)
                                 taskMessage, settings.MessageCompressionSettings, runtimeState.OrchestrationInstance,
                                 "ScheduleTask");
                             brokeredMessage.SessionId = session.SessionId;
-                            messagesToSend.Add(brokeredMessage);
+                            messagesToSend.Add(new MessageContainer(brokeredMessage, decision));
                             break;
                         case OrchestratorActionType.CreateTimer:
                             var timerOrchestratorAction = (CreateTimerOrchestratorAction) decision;
@@ -216,7 +218,7 @@ protected override async Task OnProcessWorkItem(SessionWorkItem sessionWorkItem)
                                 "Timer");
                             brokeredTimerMessage.ScheduledEnqueueTimeUtc = timerOrchestratorAction.FireAt;
                             brokeredTimerMessage.SessionId = session.SessionId;
-                            timerMessages.Add(brokeredTimerMessage);
+                            timerMessages.Add(new MessageContainer(brokeredTimerMessage, decision));
                             break;
                         case OrchestratorActionType.CreateSubOrchestration:
                             var createSubOrchestrationAction = (CreateSubOrchestrationAction) decision;
@@ -228,7 +230,7 @@ protected override async Task OnProcessWorkItem(SessionWorkItem sessionWorkItem)
                                 runtimeState.OrchestrationInstance, "Schedule Suborchestration");
                             createSubOrchestrationMessage.SessionId =
                                 createSubOrchestrationInstanceMessage.OrchestrationInstance.InstanceId;
-                            subOrchestrationMessages.Add(createSubOrchestrationMessage);
+                            subOrchestrationMessages.Add(new MessageContainer(createSubOrchestrationMessage, decision));
                             break;
                         case OrchestratorActionType.OrchestrationComplete:
                             TaskMessage workflowInstanceCompletedMessage =
@@ -252,7 +254,7 @@ protected override async Task OnProcessWorkItem(SessionWorkItem sessionWorkItem)
                                 }
                                 else
                                 {
-                                    subOrchestrationMessages.Add(workflowCompletedBrokeredMessage);
+                                    subOrchestrationMessages.Add(new MessageContainer(workflowCompletedBrokeredMessage, decision));
                                 }
                             }
                             isCompleted = !continuedAsNew;
@@ -294,7 +296,7 @@ protected override async Task OnProcessWorkItem(SessionWorkItem sessionWorkItem)
                             "MaxMessageCount Timer");
                         brokeredTimerMessage.ScheduledEnqueueTimeUtc = dummyTimer.FireAt;
                         brokeredTimerMessage.SessionId = session.SessionId;
-                        timerMessages.Add(brokeredTimerMessage);
+                        timerMessages.Add(new MessageContainer(brokeredTimerMessage, null));
                         break;
                     }
                 }
@@ -423,35 +425,40 @@ protected override async Task OnProcessWorkItem(SessionWorkItem sessionWorkItem)
                     {
                         if (messagesToSend.Count > 0)
                         {
-                            messagesToSend.ForEach(m => workerSender.Send(m));
+                            messagesToSend.ForEach(m => workerSender.Send(m.Message));
+                            this.LogSentMessages(session, "Worker outbound", messagesToSend);
                         }
 
                         if (timerMessages.Count > 0)
                         {
-                            timerMessages.ForEach(m => orchestratorQueueClient.Send(m));
+                            timerMessages.ForEach(m => orchestratorQueueClient.Send(m.Message));
+                            this.LogSentMessages(session, "Timer Message", timerMessages);
                         }
                     }
 
                     if (subOrchestrationMessages.Count > 0)
                     {
-                        subOrchestrationMessages.ForEach(m => orchestratorQueueClient.Send(m));
+                        subOrchestrationMessages.ForEach(m => orchestratorQueueClient.Send(m.Message));
+                        this.LogSentMessages(session, "Sub Orchestration", subOrchestrationMessages);
                     }
 
                     if (continuedAsNewMessage != null)
                     {
                         orchestratorQueueClient.Send(continuedAsNewMessage);
+                        this.LogSentMessages(session, "Continue as new", new List<MessageContainer> () { new MessageContainer(continuedAsNewMessage, null) });
                     }
 
                     if (isTrackingEnabled)
                     {
-                        List<BrokeredMessage> trackingMessages = CreateTrackingMessages(runtimeState);
+                        List<MessageContainer> trackingMessages = CreateTrackingMessages(runtimeState);
 
                         TraceHelper.TraceInstance(TraceEventType.Information, runtimeState.OrchestrationInstance,
                             "Created {0} tracking messages", trackingMessages.Count);
 
                         if (trackingMessages.Count > 0)
                         {
-                            trackingMessages.ForEach(m => trackingSender.Send(m));
+                            trackingMessages.ForEach(m => trackingSender.Send(m.Message));
+                            this.LogSentMessages(session, "Tracking messages", trackingMessages);
                         }
                     }
                 }
@@ -472,6 +479,15 @@ protected override async Task OnProcessWorkItem(SessionWorkItem sessionWorkItem)
             }
         }
 
+        void LogSentMessages(MessageSession session, string messageType, IList<MessageContainer> messages)
+        {
+            TraceHelper.TraceSession(
+                TraceEventType.Information,
+                session.SessionId,
+                this.GetFormattedLog($@"{messages.Count().ToString()} messages queued for {messageType}: {
+                    string.Join(",", messages.Select(m => $"{m.Message.MessageId} <{m.Action?.Id.ToString()}>"))}"));
+        }
+
         BrokeredMessage CreateForcedTerminateMessage(string instanceId, string reason)
         {
             var taskMessage = new TaskMessage
@@ -487,9 +503,9 @@ BrokeredMessage CreateForcedTerminateMessage(string instanceId, string reason)
             return message;
         }
 
-        List<BrokeredMessage> CreateTrackingMessages(OrchestrationRuntimeState runtimeState)
+        List<MessageContainer> CreateTrackingMessages(OrchestrationRuntimeState runtimeState)
         {
-            var trackingMessages = new List<BrokeredMessage>();
+            var trackingMessages = new List<MessageContainer> ();
 
             // We cannot create tracking messages if runtime state does not have Orchestration InstanceId
             // This situation can happen due to corruption of service bus session state or if somehow first message of orchestration is not execution started
@@ -516,7 +532,7 @@ List<BrokeredMessage> CreateTrackingMessages(OrchestrationRuntimeState runtimeSt
                 trackingMessage.ContentType = FrameworkConstants.TaskMessageContentType;
                 trackingMessage.SessionId = runtimeState.OrchestrationInstance.InstanceId;
                 trackingMessage.Properties[FrameworkConstants.HistoryEventIndexPropertyName] = historyEventIndex++;
-                trackingMessages.Add(trackingMessage);
+                trackingMessages.Add(new MessageContainer(trackingMessage, null));
             }
 
             var stateMessage = new StateMessage {State = BuildOrchestrationState(runtimeState)};
@@ -526,7 +542,7 @@ List<BrokeredMessage> CreateTrackingMessages(OrchestrationRuntimeState runtimeSt
                 "State Tracking Message");
             brokeredStateMessage.SessionId = runtimeState.OrchestrationInstance.InstanceId;
             brokeredStateMessage.ContentType = FrameworkConstants.StateMessageContentType;
-            trackingMessages.Add(brokeredStateMessage);
+            trackingMessages.Add(new MessageContainer(brokeredStateMessage, null));
             return trackingMessages;
         }
 
@@ -881,5 +897,17 @@ protected override int GetDelayInSecondsAfterOnFetchException(Exception exceptio
             }
             return delay;
         }
+
+        internal class MessageContainer
+        {
+            internal BrokeredMessage Message { get; set; }
+            internal OrchestratorAction Action { get; set; }
+
+            internal MessageContainer(BrokeredMessage message, OrchestratorAction action)
+            {
+                this.Message = message;
+                this.Action = action;
+            }
+        }
     }
 }

Framework/Tracing/OrchestrationConsoleTraceListener.cs

@@ -44,8 +44,7 @@ public override void TraceEvent(TraceEventCache eventCache, string source, Trace
             }
             catch (Exception exception)
             {
-                string toWrite = string.Format("Exception while parsing trace:  {0}\n\t", exception.Message,
-                    exception.StackTrace);
+                string toWrite = $"Exception while parsing trace: {message} : {exception.Message}\n\t{exception.StackTrace}";
                 Console.WriteLine(toWrite);
                 Debug.WriteLine(toWrite);
             }

Framework/Tracing/TraceHelper.cs

@@ -208,7 +208,7 @@ static void ExceptionHandlingWrapper(Action innerFunc)
                 try
                 {
                     source.TraceEvent(TraceEventType.Critical, 0,
-                        "Failed to log actual trace because one or more trace listeners threw an exception.");
+                        $"Failed to log actual trace because one or more trace listeners threw an exception: {exception}");
                 }
                 catch (Exception anotherException)
                 {