Distributed Tracing for Entities (Isolated)

Directory.Packages.props

@@ -28,8 +28,8 @@
     <PackageVersion Include="Microsoft.CodeAnalysis.CSharp" Version="3.9.0" />
     <PackageVersion Include="Microsoft.CodeAnalysis.CSharp.Workspaces" Version="3.9.0"  />
     <PackageVersion Include="Microsoft.CodeAnalysis.Workspaces.Common" Version="3.9.0" />
-    <PackageVersion Include="Microsoft.DurableTask.Client.Grpc" Version="1.10.0" />
-    <PackageVersion Include="Microsoft.DurableTask.Worker.Grpc" Version="1.10.0" />
+    <PackageVersion Include="Microsoft.DurableTask.Client.Grpc" Version="1.11.0" />
+    <PackageVersion Include="Microsoft.DurableTask.Worker.Grpc" Version="1.11.0" />
     <PackageVersion Include="Microsoft.Extensions.Azure" Version="1.7.0" />
     <PackageVersion Include="Microsoft.Extensions.Http" Version="6.0.0" />
     <PackageVersion Include="Microsoft.NET.Sdk.Functions" Version="4.2.0" />
@@ -65,7 +65,7 @@
     <PackageVersion Include="Microsoft.CodeAnalysis" Version="3.9.0" />
     <PackageVersion Include="Microsoft.CodeAnalysis.Workspaces.MSBuild" Version="3.9.0" />
     <PackageVersion Include="Microsoft.Diagnostics.Tracing.TraceEvent" Version="2.0.65" />
-    <PackageVersion Include="Microsoft.DurableTask.Abstractions" Version="1.10.0" />
+    <PackageVersion Include="Microsoft.DurableTask.Abstractions" Version="1.11.0" />
     <PackageVersion Include="Microsoft.DurableTask.Generators" Version="1.0.0-preview.1" />
     <PackageVersion Include="Microsoft.DurableTask.SqlServer.AzureFunctions" Version="1.5.0" />
     <PackageVersion Include="Microsoft.Extensions.Configuration" Version="9.0.0" />

src/WebJobs.Extensions.DurableTask/Correlation/TraceHelper.cs

@@ -17,14 +17,15 @@ internal class TraceHelper
         private const string Source = "WebJobs.Extensions.DurableTask";
 
         private static readonly ActivitySource ActivityTraceSource = new ActivitySource(Source);
-
-        internal static Activity? StartActivityForNewOrchestration(ExecutionStartedEvent startEvent, ActivityContext parentTraceContext)
+
+        internal static Activity? StartActivityForNewOrchestration(ExecutionStartedEvent startEvent, ActivityContext parentTraceContext, DateTimeOffset? startTime = default)
         {
             // Start the new activity to represent scheduling the orchestration
             Activity? newActivity = ActivityTraceSource.StartActivity(
                 Schema.SpanNames.CreateOrchestration(startEvent.Name, startEvent.Version),
-                kind: ActivityKind.Producer,
-                parentContext: parentTraceContext);
+                kind: ActivityKind.Producer,
+                parentContext: parentTraceContext,
+                startTime: startTime ?? default);
 
             if (newActivity == null)
             {
@@ -46,8 +47,8 @@ internal class TraceHelper
 
             return newActivity;
         }
-
-        internal static Activity? StartActivityForCallingOrSignalingEntity(string targetEntityId, string entityName, string operationName, bool signalEntity, DateTime? scheduledTime, ActivityContext? parentTraceContext, DateTimeOffset startTime = default, string? entityId = null)
+
+        internal static Activity? StartActivityForCallingOrSignalingEntity(string targetEntityId, string entityName, string operationName, bool signalEntity, DateTime? scheduledTime, ActivityContext? parentTraceContext, DateTimeOffset? startTime = default, string? entityId = null)
         {
             // We only want to create a trace activity for calling or signaling an entity in the case that we can successfully get the parent trace context of the request.
             // Otherwise, we will create an unlinked trace activity with no parent.
@@ -59,8 +60,8 @@ internal class TraceHelper
             Activity? newActivity = ActivityTraceSource.StartActivity(
                 Schema.SpanNames.CallOrSignalEntity(entityName, operationName),
                 kind: signalEntity ? ActivityKind.Producer : ActivityKind.Client,
-                parentContext: parentTraceContext.Value,
-                startTime: startTime);
+                parentContext: parentTraceContext.Value,
+                startTime: startTime ?? default);
 
             if (newActivity == null)
             {

src/WebJobs.Extensions.DurableTask/Grpc/orchestrator_service.proto

@@ -11,6 +11,7 @@ import "google/protobuf/timestamp.proto";
 import "google/protobuf/duration.proto";
 import "google/protobuf/wrappers.proto";
 import "google/protobuf/empty.proto";
+import "google/protobuf/struct.proto";
 
 message OrchestrationInstance {
     string instanceId = 1;
@@ -75,6 +76,7 @@ message ExecutionStartedEvent {
     google.protobuf.Timestamp scheduledStartTimestamp = 6;
     TraceContext parentTraceContext = 7;
     google.protobuf.StringValue orchestrationSpanID = 8;
+    map<string, string> tags = 9;
 }
 
 message ExecutionCompletedEvent {
@@ -93,6 +95,7 @@ message TaskScheduledEvent {
     google.protobuf.StringValue version = 2;
     google.protobuf.StringValue input = 3;
     TraceContext parentTraceContext = 4;
+    map<string, string> tags = 5;
 }
 
 message TaskCompletedEvent {
@@ -254,6 +257,7 @@ message ScheduleTaskAction {
     string name = 1;
     google.protobuf.StringValue version = 2;
     google.protobuf.StringValue input = 3;
+    map<string, string> tags = 4;
 }
 
 message CreateSubOrchestrationAction {
@@ -288,6 +292,15 @@ message TerminateOrchestrationAction {
     bool recurse = 3;
 }
 
+message SendEntityMessageAction {
+    oneof EntityMessageType {
+        EntityOperationSignaledEvent entityOperationSignaled = 1;
+        EntityOperationCalledEvent entityOperationCalled = 2;
+        EntityLockRequestedEvent entityLockRequested = 3;
+        EntityUnlockSentEvent entityUnlockSent = 4;
+    }
+}
+
 message OrchestratorAction {
     int32 id = 1;
     oneof orchestratorActionType {
@@ -297,6 +310,7 @@ message OrchestratorAction {
         SendEventAction sendEvent = 5;
         CompleteOrchestrationAction completeOrchestration = 6;
         TerminateOrchestrationAction terminateOrchestration = 7;
+        SendEntityMessageAction sendEntityMessage = 8;
     }
 }
 
@@ -307,6 +321,7 @@ message OrchestratorRequest {
     repeated HistoryEvent newEvents = 4;
     OrchestratorEntityParameters entityParameters = 5;
     bool requiresHistoryStreaming = 6;
+    map<string, google.protobuf.Value> properties = 7;
 }
 
 message OrchestratorResponse {
@@ -330,17 +345,12 @@ message CreateInstanceRequest {
     google.protobuf.StringValue executionId = 7;
     map<string, string> tags = 8;
     TraceContext parentTraceContext = 9;
+    google.protobuf.Timestamp requestTime = 10;
 }
 
 message OrchestrationIdReusePolicy {
-    repeated OrchestrationStatus operationStatus = 1;
-    CreateOrchestrationAction action = 2;
-}
-
-enum CreateOrchestrationAction {
-    ERROR = 0;
-    IGNORE = 1;
-    TERMINATE = 2;
+    repeated OrchestrationStatus replaceableStatus = 1;
+    reserved 2;
 }
 
 message CreateInstanceResponse {
@@ -381,6 +391,7 @@ message OrchestrationState {
     google.protobuf.StringValue executionId = 12;
     google.protobuf.Timestamp completedTimestamp = 13;
     google.protobuf.StringValue parentInstanceId = 14;
+    map<string, string> tags = 15;
 }
 
 message RaiseEventRequest {
@@ -457,6 +468,7 @@ message PurgeInstanceFilter {
 
 message PurgeInstancesResponse {
     int32 deletedInstanceCount = 1;
+    google.protobuf.BoolValue isComplete = 2;
 }
 
 message CreateTaskHubRequest {
@@ -481,6 +493,8 @@ message SignalEntityRequest {
     google.protobuf.StringValue input = 3;
     string requestId = 4;
     google.protobuf.Timestamp scheduledTime = 5;
+    TraceContext parentTraceContext = 6;
+    google.protobuf.Timestamp requestTime = 7;
 }
 
 message SignalEntityResponse {
@@ -551,6 +565,8 @@ message EntityBatchResult {
     repeated OperationAction actions = 2;
     google.protobuf.StringValue entityState = 3;
     TaskFailureDetails failureDetails = 4;
+    string completionToken = 5;
+    repeated OperationInfo operationInfos = 6; // used only with DTS
 }
 
 message EntityRequest {
@@ -564,6 +580,7 @@ message OperationRequest {
     string operation = 1;
     string requestId = 2;
     google.protobuf.StringValue input = 3;
+    TraceContext traceContext = 4;
 }
 
 message OperationResult {
@@ -573,12 +590,21 @@ message OperationResult {
     }
 }
 
+message OperationInfo {
+    string requestId = 1;
+    OrchestrationInstance responseDestination = 2; // null for signals
+}
+
 message OperationResultSuccess {
     google.protobuf.StringValue result = 1;
+    google.protobuf.Timestamp startTimeUtc = 2;
+    google.protobuf.Timestamp endTimeUtc = 3;
 }
 
 message OperationResultFailure {
     TaskFailureDetails failureDetails = 1;
+    google.protobuf.Timestamp startTimeUtc = 2;
+    google.protobuf.Timestamp endTimeUtc = 3;
 }
 
 message OperationAction {
@@ -594,6 +620,8 @@ message SendSignalAction {
     string name = 2;
     google.protobuf.StringValue input = 3;
     google.protobuf.Timestamp scheduledTime = 4;
+    google.protobuf.Timestamp requestTime = 5;
+    TraceContext parentTraceContext = 6;
 }
 
 message StartNewOrchestrationAction {
@@ -602,6 +630,32 @@ message StartNewOrchestrationAction {
     google.protobuf.StringValue version = 3;
     google.protobuf.StringValue input = 4;
     google.protobuf.Timestamp scheduledTime = 5;
+    google.protobuf.Timestamp requestTime = 6;
+    TraceContext parentTraceContext = 7;
+}
+
+message AbandonActivityTaskRequest {
+    string completionToken = 1;
+}
+
+message AbandonActivityTaskResponse {
+    // Empty.
+}
+
+message AbandonOrchestrationTaskRequest {
+    string completionToken = 1;
+}
+
+message AbandonOrchestrationTaskResponse {
+    // Empty.
+}
+
+message AbandonEntityTaskRequest {
+    string completionToken = 1;
+}
+
+message AbandonEntityTaskResponse {
+    // Empty.
 }
 
 service TaskHubSidecarService {
@@ -665,6 +719,15 @@ service TaskHubSidecarService {
 
     // clean entity storage
     rpc CleanEntityStorage(CleanEntityStorageRequest) returns (CleanEntityStorageResponse);
+
+    // Abandons a single work item
+    rpc AbandonTaskActivityWorkItem(AbandonActivityTaskRequest) returns (AbandonActivityTaskResponse);
+
+    // Abandon an orchestration work item
+    rpc AbandonTaskOrchestratorWorkItem(AbandonOrchestrationTaskRequest) returns (AbandonOrchestrationTaskResponse);
+
+    // Abandon an entity work item
+    rpc AbandonTaskEntityWorkItem(AbandonEntityTaskRequest) returns (AbandonEntityTaskResponse);
 }
 
 message GetWorkItemsRequest {
@@ -714,4 +777,4 @@ message StreamInstanceHistoryRequest {
 
 message HistoryChunk {
     repeated HistoryEvent events = 1;
-}
+}

src/WebJobs.Extensions.DurableTask/Grpc/versions.txt

@@ -1,2 +1,2 @@
-# The following files were downloaded from branch main at 2025-02-07 22:22:03 UTC
-https://raw.githubusercontent.com/microsoft/durabletask-protobuf/6000187e90d79fba297dfd75d42095abc1462eba/protos/orchestrator_service.proto
+# The following files were downloaded from branch main at 2025-06-05 21:25:17 UTC
+https://raw.githubusercontent.com/microsoft/durabletask-protobuf/fd9369c6a03d6af4e95285e432b7c4e943c06970/protos/orchestrator_service.proto

src/WebJobs.Extensions.DurableTask/Listener/TaskEntityShim.cs

@@ -755,7 +755,7 @@ private async Task ExecuteOutOfProcBatch()
                         request.IsSignal,
                         request.ScheduledTime,
                         parentTraceContext,
-                        request.RequestTime.Value);
+                        request.RequestTime);
                     parentTraceContext = callEntityActivity.Context;
                 }
 

src/WebJobs.Extensions.DurableTask/LocalGrpcListener.cs

@@ -236,7 +236,7 @@ public override Task<Empty> Hello(Empty request, ServerCallContext context)
 
                     // Create a new activity with the parent context
                     ActivityContext.TryParse(traceParent, traceState, out ActivityContext parentActivityContext);
-                    using Activity? scheduleOrchestrationActivity = TraceHelper.StartActivityForNewOrchestration(executionStartedEvent, parentActivityContext);
+                    using Activity? scheduleOrchestrationActivity = TraceHelper.StartActivityForNewOrchestration(executionStartedEvent, parentActivityContext, request.RequestTime?.ToDateTimeOffset());
 
                     // Schedule the orchestration
                     await this.GetDurabilityProvider(context).CreateTaskOrchestrationAsync(
@@ -319,17 +319,35 @@ private OrchestrationStatus[] GetStatusesNotToOverride()
             {
                 this.CheckEntitySupport(context, out var durabilityProvider, out var entityOrchestrationService);
 
-                EntityMessageEvent eventToSend = ClientEntityHelpers.EmitOperationSignal(
-                    new OrchestrationInstance() { InstanceId = request.InstanceId },
-                    Guid.Parse(request.RequestId),
-                    request.Name,
-                    request.Input,
-                    EntityMessageEvent.GetCappedScheduledTime(
-                        DateTime.UtcNow,
-                        entityOrchestrationService.EntityBackendProperties!.MaximumSignalDelayTime,
-                        request.ScheduledTime?.ToDateTime()));
+                Activity? signalEntityActivity = null;
+
+                // We only want to create a trace activity for signaling the entity in the case that we can successfully parse the trace context of the signal entity request.
+                // Otherwise, we will create an unlinked trace activity with no parent.
+                if (ActivityContext.TryParse(request.ParentTraceContext?.TraceParent, request.ParentTraceContext?.TraceState, out ActivityContext parentTraceContext))
+                {
+                    signalEntityActivity = TraceHelper.StartActivityForCallingOrSignalingEntity(
+                        request.InstanceId,
+                        EntityId.GetEntityIdFromSchedulerId(request.InstanceId).EntityName,
+                        request.Name,
+                        signalEntity: true,
+                        request.ScheduledTime?.ToDateTime(),
+                        parentTraceContext,
+                        request.RequestTime?.ToDateTime());
+                }
+
+                EntityMessageEvent eventToSend = ClientEntityHelpers.EmitOperationSignal(
+                    new OrchestrationInstance() { InstanceId = request.InstanceId },
+                    Guid.Parse(request.RequestId),
+                    request.Name,
+                    request.Input,
+                    EntityMessageEvent.GetCappedScheduledTime(
+                        DateTime.UtcNow,
+                        entityOrchestrationService.EntityBackendProperties!.MaximumSignalDelayTime,
+                        request.ScheduledTime?.ToDateTime()),
+                    signalEntityActivity != null ? new DTCore.Tracing.DistributedTraceContext(signalEntityActivity.Id!, signalEntityActivity.TraceStateString) : null);
 
                 await durabilityProvider.SendTaskOrchestrationMessageAsync(eventToSend.AsTaskMessage());
+                signalEntityActivity?.Dispose();
 
                 // No fields in the response
                 return new P.SignalEntityResponse();