Add trace support into task

backend/activity.go

@@ -68,6 +68,8 @@ func (p *activityProcessor) ProcessWorkItem(ctx context.Context, wi WorkItem) er
 		}()
 	}
 
+	// set the parent trace context to be the newly created activity span
+	ts.ParentTraceContext = helpers.TraceContextFromSpan(span)
 	// Execute the activity and get its result
 	result, err := p.executor.ExecuteActivity(ctx, awi.InstanceID, awi.NewEvent)
 	if err != nil {

backend/executor.go

@@ -153,6 +153,7 @@ func (executor *grpcExecutor) ExecuteActivity(ctx context.Context, iid api.Insta
 				Input:                 task.Input,
 				OrchestrationInstance: &protos.OrchestrationInstance{InstanceId: string(iid)},
 				TaskId:                e.EventId,
+				ParentTraceContext:    task.ParentTraceContext,
 			},
 		},
 	}

backend/runtimestate.go

@@ -159,7 +159,7 @@ func (s *OrchestrationRuntimeState) ApplyActions(actions []*protos.OrchestratorA
 			}
 		} else if createtimer := action.GetCreateTimer(); createtimer != nil {
 			s.AddEvent(helpers.NewTimerCreatedEvent(action.Id, createtimer.FireAt))
-			s.pendingTimers = append(s.pendingTimers, helpers.NewTimerFiredEvent(action.Id, createtimer.FireAt, currentTraceContext))
+			s.pendingTimers = append(s.pendingTimers, helpers.NewTimerFiredEvent(action.Id, createtimer.FireAt))
 		} else if scheduleTask := action.GetScheduleTask(); scheduleTask != nil {
 			scheduledEvent := helpers.NewTaskScheduledEvent(
 				action.Id,

client/worker_grpc.go

@@ -169,8 +169,12 @@ func (c *TaskHubGrpcClient) processActivityWorkItem(
 	executor backend.Executor,
 	req *protos.ActivityRequest,
 ) {
-	var tc *protos.TraceContext = nil // TODO: How to populate trace context?
-	event := helpers.NewTaskScheduledEvent(req.TaskId, req.Name, req.Version, req.Input, tc)
+	var ptc *protos.TraceContext = req.ParentTraceContext
+	ctx, err := helpers.ContextFromTraceContext(ctx, ptc)
+	if err != nil {
+		fmt.Printf("%v: failed to parse trace context: %v", req.Name, err)
+	}
+	event := helpers.NewTaskScheduledEvent(req.TaskId, req.Name, req.Version, req.Input, ptc)
 	result, err := executor.ExecuteActivity(ctx, api.InstanceID(req.OrchestrationInstance.InstanceId), event)
 
 	resp := protos.ActivityResponse{InstanceId: req.OrchestrationInstance.InstanceId, TaskId: req.TaskId}

internal/helpers/history.go

@@ -147,7 +147,6 @@ func NewTimerCreatedEvent(eventID int32, fireAt *timestamppb.Timestamp) *protos.
 func NewTimerFiredEvent(
 	timerID int32,
 	fireAt *timestamppb.Timestamp,
-	parentTraceContext *protos.TraceContext,
 ) *protos.HistoryEvent {
 	return &protos.HistoryEvent{
 		EventId:   -1,

samples/distributedtracing/distributedtracing.go

@@ -18,6 +18,8 @@ import (
 	"github.com/microsoft/durabletask-go/task"
 )
 
+var tracer = otel.Tracer("distributedtracing-example")
+
 func main() {
 	// Tracing can be configured independently of the orchestration code.
 	tp, err := ConfigureZipkinTracing()
@@ -136,6 +138,16 @@ func DoWorkActivity(ctx task.ActivityContext) (any, error) {
 		return "", err
 	}
 
+	_, childSpan := tracer.Start(ctx.Context(), "activity-subwork")
+	// Simulate doing some sub work
+	select {
+	case <-time.After(2 * time.Second):
+		// Ok
+	case <-ctx.Context().Done():
+		return nil, ctx.Context().Err()
+	}
+	childSpan.End()
+
 	// Simulate doing work
 	select {
 	case <-time.After(duration):

tests/grpc/grpc_test.go

@@ -16,15 +16,18 @@ import (
 	"github.com/microsoft/durabletask-go/client"
 	"github.com/microsoft/durabletask-go/internal/protos"
 	"github.com/microsoft/durabletask-go/task"
+	"github.com/microsoft/durabletask-go/tests/utils"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )
 
 var (
 	grpcClient *client.TaskHubGrpcClient
 	ctx        = context.Background()
+	tracer     = otel.Tracer("grpc-test")
 )
 
 // TestMain is the entry point for the test suite. We use this to set up a gRPC server and client instance
@@ -417,8 +420,56 @@ func Test_Grpc_ReuseInstanceIDError(t *testing.T) {
 
 	id, err := grpcClient.ScheduleNewOrchestration(ctx, "SingleActivity", api.WithInput("世界"), api.WithInstanceID(instanceID))
 	require.NoError(t, err)
-	id, err = grpcClient.ScheduleNewOrchestration(ctx, "SingleActivity", api.WithInput("World"), api.WithInstanceID(id))
+	_, err = grpcClient.ScheduleNewOrchestration(ctx, "SingleActivity", api.WithInput("World"), api.WithInstanceID(id))
 	if assert.Error(t, err) {
 		assert.Contains(t, err.Error(), "orchestration instance already exists")
 	}
 }
+
+func Test_SingleActivity_TaskSpan(t *testing.T) {
+	// Registration
+	r := task.NewTaskRegistry()
+	r.AddOrchestratorN("SingleActivity_TestSpan", func(ctx *task.OrchestrationContext) (any, error) {
+		var input string
+		if err := ctx.GetInput(&input); err != nil {
+			return nil, err
+		}
+		var output string
+		err := ctx.CallActivity("SayHello", task.WithActivityInput(input)).Await(&output)
+		return output, err
+	})
+	r.AddActivityN("SayHello", func(ctx task.ActivityContext) (any, error) {
+		var name string
+		if err := ctx.GetInput(&name); err != nil {
+			return nil, err
+		}
+		_, childSpan := tracer.Start(ctx.Context(), "activityChild_TestSpan")
+		childSpan.End()
+		return fmt.Sprintf("Hello, %s!", name), nil
+	})
+
+	exporter := utils.InitTracing()
+	cancelListener := startGrpcListener(t, r)
+	defer cancelListener()
+
+	// Run the orchestration
+	id, err := grpcClient.ScheduleNewOrchestration(ctx, "SingleActivity_TestSpan", api.WithInput("世界"))
+	if assert.NoError(t, err) {
+		metadata, err := grpcClient.WaitForOrchestrationCompletion(ctx, id)
+		if assert.NoError(t, err) {
+			assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus)
+			assert.Equal(t, `"Hello, 世界!"`, metadata.SerializedOutput)
+		}
+	}
+
+	// Validate the exported OTel traces
+	spans := exporter.GetSpans().Snapshots()
+	utils.AssertSpanSequence(t, spans,
+		utils.AssertOrchestratorCreated("SingleActivity_TestSpan", id),
+		utils.AssertSpan("activityChild_TestSpan"),
+		utils.AssertActivity("SayHello", id, 0),
+		utils.AssertOrchestratorExecuted("SingleActivity_TestSpan", id, "COMPLETED"),
+	)
+	// assert child-parent relationship
+	assert.Equal(t, spans[1].Parent().SpanID(), spans[2].SpanContext().SpanID())
+}