Beginning of replay-aware workflow tracer

Author: cschleiden

internal/workflow/executor.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/task"
 	"github.com/cschleiden/go-workflows/internal/tracing"
 	"github.com/cschleiden/go-workflows/internal/workflowstate"
+	"github.com/cschleiden/go-workflows/internal/workflowtracer"
 	"github.com/cschleiden/go-workflows/log"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
@@ -42,7 +43,7 @@ type executor struct {
 	registry          *Registry
 	historyProvider   WorkflowHistoryProvider
 	workflow          *workflow
-	workflowTracer    *tracing.WorkflowTracer
+	workflowTracer    *workflowtracer.WorkflowTracer
 	workflowState     *workflowstate.WfState
 	workflowCtx       sync.Context
 	workflowCtxCancel sync.CancelFunc
@@ -55,11 +56,11 @@ type executor struct {
 func NewExecutor(logger log.Logger, tracer trace.Tracer, registry *Registry, historyProvider WorkflowHistoryProvider, instance *core.WorkflowInstance, clock clock.Clock) (WorkflowExecutor, error) {
 	s := workflowstate.NewWorkflowState(instance, logger, clock)
 
-	wfTracer := tracing.NewWorkflowTracer(tracer)
+	wfTracer := workflowtracer.New(tracer)
 
 	wfCtx, cancel := sync.WithCancel(
 		workflowstate.WithWorkflowState(
-			tracing.WithWorkflowTracer(
+			workflowtracer.WithWorkflowTracer(
 				sync.Background(),
 				wfTracer,
 			),

internal/tracing/tracer.go renamed to internal/workflowtracer/tracer.go

@@ -1,9 +1,10 @@
-package tracing
+package workflowtracer
 
 import (
 	"context"
 
 	"github.com/cschleiden/go-workflows/internal/sync"
+	"github.com/cschleiden/go-workflows/internal/workflowstate"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -28,7 +29,7 @@ type WorkflowTracer struct {
 	tracer     trace.Tracer
 }
 
-func NewWorkflowTracer(tracer trace.Tracer) *WorkflowTracer {
+func New(tracer trace.Tracer) *WorkflowTracer {
 	return &WorkflowTracer{
 		tracer: tracer,
 	}
@@ -38,8 +39,15 @@ func (wt *WorkflowTracer) UpdateExecution(span trace.Span) {
 	wt.parentSpan = span
 }
 
-func (wt *WorkflowTracer) Start(name string, opts ...trace.SpanStartOption) trace.Span {
-	ctx := trace.ContextWithSpan(context.Background(), wt.parentSpan)
-	_, span := wt.tracer.Start(ctx, name, opts...)
+func (wt *WorkflowTracer) Start(ctx sync.Context, name string, opts ...trace.SpanStartOption) trace.Span {
+	sctx := trace.ContextWithSpan(context.Background(), wt.parentSpan)
+	sctx, span := wt.tracer.Start(sctx, name, opts...)
+
+	state := workflowstate.WorkflowState(ctx)
+	if state.Replaying() {
+		sctx = trace.ContextWithSpanContext(sctx, span.SpanContext())
+		return trace.SpanFromContext(sctx)
+	}
+
 	return span
 }

workflow/activity.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/sync"
 	"github.com/cschleiden/go-workflows/internal/tracing"
 	"github.com/cschleiden/go-workflows/internal/workflowstate"
+	"github.com/cschleiden/go-workflows/internal/workflowtracer"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -51,7 +52,8 @@ func executeActivity[TResult any](ctx Context, options ActivityOptions, attempt
 	wfState.AddCommand(cmd)
 	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(f))
 
-	span := tracing.Tracer(ctx).Start(
+	span := workflowtracer.Tracer(ctx).Start(
+		ctx,
 		"ExecuteActivity", trace.WithAttributes(
 			attribute.String("name", name),
 			attribute.Int64(tracing.ScheduleEventID, scheduleEventID),

workflow/sideeffect.go

@@ -4,12 +4,12 @@ import (
 	"github.com/cschleiden/go-workflows/internal/command"
 	"github.com/cschleiden/go-workflows/internal/converter"
 	"github.com/cschleiden/go-workflows/internal/sync"
-	"github.com/cschleiden/go-workflows/internal/tracing"
 	"github.com/cschleiden/go-workflows/internal/workflowstate"
+	"github.com/cschleiden/go-workflows/internal/workflowtracer"
 )
 
 func SideEffect[TResult any](ctx Context, f func(ctx Context) TResult) Future[TResult] {
-	span := tracing.Tracer(ctx).Start("SideEffect")
+	span := workflowtracer.Tracer(ctx).Start(ctx, "SideEffect")
 	defer span.End()
 
 	future := sync.NewFuture[TResult]()

workflow/sleep.go

@@ -4,13 +4,13 @@ import (
 	"time"
 
 	"github.com/cschleiden/go-workflows/internal/sync"
-	"github.com/cschleiden/go-workflows/internal/tracing"
+	"github.com/cschleiden/go-workflows/internal/workflowtracer"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 )
 
 func Sleep(ctx sync.Context, d time.Duration) error {
-	span := tracing.Tracer(ctx).Start("Sleep",
+	span := workflowtracer.Tracer(ctx).Start(ctx, "Sleep",
 		trace.WithAttributes(attribute.Int64("duration_s", int64(d/time.Second))))
 	defer span.End()
 

workflow/subworkflow.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/sync"
 	"github.com/cschleiden/go-workflows/internal/tracing"
 	"github.com/cschleiden/go-workflows/internal/workflowstate"
+	"github.com/cschleiden/go-workflows/internal/workflowtracer"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -30,7 +31,7 @@ func CreateSubWorkflowInstance[TResult any](ctx sync.Context, options SubWorkflo
 	})
 }
 
-func createSubWorkflowInstance[TResult any](ctx sync.Context, options SubWorkflowOptions, attempt int, workflow interface{}, args ...interface{}) Future[TResult] {
+func createSubWorkflowInstance[TResult any](ctx sync.Context, options SubWorkflowOptions, attempt int, wf interface{}, args ...interface{}) Future[TResult] {
 	f := sync.NewFuture[TResult]()
 
 	// If the context is already canceled, return immediately.
@@ -39,7 +40,7 @@ func createSubWorkflowInstance[TResult any](ctx sync.Context, options SubWorkflo
 		return f
 	}
 
-	name := fn.Name(workflow)
+	name := fn.Name(wf)
 
 	inputs, err := a.ArgsToInputs(converter.DefaultConverter, args...)
 	if err != nil {
@@ -54,7 +55,7 @@ func createSubWorkflowInstance[TResult any](ctx sync.Context, options SubWorkflo
 
 	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(f))
 
-	span := tracing.Tracer(ctx).Start(
+	span := workflowtracer.Tracer(ctx).Start(ctx,
 		"CreateSubworkflowInstance", trace.WithAttributes(
 			attribute.String("name", name),
 			attribute.Int64(tracing.ScheduleEventID, scheduleEventID),

workflow/timer.go

@@ -5,8 +5,8 @@ import (
 
 	"github.com/cschleiden/go-workflows/internal/command"
 	"github.com/cschleiden/go-workflows/internal/sync"
-	"github.com/cschleiden/go-workflows/internal/tracing"
 	"github.com/cschleiden/go-workflows/internal/workflowstate"
+	"github.com/cschleiden/go-workflows/internal/workflowtracer"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -28,7 +28,7 @@ func ScheduleTimer(ctx Context, delay time.Duration) Future[struct{}] {
 
 	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(f))
 
-	span := tracing.Tracer(ctx).Start("ScheduleTimer",
+	span := workflowtracer.Tracer(ctx).Start(ctx, "ScheduleTimer",
 		trace.WithAttributes(attribute.Int64("duration_s", int64(delay/time.Second))))
 	defer span.End()