diff --git a/src/Shared/Grpc/Tracing/TraceHelper.cs b/src/Shared/Grpc/Tracing/TraceHelper.cs index 1283ff12..ec5ecf5f 100644 --- a/src/Shared/Grpc/Tracing/TraceHelper.cs +++ b/src/Shared/Grpc/Tracing/TraceHelper.cs @@ -20,6 +20,17 @@ static class TraceHelper static readonly ActivitySource ActivityTraceSource = new ActivitySource(Source); + /// + /// Gets a value indicating whether any listener is currently registered for the Durable Task + /// . + /// + /// + /// This is a cheap check that callers can use to skip trace-event lookup work (such as scanning + /// orchestration history to correlate scheduling events) when no listener is registered and any resulting + /// would be discarded anyway. + /// + public static bool HasListeners => ActivityTraceSource.HasListeners(); + /// /// Starts a new trace activity for scheduling an orchestration from the client. /// diff --git a/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs b/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs new file mode 100644 index 00000000..2e7c4ec6 --- /dev/null +++ b/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs @@ -0,0 +1,92 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Tracing; + +/// +/// Provides indexed lookups of past history events by event ID, used to correlate new completion/failure +/// events (e.g. "TaskCompleted") back to the history event that scheduled them (e.g. "TaskScheduled"). +/// +/// +/// The indexes are built lazily, at most once per instance, and cached for the lifetime of the instance. This +/// avoids re-scanning the full set of past events for every new event being processed in a work item, which +/// would otherwise be O(new events x past events) for work items with many new events. +/// +sealed class TraceHistoryEventLookup +{ + readonly IEnumerable pastEvents; + + Dictionary? taskScheduledEventsByEventId; + Dictionary? subOrchestrationInstanceCreatedEventsByEventId; + + /// + /// Initializes a new instance of the class. + /// + /// The past history events for the current orchestrator work item. + public TraceHistoryEventLookup(IEnumerable pastEvents) + { + this.pastEvents = pastEvents; + } + + /// + /// Gets the "TaskScheduled" history event with the given event ID, if any. + /// + /// The event ID to look up. + /// The matching event, or if none is found. + /// + /// If more than one "TaskScheduled" event shares the given event ID, the last one encountered (in history + /// order) is returned, matching the original LastOrDefault lookup semantics. + /// + public P.HistoryEvent? GetTaskScheduledEvent(int eventId) + { + this.taskScheduledEventsByEventId ??= BuildIndex( + this.pastEvents, P.HistoryEvent.EventTypeOneofCase.TaskScheduled, keepFirst: false); + return this.taskScheduledEventsByEventId.TryGetValue(eventId, out P.HistoryEvent? historyEvent) + ? historyEvent + : null; + } + + /// + /// Gets the "SubOrchestrationInstanceCreated" history event with the given event ID, if any. + /// + /// The event ID to look up. + /// The matching event, or if none is found. + /// + /// If more than one "SubOrchestrationInstanceCreated" event shares the given event ID, the first one + /// encountered (in history order) is returned, matching the original FirstOrDefault lookup semantics. + /// + public P.HistoryEvent? GetSubOrchestrationInstanceCreatedEvent(int eventId) + { + this.subOrchestrationInstanceCreatedEventsByEventId ??= BuildIndex( + this.pastEvents, P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated, keepFirst: true); + return this.subOrchestrationInstanceCreatedEventsByEventId.TryGetValue(eventId, out P.HistoryEvent? historyEvent) + ? historyEvent + : null; + } + + static Dictionary BuildIndex( + IEnumerable events, P.HistoryEvent.EventTypeOneofCase eventType, bool keepFirst) + { + Dictionary index = new(); + foreach (P.HistoryEvent historyEvent in events) + { + if (historyEvent.EventTypeCase != eventType) + { + continue; + } + + if (keepFirst && index.ContainsKey(historyEvent.EventId)) + { + // Preserve first-match-wins semantics for duplicate event IDs. + continue; + } + + // Last write wins for duplicate event IDs, preserving last-match-wins semantics. + index[historyEvent.EventId] = historyEvent; + } + + return index; + } +} diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 6d63f6af..c4a7080e 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -586,107 +586,110 @@ async Task OnRunOrchestratorAsync( string completionToken, CancellationToken cancellationToken) { - var executionStartedEvent = - request - .NewEvents - .Concat(request.PastEvents) - .Where(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) - .Select(e => e.ExecutionStarted) - .FirstOrDefault(); - - Activity? traceActivity = TraceHelper.StartTraceActivityForOrchestrationExecution( - executionStartedEvent, - request.OrchestrationTraceContext); - - if (executionStartedEvent is not null) + // Avoid the cost of scanning orchestration history for tracing purposes (potentially O(new events x + // past events) for work items with many new events) when no listener is registered for the Durable + // Task ActivitySource. In that case any Activity created below would be discarded anyway. + Activity? traceActivity = null; + if (TraceHelper.HasListeners) { - P.HistoryEvent? GetSuborchestrationInstanceCreatedEvent(int eventId) - { - var subOrchestrationEvent = - request - .PastEvents - .Where(x => x.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated) - .FirstOrDefault(x => x.EventId == eventId); + P.ExecutionStartedEvent? executionStartedEvent = FindExecutionStartedEvent(request); - return subOrchestrationEvent; - } + traceActivity = TraceHelper.StartTraceActivityForOrchestrationExecution( + executionStartedEvent, + request.OrchestrationTraceContext); - P.HistoryEvent? GetTaskScheduledEvent(int eventId) + if (executionStartedEvent is not null) { - var taskScheduledEvent = - request - .PastEvents - .Where(x => x.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskScheduled) - .LastOrDefault(x => x.EventId == eventId); + // Build lookups once per work item instead of rescanning PastEvents for every new event. + TraceHistoryEventLookup historyLookup = new(request.PastEvents); - return taskScheduledEvent; - } - - foreach (var newEvent in request.NewEvents) - { - switch (newEvent.EventTypeCase) + foreach (var newEvent in request.NewEvents) { - case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted: - { - P.HistoryEvent? subOrchestrationInstanceCreatedEvent = - GetSuborchestrationInstanceCreatedEvent( - newEvent.SubOrchestrationInstanceCompleted.TaskScheduledId); - - TraceHelper.EmitTraceActivityForSubOrchestrationCompleted( - request.InstanceId, - subOrchestrationInstanceCreatedEvent, - subOrchestrationInstanceCreatedEvent?.SubOrchestrationInstanceCreated); - break; - } - - case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed: - { - P.HistoryEvent? subOrchestrationInstanceCreatedEvent = - GetSuborchestrationInstanceCreatedEvent( - newEvent.SubOrchestrationInstanceFailed.TaskScheduledId); - - TraceHelper.EmitTraceActivityForSubOrchestrationFailed( - request.InstanceId, - subOrchestrationInstanceCreatedEvent, - subOrchestrationInstanceCreatedEvent?.SubOrchestrationInstanceCreated, - newEvent.SubOrchestrationInstanceFailed); - break; - } + switch (newEvent.EventTypeCase) + { + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted: + { + P.HistoryEvent? subOrchestrationInstanceCreatedEvent = + historyLookup.GetSubOrchestrationInstanceCreatedEvent( + newEvent.SubOrchestrationInstanceCompleted.TaskScheduledId); + + TraceHelper.EmitTraceActivityForSubOrchestrationCompleted( + request.InstanceId, + subOrchestrationInstanceCreatedEvent, + subOrchestrationInstanceCreatedEvent?.SubOrchestrationInstanceCreated); + break; + } + + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed: + { + P.HistoryEvent? subOrchestrationInstanceCreatedEvent = + historyLookup.GetSubOrchestrationInstanceCreatedEvent( + newEvent.SubOrchestrationInstanceFailed.TaskScheduledId); + + TraceHelper.EmitTraceActivityForSubOrchestrationFailed( + request.InstanceId, + subOrchestrationInstanceCreatedEvent, + subOrchestrationInstanceCreatedEvent?.SubOrchestrationInstanceCreated, + newEvent.SubOrchestrationInstanceFailed); + break; + } + + case P.HistoryEvent.EventTypeOneofCase.TaskCompleted: + { + P.HistoryEvent? taskScheduledEvent = + historyLookup.GetTaskScheduledEvent(newEvent.TaskCompleted.TaskScheduledId); - case P.HistoryEvent.EventTypeOneofCase.TaskCompleted: - { - P.HistoryEvent? taskScheduledEvent = - GetTaskScheduledEvent(newEvent.TaskCompleted.TaskScheduledId); + TraceHelper.EmitTraceActivityForTaskCompleted( + request.InstanceId, + taskScheduledEvent, + taskScheduledEvent?.TaskScheduled); + break; + } - TraceHelper.EmitTraceActivityForTaskCompleted( + case P.HistoryEvent.EventTypeOneofCase.TaskFailed: + { + P.HistoryEvent? taskScheduledEvent = + historyLookup.GetTaskScheduledEvent(newEvent.TaskFailed.TaskScheduledId); + + TraceHelper.EmitTraceActivityForTaskFailed( + request.InstanceId, + taskScheduledEvent, + taskScheduledEvent?.TaskScheduled, + newEvent.TaskFailed); + break; + } + + case P.HistoryEvent.EventTypeOneofCase.TimerFired: + TraceHelper.EmitTraceActivityForTimer( request.InstanceId, - taskScheduledEvent, - taskScheduledEvent?.TaskScheduled); + executionStartedEvent.Name, + newEvent.Timestamp.ToDateTime(), + newEvent.TimerFired); break; - } + } + } + } + } - case P.HistoryEvent.EventTypeOneofCase.TaskFailed: - { - P.HistoryEvent? taskScheduledEvent = - GetTaskScheduledEvent(newEvent.TaskFailed.TaskScheduledId); + static P.ExecutionStartedEvent? FindExecutionStartedEvent(P.OrchestratorRequest request) + { + foreach (P.HistoryEvent newEvent in request.NewEvents) + { + if (newEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) + { + return newEvent.ExecutionStarted; + } + } - TraceHelper.EmitTraceActivityForTaskFailed( - request.InstanceId, - taskScheduledEvent, - taskScheduledEvent?.TaskScheduled, - newEvent.TaskFailed); - break; - } - - case P.HistoryEvent.EventTypeOneofCase.TimerFired: - TraceHelper.EmitTraceActivityForTimer( - request.InstanceId, - executionStartedEvent.Name, - newEvent.Timestamp.ToDateTime(), - newEvent.TimerFired); - break; + foreach (P.HistoryEvent pastEvent in request.PastEvents) + { + if (pastEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) + { + return pastEvent.ExecutionStarted; } } + + return null; } OrchestratorExecutionResult? result = null; diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs index bc9faab6..8f5d6e55 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs @@ -2,12 +2,14 @@ // Licensed under the MIT License. using System.Collections.Concurrent; +using System.Diagnostics; using System.IO; using System.Reflection; using Google.Protobuf.WellKnownTypes; using Grpc.Core; using Microsoft.DurableTask; using Microsoft.DurableTask.Tests.Logging; +using Microsoft.DurableTask.Tracing; using Microsoft.DurableTask.Worker; using Microsoft.DurableTask.Worker.Grpc.Internal; using Microsoft.Extensions.Logging; @@ -359,6 +361,154 @@ public async Task DispatchWorkItem_ActivityRequest_NotificationFailure_Completes logs.Should().Contain(log => log.Message.Contains("Activity notification callback failed for phase 'Completed'")); } + // The following two tests both touch the process-wide "Microsoft.DurableTask" ActivitySource used by + // TraceHelper, so they are kept in this class (whose test methods xunit runs sequentially by default) to + // avoid flaky interference between them. + [Fact] + public async Task DispatchWorkItem_OrchestratorRequest_NoActivityListeners_SkipsTracingWorkAndCompletes() + { + // Arrange: verify no listener is registered for the Durable Task ActivitySource, so that + // OnRunOrchestratorAsync takes the fast path that skips all trace-event lookup work. + TraceHelper.HasListeners.Should().BeFalse(); + + P.WorkItem orchestratorWorkItem = CreateOrchestratorWorkItemWithDuplicateEventIds(); + + TaskCompletionSource completed = new(TaskCreationOptions.RunContinuationsAsynchronously); + GrpcDurableTaskWorker worker = CreateActivityWorker(new GrpcDurableTaskWorkerOptions()); + Mock clientMock = new( + MockBehavior.Strict, + new object[] { Mock.Of() }); + clientMock + .Setup(client => client.CompleteOrchestratorTaskAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback(() => completed.TrySetResult()) + .Returns(CreateUnaryCall(Task.FromResult(new P.CompleteTaskResponse()))); + object processor = CreateProcessor(worker, clientMock.Object); + + // Act + InvokeDispatchWorkItem(processor, orchestratorWorkItem, CancellationToken.None); + await completed.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert: work item processing still completes normally with no listener registered. + clientMock.VerifyAll(); + } + + [Fact] + public async Task DispatchWorkItem_OrchestratorRequest_WithActivityListener_UsesFirstAndLastWinsSemantics() + { + // Arrange: register a listener so OnRunOrchestratorAsync performs the tracing-correlation work, and + // craft history containing duplicate event IDs to verify the last/first-wins lookup semantics are + // preserved by the new indexed TraceHistoryEventLookup. + ConcurrentQueue stoppedActivities = new(); + using ActivityListener listener = new() + { + ShouldListenTo = source => source.Name == "Microsoft.DurableTask", + Sample = (ref ActivityCreationOptions options) => ActivitySamplingResult.AllDataAndRecorded, + ActivityStopped = activity => stoppedActivities.Enqueue(activity), + }; + ActivitySource.AddActivityListener(listener); + TraceHelper.HasListeners.Should().BeTrue(); + + P.WorkItem orchestratorWorkItem = CreateOrchestratorWorkItemWithDuplicateEventIds(); + + TaskCompletionSource completed = new(TaskCreationOptions.RunContinuationsAsynchronously); + GrpcDurableTaskWorker worker = CreateActivityWorker(new GrpcDurableTaskWorkerOptions()); + Mock clientMock = new( + MockBehavior.Strict, + new object[] { Mock.Of() }); + clientMock + .Setup(client => client.CompleteOrchestratorTaskAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback(() => completed.TrySetResult()) + .Returns(CreateUnaryCall(Task.FromResult(new P.CompleteTaskResponse()))); + object processor = CreateProcessor(worker, clientMock.Object); + + // Act + InvokeDispatchWorkItem(processor, orchestratorWorkItem, CancellationToken.None); + await completed.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert: the "TaskScheduled" event with EventId=1 was duplicated; the last one (by history order) + // should win, matching the original LastOrDefault lookup semantics. + Activity taskActivity = stoppedActivities.Should().ContainSingle( + a => a.GetTagItem(Schema.Task.Type) as string == TraceActivityConstants.Activity).Subject; + taskActivity.GetTagItem(Schema.Task.Name).Should().Be("SecondScheduled"); + + // The "SubOrchestrationInstanceCreated" event with EventId=2 was duplicated; the first one (by history + // order) should win, matching the original FirstOrDefault lookup semantics. + Activity subOrchestrationActivity = stoppedActivities.Should().ContainSingle( + a => a.GetTagItem(Schema.Task.Type) as string == TraceActivityConstants.Orchestration + && a.OperationName.Contains("FirstSub", StringComparison.Ordinal)).Subject; + subOrchestrationActivity.GetTagItem(Schema.Task.Name).Should().Be("FirstSub"); + } + + static P.WorkItem CreateOrchestratorWorkItemWithDuplicateEventIds() + { + P.OrchestratorRequest request = new() + { + InstanceId = "instance1", + ExecutionId = "execution1", + }; + request.PastEvents.Add(new P.HistoryEvent + { + EventId = -1, + ExecutionStarted = new P.ExecutionStartedEvent + { + Name = "TestOrchestration", + OrchestrationInstance = new P.OrchestrationInstance { InstanceId = "instance1", ExecutionId = "execution1" }, + }, + }); + request.PastEvents.Add(new P.HistoryEvent + { + EventId = 1, + TaskScheduled = new P.TaskScheduledEvent { Name = "FirstScheduled" }, + }); + request.PastEvents.Add(new P.HistoryEvent + { + EventId = 1, + TaskScheduled = new P.TaskScheduledEvent { Name = "SecondScheduled" }, + }); + request.PastEvents.Add(new P.HistoryEvent + { + EventId = 2, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent + { + InstanceId = "sub1", + Name = "FirstSub", + }, + }); + request.PastEvents.Add(new P.HistoryEvent + { + EventId = 2, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent + { + InstanceId = "sub2", + Name = "SecondSub", + }, + }); + request.NewEvents.Add(new P.HistoryEvent + { + EventId = 10, + TaskCompleted = new P.TaskCompletedEvent { TaskScheduledId = 1 }, + }); + request.NewEvents.Add(new P.HistoryEvent + { + EventId = 11, + SubOrchestrationInstanceCompleted = new P.SubOrchestrationInstanceCompletedEvent { TaskScheduledId = 2 }, + }); + + return new P.WorkItem + { + OrchestratorRequest = request, + CompletionToken = "completion1", + }; + } + [Fact] public async Task ProcessorExecuteAsync_HelloDeadlineExceeded_ReturnsChannelRecreateRequested() { diff --git a/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs new file mode 100644 index 00000000..ab9f6544 --- /dev/null +++ b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs @@ -0,0 +1,134 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Tracing; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +public class TraceHistoryEventLookupTests +{ + [Fact] + public void GetTaskScheduledEvent_DuplicateEventIds_ReturnsLastMatch() + { + // Arrange + List pastEvents = + [ + CreateTaskScheduled(eventId: 1, name: "FirstScheduled"), + CreateTaskScheduled(eventId: 1, name: "SecondScheduled"), + ]; + TraceHistoryEventLookup lookup = new(pastEvents); + + // Act + P.HistoryEvent? result = lookup.GetTaskScheduledEvent(1); + + // Assert + result.Should().NotBeNull(); + result!.TaskScheduled.Name.Should().Be("SecondScheduled"); + } + + [Fact] + public void GetSubOrchestrationInstanceCreatedEvent_DuplicateEventIds_ReturnsFirstMatch() + { + // Arrange + List pastEvents = + [ + CreateSubOrchestrationInstanceCreated(eventId: 2, name: "FirstSub"), + CreateSubOrchestrationInstanceCreated(eventId: 2, name: "SecondSub"), + ]; + TraceHistoryEventLookup lookup = new(pastEvents); + + // Act + P.HistoryEvent? result = lookup.GetSubOrchestrationInstanceCreatedEvent(2); + + // Assert + result.Should().NotBeNull(); + result!.SubOrchestrationInstanceCreated.Name.Should().Be("FirstSub"); + } + + [Fact] + public void GetTaskScheduledEvent_NoMatch_ReturnsNull() + { + // Arrange + List pastEvents = [CreateTaskScheduled(eventId: 1, name: "Scheduled")]; + TraceHistoryEventLookup lookup = new(pastEvents); + + // Act + P.HistoryEvent? result = lookup.GetTaskScheduledEvent(99); + + // Assert + result.Should().BeNull(); + } + + [Fact] + public void GetSubOrchestrationInstanceCreatedEvent_NoMatch_ReturnsNull() + { + // Arrange + List pastEvents = [CreateSubOrchestrationInstanceCreated(eventId: 2, name: "Sub")]; + TraceHistoryEventLookup lookup = new(pastEvents); + + // Act + P.HistoryEvent? result = lookup.GetSubOrchestrationInstanceCreatedEvent(99); + + // Assert + result.Should().BeNull(); + } + + [Fact] + public void GetTaskScheduledEvent_IgnoresOtherEventTypesWithSameEventId() + { + // Arrange: a SubOrchestrationInstanceCreated event shares the event ID of the TaskScheduled event we + // look up, but must not be returned since it is a different history event type. + List pastEvents = + [ + CreateSubOrchestrationInstanceCreated(eventId: 5, name: "Sub"), + CreateTaskScheduled(eventId: 5, name: "Scheduled"), + ]; + TraceHistoryEventLookup lookup = new(pastEvents); + + // Act + P.HistoryEvent? taskScheduled = lookup.GetTaskScheduledEvent(5); + P.HistoryEvent? subOrchestrationCreated = lookup.GetSubOrchestrationInstanceCreatedEvent(5); + + // Assert + taskScheduled.Should().NotBeNull(); + taskScheduled!.TaskScheduled.Name.Should().Be("Scheduled"); + subOrchestrationCreated.Should().NotBeNull(); + subOrchestrationCreated!.SubOrchestrationInstanceCreated.Name.Should().Be("Sub"); + } + + [Fact] + public void GetTaskScheduledEvent_EmptyPastEvents_ReturnsNull() + { + // Arrange + TraceHistoryEventLookup lookup = new([]); + + // Act + P.HistoryEvent? result = lookup.GetTaskScheduledEvent(0); + + // Assert + result.Should().BeNull(); + } + + static P.HistoryEvent CreateTaskScheduled(int eventId, string name) + { + return new P.HistoryEvent + { + EventId = eventId, + TaskScheduled = new P.TaskScheduledEvent { Name = name }, + }; + } + + static P.HistoryEvent CreateSubOrchestrationInstanceCreated(int eventId, string name) + { + return new P.HistoryEvent + { + EventId = eventId, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent + { + InstanceId = $"sub-{eventId}", + Name = name, + }, + }; + } +}