Correctly trace timers and adjust subworkflows

Author: cschleiden

backend/history/timer_fired.go

@@ -3,5 +3,6 @@ package history
 import "time"
 
 type TimerFiredAttributes struct {
-	At time.Time `json:"at,omitempty"`
+	ScheduledAt time.Time `json:"scheduled_at,omitempty"`
+	At          time.Time `json:"at,omitempty"`
 }

backend/test/e2e_tracing.go

@@ -58,6 +58,43 @@ var e2eTracingTests = []backendTest{
 			)
 		},
 	},
+	{
+		name: "Tracing/TimersHaveSpans",
+		f: func(t *testing.T, ctx context.Context, c *client.Client, w *worker.Worker, b TestBackend) {
+			exporter := setupTracing(b)
+
+			wf := func(ctx workflow.Context) error {
+				workflow.ScheduleTimer(ctx, time.Millisecond*20).Get(ctx)
+
+				return nil
+			}
+			register(t, ctx, w, []interface{}{wf}, nil)
+
+			instance := runWorkflow(t, ctx, c, wf)
+			_, err := client.GetWorkflowResult[any](ctx, c, instance, time.Second*5)
+			require.NoError(t, err)
+
+			spans := exporter.GetSpans().Snapshots()
+
+			workflow1Span := findSpan(spans, func(span trace.ReadOnlySpan) bool {
+				return strings.Contains(span.Name(), "Workflow: 1")
+			})
+			require.NotNil(t, workflow1Span)
+
+			timerSpan := findSpan(spans, func(span trace.ReadOnlySpan) bool {
+				return strings.Contains(span.Name(), "Timer")
+			})
+			require.NotNil(t, workflow1Span)
+			require.InEpsilon(t, time.Duration(20*time.Millisecond),
+				timerSpan.EndTime().Sub(timerSpan.StartTime())/time.Millisecond,
+				float64(5*time.Millisecond))
+
+			require.Equal(t,
+				workflow1Span.SpanContext().SpanID().String(),
+				timerSpan.Parent().SpanID().String(),
+			)
+		},
+	},
 	{
 		name: "Tracing/SubWorkflowsHaveSpansWithCorrectParent",
 		f: func(t *testing.T, ctx context.Context, c *client.Client, w *worker.Worker, b TestBackend) {
@@ -96,13 +133,18 @@ var e2eTracingTests = []backendTest{
 			})
 			require.NotNil(t, workflow1Span)
 
+			createSWFSpan := findSpan(spans, func(span trace.ReadOnlySpan) bool {
+				return strings.Contains(span.Name(), "CreateSubworkflowInstance: swf")
+			})
+			require.NotNil(t, createSWFSpan)
+
 			workflow2Span := findSpan(spans, func(span trace.ReadOnlySpan) bool {
 				return strings.Contains(span.Name(), "Workflow: swf")
 			})
 			require.NotNil(t, workflow1Span)
 
 			require.Equal(t,
-				workflow1Span.SpanContext().SpanID().String(),
+				createSWFSpan.SpanContext().SpanID().String(),
 				workflow2Span.Parent().SpanID().String(),
 			)
 		},

internal/command/schedule_timer.go

@@ -33,10 +33,12 @@ func (c *ScheduleTimerCommand) Execute(clock clock.Clock) *CommandResult {
 	case CommandState_Pending:
 		c.state = CommandState_Committed
 
+		now := clock.Now()
+
 		return &CommandResult{
 			Events: []*history.Event{
 				history.NewPendingEvent(
-					clock.Now(),
+					now,
 					history.EventType_TimerScheduled,
 					&history.TimerScheduledAttributes{
 						At: c.at,
@@ -50,7 +52,8 @@ func (c *ScheduleTimerCommand) Execute(clock clock.Clock) *CommandResult {
 					clock.Now(),
 					history.EventType_TimerFired,
 					&history.TimerFiredAttributes{
-						At: c.at,
+						ScheduledAt: now,
+						At:          c.at,
 					},
 					history.ScheduleEventID(c.id),
 					history.VisibleAt(c.at),

workflow/executor/executor.go

@@ -8,6 +8,7 @@ import (
 	"reflect"
 	"slices"
 	"testing"
+	"time"
 
 	"github.com/benbjohnson/clock"
 	"github.com/cschleiden/go-workflows/backend"
@@ -26,6 +27,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/workflowstate"
 	"github.com/cschleiden/go-workflows/registry"
 	wf "github.com/cschleiden/go-workflows/workflow"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -530,6 +532,18 @@ func (e *executor) handleTimerFired(event *history.Event, a *history.TimerFiredA
 		return nil
 	}
 
+	if !e.workflowState.Replaying() {
+		// Trace timer
+		parentSpan := tracing.SpanFromContext(e.workflowCtx)
+		ctx := trace.ContextWithSpan(context.Background(), parentSpan)
+		_, span := e.tracer.Start(ctx, "Timer", trace.WithAttributes(
+			attribute.Int64(log.DurationKey, int64(a.At.Sub(a.ScheduledAt)/time.Millisecond)),
+			attribute.String(log.NowKey, a.ScheduledAt.String()),
+			attribute.String(log.AtKey, a.At.String()),
+		), trace.WithTimestamp(a.ScheduledAt))
+		span.End()
+	}
+
 	if err := f.Set(nil, nil); err != nil {
 		return fmt.Errorf("setting timer fired result: %w", err)
 	}

workflow/subworkflow.go

@@ -90,13 +90,7 @@ func createSubWorkflowInstance[TResult any](ctx Context, options SubWorkflowOpti
 			attribute.Int64(log.ScheduleEventIDKey, scheduleEventID),
 			attribute.Int(log.AttemptKey, attempt),
 		))
-
-	tf := sync.NewFuture[TResult]()
-	Go(ctx, func(ctx Context) {
-		r, err := f.Get(ctx)
-		span.End()
-		tf.Set(r, err)
-	})
+	defer span.End()
 
 	// Capture context
 	propagators := propagators(ctx)
@@ -146,5 +140,5 @@ func createSubWorkflowInstance[TResult any](ctx Context, options SubWorkflowOpti
 		})
 	}
 
-	return tf
+	return f
 }

workflow/timer.go

@@ -6,11 +6,8 @@ import (
 
 	"github.com/cschleiden/go-workflows/internal/command"
 	"github.com/cschleiden/go-workflows/internal/contextvalue"
-	"github.com/cschleiden/go-workflows/internal/log"
 	"github.com/cschleiden/go-workflows/internal/sync"
 	"github.com/cschleiden/go-workflows/internal/workflowstate"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 )
 
 // ScheduleTimer schedules a timer to fire after the given delay.
@@ -48,19 +45,6 @@ func ScheduleTimer(ctx Context, delay time.Duration) Future[any] {
 		},
 	}
 
-	ctx, span := Tracer(ctx).Start(ctx, "ScheduleTimer",
-		trace.WithAttributes(
-			attribute.Int64(log.DurationKey, int64(delay/time.Millisecond)),
-			attribute.String(log.NowKey, Now(ctx).String()),
-			attribute.String(log.AtKey, at.String()),
-		))
-	tf := sync.NewFuture[any]()
-	Go(ctx, func(ctx Context) {
-		r, err := f.Get(ctx)
-		span.End()
-		tf.Set(r, err)
-	})
-
 	// Check if the context is cancelable
 	if c, cancelable := ctx.Done().(sync.CancelChannel); cancelable {
 		// Register a callback for when it's canceled. The only operation on the `Done` channel
@@ -72,5 +56,5 @@ func ScheduleTimer(ctx Context, delay time.Duration) Future[any] {
 		})
 	}
 
-	return tf
+	return f
 }

workflow/tracer.go

@@ -2,7 +2,6 @@ package workflow
 
 import (
 	"context"
-	"log"
 
 	"github.com/cschleiden/go-workflows/internal/command"
 	"github.com/cschleiden/go-workflows/internal/contextvalue"
@@ -21,8 +20,6 @@ func (s *wfSpan) End() {
 	if !s.state.Replaying() {
 		// Only end the trace when we are not replaying
 		s.span.End()
-	} else {
-		log.Println("Not ending span as we are replaying")
 	}
 }