Merge pull request #89 from cschleiden/opentelemetry

Basic tracing using OpenTelemetry

Author: cschleiden

.vscode/launch.json

@@ -10,7 +10,7 @@
       "type": "go",
       "request": "launch",
       "mode": "debug",
-      "program": "${file}"
+      "program": "${fileDirname}"
     },
     {
       "name": "Launch simple sample",

.vscode/settings.json

@@ -2,18 +2,10 @@
   // Some integration tests take quite long to run, increase the overall limit for now
   "go.testFlags": ["-timeout", "120s", "-race", "-count", "1"], // , "-short"],
   "workbench.colorCustomizations": {
-    "sash.hoverBorder": "#3399ff",
     "titleBar.activeBackground": "#007fff",
     "titleBar.activeForeground": "#e7e7e7",
     "titleBar.inactiveBackground": "#007fff99",
     "titleBar.inactiveForeground": "#e7e7e799",
-    "activityBar.activeBackground": "#3399ff",
-    "activityBar.activeBorder": "#bf0060",
-    "activityBar.background": "#3399ff",
-    "activityBar.foreground": "#15202b",
-    "activityBar.inactiveForeground": "#15202b99",
-    "activityBarBadge.background": "#bf0060",
-    "activityBarBadge.foreground": "#e7e7e7"
   },
 
   "files.exclude": {
@@ -32,5 +24,7 @@
     ".github/workflows/go.yml"
   ],
   "go.testExplorer.showDynamicSubtestsInEditor": true,
-  "go.lintTool": "golangci-lint"
+  "go.lintTool": "golangci-lint",
+  "todo-tree.tree.showBadges": true,
+  "todo-tree.tree.disableCompactFolders": true
 }

backend/backend.go

@@ -9,6 +9,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/task"
 	"github.com/cschleiden/go-workflows/log"
 	"github.com/cschleiden/go-workflows/workflow"
+	"go.opentelemetry.io/otel/trace"
 )
 
 var ErrInstanceNotFound = errors.New("workflow instance not found")
@@ -21,13 +22,15 @@ const (
 	WorkflowStateFinished
 )
 
+const TracerName = "go-workflow"
+
 //go:generate mockery --name=Backend --inpackage
 type Backend interface {
 	// CreateWorkflowInstance creates a new workflow instance
-	CreateWorkflowInstance(ctx context.Context, event history.WorkflowEvent) error
+	CreateWorkflowInstance(ctx context.Context, instance *workflow.Instance, event history.Event) error
 
 	// CancelWorkflowInstance cancels a running workflow instance
-	CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error
+	CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance, cancelEvent *history.Event) error
 
 	// GetWorkflowInstanceState returns the state of the given workflow instance
 	GetWorkflowInstanceState(ctx context.Context, instance *workflow.Instance) (WorkflowState, error)
@@ -65,4 +68,7 @@ type Backend interface {
 
 	// Logger returns the configured logger for the backend
 	Logger() log.Logger
+
+	// Tracer returns th configured trace provider for the backend
+	Tracer() trace.Tracer
 }

backend/mock_Backend.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	_ "embed"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"strings"
@@ -17,6 +18,7 @@ import (
 	"github.com/cschleiden/go-workflows/workflow"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/google/uuid"
+	"go.opentelemetry.io/otel/trace"
 )
 
 //go:embed schema.sql
@@ -58,7 +60,7 @@ type mysqlBackend struct {
 }
 
 // CreateWorkflowInstance creates a new workflow instance
-func (b *mysqlBackend) CreateWorkflowInstance(ctx context.Context, m history.WorkflowEvent) error {
+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,
 	})
@@ -68,12 +70,12 @@ func (b *mysqlBackend) CreateWorkflowInstance(ctx context.Context, m history.Wor
 	defer tx.Rollback()
 
 	// Create workflow instance
-	if err := createInstance(ctx, tx, m.WorkflowInstance, false); err != nil {
+	if err := createInstance(ctx, tx, instance, event.Attributes.(*history.ExecutionStartedAttributes).Metadata, false); err != nil {
 		return err
 	}
 
 	// Initial history is empty, store only new events
-	if err := insertPendingEvents(ctx, tx, m.WorkflowInstance.InstanceID, []history.Event{m.HistoryEvent}); err != nil {
+	if err := insertPendingEvents(ctx, tx, instance.InstanceID, []history.Event{event}); err != nil {
 		return fmt.Errorf("inserting new event: %w", err)
 	}
 
@@ -88,6 +90,10 @@ func (b *mysqlBackend) Logger() log.Logger {
 	return b.options.Logger
 }
 
+func (b *mysqlBackend) Tracer() trace.Tracer {
+	return b.options.TracerProvider.Tracer(backend.TracerName)
+}
+
 func (b *mysqlBackend) CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {
 	tx, err := b.db.BeginTx(ctx, &sql.TxOptions{
 		Isolation: sql.LevelReadCommitted,
@@ -199,7 +205,7 @@ func (b *mysqlBackend) GetWorkflowInstanceState(ctx context.Context, instance *w
 	return backend.WorkflowStateActive, nil
 }
 
-func createInstance(ctx context.Context, tx *sql.Tx, wfi *workflow.Instance, ignoreDuplicate bool) error {
+func createInstance(ctx context.Context, tx *sql.Tx, wfi *workflow.Instance, metadata *workflow.Metadata, ignoreDuplicate bool) error {
 	var parentInstanceID *string
 	var parentEventID *int64
 	if wfi.SubWorkflow() {
@@ -210,13 +216,19 @@ func createInstance(ctx context.Context, tx *sql.Tx, wfi *workflow.Instance, ign
 		parentEventID = &n
 	}
 
+	metadataJson, err := json.Marshal(metadata)
+	if err != nil {
+		return fmt.Errorf("marshaling metadata: %w", err)
+	}
+
 	res, err := tx.ExecContext(
 		ctx,
-		"INSERT IGNORE INTO `instances` (instance_id, execution_id, parent_instance_id, parent_schedule_event_id) VALUES (?, ?, ?, ?)",
+		"INSERT IGNORE INTO `instances` (instance_id, execution_id, parent_instance_id, parent_schedule_event_id, metadata) VALUES (?, ?, ?, ?, ?)",
 		wfi.InstanceID,
 		wfi.ExecutionID,
 		parentInstanceID,
 		parentEventID,
+		string(metadataJson),
 	)
 	if err != nil {
 		return fmt.Errorf("inserting workflow instance: %w", err)
@@ -273,7 +285,7 @@ func (b *mysqlBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, err
 	now := time.Now()
 	row := tx.QueryRowContext(
 		ctx,
-		`SELECT i.id, i.instance_id, i.execution_id, i.parent_instance_id, i.parent_schedule_event_id, i.sticky_until
+		`SELECT i.id, i.instance_id, i.execution_id, i.parent_instance_id, i.parent_schedule_event_id, i.metadata, i.sticky_until
 			FROM instances i
 			INNER JOIN pending_events pe ON i.instance_id = pe.instance_id
 			WHERE
@@ -293,8 +305,9 @@ func (b *mysqlBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, err
 	var instanceID, executionID string
 	var parentInstanceID *string
 	var parentEventID *int64
+	var metadataJson sql.NullString
 	var stickyUntil *time.Time
-	if err := row.Scan(&id, &instanceID, &executionID, &parentInstanceID, &parentEventID, &stickyUntil); err != nil {
+	if err := row.Scan(&id, &instanceID, &executionID, &parentInstanceID, &parentEventID, &metadataJson, &stickyUntil); err != nil {
 		if err == sql.ErrNoRows {
 			return nil, nil
 		}
@@ -329,9 +342,17 @@ func (b *mysqlBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, err
 		wfi = core.NewWorkflowInstance(instanceID, executionID)
 	}
 
+	var metadata *core.WorkflowMetadata
+	if metadataJson.Valid {
+		if err := json.Unmarshal([]byte(metadataJson.String), &metadata); err != nil {
+			return nil, fmt.Errorf("parsing workflow metadata: %w", err)
+		}
+	}
+
 	t := &task.Workflow{
 		ID:               wfi.InstanceID,
 		WorkflowInstance: wfi,
+		Metadata:         metadata,
 		NewEvents:        []history.Event{},
 	}
 
@@ -489,20 +510,18 @@ func (b *mysqlBackend) CompleteWorkflowTask(
 	}
 
 	// Insert new workflow events
-	groupedEvents := make(map[*workflow.Instance][]history.Event)
-	for _, m := range workflowEvents {
-		if _, ok := groupedEvents[m.WorkflowInstance]; !ok {
-			groupedEvents[m.WorkflowInstance] = []history.Event{}
-		}
-
-		groupedEvents[m.WorkflowInstance] = append(groupedEvents[m.WorkflowInstance], m.HistoryEvent)
-	}
+	groupedEvents := history.EventsByWorkflowInstance(workflowEvents)
 
 	for targetInstance, events := range groupedEvents {
-		if targetInstance.InstanceID != instance.InstanceID {
-			// Create new instance
-			if err := createInstance(ctx, tx, targetInstance, true); err != nil {
-				return err
+		for _, event := range events {
+			if event.Type == history.EventType_WorkflowExecutionStarted {
+				a := event.Attributes.(*history.ExecutionStartedAttributes)
+				// Create new instance
+				if err := createInstance(ctx, tx, targetInstance, a.Metadata, true); err != nil {
+					return err
+				}
+
+				break
 			}
 		}
 
@@ -561,9 +580,11 @@ func (b *mysqlBackend) GetActivityTask(ctx context.Context) (*task.Activity, err
 	now := time.Now()
 	res := tx.QueryRowContext(
 		ctx,
-		`SELECT id, activity_id, instance_id, execution_id, event_type, timestamp, schedule_event_id, attributes, visible_at
+		`SELECT activities.id, activity_id, activities.instance_id, activities.execution_id,
+			instances.metadata, event_type, timestamp, schedule_event_id, attributes, visible_at
 			FROM activities
-			WHERE locked_until IS NULL OR locked_until < ?
+				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`,
 		now,
@@ -572,16 +593,24 @@ func (b *mysqlBackend) GetActivityTask(ctx context.Context) (*task.Activity, err
 	var id int64
 	var instanceID, executionID string
 	var attributes []byte
+	var metadataJson sql.NullString
 	event := history.Event{}
 
-	if err := res.Scan(&id, &event.ID, &instanceID, &executionID, &event.Type, &event.Timestamp, &event.ScheduleEventID, &attributes, &event.VisibleAt); err != nil {
+	if err := res.Scan(
+		&id, &event.ID, &instanceID, &executionID, &metadataJson, &event.Type,
+		&event.Timestamp, &event.ScheduleEventID, &attributes, &event.VisibleAt); err != nil {
 		if err == sql.ErrNoRows {
 			return nil, nil
 		}
 
 		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)
@@ -602,6 +631,7 @@ func (b *mysqlBackend) GetActivityTask(ctx context.Context) (*task.Activity, err
 	t := &task.Activity{
 		ID:               event.ID,
 		WorkflowInstance: core.NewWorkflowInstance(instanceID, executionID),
+		Metadata:         metadata,
 		Event:            event,
 	}
 

backend/mysql/mysql.go

@@ -4,6 +4,7 @@ CREATE TABLE IF NOT EXISTS `instances` (
   `execution_id` NVARCHAR(128) NOT NULL,
   `parent_instance_id` NVARCHAR(128) NULL,
   `parent_schedule_event_id` BIGINT NULL,
+  `metadata` BLOB NULL,
   `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
   `completed_at` DATETIME NULL,
   `locked_until` DATETIME NULL,

backend/mysql/schema.sql

@@ -5,11 +5,14 @@ import (
 
 	"github.com/cschleiden/go-workflows/internal/logger"
 	"github.com/cschleiden/go-workflows/log"
+	"go.opentelemetry.io/otel/trace"
 )
 
 type Options struct {
 	Logger log.Logger
 
+	TracerProvider trace.TracerProvider
+
 	StickyTimeout time.Duration
 
 	WorkflowLockTimeout time.Duration
@@ -21,6 +24,9 @@ var DefaultOptions Options = Options{
 	StickyTimeout:       30 * time.Second,
 	WorkflowLockTimeout: time.Minute,
 	ActivityLockTimeout: time.Minute * 2,
+
+	Logger:         logger.NewDefaultLogger(),
+	TracerProvider: trace.NewNoopTracerProvider(),
 }
 
 type BackendOption func(*Options)
@@ -37,6 +43,12 @@ func WithLogger(logger log.Logger) BackendOption {
 	}
 }
 
+func WithTracerProvider(tp trace.TracerProvider) BackendOption {
+	return func(o *Options) {
+		o.TracerProvider = tp
+	}
+}
+
 func ApplyOptions(opts ...BackendOption) Options {
 	options := DefaultOptions