Merge branch 'main' into nytian/restart-api

Author: nytian

.gitignore

@@ -348,3 +348,6 @@ MigrationBackup/
 
 # Ionide (cross platform F# VS Code tools) working folder
 .ionide/
+
+# Rider (cross platform .NET/C# tools) working folder
+.idea/

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -5,6 +5,7 @@
 using System.Text;
 using Google.Protobuf.WellKnownTypes;
 using Microsoft.DurableTask.Client.Entities;
+using Microsoft.DurableTask.Tracing;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
@@ -107,24 +108,6 @@ public override async Task<string> ScheduleNewOrchestrationInstanceAsync(
             }
         }
 
-        if (Activity.Current?.Id != null || Activity.Current?.TraceStateString != null)
-        {
-            if (request.ParentTraceContext == null)
-            {
-                request.ParentTraceContext = new P.TraceContext();
-            }
-
-            if (Activity.Current?.Id != null)
-            {
-                request.ParentTraceContext.TraceParent = Activity.Current?.Id;
-            }
-
-            if (Activity.Current?.TraceStateString != null)
-            {
-                request.ParentTraceContext.TraceState = Activity.Current?.TraceStateString;
-            }
-        }
-
         DateTimeOffset? startAt = options?.StartAt;
         this.logger.SchedulingOrchestration(
             request.InstanceId,
@@ -138,6 +121,8 @@ public override async Task<string> ScheduleNewOrchestrationInstanceAsync(
             request.ScheduledStartTimestamp = Timestamp.FromDateTimeOffset(startAt.Value.ToUniversalTime());
         }
 
+        using Activity? newActivity = TraceHelper.StartActivityForNewOrchestration(request);
+
         P.CreateInstanceResponse? result = await this.sidecarClient.StartInstanceAsync(
             request, cancellationToken: cancellation);
         return result.InstanceId;
@@ -159,6 +144,8 @@ public override async Task RaiseEventAsync(
             Input = this.DataConverter.Serialize(eventPayload),
         };
 
+        using Activity? traceActivity = TraceHelper.StartActivityForNewEventRaisedFromClient(request, instanceId);
+
         await this.sidecarClient.RaiseEventAsync(request, cancellationToken: cancellation);
     }
 

src/Grpc/orchestrator_service.proto

@@ -193,7 +193,7 @@ message EntityOperationCalledEvent {
 }
 
 message EntityLockRequestedEvent {
-    string criticalSectionId = 1;  
+    string criticalSectionId = 1;
     repeated string lockSet = 2;
     int32 position = 3;
     google.protobuf.StringValue parentInstanceId = 4; // used only within messages, null in histories 
@@ -218,7 +218,7 @@ message EntityUnlockSentEvent {
 message EntityLockGrantedEvent {
     string criticalSectionId = 1;
 }
- 
+
 message HistoryEvent {
     int32 eventId = 1;
     google.protobuf.Timestamp timestamp = 2;
@@ -245,8 +245,8 @@ message HistoryEvent {
         ExecutionResumedEvent executionResumed = 22;
         EntityOperationSignaledEvent entityOperationSignaled = 23;
         EntityOperationCalledEvent entityOperationCalled = 24;
-        EntityOperationCompletedEvent entityOperationCompleted = 25; 
-        EntityOperationFailedEvent entityOperationFailed = 26; 
+        EntityOperationCompletedEvent entityOperationCompleted = 25;
+        EntityOperationFailedEvent entityOperationFailed = 26;
         EntityLockRequestedEvent entityLockRequested = 27;
         EntityLockGrantedEvent entityLockGranted = 28;
         EntityUnlockSentEvent entityUnlockSent = 29;
@@ -258,13 +258,15 @@ message ScheduleTaskAction {
     google.protobuf.StringValue version = 2;
     google.protobuf.StringValue input = 3;
     map<string, string> tags = 4;
+    TraceContext parentTraceContext = 5;
 }
 
 message CreateSubOrchestrationAction {
     string instanceId = 1;
     string name = 2;
     google.protobuf.StringValue version = 3;
     google.protobuf.StringValue input = 4;
+    TraceContext parentTraceContext = 5;
 }
 
 message CreateTimerAction {
@@ -314,6 +316,11 @@ message OrchestratorAction {
     }
 }
 
+message OrchestrationTraceContext {
+    google.protobuf.StringValue spanID = 1;
+    google.protobuf.Timestamp spanStartTime = 2;   
+}
+
 message OrchestratorRequest {
     string instanceId = 1;
     google.protobuf.StringValue executionId = 2;
@@ -322,6 +329,8 @@ message OrchestratorRequest {
     OrchestratorEntityParameters entityParameters = 5;
     bool requiresHistoryStreaming = 6;
     map<string, google.protobuf.Value> properties = 7;
+
+    OrchestrationTraceContext orchestrationTraceContext = 8;
 }
 
 message OrchestratorResponse {
@@ -333,6 +342,8 @@ message OrchestratorResponse {
     // The number of work item events that were processed by the orchestrator.
     // This field is optional. If not set, the service should assume that the orchestrator processed all events.
     google.protobuf.Int32Value numEventsProcessed = 5;
+
+    OrchestrationTraceContext orchestrationTraceContext = 6;
 }
 
 message CreateInstanceRequest {
@@ -507,7 +518,7 @@ message SignalEntityRequest {
 }
 
 message SignalEntityResponse {
-     // no payload
+    // no payload
 }
 
 message GetEntityRequest {
@@ -682,16 +693,16 @@ service TaskHubSidecarService {
 
     // Waits for an orchestration instance to reach a running or completion state.
     rpc WaitForInstanceStart(GetInstanceRequest) returns (GetInstanceResponse);
-    
+
     // Waits for an orchestration instance to reach a completion state (completed, failed, terminated, etc.).
     rpc WaitForInstanceCompletion(GetInstanceRequest) returns (GetInstanceResponse);
 
     // Raises an event to a running orchestration instance.
     rpc RaiseEvent(RaiseEventRequest) returns (RaiseEventResponse);
-    
+
     // Terminates a running orchestration instance.
     rpc TerminateInstance(TerminateRequest) returns (TerminateResponse);
-    
+
     // Suspends a running orchestration instance.
     rpc SuspendInstance(SuspendRequest) returns (SuspendResponse);
 
@@ -776,7 +787,7 @@ message CompleteTaskResponse {
 }
 
 message HealthPing {
-	// No payload
+    // No payload
 }
 
 message StreamInstanceHistoryRequest {
@@ -789,4 +800,4 @@ message StreamInstanceHistoryRequest {
 
 message HistoryChunk {
     repeated HistoryEvent events = 1;
-}
+}

src/Grpc/versions.txt

@@ -1,2 +1,2 @@
-# The following files were downloaded from branch main at 2025-06-02 21:12:34 UTC
-https://raw.githubusercontent.com/microsoft/durabletask-protobuf/fd9369c6a03d6af4e95285e432b7c4e943c06970/protos/orchestrator_service.proto
+# The following files were downloaded from branch main at 2025-08-08 16:46:11 UTC
+https://raw.githubusercontent.com/microsoft/durabletask-protobuf/e88acbd07ae38b499dbe8c4e333e9e3feeb2a9cc/protos/orchestrator_service.proto

src/Shared/Grpc/ProtoUtils.cs

@@ -3,6 +3,7 @@
 
 using System.Buffers;
 using System.Buffers.Text;
+using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using System.Text;
 using DurableTask.Core;
@@ -15,6 +16,7 @@
 using Google.Protobuf.WellKnownTypes;
 using DTCore = DurableTask.Core;
 using P = Microsoft.DurableTask.Protobuf;
+using TraceHelper = Microsoft.DurableTask.Tracing.TraceHelper;
 
 namespace Microsoft.DurableTask;
 
@@ -268,43 +270,72 @@ internal static Timestamp ToTimestamp(this DateTime dateTime)
     /// Constructs a <see cref="P.OrchestratorResponse" />.
     /// </summary>
     /// <param name="instanceId">The orchestrator instance ID.</param>
+    /// <param name="executionId">The orchestrator execution ID.</param>
     /// <param name="customStatus">The orchestrator customer status or <c>null</c> if no custom status.</param>
     /// <param name="actions">The orchestrator actions.</param>
     /// <param name="completionToken">
     /// The completion token for the work item. It must be the exact same <see cref="P.WorkItem.CompletionToken" />
     /// value that was provided by the corresponding <see cref="P.WorkItem"/> that triggered the orchestrator execution.
     /// </param>
     /// <param name="entityConversionState">The entity conversion state, or null if no conversion is required.</param>
+    /// <param name="orchestrationActivity">The <see cref="Activity" /> that represents orchestration execution.</param>
     /// <returns>The orchestrator response.</returns>
     /// <exception cref="NotSupportedException">When an orchestrator action is unknown.</exception>
     internal static P.OrchestratorResponse ConstructOrchestratorResponse(
         string instanceId,
+        string executionId,
         string? customStatus,
         IEnumerable<OrchestratorAction> actions,
         string completionToken,
-        EntityConversionState? entityConversionState)
+        EntityConversionState? entityConversionState,
+        Activity? orchestrationActivity)
     {
         Check.NotNull(actions);
         var response = new P.OrchestratorResponse
         {
             InstanceId = instanceId,
             CustomStatus = customStatus,
             CompletionToken = completionToken,
+            OrchestrationTraceContext =
+                new()
+                {
+                    SpanID = orchestrationActivity?.SpanId.ToString(),
+                    SpanStartTime = orchestrationActivity?.StartTimeUtc.ToTimestamp(),
+                },
         };
 
         foreach (OrchestratorAction action in actions)
         {
             var protoAction = new P.OrchestratorAction { Id = action.Id };
 
+            P.TraceContext? CreateTraceContext()
+            {
+                if (orchestrationActivity is null)
+                {
+                    return null;
+                }
+
+                ActivitySpanId clientSpanId = ActivitySpanId.CreateRandom();
+                ActivityContext clientActivityContext = new(orchestrationActivity.TraceId, clientSpanId, orchestrationActivity.ActivityTraceFlags, orchestrationActivity.TraceStateString);
+
+                return new P.TraceContext
+                {
+                    TraceParent = $"00-{clientActivityContext.TraceId}-{clientActivityContext.SpanId}-0{clientActivityContext.TraceFlags:d}",
+                    TraceState = clientActivityContext.TraceState,
+                };
+            }
+
             switch (action.OrchestratorActionType)
             {
                 case OrchestratorActionType.ScheduleOrchestrator:
                     var scheduleTaskAction = (ScheduleTaskOrchestratorAction)action;
+
                     protoAction.ScheduleTask = new P.ScheduleTaskAction
                     {
                         Name = scheduleTaskAction.Name,
                         Version = scheduleTaskAction.Version,
                         Input = scheduleTaskAction.Input,
+                        ParentTraceContext = CreateTraceContext(),
                     };
 
                     if (scheduleTaskAction.Tags != null)
@@ -324,6 +355,7 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse(
                         InstanceId = subOrchestrationAction.InstanceId,
                         Name = subOrchestrationAction.Name,
                         Version = subOrchestrationAction.Version,
+                        ParentTraceContext = CreateTraceContext(),
                     };
                     break;
                 case OrchestratorActionType.CreateTimer:
@@ -378,6 +410,12 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse(
                             Name = sendEventAction.EventName,
                             Data = sendEventAction.EventData,
                         };
+
+                        // Distributed Tracing: start a new trace activity derived from the orchestration
+                        // for an EventRaisedEvent (external event)
+                        using Activity? traceActivity = TraceHelper.StartTraceActivityForEventRaisedFromWorker(sendEventAction, instanceId, executionId);
+
+                        traceActivity?.Stop();
                     }
 
                     break;