Fix telemetry reporting 500 for successful activities/orchestrations with caught exceptions

Test/DurableTask.Core.Tests/TraceHelperTests.cs

@@ -0,0 +1,69 @@
+//  ----------------------------------------------------------------------------------
+//  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.
+//  ----------------------------------------------------------------------------------
+#if !NET462
+#nullable enable
+namespace DurableTask.Core.Tests
+{
+    using System.Collections.Generic;
+    using System.Diagnostics;
+    using DurableTask.Core.Entities.OperationFormat;
+    using DurableTask.Core.Tracing;
+    using Microsoft.VisualStudio.TestTools.UnitTesting;
+    using DiagnosticsActivityStatusCode = System.Diagnostics.ActivityStatusCode;
+    using TraceActivityStatusCode = DurableTask.Core.Tracing.ActivityStatusCode;
+
+    [TestClass]
+    public class TraceHelperTests
+    {
+        [TestMethod]
+        public void EndActivitiesForEntityInvocationResetsSuccessfulStatus()
+        {
+            var activities = new List<Activity>
+            {
+                new Activity("entityOperation").Start()
+            };
+            activities[0].SetStatus(TraceActivityStatusCode.Error, "instrumented error");
+
+            var results = new List<OperationResult>
+            {
+                new OperationResult()
+            };
+
+            TraceHelper.EndActivitiesForProcessingEntityInvocation(activities, results, batchFailureDetails: null);
+
+            Assert.AreEqual(DiagnosticsActivityStatusCode.Ok, activities[0].Status);
+        }
+
+        [TestMethod]
+        public void EndActivitiesForEntityInvocationMarksFailures()
+        {
+            var activities = new List<Activity>
+            {
+                new Activity("entityOperation").Start()
+            };
+
+            var failingResults = new List<OperationResult>
+            {
+                new OperationResult
+                {
+                    ErrorMessage = "entity failure"
+                }
+            };
+
+            TraceHelper.EndActivitiesForProcessingEntityInvocation(activities, failingResults, batchFailureDetails: null);
+
+            Assert.AreEqual(DiagnosticsActivityStatusCode.Error, activities[0].Status);
+        }
+    }
+}
+#endif

src/DurableTask.Core/TaskActivityDispatcher.cs

@@ -267,6 +267,12 @@ await this.dispatchPipeline.RunAsync(dispatchContext, async _ =>
                     eventToRespond = new TaskCompletedEvent(-1, scheduledEvent.EventId, null);
                 }
 
+                if (traceActivity != null && eventToRespond is TaskCompletedEvent)
+                {
+                    // Ensure successful executions don't preserve a prior error status from custom instrumentation.
+                    traceActivity.SetStatus(ActivityStatusCode.OK, "Completed");
+                }
+
                 var responseTaskMessage = new TaskMessage
                 {
                     Event = eventToRespond,

src/DurableTask.Core/TaskOrchestrationDispatcher.cs

@@ -1043,6 +1043,8 @@ internal static bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workIt
                 runtimeState.OrchestrationInstance!,
                 () => Utils.EscapeJson(JsonDataConverter.Default.Serialize(runtimeState.GetOrchestrationRuntimeStateDump(), true)));
 
+            SetOrchestrationActivityStatus(completeOrchestratorAction);
+
             // Check to see if we need to start a new execution
             if (completeOrchestratorAction.OrchestrationStatus == OrchestrationStatus.ContinuedAsNew)
             {
@@ -1090,12 +1092,6 @@ internal static bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workIt
                     subOrchestrationFailedEvent.FailureDetails = completeOrchestratorAction.FailureDetails;
 
                     taskMessage.Event = subOrchestrationFailedEvent;
-
-                    if (completeOrchestratorAction.OrchestrationStatus == OrchestrationStatus.Failed)
-                    {
-                        DistributedTraceActivity.Current?.SetStatus(
-                            ActivityStatusCode.Error, completeOrchestratorAction.Result);
-                    }
                 }
 
                 ResetDistributedTraceActivity(runtimeState);
@@ -1107,12 +1103,6 @@ internal static bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workIt
                 }
             }
 
-            if (completeOrchestratorAction.OrchestrationStatus == OrchestrationStatus.Failed)
-            {
-                DistributedTraceActivity.Current?.SetStatus(
-                    ActivityStatusCode.Error, completeOrchestratorAction.Result);
-            }
-
             ResetDistributedTraceActivity(runtimeState);
 
             return null;
@@ -1125,6 +1115,34 @@ private void ResetDistributedTraceActivity(OrchestrationRuntimeState runtimeStat
             DistributedTraceActivity.Current = null;
         }
 
+        private static void SetOrchestrationActivityStatus(OrchestrationCompleteOrchestratorAction completeOrchestratorAction)
+        {
+            if (DistributedTraceActivity.Current == null)
+            {
+                return;
+            }
+
+            string failureDescription = completeOrchestratorAction.FailureDetails?.ErrorMessage
+                ?? completeOrchestratorAction.Result
+                ?? completeOrchestratorAction.OrchestrationStatus.ToString();
+
+            switch (completeOrchestratorAction.OrchestrationStatus)
+            {
+                case OrchestrationStatus.Completed:
+                    DistributedTraceActivity.Current.SetStatus(ActivityStatusCode.OK, OrchestrationStatus.Completed.ToString());
+                    break;
+                case OrchestrationStatus.ContinuedAsNew:
+                    DistributedTraceActivity.Current.SetStatus(ActivityStatusCode.OK, OrchestrationStatus.ContinuedAsNew.ToString());
+                    break;
+                case OrchestrationStatus.Failed:
+                    DistributedTraceActivity.Current.SetStatus(ActivityStatusCode.Error, failureDescription);
+                    break;
+                case OrchestrationStatus.Terminated:
+                    DistributedTraceActivity.Current.SetStatus(ActivityStatusCode.Error, OrchestrationStatus.Terminated.ToString());
+                    break;
+            }
+        }
+
         TaskMessage ProcessScheduleTaskDecision(
             ScheduleTaskOrchestratorAction scheduleTaskOrchestratorAction,
             OrchestrationRuntimeState runtimeState,

src/DurableTask.Core/Tracing/TraceHelper.cs

@@ -602,7 +602,13 @@ internal static void EndActivitiesForProcessingEntityInvocation(List<Activity> t
                     {
                         if (result.ErrorMessage != null || result.FailureDetails != null)
                         {
-                            activity.SetTag(Schema.Task.ErrorMessage, result.ErrorMessage ?? result.FailureDetails!.ErrorMessage);
+                            string errorDetails = result.ErrorMessage ?? result.FailureDetails!.ErrorMessage;
+                            activity.SetTag(Schema.Task.ErrorMessage, errorDetails);
+                            activity.SetStatus(ActivityStatusCode.Error, errorDetails);
+                        }
+                        else
+                        {
+                            activity.SetStatus(ActivityStatusCode.OK, "Completed");
                         }
                         if (result.StartTimeUtc is DateTime startTime)
                         {
@@ -630,6 +636,7 @@ internal static void EndActivitiesForProcessingEntityInvocation(List<Activity> t
                     if (activity != null)
                     {
                         activity.SetTag(Schema.Task.ErrorMessage, errorMessage);
+                        activity.SetStatus(ActivityStatusCode.Error, errorMessage);
                         activity.Dispose();
                     }
                 }

test/DurableTask.Core.Tests/DispatcherMiddlewareTests.cs

@@ -26,11 +26,14 @@ namespace DurableTask.Core.Tests
     using System.Xml;
     using DurableTask.Core.Command;
     using DurableTask.Core.History;
+    using DurableTask.Core.Tracing;
     using DurableTask.Emulator;
     using DurableTask.Test.Orchestrations;
     using Microsoft.Extensions.Logging;
     using Microsoft.Extensions.Logging.Console;
     using Microsoft.VisualStudio.TestTools.UnitTesting;
+    using DiagnosticsActivityStatusCode = System.Diagnostics.ActivityStatusCode;
+    using TraceActivityStatusCode = DurableTask.Core.Tracing.ActivityStatusCode;
 
     [TestClass]
     public class DispatcherMiddlewareTests
@@ -53,8 +56,15 @@ public void InitializeTests()
             // We use `GetAwaiter().GetResult()` because otherwise this method will fail with:
             // "X has wrong signature. The method must be non-static, public, does not return a value and should not take any parameter."
             this.worker
-                .AddTaskOrchestrations(typeof(SimplestGreetingsOrchestration), typeof(ParentWorkflow), typeof(ChildWorkflow))
-                .AddTaskActivities(typeof(SimplestGetUserTask), typeof(SimplestSendGreetingTask))
+                .AddTaskOrchestrations(
+                    typeof(SimplestGreetingsOrchestration),
+                    typeof(ParentWorkflow),
+                    typeof(ChildWorkflow),
+                    typeof(ActivityStatusResetOrchestration))
+                .AddTaskActivities(
+                    typeof(SimplestGetUserTask),
+                    typeof(SimplestSendGreetingTask),
+                    typeof(ActivityStatusResetActivity))
                 .StartAsync().GetAwaiter().GetResult();
 
             this.client = new TaskHubClient(service);
@@ -451,6 +461,82 @@ public async Task MockActivityOrchestration()
             Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus);
             Assert.AreEqual("FakeActivity,FakeActivityVersion,SomeInput", state.Output);
         }
+
+        [TestMethod]
+        public async Task ActivityAndOrchestrationSpansResetStatuses()
+        {
+            using var activityCollector = new DurableActivityCollector();
+
+            OrchestrationInstance instance = await this.client.CreateOrchestrationInstanceAsync(
+                typeof(ActivityStatusResetOrchestration),
+                "payload");
+
+            TimeSpan timeout = TimeSpan.FromSeconds(Debugger.IsAttached ? 1000 : 10);
+            OrchestrationState finalState = await this.client.WaitForOrchestrationAsync(instance, timeout);
+            Assert.AreEqual(OrchestrationStatus.Completed, finalState.OrchestrationStatus);
+
+            string orchestrationName = NameVersionHelper.GetDefaultName(typeof(ActivityStatusResetOrchestration));
+            Activity? orchestrationActivity = activityCollector.Find(TraceActivityConstants.Orchestration, orchestrationName, ActivityKind.Server);
+            Assert.IsNotNull(orchestrationActivity, "Expected orchestration server trace to be captured.");
+            Assert.AreEqual(DiagnosticsActivityStatusCode.Ok, orchestrationActivity!.Status);
+
+            string activityName = NameVersionHelper.GetDefaultName(typeof(ActivityStatusResetActivity));
+            Activity? activitySpan = activityCollector.Find(TraceActivityConstants.Activity, activityName, ActivityKind.Server);
+            Assert.IsNotNull(activitySpan, "Expected activity server trace to be captured.");
+            Assert.AreEqual(DiagnosticsActivityStatusCode.Ok, activitySpan!.Status);
+        }
+
+        private sealed class DurableActivityCollector : IDisposable
+        {
+            private readonly ActivityListener listener;
+            private readonly ConcurrentBag<Activity> activities = new ConcurrentBag<Activity>();
+
+            public DurableActivityCollector()
+            {
+                this.listener = new ActivityListener
+                {
+                    ShouldListenTo = source => source.Name == "DurableTask.Core",
+                    Sample = (ref ActivityCreationOptions<ActivityContext> _) => ActivitySamplingResult.AllData,
+                    SampleUsingParentId = (ref ActivityCreationOptions<string> _) => ActivitySamplingResult.AllData,
+                    ActivityStopped = activity => this.activities.Add(activity),
+                };
+
+                ActivitySource.AddActivityListener(this.listener);
+            }
+
+            public Activity? Find(string taskType, string taskName, ActivityKind kind)
+            {
+                return this.activities
+                    .Where(activity => activity.Kind == kind)
+                    .Where(activity => string.Equals(Convert.ToString(activity.GetTagItem(Schema.Task.Type)), taskType, StringComparison.Ordinal))
+                    .Where(activity => string.Equals(Convert.ToString(activity.GetTagItem(Schema.Task.Name)), taskName, StringComparison.Ordinal))
+                    .LastOrDefault();
+            }
+
+            public void Dispose()
+            {
+                this.listener.Dispose();
+            }
+        }
+
+        private sealed class ActivityStatusResetOrchestration : TaskOrchestration<string, string>
+        {
+            public override async Task<string> RunTask(OrchestrationContext context, string input)
+            {
+                DistributedTraceActivity.Current?.SetStatus(TraceActivityStatusCode.Error, "orchestration instrumentation error");
+                string? activityOutput = await context.ScheduleTask<string>(typeof(ActivityStatusResetActivity), input ?? "ok");
+                return activityOutput ?? string.Empty;
+            }
+        }
+
+        private sealed class ActivityStatusResetActivity : TaskActivity<string, string>
+        {
+            protected override string Execute(TaskContext context, string input)
+            {
+                Activity.Current?.SetStatus(TraceActivityStatusCode.Error, "activity instrumentation error");
+                return input ?? "ok";
+            }
+        }
     }
 }
 #endif