Logical tracing

Author: cschleiden

backend/history/history.go

@@ -55,6 +55,9 @@ const (
 
 	// Recorded result of a side-efect
 	EventType_SideEffectResult
+
+	// Distributed tracing span has been started
+	EventType_TraceStarted
 )
 
 func (et EventType) String() string {
@@ -102,6 +105,9 @@ func (et EventType) String() string {
 	case EventType_SideEffectResult:
 		return "SideEffectResult"
 
+	case EventType_TraceStarted:
+		return "TraceStarted"
+
 	default:
 		return "Unknown"
 	}

backend/history/serialization.go

@@ -59,6 +59,9 @@ func DeserializeAttributes(eventType EventType, attributes []byte) (attr interfa
 	case EventType_SideEffectResult:
 		attr = &SideEffectResultAttributes{}
 
+	case EventType_TraceStarted:
+		attr = &TraceStartedAttributes{}
+
 	case EventType_TimerScheduled:
 		attr = &TimerScheduledAttributes{}
 	case EventType_TimerFired:

backend/history/span_started.go

@@ -0,0 +1,7 @@
+package history
+
+import "github.com/cschleiden/go-workflows/backend/payload"
+
+type TraceStartedAttributes struct {
+	SpanID payload.Payload `json:"spanID"`
+}

backend/history/subworkflow_scheduled.go

@@ -16,4 +16,6 @@ type SubWorkflowScheduledAttributes struct {
 	Inputs []payload.Payload `json:"inputs,omitempty"`
 
 	Metadata *metadata.WorkflowMetadata `json:"metadata,omitempty"`
+
+	WorkflowSpanID [8]byte `json:"workflow_span_id,omitempty"`
 }

backend/history/workflow_started.go

@@ -14,4 +14,6 @@ type ExecutionStartedAttributes struct {
 	Metadata *metadata.WorkflowMetadata `json:"metadata,omitempty"`
 
 	Inputs []payload.Payload `json:"inputs,omitempty"`
+
+	WorkflowSpanID [8]byte `json:"workflowSpanID,omitempty"`
 }

backend/options.go

@@ -7,9 +7,10 @@ import (
 	"github.com/cschleiden/go-workflows/backend/converter"
 	"github.com/cschleiden/go-workflows/backend/metrics"
 	mi "github.com/cschleiden/go-workflows/internal/metrics"
-	"github.com/cschleiden/go-workflows/internal/tracing"
+	"github.com/cschleiden/go-workflows/internal/propagators"
 	"github.com/cschleiden/go-workflows/workflow"
 	"go.opentelemetry.io/otel/trace"
+	"go.opentelemetry.io/otel/trace/noop"
 )
 
 type Options struct {
@@ -51,10 +52,10 @@ var DefaultOptions Options = Options{
 
 	Logger:         slog.Default(),
 	Metrics:        mi.NewNoopMetricsClient(),
-	TracerProvider: trace.NewNoopTracerProvider(),
+	TracerProvider: noop.NewTracerProvider(),
 	Converter:      converter.DefaultConverter,
 
-	ContextPropagators: []workflow.ContextPropagator{&tracing.TracingContextPropagator{}},
+	ContextPropagators: []workflow.ContextPropagator{&propagators.TracingContextPropagator{}},
 
 	RemoveContinuedAsNewInstances: false,
 }

backend/redis/instance.go

@@ -133,10 +133,10 @@ func (rb *redisBackend) CancelWorkflowInstance(ctx context.Context, instance *co
 	}
 
 	// Cancel instance
-	if cmds, err := rb.rdb.Pipelined(ctx, func(p redis.Pipeliner) error {
+	if _, err := rb.rdb.Pipelined(ctx, func(p redis.Pipeliner) error {
 		return rb.addWorkflowInstanceEventP(ctx, p, workflow.Queue(instanceState.Queue), instance, event)
 	}); err != nil {
-		fmt.Println(cmds)
+		// fmt.Println(cmds)
 		return fmt.Errorf("adding cancellation event to workflow instance: %w", err)
 	}
 

backend/redis/signal.go

@@ -6,12 +6,8 @@ import (
 
 	"github.com/cschleiden/go-workflows/backend"
 	"github.com/cschleiden/go-workflows/backend/history"
-	"github.com/cschleiden/go-workflows/internal/log"
-	"github.com/cschleiden/go-workflows/internal/propagators"
 	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/redis/go-redis/v9"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 )
 
 func (rb *redisBackend) SignalWorkflow(ctx context.Context, instanceID string, event *history.Event) error {
@@ -30,19 +26,6 @@ func (rb *redisBackend) SignalWorkflow(ctx context.Context, instanceID string, e
 		return err
 	}
 
-	// TODO: Can we do this in the client?
-	ctx, err = (&propagators.TracingContextPropagator{}).Extract(ctx, instanceState.Metadata)
-	if err != nil {
-		rb.Options().Logger.Error("extracting tracing context", log.ErrorKey, err)
-	}
-
-	a := event.Attributes.(*history.SignalReceivedAttributes)
-	ctx, span := rb.Tracer().Start(ctx, fmt.Sprintf("SignalWorkflow: %s", a.Name), trace.WithAttributes(
-		attribute.String(log.InstanceIDKey, instanceID),
-		attribute.String(log.SignalNameKey, event.Attributes.(*history.SignalReceivedAttributes).Name),
-	))
-	defer span.End()
-
 	if _, err = rb.rdb.TxPipelined(ctx, func(p redis.Pipeliner) error {
 		if err := rb.addWorkflowInstanceEventP(ctx, p, workflow.Queue(instanceState.Queue), instanceState.Instance, event); err != nil {
 			return fmt.Errorf("adding event to stream: %w", err)

backend/redis/workflow.go

@@ -15,8 +15,6 @@ import (
 	"github.com/cschleiden/go-workflows/internal/workflowerrors"
 	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/redis/go-redis/v9"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 )
 
 func (rb *redisBackend) PrepareWorkflowQueues(ctx context.Context, queues []workflow.Queue) error {
@@ -306,12 +304,6 @@ func (rb *redisBackend) CompleteWorkflowTask(
 			rb.options.Logger.Error("extracting tracing context", log.ErrorKey, err)
 		}
 
-		_, span := rb.Tracer().Start(ctx, "WorkflowComplete",
-			trace.WithAttributes(
-				attribute.String(log.NamespaceKey+log.InstanceIDKey, task.WorkflowInstance.InstanceID),
-			))
-		span.End()
-
 		// Auto expiration
 		expiration := rb.options.AutoExpiration
 		if state == core.WorkflowInstanceStateContinuedAsNew && rb.options.AutoExpirationContinueAsNew > 0 {

backend/test/e2e.go

@@ -622,6 +622,7 @@ func EndToEndBackendTest(t *testing.T, setup func(options ...backend.BackendOpti
 	tests = append(tests, e2eQueueTests...)
 	tests = append(tests, e2eRemovalTests...)
 	tests = append(tests, e2eContinueAsNewTests...)
+	tests = append(tests, e2eTracingTests...)
 
 	run := func(suffix string, workerOptions worker.Options) {
 		for _, tt := range tests {