Merge pull request #365 from cschleiden/improve-tracing

Improve tracing

Author: cschleiden

.github/workflows/go.yml

@@ -16,12 +16,12 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
 
     - name: Set up Go
-      uses: actions/setup-go@v3
+      uses: actions/setup-go@v5
       with:
-        go-version: 1.21
+        go-version: 1.22
         check-latest: true
         cache: true
 

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/timer_fired.go

@@ -1,7 +1,14 @@
 package history
 
-import "time"
+import (
+	"time"
+
+	"github.com/cschleiden/go-workflows/internal/tracing"
+)
 
 type TimerFiredAttributes struct {
-	At time.Time `json:"at,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"`
 }

backend/history/timer_scheduled.go

@@ -3,5 +3,6 @@ package history
 import "time"
 
 type TimerScheduledAttributes struct {
-	At time.Time `json:"at,omitempty"`
+	At   time.Time `json:"at,omitempty"`
+	Name string    `json:"name,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)
 	}