Merge pull request #204 from cschleiden/namespace-log-fields

Namespace fields in logs

Author: cschleiden

backend/redis/diagnostics.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"github.com/cschleiden/go-workflows/diag"
+	"github.com/cschleiden/go-workflows/log"
 	redis "github.com/redis/go-redis/v9"
 )
 
@@ -21,7 +22,7 @@ func (rb *redisBackend) GetWorkflowInstances(ctx context.Context, afterInstanceI
 		}
 
 		if len(scores) == 0 {
-			rb.Logger().Error("could not find instance %v", "afterInstanceID", afterInstanceID)
+			rb.Logger().Error("could not find instance %v", log.NamespaceKey+".redis.afterInstanceID", afterInstanceID)
 			return nil, nil
 		}
 

backend/redis/signal.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/tracing"
+	"github.com/cschleiden/go-workflows/log"
 	"github.com/redis/go-redis/v9"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
@@ -20,8 +21,8 @@ func (rb *redisBackend) SignalWorkflow(ctx context.Context, instanceID string, e
 	ctx = tracing.UnmarshalSpan(ctx, instanceState.Metadata)
 	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),
+		attribute.String(log.InstanceIDKey, instanceID),
+		attribute.String(log.SignalNameKey, event.Attributes.(*history.SignalReceivedAttributes).Name),
 	))
 	defer span.End()
 

backend/redis/workflow.go

@@ -11,6 +11,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/task"
 	"github.com/cschleiden/go-workflows/internal/tracing"
+	"github.com/cschleiden/go-workflows/log"
 	"github.com/redis/go-redis/v9"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
@@ -265,7 +266,7 @@ func (rb *redisBackend) CompleteWorkflowTask(
 
 		for _, cmd := range executedCmds {
 			if cmdErr := cmd.Err(); cmdErr != nil {
-				rb.Logger().Debug("redis command error", "cmd", cmd.FullName(), "cmdErr", cmdErr.Error())
+				rb.Logger().Debug("redis command error", log.NamespaceKey+".redis.cmd", cmd.FullName(), log.NamespaceKey+".redis.cmdErr", cmdErr.Error())
 			}
 		}
 
@@ -277,7 +278,7 @@ func (rb *redisBackend) CompleteWorkflowTask(
 		ctx = tracing.UnmarshalSpan(ctx, instanceState.Metadata)
 		_, span := rb.Tracer().Start(ctx, "WorkflowComplete",
 			trace.WithAttributes(
-				attribute.String("workflow_instance_id", instanceState.Instance.InstanceID),
+				attribute.String(log.NamespaceKey+log.InstanceIDKey, instanceState.Instance.InstanceID),
 			))
 		span.End()
 

client/client.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/metrickeys"
 	"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/attribute"
@@ -73,8 +74,8 @@ func (c *client) CreateWorkflowInstance(ctx context.Context, options WorkflowIns
 
 	// Start new span and add to metadata
 	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),
+		attribute.String(log.InstanceIDKey, wfi.InstanceID),
+		attribute.String(log.WorkflowNameKey, workflowName),
 	))
 	defer span.End()
 
@@ -93,19 +94,30 @@ func (c *client) CreateWorkflowInstance(ctx context.Context, options WorkflowIns
 		return nil, fmt.Errorf("creating workflow instance: %w", err)
 	}
 
-	c.backend.Logger().Debug("Created workflow instance", "instance_id", wfi.InstanceID)
+	c.backend.Logger().Debug("Created workflow instance", log.InstanceIDKey, wfi.InstanceID)
 
 	c.backend.Metrics().Counter(metrickeys.WorkflowInstanceCreated, metrics.Tags{}, 1)
 
 	return wfi, nil
 }
 
 func (c *client) CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance) error {
+	ctx, span := c.backend.Tracer().Start(ctx, "CancelWorkflowInstance", trace.WithAttributes(
+		attribute.String(log.InstanceIDKey, instance.InstanceID),
+	))
+	defer span.End()
+
 	cancellationEvent := history.NewWorkflowCancellationEvent(time.Now())
 	return c.backend.CancelWorkflowInstance(ctx, instance, cancellationEvent)
 }
 
 func (c *client) SignalWorkflow(ctx context.Context, instanceID string, name string, arg interface{}) error {
+	ctx, span := c.backend.Tracer().Start(ctx, "SignalWorkflow", trace.WithAttributes(
+		attribute.String(log.InstanceIDKey, instanceID),
+		attribute.String(log.SignalNameKey, name),
+	))
+	defer span.End()
+
 	input, err := c.backend.Converter().To(arg)
 	if err != nil {
 		return fmt.Errorf("converting arguments: %w", err)
@@ -125,7 +137,7 @@ func (c *client) SignalWorkflow(ctx context.Context, instanceID string, name str
 		return err
 	}
 
-	c.backend.Logger().Debug("Signaled workflow instance", "instance_id", instanceID)
+	c.backend.Logger().Debug("Signaled workflow instance", log.InstanceIDKey, instanceID)
 
 	return nil
 }
@@ -135,6 +147,11 @@ func (c *client) WaitForWorkflowInstance(ctx context.Context, instance *workflow
 		timeout = time.Second * 20
 	}
 
+	ctx, span := c.backend.Tracer().Start(ctx, "WaitForWorkflowInstance", trace.WithAttributes(
+		attribute.String(log.InstanceIDKey, instance.InstanceID),
+	))
+	defer span.End()
+
 	b := backoff.ExponentialBackOff{
 		InitialInterval:     time.Millisecond * 1,
 		MaxInterval:         time.Second * 1,
@@ -166,13 +183,18 @@ func (c *client) WaitForWorkflowInstance(ctx context.Context, instance *workflow
 // GetWorkflowResult gets the workflow result for the given workflow result. It first waits for the workflow to finish or until
 // the given timeout has expired.
 func GetWorkflowResult[T any](ctx context.Context, c Client, instance *workflow.Instance, timeout time.Duration) (T, error) {
+	ic := c.(*client)
+	b := ic.backend
+
+	ctx, span := b.Tracer().Start(ctx, "GetWorkflowResult", trace.WithAttributes(
+		attribute.String(log.InstanceIDKey, instance.InstanceID),
+	))
+	defer span.End()
+
 	if err := c.WaitForWorkflowInstance(ctx, instance, timeout); err != nil {
 		return *new(T), fmt.Errorf("workflow did not finish in time: %w", err)
 	}
 
-	ic := c.(*client)
-	b := ic.backend
-
 	h, err := b.GetWorkflowInstanceHistory(ctx, instance, nil)
 	if err != nil {
 		return *new(T), fmt.Errorf("getting workflow history: %w", err)
@@ -207,5 +229,10 @@ func GetWorkflowResult[T any](ctx context.Context, c Client, instance *workflow.
 }
 
 func (c *client) RemoveWorkflowInstance(ctx context.Context, instance *core.WorkflowInstance) error {
+	ctx, span := c.backend.Tracer().Start(ctx, "RemoveWorkflowInstance", trace.WithAttributes(
+		attribute.String(log.InstanceIDKey, instance.InstanceID),
+	))
+	defer span.End()
+
 	return c.backend.RemoveWorkflowInstance(ctx, instance)
 }

client/client_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/trace"
 )
 
 func Test_Client_CreateWorkflowInstance_ParamMismatch(t *testing.T) {
@@ -45,6 +46,7 @@ func Test_Client_GetWorkflowResultTimeout(t *testing.T) {
 	ctx := context.Background()
 
 	b := &backend.MockBackend{}
+	b.On("Tracer").Return(trace.NewNoopTracerProvider().Tracer("test"))
 	b.On("GetWorkflowInstanceState", mock.Anything, instance).Return(core.WorkflowInstanceStateActive, nil)
 
 	c := &client{
@@ -68,6 +70,7 @@ func Test_Client_GetWorkflowResultSuccess(t *testing.T) {
 	r, _ := converter.DefaultConverter.To(42)
 
 	b := &backend.MockBackend{}
+	b.On("Tracer").Return(trace.NewNoopTracerProvider().Tracer("test"))
 	b.On("GetWorkflowInstanceState", mock.Anything, instance).Return(core.WorkflowInstanceStateActive, nil).Once().Run(func(args mock.Arguments) {
 		// After the first call, advance the clock to immediately go to the second call below
 		mockClock.Add(time.Second)
@@ -99,9 +102,10 @@ func Test_Client_SignalWorkflow(t *testing.T) {
 	ctx := context.Background()
 
 	b := &backend.MockBackend{}
+	b.On("Tracer").Return(trace.NewNoopTracerProvider().Tracer("test"))
 	b.On("Logger").Return(logger.NewDefaultLogger())
 	b.On("Converter").Return(converter.DefaultConverter)
-	b.On("SignalWorkflow", ctx, instanceID, mock.MatchedBy(func(event *history.Event) bool {
+	b.On("SignalWorkflow", mock.Anything, instanceID, mock.MatchedBy(func(event *history.Event) bool {
 		return event.Type == history.EventType_SignalReceived &&
 			event.Attributes.(*history.SignalReceivedAttributes).Name == "test"
 	})).Return(nil)
@@ -127,9 +131,10 @@ func Test_Client_SignalWorkflow_WithArgs(t *testing.T) {
 	input, _ := converter.DefaultConverter.To(arg)
 
 	b := &backend.MockBackend{}
+	b.On("Tracer").Return(trace.NewNoopTracerProvider().Tracer("test"))
 	b.On("Logger").Return(logger.NewDefaultLogger())
 	b.On("Converter").Return(converter.DefaultConverter)
-	b.On("SignalWorkflow", ctx, instanceID, mock.MatchedBy(func(event *history.Event) bool {
+	b.On("SignalWorkflow", mock.Anything, instanceID, mock.MatchedBy(func(event *history.Event) bool {
 		return event.Type == history.EventType_SignalReceived &&
 			event.Attributes.(*history.SignalReceivedAttributes).Name == "test" &&
 			bytes.Equal(event.Attributes.(*history.SignalReceivedAttributes).Arg, input)

internal/activity/activitystate.go

@@ -18,8 +18,8 @@ func NewActivityState(activityID string, instance *workflow.Instance, logger log
 		activityID,
 		instance,
 		logger.With(
-			"activity_id", activityID,
-			"instance_id", instance.InstanceID,
+			log.ActivityIDKey, activityID,
+			log.InstanceIDKey, instance.InstanceID,
 		)}
 }
 

internal/activity/executor.go

@@ -60,10 +60,10 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *task.Activity) (pa
 	activityCtx := WithActivityState(ctx, as)
 
 	activityCtx = tracing.UnmarshalSpan(activityCtx, task.Metadata)
-	activityCtx, span := e.tracer.Start(activityCtx, "ActivityTaskExecution", trace.WithAttributes(
-		attribute.String("activity", a.Name),
-		attribute.String(tracing.WorkflowInstanceID, task.WorkflowInstance.InstanceID),
-		attribute.String(tracing.ActivityTaskID, task.ID),
+	activityCtx, span := e.tracer.Start(activityCtx, fmt.Sprintf("ActivityTaskExecution: %s", a.Name), trace.WithAttributes(
+		attribute.String(log.ActivityNameKey, a.Name),
+		attribute.String(log.InstanceIDKey, task.WorkflowInstance.InstanceID),
+		attribute.String(log.ActivityIDKey, task.ID),
 	))
 	defer span.End()
 

internal/tracing/attributes.go

@@ -81,9 +81,9 @@ func NewExecutor(logger log.Logger, tracer trace.Tracer, registry *Registry, cv
 func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*ExecutionResult, error) {
 	ctx = tracing.UnmarshalSpan(ctx, t.Metadata)
 	ctx, span := e.tracer.Start(ctx, "WorkflowTaskExecution", trace.WithAttributes(
-		attribute.String(tracing.WorkflowInstanceID, t.WorkflowInstance.InstanceID),
-		attribute.String(tracing.WorkflowTaskID, t.ID),
-		attribute.Int(tracing.WorkflowTaskEvents, len(t.NewEvents)),
+		attribute.String(log.InstanceIDKey, t.WorkflowInstance.InstanceID),
+		attribute.String(log.TaskIDKey, t.ID),
+		attribute.Int(log.NewEventsKey, len(t.NewEvents)),
 	))
 	defer span.End()
 
@@ -92,17 +92,22 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 	// execution to be associated with the span for the WorkflowTaskExecution.
 	e.workflowTracer.UpdateExecution(span)
 
-	logger := e.logger.With("task_id", t.ID, "instance_id", t.WorkflowInstance.InstanceID)
+	logger := e.logger.With(
+		log.TaskIDKey, t.ID,
+		log.InstanceIDKey, t.WorkflowInstance.InstanceID)
 
-	logger.Debug("Executing workflow task", "task_last_sequence_id", t.LastSequenceID)
+	logger.Debug("Executing workflow task", log.TaskLastSequenceIDKey, t.LastSequenceID)
 
 	if t.WorkflowInstanceState == core.WorkflowInstanceStateFinished {
 		// This could happen if signals are delivered after the workflow is finished
 		logger.Error("Received workflow task for finished workflow instance, discarding events")
 
 		// Log events that caused this task to be scheduled
 		for _, event := range t.NewEvents {
-			logger.Debug("Discarded event:", "id", event.ID, "event_type", event.Type.String(), "schedule_event_id", event.ScheduleEventID)
+			logger.Debug("Discarded event:",
+				log.EventIDKey, event.ID,
+				log.EventTypeKey, event.Type.String(),
+				log.ScheduleEventIDKey, event.ScheduleEventID)
 		}
 
 		return &ExecutionResult{
@@ -113,7 +118,9 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 	skipNewEvents := false
 
 	if t.LastSequenceID > e.lastSequenceID {
-		logger.Debug("Task has newer history than current state, fetching and replaying history", "task_sequence_id", t.LastSequenceID, "local_sequence_id", e.lastSequenceID)
+		logger.Debug("Task has newer history than current state, fetching and replaying history",
+			log.TaskSequenceIDKey, t.LastSequenceID,
+			log.LocalSequenceIDKey, e.lastSequenceID)
 
 		h, err := e.historyProvider.GetWorkflowInstanceHistory(ctx, t.WorkflowInstance, &e.lastSequenceID)
 		if err != nil {
@@ -130,7 +137,9 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 			// With an error occurred during replay, we need to ensure new events don't get duplicate sequence ids
 			e.lastSequenceID = t.LastSequenceID
 		} else if t.LastSequenceID != e.lastSequenceID {
-			logger.Error("After replaying history, task still has newer history than current state", "task_sequence_id", t.LastSequenceID, "local_sequence_id", e.lastSequenceID)
+			logger.Error("After replaying history, task still has newer history than current state",
+				log.TaskSequenceIDKey, t.LastSequenceID,
+				log.LocalSequenceIDKey, e.lastSequenceID)
 
 			return nil, errors.New("even after fetching history and replaying history executor state does not match task")
 		}
@@ -188,9 +197,9 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 	}
 
 	logger.Debug("Finished workflow task",
-		"executed", len(executedEvents),
-		"last_sequence_id", e.lastSequenceID,
-		"completed", completed,
+		log.ExecutedEventsKey, len(executedEvents),
+		log.TaskLastSequenceIDKey, e.lastSequenceID,
+		log.WorkflowCompletedKey, completed,
 	)
 
 	return &ExecutionResult{
@@ -242,7 +251,7 @@ func (e *executor) executeNewEvents(newEvents []*history.Event) ([]*history.Even
 
 func (e *executor) Close() {
 	if e.workflow != nil {
-		e.logger.Debug("Stopping workflow executor", "instance_id", e.workflowState.Instance().InstanceID)
+		e.logger.Debug("Stopping workflow executor", log.InstanceIDKey, e.workflowState.Instance().InstanceID)
 
 		// End workflow if running to prevent leaking goroutines
 		e.workflow.Close()
@@ -251,12 +260,12 @@ func (e *executor) Close() {
 
 func (e *executor) executeEvent(event *history.Event) error {
 	e.logger.Debug("Executing event",
-		"instance_id", e.workflowState.Instance().InstanceID,
-		"event_id", event.ID,
-		"seq_id", event.SequenceID,
-		"event_type", event.Type,
-		"schedule_event_id", event.ScheduleEventID,
-		"is_replaying", e.workflowState.Replaying(),
+		log.InstanceIDKey, e.workflowState.Instance().InstanceID,
+		log.EventIDKey, event.ID,
+		log.SeqIDKey, event.SequenceID,
+		log.EventTypeKey, event.Type,
+		log.ScheduleEventIDKey, event.ScheduleEventID,
+		log.IsReplayingKey, e.workflowState.Replaying(),
 	)
 
 	var err error

internal/workflow/executor.go

@@ -76,7 +76,7 @@ func NewWorkflowState(instance *core.WorkflowInstance, logger log.Logger, clock
 	}
 
 	state.logger = NewReplayLogger(state, logger.With(
-		"instance_id", instance.InstanceID,
+		log.InstanceIDKey, instance.InstanceID,
 	))
 
 	return state