From f81615af846910d65079b6d05b2869975dc920c2 Mon Sep 17 00:00:00 2001 From: Relay Date: Fri, 25 Sep 2026 17:26:48 +0000 Subject: [PATCH 1/4] Optimize orchestration history scans for tracing performance --- src/Shared/Grpc/Tracing/TraceHelper.cs | 6 ++ .../Grpc/GrpcDurableTaskWorker.Processor.cs | 58 ++++++++---------- src/Worker/Grpc/TracingHistoryEventIndex.cs | 44 ++++++++++++++ test/Worker/Grpc.Tests/TraceHelperTests.cs | 33 +++++++++++ .../TracingHistoryEventIndexTests.cs | 59 +++++++++++++++++++ 5 files changed, 166 insertions(+), 34 deletions(-) create mode 100644 src/Worker/Grpc/TracingHistoryEventIndex.cs create mode 100644 test/Worker/Grpc.Tests/TraceHelperTests.cs create mode 100644 test/Worker/Grpc.Tests/TracingHistoryEventIndexTests.cs diff --git a/src/Shared/Grpc/Tracing/TraceHelper.cs b/src/Shared/Grpc/Tracing/TraceHelper.cs index 1283ff12..2f80cdee 100644 --- a/src/Shared/Grpc/Tracing/TraceHelper.cs +++ b/src/Shared/Grpc/Tracing/TraceHelper.cs @@ -20,6 +20,12 @@ static class TraceHelper static readonly ActivitySource ActivityTraceSource = new ActivitySource(Source); + /// + /// Gets whether any listener is subscribed to Durable Task tracing activities. + /// + /// when the activity source has at least one listener; otherwise, . + public static bool HasListeners() => ActivityTraceSource.HasListeners(); + /// /// Starts a new trace activity for scheduling an orchestration from the client. /// diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index abfba18a..5b18916a 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -659,13 +659,19 @@ async Task OnRunOrchestratorAsync( } IReadOnlyList pastEvents = materializedPastEvents ?? request.PastEvents; - var executionStartedEvent = - request - .NewEvents - .Concat(pastEvents) - .Where(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) - .Select(e => e.ExecutionStarted) - .FirstOrDefault(); + bool hasTraceListeners = TraceHelper.HasListeners(); + + P.ExecutionStartedEvent? executionStartedEvent = null; + if (hasTraceListeners || isInitialRewind) + { + executionStartedEvent = + request + .NewEvents + .Concat(pastEvents) + .Where(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) + .Select(e => e.ExecutionStarted) + .FirstOrDefault(); + } if (isInitialRewind) { @@ -684,9 +690,11 @@ async Task OnRunOrchestratorAsync( // A rewind starts a new orchestration span instead of continuing the failed execution's stored span. P.OrchestrationTraceContext? orchestrationTraceContext = isInitialRewind ? null : request.OrchestrationTraceContext; - Activity? traceActivity = TraceHelper.StartTraceActivityForOrchestrationExecution( - executionStartedEvent, - orchestrationTraceContext); + Activity? traceActivity = hasTraceListeners + ? TraceHelper.StartTraceActivityForOrchestrationExecution( + executionStartedEvent, + orchestrationTraceContext) + : null; if (isInitialRewind) { @@ -701,27 +709,9 @@ await this.CompleteOrchestratorTaskWithChunkingAsync( return; } - if (executionStartedEvent is not null) + if (hasTraceListeners && executionStartedEvent is not null) { - P.HistoryEvent? GetSuborchestrationInstanceCreatedEvent(int eventId) - { - var subOrchestrationEvent = - pastEvents - .Where(x => x.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated) - .FirstOrDefault(x => x.EventId == eventId); - - return subOrchestrationEvent; - } - - P.HistoryEvent? GetTaskScheduledEvent(int eventId) - { - var taskScheduledEvent = - pastEvents - .Where(x => x.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskScheduled) - .LastOrDefault(x => x.EventId == eventId); - - return taskScheduledEvent; - } + TracingHistoryEventIndex historyEventIndex = new(pastEvents); foreach (var newEvent in request.NewEvents) { @@ -730,7 +720,7 @@ await this.CompleteOrchestratorTaskWithChunkingAsync( case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted: { P.HistoryEvent? subOrchestrationInstanceCreatedEvent = - GetSuborchestrationInstanceCreatedEvent( + historyEventIndex.GetSubOrchestrationInstanceCreatedEvent( newEvent.SubOrchestrationInstanceCompleted.TaskScheduledId); TraceHelper.EmitTraceActivityForSubOrchestrationCompleted( @@ -743,7 +733,7 @@ await this.CompleteOrchestratorTaskWithChunkingAsync( case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed: { P.HistoryEvent? subOrchestrationInstanceCreatedEvent = - GetSuborchestrationInstanceCreatedEvent( + historyEventIndex.GetSubOrchestrationInstanceCreatedEvent( newEvent.SubOrchestrationInstanceFailed.TaskScheduledId); TraceHelper.EmitTraceActivityForSubOrchestrationFailed( @@ -757,7 +747,7 @@ await this.CompleteOrchestratorTaskWithChunkingAsync( case P.HistoryEvent.EventTypeOneofCase.TaskCompleted: { P.HistoryEvent? taskScheduledEvent = - GetTaskScheduledEvent(newEvent.TaskCompleted.TaskScheduledId); + historyEventIndex.GetTaskScheduledEvent(newEvent.TaskCompleted.TaskScheduledId); TraceHelper.EmitTraceActivityForTaskCompleted( request.InstanceId, @@ -769,7 +759,7 @@ await this.CompleteOrchestratorTaskWithChunkingAsync( case P.HistoryEvent.EventTypeOneofCase.TaskFailed: { P.HistoryEvent? taskScheduledEvent = - GetTaskScheduledEvent(newEvent.TaskFailed.TaskScheduledId); + historyEventIndex.GetTaskScheduledEvent(newEvent.TaskFailed.TaskScheduledId); TraceHelper.EmitTraceActivityForTaskFailed( request.InstanceId, diff --git a/src/Worker/Grpc/TracingHistoryEventIndex.cs b/src/Worker/Grpc/TracingHistoryEventIndex.cs new file mode 100644 index 00000000..249f6436 --- /dev/null +++ b/src/Worker/Grpc/TracingHistoryEventIndex.cs @@ -0,0 +1,44 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Worker.Grpc; + +/// +/// Indexes the orchestration history events used to reconstruct tracing spans. +/// +sealed class TracingHistoryEventIndex +{ + readonly Dictionary subOrchestrationCreatedEvents = new(); + readonly Dictionary taskScheduledEvents = new(); + + public TracingHistoryEventIndex(IEnumerable pastEvents) + { + foreach (P.HistoryEvent historyEvent in pastEvents) + { + switch (historyEvent.EventTypeCase) + { + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated: + // Preserve the previous FirstOrDefault semantics for duplicate IDs. + this.subOrchestrationCreatedEvents.TryAdd(historyEvent.EventId, historyEvent); + break; + + case P.HistoryEvent.EventTypeOneofCase.TaskScheduled: + // Preserve the previous LastOrDefault semantics for duplicate IDs. + this.taskScheduledEvents[historyEvent.EventId] = historyEvent; + break; + } + } + } + + public P.HistoryEvent? GetSubOrchestrationInstanceCreatedEvent(int eventId) + => this.subOrchestrationCreatedEvents.TryGetValue(eventId, out P.HistoryEvent? historyEvent) + ? historyEvent + : null; + + public P.HistoryEvent? GetTaskScheduledEvent(int eventId) + => this.taskScheduledEvents.TryGetValue(eventId, out P.HistoryEvent? historyEvent) + ? historyEvent + : null; +} diff --git a/test/Worker/Grpc.Tests/TraceHelperTests.cs b/test/Worker/Grpc.Tests/TraceHelperTests.cs new file mode 100644 index 00000000..0ba7e616 --- /dev/null +++ b/test/Worker/Grpc.Tests/TraceHelperTests.cs @@ -0,0 +1,33 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Diagnostics; +using Microsoft.DurableTask.Tracing; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +public class TraceHelperTests +{ + [Fact] + public void HasListeners_TracksMatchingActivityListener() + { + bool initialHasListeners = TraceHelper.HasListeners(); + ActivityListener listener = new() + { + ShouldListenTo = source => source.Name == "Microsoft.DurableTask", + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.None, + }; + + try + { + ActivitySource.AddActivityListener(listener); + TraceHelper.HasListeners().Should().BeTrue(); + } + finally + { + listener.Dispose(); + } + + TraceHelper.HasListeners().Should().Be(initialHasListeners); + } +} diff --git a/test/Worker/Grpc.Tests/TracingHistoryEventIndexTests.cs b/test/Worker/Grpc.Tests/TracingHistoryEventIndexTests.cs new file mode 100644 index 00000000..38767c0c --- /dev/null +++ b/test/Worker/Grpc.Tests/TracingHistoryEventIndexTests.cs @@ -0,0 +1,59 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +public class TracingHistoryEventIndexTests +{ + [Fact] + public void GetSubOrchestrationInstanceCreatedEvent_DuplicateIds_ReturnsFirstEvent() + { + P.HistoryEvent first = new() + { + EventId = 7, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent { Name = "first" }, + }; + P.HistoryEvent second = new() + { + EventId = 7, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent { Name = "second" }, + }; + + TracingHistoryEventIndex index = new([first, second]); + index.GetSubOrchestrationInstanceCreatedEvent(7).Should().BeSameAs(first); + } + + [Fact] + public void GetTaskScheduledEvent_DuplicateIds_ReturnsLastEvent() + { + P.HistoryEvent first = new() + { + EventId = 11, + TaskScheduled = new P.TaskScheduledEvent { Name = "first" }, + }; + P.HistoryEvent second = new() + { + EventId = 11, + TaskScheduled = new P.TaskScheduledEvent { Name = "second" }, + }; + + TracingHistoryEventIndex index = new([first, second]); + index.GetTaskScheduledEvent(11).Should().BeSameAs(second); + } + + [Fact] + public void Lookups_MissingIds_ReturnNull() + { + P.HistoryEvent unrelated = new() + { + EventId = 3, + TimerCreated = new P.TimerCreatedEvent(), + }; + + TracingHistoryEventIndex index = new([unrelated]); + index.GetSubOrchestrationInstanceCreatedEvent(3).Should().BeNull(); + index.GetTaskScheduledEvent(3).Should().BeNull(); + } +} From 7537328eca6fe40b25072466dc823e0c7682fd86 Mon Sep 17 00:00:00 2001 From: Relay Date: Sat, 26 Sep 2026 17:39:08 -0400 Subject: [PATCH 2/4] Fix netstandard2.0 tracing history compatibility --- src/Worker/Grpc/TracingHistoryEventIndex.cs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/Worker/Grpc/TracingHistoryEventIndex.cs b/src/Worker/Grpc/TracingHistoryEventIndex.cs index 249f6436..66283877 100644 --- a/src/Worker/Grpc/TracingHistoryEventIndex.cs +++ b/src/Worker/Grpc/TracingHistoryEventIndex.cs @@ -21,7 +21,10 @@ public TracingHistoryEventIndex(IEnumerable pastEvents) { case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated: // Preserve the previous FirstOrDefault semantics for duplicate IDs. - this.subOrchestrationCreatedEvents.TryAdd(historyEvent.EventId, historyEvent); + if (!this.subOrchestrationCreatedEvents.ContainsKey(historyEvent.EventId)) + { + this.subOrchestrationCreatedEvents.Add(historyEvent.EventId, historyEvent); + } break; case P.HistoryEvent.EventTypeOneofCase.TaskScheduled: From 315993aa5c0e39c6c54404ba5cf2fae3b0df9e59 Mon Sep 17 00:00:00 2001 From: Relay Date: Sat, 26 Sep 2026 17:39:10 -0400 Subject: [PATCH 3/4] Serialize ActivitySource listener tests --- test/Worker/Grpc.Tests/TraceHelperTests.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/test/Worker/Grpc.Tests/TraceHelperTests.cs b/test/Worker/Grpc.Tests/TraceHelperTests.cs index 0ba7e616..4aa2155e 100644 --- a/test/Worker/Grpc.Tests/TraceHelperTests.cs +++ b/test/Worker/Grpc.Tests/TraceHelperTests.cs @@ -6,6 +6,7 @@ namespace Microsoft.DurableTask.Worker.Grpc.Tests; +[Collection("ActivitySource listener tests")] public class TraceHelperTests { [Fact] From 505ebcaf8119b58f763dd2107077ddb1908332f5 Mon Sep 17 00:00:00 2001 From: Relay Date: Sat, 26 Sep 2026 17:39:12 -0400 Subject: [PATCH 4/4] Serialize worker ActivitySource listener tests --- test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs index 61c3cdf8..71a4d37a 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs @@ -20,6 +20,7 @@ namespace Microsoft.DurableTask.Worker.Grpc.Tests; +[Collection("ActivitySource listener tests")] public class GrpcDurableTaskWorkerTests { const string Category = "Microsoft.DurableTask.Worker.Grpc";