- Trace continued as new

- Trace timers with name

Author: cschleiden

backend/history/timer_fired.go

@@ -3,12 +3,12 @@ package history
 import (
 	"time"
 
-	"github.com/cschleiden/go-workflows/backend/metadata"
+	"github.com/cschleiden/go-workflows/internal/tracing"
 )
 
 type TimerFiredAttributes struct {
-	ScheduledAt  time.Time                 `json:"scheduled_at,omitempty"`
-	At           time.Time                 `json:"at,omitempty"`
-	Name         string                    `json:"name,omitempty"`
-	SpanMetadata metadata.WorkflowMetadata `json:"span_metadata,omitempty"`
+	ScheduledAt  time.Time       `json:"scheduled_at,omitempty"`
+	At           time.Time       `json:"at,omitempty"`
+	Name         string          `json:"name,omitempty"`
+	TraceContext tracing.Context `json:"span_metadata,omitempty"`
 }

client/client.go

@@ -87,7 +87,7 @@ func (c *Client) CreateWorkflowInstance(ctx context.Context, options WorkflowIns
 		attribute.String(log.InstanceIDKey, wfi.InstanceID),
 		attribute.String(log.ExecutionIDKey, wfi.ExecutionID),
 		attribute.String(log.WorkflowNameKey, workflowName),
-	))
+	), trace.WithSpanKind(trace.SpanKindProducer))
 	defer span.End()
 
 	// Inject state from any propagators

internal/activity/executor.go

@@ -13,6 +13,7 @@ import (
 	"github.com/cschleiden/go-workflows/backend/payload"
 	"github.com/cschleiden/go-workflows/internal/args"
 	"github.com/cschleiden/go-workflows/internal/log"
+	"github.com/cschleiden/go-workflows/internal/tracing"
 	"github.com/cschleiden/go-workflows/internal/workflowerrors"
 	"github.com/cschleiden/go-workflows/registry"
 	wf "github.com/cschleiden/go-workflows/workflow"
@@ -47,21 +48,6 @@ func NewExecutor(
 func (e *Executor) ExecuteActivity(ctx context.Context, task *backend.ActivityTask) (payload.Payload, error) {
 	a := task.Event.Attributes.(*history.ActivityScheduledAttributes)
 
-	activity, err := e.r.GetActivity(a.Name)
-	if err != nil {
-		return nil, workflowerrors.NewPermanentError(fmt.Errorf("activity not found: %w", err))
-	}
-
-	activityFn := reflect.ValueOf(activity)
-	if activityFn.Type().Kind() != reflect.Func {
-		return nil, workflowerrors.NewPermanentError(errors.New("activity not a function"))
-	}
-
-	args, addContext, err := args.InputsToArgs(e.converter, activityFn, a.Inputs)
-	if err != nil {
-		return nil, workflowerrors.NewPermanentError(fmt.Errorf("converting activity inputs: %w", err))
-	}
-
 	// Add activity state to context
 	as := NewActivityState(
 		task.Event.ID,
@@ -71,6 +57,7 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *backend.ActivityTa
 	activityCtx := WithActivityState(ctx, as)
 
 	for _, propagator := range e.propagators {
+		var err error
 		activityCtx, err = propagator.Extract(activityCtx, a.Metadata)
 		if err != nil {
 			return nil, workflowerrors.NewPermanentError(fmt.Errorf("extracting context from propagator: %w", err))
@@ -83,6 +70,22 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *backend.ActivityTa
 		attribute.String(log.ActivityIDKey, task.ID),
 		attribute.Int(log.AttemptKey, a.Attempt),
 	))
+
+	activity, err := e.r.GetActivity(a.Name)
+	if err != nil {
+		return nil, workflowerrors.NewPermanentError(tracing.WithSpanError(span, fmt.Errorf("activity not found: %w", err)))
+	}
+
+	activityFn := reflect.ValueOf(activity)
+	if activityFn.Type().Kind() != reflect.Func {
+		return nil, workflowerrors.NewPermanentError(tracing.WithSpanError(span, errors.New("activity not a function")))
+	}
+
+	args, addContext, err := args.InputsToArgs(e.converter, activityFn, a.Inputs)
+	if err != nil {
+		return nil, workflowerrors.NewPermanentError(tracing.WithSpanError(span, fmt.Errorf("converting activity inputs: %w", err)))
+	}
+
 	defer span.End()
 
 	// Execute activity
@@ -97,7 +100,7 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *backend.ActivityTa
 		// Recover any panic encountered during activity execution
 		defer func() {
 			if r := recover(); r != nil {
-				err = workflowerrors.NewPanicError(fmt.Sprintf("panic: %v", r))
+				err := workflowerrors.NewPanicError(fmt.Sprintf("panic: %v", r))
 				rv = []reflect.Value{reflect.ValueOf(err)}
 			}
 
@@ -110,7 +113,8 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *backend.ActivityTa
 	<-done
 
 	if len(rv) < 1 || len(rv) > 2 {
-		return nil, workflowerrors.NewPermanentError(errors.New("activity has to return either (error) or (<result>, error)"))
+		return nil, workflowerrors.NewPermanentError(
+			tracing.WithSpanError(span, errors.New("activity has to return either (error) or (<result>, error)")))
 	}
 
 	var result payload.Payload
@@ -120,7 +124,7 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *backend.ActivityTa
 		var err error
 		result, err = e.converter.To(rv[0].Interface())
 		if err != nil {
-			return nil, workflowerrors.NewPermanentError(fmt.Errorf("converting activity result: %w", err))
+			return nil, workflowerrors.NewPermanentError(tracing.WithSpanError(span, fmt.Errorf("converting activity result: %w", err)))
 		}
 	}
 
@@ -133,8 +137,9 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *backend.ActivityTa
 
 	err, ok := errResult.Interface().(error)
 	if !ok {
-		return nil, workflowerrors.NewPermanentError(fmt.Errorf("activity error result does not satisfy error interface (%T): %v", errResult, errResult))
+		return nil, workflowerrors.NewPermanentError(
+			tracing.WithSpanError(span, fmt.Errorf("activity error result does not satisfy error interface (%T): %v", errResult, errResult)))
 	}
 
-	return result, workflowerrors.FromError(err)
+	return result, workflowerrors.FromError(tracing.WithSpanError(span, err))
 }

internal/command/continueasnew.go

@@ -17,6 +17,8 @@ type ContinueAsNewCommand struct {
 	Metadata *metadata.WorkflowMetadata
 	Inputs   []payload.Payload
 	Result   payload.Payload
+
+	ContinuedExecutionID string
 }
 
 var _ Command = (*ContinueAsNewCommand)(nil)
@@ -28,27 +30,26 @@ func NewContinueAsNewCommand(id int64, instance *core.WorkflowInstance, result p
 			name:  "ContinueAsNew",
 			state: CommandState_Pending,
 		},
-		Instance: instance,
-		Name:     name,
-		Metadata: metadata,
-		Inputs:   inputs,
-		Result:   result,
+		Instance:             instance,
+		Name:                 name,
+		Metadata:             metadata,
+		Inputs:               inputs,
+		Result:               result,
+		ContinuedExecutionID: uuid.NewString(),
 	}
 }
 
 func (c *ContinueAsNewCommand) Execute(clock clock.Clock) *CommandResult {
 	switch c.state {
 	case CommandState_Pending:
-		continuedExecutionID := uuid.NewString()
-
 		var continuedInstance *core.WorkflowInstance
 		if c.Instance.SubWorkflow() {
 			// If the current workflow execution was a sub-workflow, ensure the new workflow execution is also a sub-workflow.
 			// This will guarantee that the finished event for the new execution will be delivered to the right parent instance
 			continuedInstance = core.NewSubWorkflowInstance(
-				c.Instance.InstanceID, continuedExecutionID, c.Instance.Parent, c.Instance.ParentEventID)
+				c.Instance.InstanceID, c.ContinuedExecutionID, c.Instance.Parent, c.Instance.ParentEventID)
 		} else {
-			continuedInstance = core.NewWorkflowInstance(c.Instance.InstanceID, continuedExecutionID)
+			continuedInstance = core.NewWorkflowInstance(c.Instance.InstanceID, c.ContinuedExecutionID)
 		}
 
 		c.state = CommandState_Committed
@@ -61,7 +62,7 @@ func (c *ContinueAsNewCommand) Execute(clock clock.Clock) *CommandResult {
 					history.EventType_WorkflowExecutionContinuedAsNew,
 					&history.ExecutionContinuedAsNewAttributes{
 						Result:               c.Result,
-						ContinuedExecutionID: continuedExecutionID,
+						ContinuedExecutionID: c.ContinuedExecutionID,
 					},
 				),
 			},

internal/command/schedule_timer.go

@@ -5,20 +5,20 @@ import (
 
 	"github.com/benbjohnson/clock"
 	"github.com/cschleiden/go-workflows/backend/history"
-	"github.com/cschleiden/go-workflows/backend/metadata"
+	"github.com/cschleiden/go-workflows/internal/tracing"
 )
 
 type ScheduleTimerCommand struct {
 	cancelableCommand
 
 	at           time.Time
 	name         string
-	spanMetadata metadata.WorkflowMetadata
+	traceContext tracing.Context
 }
 
 var _ CancelableCommand = (*ScheduleTimerCommand)(nil)
 
-func NewScheduleTimerCommand(id int64, at time.Time, name string, carrier metadata.WorkflowMetadata) *ScheduleTimerCommand {
+func NewScheduleTimerCommand(id int64, at time.Time, name string, traceContext tracing.Context) *ScheduleTimerCommand {
 	return &ScheduleTimerCommand{
 		cancelableCommand: cancelableCommand{
 			command: command{
@@ -29,7 +29,7 @@ func NewScheduleTimerCommand(id int64, at time.Time, name string, carrier metada
 		},
 		at:           at,
 		name:         name,
-		spanMetadata: carrier,
+		traceContext: traceContext,
 	}
 }
 
@@ -61,7 +61,7 @@ func (c *ScheduleTimerCommand) Execute(clock clock.Clock) *CommandResult {
 						ScheduledAt:  now,
 						At:           c.at,
 						Name:         c.name,
-						SpanMetadata: c.spanMetadata,
+						TraceContext: c.traceContext,
 					},
 					history.ScheduleEventID(c.id),
 					history.VisibleAt(c.at),

internal/log/fields.go

@@ -10,6 +10,8 @@ const (
 	InstanceIDKey   = NamespaceKey + ".instance.id"
 	ExecutionIDKey  = NamespaceKey + ".execution.id"
 
+	ContinuedExecutionIDKey = NamespaceKey + ".continued_execution.id"
+
 	WorkflowNameKey = NamespaceKey + ".workflow.name"
 
 	SignalNameKey = NamespaceKey + ".signal.name"

internal/tracing/spanerror.go

@@ -0,0 +1,14 @@
+package tracing
+
+import (
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
+)
+
+func WithSpanError(span trace.Span, err error) error {
+	if err != nil {
+		span.SetStatus(codes.Error, err.Error())
+	}
+
+	return err
+}

internal/tracing/tracecontext.go

@@ -0,0 +1,43 @@
+package tracing
+
+import (
+	"context"
+
+	"github.com/cschleiden/go-workflows/internal/sync"
+	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/trace"
+)
+
+type Context map[string]string
+
+func (wim Context) Get(key string) string {
+	return wim[key]
+}
+
+func (wim Context) Set(key string, value string) {
+	wim[key] = value
+}
+
+func (wim Context) Keys() []string {
+	r := make([]string, 0, len(wim))
+
+	for k := range wim {
+		r = append(r, k)
+	}
+
+	return r
+}
+
+var propagator propagation.TraceContext
+
+func ContextFromWfCtx(ctx sync.Context) Context {
+	span := SpanFromContext(ctx)
+	spanCtx := trace.ContextWithSpan(context.Background(), span)
+	carrier := make(Context)
+	propagator.Inject(spanCtx, carrier)
+	return carrier
+}
+
+func SpanContextFromContext(ctx context.Context, tctx Context) context.Context {
+	return propagator.Extract(ctx, tctx)
+}

internal/tracing/workflowspan.go renamed to internal/tracing/tracing.go

@@ -6,15 +6,16 @@ import (
 	"time"
 	"unsafe"
 
-	"github.com/cschleiden/go-workflows/internal/sync"
-	"github.com/cschleiden/go-workflows/internal/workflowstate"
 	"go.opentelemetry.io/otel/trace"
 )
 
-func SpanWithStartTime(ctx context.Context, tracer trace.Tracer, name string, spanID trace.SpanID, startTime time.Time) trace.Span {
+func SpanWithStartTime(
+	ctx context.Context, tracer trace.Tracer, name string, spanID trace.SpanID, startTime time.Time, opts ...trace.SpanStartOption) trace.Span {
+
+	opts = append(opts, trace.WithTimestamp(startTime), trace.WithSpanKind(trace.SpanKindConsumer))
 	_, span := tracer.Start(ctx,
 		name,
-		trace.WithTimestamp(startTime),
+		opts...,
 	)
 
 	SetSpanID(span, spanID)
@@ -33,11 +34,6 @@ func GetNewSpanID(tracer trace.Tracer) trace.SpanID {
 	return span.SpanContext().SpanID()
 }
 
-func GetNewSpanIDWF(ctx sync.Context) trace.SpanID {
-	tracer := workflowstate.WorkflowState(ctx).Tracer()
-	return GetNewSpanID(tracer)
-}
-
 func SetSpanID(span trace.Span, sid trace.SpanID) {
 	sc := span.SpanContext()
 	sc = sc.WithSpanID(sid)

samples/tracing/tracing.go

@@ -139,6 +139,7 @@ func RunWorker(ctx context.Context, mb backend.Backend) *worker.Worker {
 	w.RegisterWorkflow(Subworkflow)
 
 	w.RegisterActivity(Activity1)
+	w.RegisterActivity(RetriedActivity)
 
 	if err := w.Start(ctx); err != nil {
 		panic("could not start worker")