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

@@ -148,6 +148,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

@@ -158,7 +158,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

@@ -94,8 +94,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

@@ -145,7 +145,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,

main.go

@@ -7,6 +7,11 @@ import (
 	"log"
 	"net"
 
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/exporters/zipkin"
+	"go.opentelemetry.io/otel/sdk/resource"
+	"go.opentelemetry.io/otel/sdk/trace"
 	"google.golang.org/grpc"
 
 	"github.com/microsoft/durabletask-go/backend"
@@ -23,6 +28,17 @@ func main() {
 	// Parse command-line arguments
 	flag.Parse()
 
+	// Tracing can be configured independently of the orchestration code.
+	tp, err := ConfigureZipkinTracing()
+	if err != nil {
+		log.Fatalf("Failed to create tracer: %v", err)
+	}
+	defer func() {
+		if err := tp.Shutdown(context.Background()); err != nil {
+			log.Fatalf("Failed to stop tracer: %v", err)
+		}
+	}()
+
 	grpcServer := grpc.NewServer()
 	worker := createTaskHubWorker(grpcServer, *dbFilePath, backend.DefaultLogger())
 	if err := worker.Start(ctx); err != nil {
@@ -40,6 +56,30 @@ func main() {
 	}
 }
 
+func ConfigureZipkinTracing() (*trace.TracerProvider, error) {
+	// Inspired by this sample: https://github.com/open-telemetry/opentelemetry-go/blob/main/example/zipkin/main.go
+	exp, err := zipkin.New("http://localhost:9411/api/v2/spans")
+	if err != nil {
+		return nil, err
+	}
+
+	// NOTE: The simple span processor is not recommended for production.
+	//       Instead, the batch span processor should be used for production.
+	processor := trace.NewSimpleSpanProcessor(exp)
+	// processor := trace.NewBatchSpanProcessor(exp)
+
+	tp := trace.NewTracerProvider(
+		trace.WithSpanProcessor(processor),
+		trace.WithSampler(trace.AlwaysSample()),
+		trace.WithResource(resource.NewWithAttributes(
+			"durabletask.io",
+			attribute.KeyValue{Key: "service.name", Value: attribute.StringValue("sample-app")},
+		)),
+	)
+	otel.SetTracerProvider(tp)
+	return tp, nil
+}
+
 func createTaskHubWorker(server *grpc.Server, sqliteFilePath string, logger backend.Logger) backend.TaskHubWorker {
 	sqliteOptions := sqlite.NewSqliteOptions(sqliteFilePath)
 	be := sqlite.NewSqliteBackend(sqliteOptions, logger)

samples/distributedtracing/distributedtracing.go

@@ -18,11 +18,13 @@ 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()
 	if err != nil {
-		log.Fatalf("Failed to create tracer: %w", err)
+		log.Fatalf("Failed to create tracer: %v", err)
 	}
 	defer func() {
 		if err := tp.Shutdown(context.Background()); err != nil {
@@ -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
@@ -351,8 +354,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())
+}