Include activity and workflow names in spans

Author: cschleiden

backend/redis/events.go

@@ -3,6 +3,7 @@ package redis
 import (
 	"context"
 	"encoding/json"
+	"strconv"
 
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
@@ -71,7 +72,7 @@ func addFutureEventP(ctx context.Context, p redis.Pipeliner, instance *core.Work
 	addFutureEventCmd.Run(
 		ctx, p,
 		[]string{futureEventsKey(), futureEventKey(instance.InstanceID, event.ScheduleEventID)},
-		event.VisibleAt.Unix(),
+		strconv.FormatInt(event.VisibleAt.UnixMilli(), 10),
 		instance.InstanceID,
 		string(eventData),
 	)

backend/redis/signal.go

@@ -18,7 +18,8 @@ func (rb *redisBackend) SignalWorkflow(ctx context.Context, instanceID string, e
 	}
 
 	ctx = tracing.UnmarshalSpan(ctx, instanceState.Metadata)
-	_, span := rb.Tracer().Start(ctx, "SignalWorkflow", trace.WithAttributes(
+	a := event.Attributes.(*history.SignalReceivedAttributes)
+	_, span := rb.Tracer().Start(ctx, fmt.Sprintf("SignalWorkflow: %s", a.Name), trace.WithAttributes(
 		attribute.String(tracing.WorkflowInstanceID, instanceID),
 		attribute.String("signal.name", event.Attributes.(*history.SignalReceivedAttributes).Name),
 	))

backend/redis/workflow.go

@@ -57,8 +57,8 @@ var futureEventsCmd = redis.NewScript(`
 
 func (rb *redisBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, error) {
 	// Check for future events
-	now := time.Now().Unix()
-	nowStr := strconv.Itoa(int(now))
+	now := time.Now().UnixMilli()
+	nowStr := strconv.FormatInt(now, 10)
 
 	queueKeys := rb.workflowQueue.Keys()
 

client/client.go

@@ -73,7 +73,7 @@ func (c *client) CreateWorkflowInstance(ctx context.Context, options WorkflowIns
 	metadata := &workflow.Metadata{}
 
 	// Start new span and add to metadata
-	sctx, span := c.backend.Tracer().Start(ctx, "CreateWorkflowInstance", trace.WithAttributes(
+	sctx, span := c.backend.Tracer().Start(ctx, fmt.Sprintf("CreateWorkflowInstance: %s", workflowName), trace.WithAttributes(
 		attribute.String(tracing.WorkflowInstanceID, wfi.InstanceID),
 		attribute.String(tracing.WorkflowName, workflowName),
 	))

samples/runner.go

@@ -34,7 +34,7 @@ func GetBackend(name string, opt ...backend.BackendOption) backend.Backend {
 			WriteTimeout: time.Second * 30,
 			ReadTimeout:  time.Second * 30,
 		})
-		b, err := redis.NewRedisBackend(rclient)
+		b, err := redis.NewRedisBackend(rclient, redis.WithBackendOptions(opt...))
 		if err != nil {
 			panic(err)
 		}

samples/tracing/tracing.go

@@ -16,7 +16,6 @@ import (
 
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
-	"go.opentelemetry.io/otel/exporters/stdout/stdouttrace"
 
 	"github.com/cschleiden/go-workflows/worker"
 
@@ -33,19 +32,24 @@ func main() {
 		attribute.String("environment", "sample"),
 	)
 
-	stdoutexp, err := stdouttrace.New(stdouttrace.WithPrettyPrint())
-	if err != nil {
-		panic(err)
-	}
+	// stdoutexp, err := stdouttrace.New(stdouttrace.WithPrettyPrint())
+	// if err != nil {
+	// 	panic(err)
+	// }
 
-	oclient := otlptracehttp.NewClient(otlptracehttp.WithEndpoint("localhost:8360"), otlptracehttp.WithURLPath("/traces/otlp/v0.9"), otlptracehttp.WithInsecure())
+	oclient := otlptracehttp.NewClient(
+		// otlptracehttp.WithEndpoint("localhost:8360"),
+		// otlptracehttp.WithURLPath("/traces/otlp/v0.9"),
+		otlptracehttp.WithEndpoint("localhost:4318"),
+		otlptracehttp.WithInsecure(),
+	)
 	exp, err := otlptrace.New(ctx, oclient)
 	if err != nil {
 		panic(err)
 	}
 
 	tp := trace.NewTracerProvider(
-		trace.WithSyncer(stdoutexp),
+		// trace.WithSyncer(stdoutexp),
 		trace.WithBatcher(exp),
 		trace.WithResource(r),
 	)

workflow/activity.go

@@ -52,9 +52,9 @@ func executeActivity[TResult any](ctx Context, options ActivityOptions, attempt
 	wfState.AddCommand(cmd)
 	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(f))
 
-	span := workflowtracer.Tracer(ctx).Start(
-		ctx,
-		"ExecuteActivity", trace.WithAttributes(
+	span := workflowtracer.Tracer(ctx).Start(ctx,
+		fmt.Sprintf("ExecuteActivity: %s", name),
+		trace.WithAttributes(
 			attribute.String("name", name),
 			attribute.Int64(tracing.ScheduleEventID, scheduleEventID),
 			attribute.Int("attempt", attempt),

workflow/sleep.go

@@ -11,7 +11,7 @@ import (
 
 func Sleep(ctx sync.Context, d time.Duration) error {
 	span := workflowtracer.Tracer(ctx).Start(ctx, "Sleep",
-		trace.WithAttributes(attribute.Int64("duration_s", int64(d/time.Second))))
+		trace.WithAttributes(attribute.Int64("duration_ms", int64(d/time.Millisecond))))
 	defer span.End()
 
 	_, err := ScheduleTimer(ctx, d).Get(ctx)

workflow/subworkflow.go

@@ -56,7 +56,8 @@ func createSubWorkflowInstance[TResult any](ctx sync.Context, options SubWorkflo
 	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(f))
 
 	span := workflowtracer.Tracer(ctx).Start(ctx,
-		"CreateSubworkflowInstance", trace.WithAttributes(
+		fmt.Sprintf("CreateSubworkflowInstance: %s", name),
+		trace.WithAttributes(
 			attribute.String("name", name),
 			attribute.Int64(tracing.ScheduleEventID, scheduleEventID),
 			attribute.Int("attempt", attempt),

workflow/timer.go

@@ -23,13 +23,18 @@ func ScheduleTimer(ctx Context, delay time.Duration) Future[struct{}] {
 	wfState := workflowstate.WorkflowState(ctx)
 
 	scheduleEventID := wfState.GetNextScheduleEventID()
-	timerCmd := command.NewScheduleTimerCommand(scheduleEventID, Now(ctx).Add(delay))
+	at := Now(ctx).Add(delay)
+	timerCmd := command.NewScheduleTimerCommand(scheduleEventID, at)
 	wfState.AddCommand(timerCmd)
 
 	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(f))
 
 	span := workflowtracer.Tracer(ctx).Start(ctx, "ScheduleTimer",
-		trace.WithAttributes(attribute.Int64("duration_s", int64(delay/time.Second))))
+		trace.WithAttributes(
+			attribute.Int64("duration_ms", int64(delay/time.Millisecond)),
+			attribute.String("now", Now(ctx).String()),
+			attribute.String("at", at.String()),
+		))
 	defer span.End()
 
 	// Check if the context is cancelable