Merge pull request #213 from yaananth/yaananth-metrics

Special handle timer fired events for workflow event delay metrics + add event name as tag for workflow task metrics

Author: cschleiden

internal/metrickeys/metrickeys.go

@@ -7,7 +7,6 @@ const (
 	WorkflowInstanceCreated  = Prefix + "workflow.created"
 	WorkflowInstanceFinished = Prefix + "workflow.finished"
 
-	WorkflowTaskScheduled = Prefix + "workflow.task.scheduled"
 	WorkflowTaskProcessed = Prefix + "workflow.task.processed"
 	WorkflowTaskDelay     = Prefix + "workflow.task.time_in_queue"
 
@@ -32,4 +31,5 @@ const (
 	ContinuedAsNew = "continued_as_new"
 
 	ActivityName = "activity"
+	EventName    = "event"
 )

internal/worker/workflow.go

@@ -10,6 +10,7 @@ import (
 	"github.com/benbjohnson/clock"
 	"github.com/cschleiden/go-workflows/backend"
 	"github.com/cschleiden/go-workflows/internal/core"
+	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/metrickeys"
 	"github.com/cschleiden/go-workflows/internal/task"
 	"github.com/cschleiden/go-workflows/internal/workflow"
@@ -133,11 +134,25 @@ func (ww *WorkflowWorker) runDispatcher() {
 
 func (ww *WorkflowWorker) handle(ctx context.Context, t *task.Workflow) {
 	// Record how long this task was in the queue
-	scheduledAt := t.NewEvents[0].Timestamp // Use the timestamp of the first event as the schedule time
+	firstEvent := t.NewEvents[0]
+	var scheduledAt time.Time
+	if firstEvent.Type == history.EventType_TimerFired {
+		timerFiredAttributes := firstEvent.Attributes.(*history.TimerFiredAttributes)
+		scheduledAt = timerFiredAttributes.At // Use the timestamp of the timer fired event as the schedule time
+	} else {
+		scheduledAt = firstEvent.Timestamp // Use the timestamp of the first event as the schedule time
+	}
+
+	eventName := fmt.Sprint(firstEvent.Type)
+
 	timeInQueue := time.Since(scheduledAt)
-	ww.backend.Metrics().Distribution(metrickeys.WorkflowTaskDelay, metrics.Tags{}, float64(timeInQueue/time.Millisecond))
+	ww.backend.Metrics().Distribution(metrickeys.WorkflowTaskDelay, metrics.Tags{
+		metrickeys.EventName: eventName,
+	}, float64(timeInQueue/time.Millisecond))
 
-	timer := metrics.Timer(ww.backend.Metrics(), metrickeys.WorkflowTaskProcessed, metrics.Tags{})
+	timer := metrics.Timer(ww.backend.Metrics(), metrickeys.WorkflowTaskProcessed, metrics.Tags{
+		metrickeys.EventName: eventName,
+	})
 
 	result, err := ww.handleTask(ctx, t)
 	if err != nil {