Merge pull request #206 from cschleiden/context-propagators

Support context propagation

Author: cschleiden

README.md

@@ -632,6 +632,22 @@ func Workflow(ctx workflow.Context) error {
 	span.End()
 ```
 
+### Context Propagation
+
+In go programs it is common to use `context.Context` to pass around request-scoped data. This library supports context propagation between activities and workflows. When you create a workflow, you can pass a `ContextPropagator` to the backend to propagate context values. The interface is:
+
+```go
+type ContextPropagator interface {
+	Inject(context.Context, *Metadata) error
+	Extract(context.Context, *Metadata) (context.Context, error)
+
+	InjectFromWorkflow(Context, *Metadata) error
+	ExtractToWorkflow(Context, *Metadata) (Context, error)
+}
+```
+
+The `context-propagation` sample shows an example of how to use this.
+
 ## Tools
 
 ### Analyzer

backend/backend.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 
+	"github.com/cschleiden/go-workflows/internal/contextpropagation"
 	"github.com/cschleiden/go-workflows/internal/converter"
 	core "github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
@@ -78,4 +79,7 @@ type Backend interface {
 
 	// Converter returns the configured converter for the backend
 	Converter() converter.Converter
+
+	// ContextPropagators returns the configured context propagators for the backend
+	ContextPropagators() []contextpropagation.ContextPropagator
 }

backend/mock_Backend.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/internal/contextpropagation"
 	"github.com/cschleiden/go-workflows/internal/converter"
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
@@ -78,6 +79,10 @@ func (b *mysqlBackend) Converter() converter.Converter {
 	return b.options.Converter
 }
 
+func (b *mysqlBackend) ContextPropagators() []contextpropagation.ContextPropagator {
+	return b.options.ContextPropagators
+}
+
 func (b *mysqlBackend) CreateWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {
 	tx, err := b.db.BeginTx(ctx, &sql.TxOptions{
 		Isolation: sql.LevelReadCommitted,
@@ -626,9 +631,8 @@ func (b *mysqlBackend) GetActivityTask(ctx context.Context) (*task.Activity, err
 	res := tx.QueryRowContext(
 		ctx,
 		`SELECT activities.id, activity_id, activities.instance_id,
-			instances.metadata, event_type, timestamp, schedule_event_id, attributes, visible_at
+			event_type, timestamp, schedule_event_id, attributes, visible_at
 			FROM activities
-				INNER JOIN instances ON activities.instance_id = instances.instance_id
 			WHERE activities.locked_until IS NULL OR activities.locked_until < ?
 			LIMIT 1
 			FOR UPDATE SKIP LOCKED`,
@@ -638,11 +642,10 @@ func (b *mysqlBackend) GetActivityTask(ctx context.Context) (*task.Activity, err
 	var id int64
 	var instanceID string
 	var attributes []byte
-	var metadataJson sql.NullString
 	event := &history.Event{}
 
 	if err := res.Scan(
-		&id, &event.ID, &instanceID, &metadataJson, &event.Type,
+		&id, &event.ID, &instanceID, &event.Type,
 		&event.Timestamp, &event.ScheduleEventID, &attributes, &event.VisibleAt); err != nil {
 		if err == sql.ErrNoRows {
 			return nil, nil
@@ -651,11 +654,6 @@ func (b *mysqlBackend) GetActivityTask(ctx context.Context) (*task.Activity, err
 		return nil, fmt.Errorf("finding activity task to lock: %w", err)
 	}
 
-	var metadata *workflow.Metadata
-	if err := json.Unmarshal([]byte(metadataJson.String), &metadata); err != nil {
-		return nil, fmt.Errorf("unmarshaling metadata: %w", err)
-	}
-
 	a, err := history.DeserializeAttributes(event.Type, attributes)
 	if err != nil {
 		return nil, fmt.Errorf("deserializing attributes: %w", err)
@@ -676,7 +674,6 @@ func (b *mysqlBackend) GetActivityTask(ctx context.Context) (*task.Activity, err
 	t := &task.Activity{
 		ID:               event.ID,
 		WorkflowInstance: core.NewWorkflowInstance(instanceID),
-		Metadata:         metadata,
 		Event:            event,
 	}
 

backend/mysql/mysql.go

@@ -26,7 +26,7 @@ func Test_MysqlBackend(t *testing.T) {
 
 	var dbName string
 
-	test.BackendTest(t, func() test.TestBackend {
+	test.BackendTest(t, func(options ...backend.BackendOption) test.TestBackend {
 		db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@/?parseTime=true&interpolateParams=true", testUser, testPassword))
 		if err != nil {
 			panic(err)
@@ -41,7 +41,9 @@ func Test_MysqlBackend(t *testing.T) {
 			panic(err)
 		}
 
-		return NewMysqlBackend("localhost", 3306, testUser, testPassword, dbName, backend.WithStickyTimeout(0))
+		options = append(options, backend.WithStickyTimeout(0))
+
+		return NewMysqlBackend("localhost", 3306, testUser, testPassword, dbName, options...)
 	}, func(b test.TestBackend) {
 		db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@/?parseTime=true&interpolateParams=true", testUser, testPassword))
 		if err != nil {
@@ -65,7 +67,7 @@ func TestMySqlBackendE2E(t *testing.T) {
 
 	var dbName string
 
-	test.EndToEndBackendTest(t, func() test.TestBackend {
+	test.EndToEndBackendTest(t, func(options ...backend.BackendOption) test.TestBackend {
 		db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@/?parseTime=true&interpolateParams=true", testUser, testPassword))
 		if err != nil {
 			panic(err)
@@ -80,7 +82,9 @@ func TestMySqlBackendE2E(t *testing.T) {
 			panic(err)
 		}
 
-		return NewMysqlBackend("localhost", 3306, testUser, testPassword, dbName, backend.WithStickyTimeout(0))
+		options = append(options, backend.WithStickyTimeout(0))
+
+		return NewMysqlBackend("localhost", 3306, testUser, testPassword, dbName, options...)
 	}, func(b test.TestBackend) {
 		db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@/?parseTime=true&interpolateParams=true", testUser, testPassword))
 		if err != nil {

backend/mysql/mysql_test.go

@@ -3,11 +3,14 @@ package backend
 import (
 	"time"
 
+	"github.com/cschleiden/go-workflows/internal/contextpropagation"
 	"github.com/cschleiden/go-workflows/internal/converter"
 	"github.com/cschleiden/go-workflows/internal/logger"
 	mi "github.com/cschleiden/go-workflows/internal/metrics"
+	"github.com/cschleiden/go-workflows/internal/tracing"
 	"github.com/cschleiden/go-workflows/log"
 	"github.com/cschleiden/go-workflows/metrics"
+	"github.com/cschleiden/go-workflows/workflow"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -22,6 +25,9 @@ type Options struct {
 	// converter.DefaultConverter is used.
 	Converter converter.Converter
 
+	// ContextPropagators is a list of context propagators to use for passing context into workflows and activities.
+	ContextPropagators []contextpropagation.ContextPropagator
+
 	StickyTimeout time.Duration
 
 	// WorkflowLockTimeout determines how long a workflow task can be locked for. If the workflow task is not completed
@@ -44,6 +50,8 @@ var DefaultOptions Options = Options{
 	Metrics:        mi.NewNoopMetricsClient(),
 	TracerProvider: trace.NewNoopTracerProvider(),
 	Converter:      converter.DefaultConverter,
+
+	ContextPropagators: []contextpropagation.ContextPropagator{&tracing.TracingContextPropagator{}},
 }
 
 type BackendOption func(*Options)
@@ -78,6 +86,12 @@ func WithConverter(converter converter.Converter) BackendOption {
 	}
 }
 
+func WithContextPropagator(prop workflow.ContextPropagator) BackendOption {
+	return func(o *Options) {
+		o.ContextPropagators = append(o.ContextPropagators, prop)
+	}
+}
+
 func ApplyOptions(opts ...BackendOption) Options {
 	options := DefaultOptions
 

backend/options.go

@@ -2,7 +2,6 @@ package redis
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
@@ -19,14 +18,8 @@ func (rb *redisBackend) GetActivityTask(ctx context.Context) (*task.Activity, er
 		return nil, nil
 	}
 
-	instanceState, err := readInstance(ctx, rb.rdb, activityTask.Data.Instance.InstanceID)
-	if err != nil {
-		return nil, fmt.Errorf("reading workflow instance for activity task: %w", err)
-	}
-
 	return &task.Activity{
 		WorkflowInstance: activityTask.Data.Instance,
-		Metadata:         instanceState.Metadata,
 		ID:               activityTask.TaskID, // Use the queue generated ID here
 		Event:            activityTask.Data.Event,
 	}, nil

backend/redis/activity.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/internal/contextpropagation"
 	"github.com/cschleiden/go-workflows/internal/converter"
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
@@ -101,6 +102,10 @@ func (rb *redisBackend) Converter() converter.Converter {
 	return rb.options.Converter
 }
 
+func (rb *redisBackend) ContextPropagators() []contextpropagation.ContextPropagator {
+	return rb.options.ContextPropagators
+}
+
 func (rb *redisBackend) Close() error {
 	return rb.rdb.Close()
 }

backend/redis/redis.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cschleiden/go-workflows/backend"
 	"github.com/cschleiden/go-workflows/backend/test"
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/redis/go-redis/v9"
@@ -52,8 +53,8 @@ func getClient() redis.UniversalClient {
 	return client
 }
 
-func getCreateBackend(client redis.UniversalClient, additionalOptions ...RedisBackendOption) func() test.TestBackend {
-	return func() test.TestBackend {
+func getCreateBackend(client redis.UniversalClient, additionalOptions ...RedisBackendOption) func(options ...backend.BackendOption) test.TestBackend {
+	return func(options ...backend.BackendOption) test.TestBackend {
 		// Flush database
 		if err := client.FlushDB(context.Background()).Err(); err != nil {
 			panic(err)
@@ -68,13 +69,14 @@ func getCreateBackend(client redis.UniversalClient, additionalOptions ...RedisBa
 			panic("Keys should've been empty" + strings.Join(r, ", "))
 		}
 
-		options := []RedisBackendOption{
+		redisOptions := []RedisBackendOption{
 			WithBlockTimeout(time.Millisecond * 10),
+			WithBackendOptions(options...),
 		}
 
-		options = append(options, additionalOptions...)
+		redisOptions = append(redisOptions, additionalOptions...)
 
-		b, err := NewRedisBackend(client, options...)
+		b, err := NewRedisBackend(client, redisOptions...)
 		if err != nil {
 			panic(err)
 		}

backend/redis/redis_test.go

@@ -18,9 +18,13 @@ func (rb *redisBackend) SignalWorkflow(ctx context.Context, instanceID string, e
 		return err
 	}
 
-	ctx = tracing.UnmarshalSpan(ctx, instanceState.Metadata)
+	ctx, err = (&tracing.TracingContextPropagator{}).Extract(ctx, instanceState.Metadata)
+	if err != nil {
+		rb.Logger().Error("extracting tracing context", log.ErrorKey, err)
+	}
+
 	a := event.Attributes.(*history.SignalReceivedAttributes)
-	_, span := rb.Tracer().Start(ctx, fmt.Sprintf("SignalWorkflow: %s", a.Name), trace.WithAttributes(
+	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),
 	))