Support emitting metrics

Author: cschleiden

client/client.go

@@ -14,7 +14,9 @@ import (
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/fn"
 	"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/metrics"
 	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/google/uuid"
 	"go.opentelemetry.io/otel/attribute"
@@ -88,6 +90,8 @@ func (c *client) CreateWorkflowInstance(ctx context.Context, options WorkflowIns
 
 	c.backend.Logger().Debug("Created workflow instance", "instance_id", wfi.InstanceID, "execution_id", wfi.ExecutionID)
 
+	c.backend.Metrics().Counter(metrickeys.WorkflowInstanceCreated, metrics.Tags{}, 1)
+
 	return wfi, nil
 }
 

internal/metrickeys/metrickeys.go

@@ -0,0 +1,32 @@
+package metrickeys
+
+const (
+	Prefix = "workflows."
+
+	// Workflows
+	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"
+
+	WorkflowInstanceCacheSize     = Prefix + "workflow.cache.size"
+	WorkflowInstanceCacheEviction = Prefix + "workflow.cache.eviction"
+
+	// Activities
+	ActivityTaskScheduled = Prefix + "activity.task.scheduled"
+	ActivityTaskProcessed = Prefix + "activity.task.processed"
+	ActivityTaskDelay     = Prefix + "activity.task.time_in_queue"
+)
+
+// Tag names
+const (
+	// Backend being used
+	Backend = "backend"
+
+	// Reason for evicting an entry from the workflow instance cache
+	EvictionReason = "reason"
+
+	ActivityName = "activity"
+)

internal/metrics/noop.go

@@ -3,7 +3,7 @@ package metrics
 import (
 	"time"
 
-	m "github.com/cschleiden/go-workflows/metrics"
+	"github.com/cschleiden/go-workflows/metrics"
 )
 
 type noopMetricsClient struct {
@@ -13,20 +13,21 @@ func NewNoopMetricsClient() *noopMetricsClient {
 	return &noopMetricsClient{}
 }
 
-var _ m.Client = (*noopMetricsClient)(nil)
+var _ metrics.Client = (*noopMetricsClient)(nil)
 
-func (*noopMetricsClient) Counter(name string, tags m.Tags, value float64) {
+func (*noopMetricsClient) Counter(name string, tags metrics.Tags, value int64) {
 }
 
-// Distribution implements metrics.Client
-func (*noopMetricsClient) Distribution(name string, tags m.Tags, value float64) {
+func (*noopMetricsClient) Distribution(name string, tags metrics.Tags, value float64) {
 }
 
-// Timing implements metrics.Client
-func (*noopMetricsClient) Timing(name string, tags m.Tags, duration time.Duration) {
+func (*noopMetricsClient) Gauge(name string, tags metrics.Tags, value int64) {
+
+}
+
+func (*noopMetricsClient) Timing(name string, tags metrics.Tags, duration time.Duration) {
 }
 
-// WithTags implements metrics.Client
-func (nmc *noopMetricsClient) WithTags(tags m.Tags) m.Client {
+func (nmc *noopMetricsClient) WithTags(tags metrics.Tags) metrics.Client {
 	return nmc
 }

internal/worker/activity.go

@@ -10,8 +10,10 @@ import (
 	"github.com/cschleiden/go-workflows/backend"
 	"github.com/cschleiden/go-workflows/internal/activity"
 	"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"
+	"github.com/cschleiden/go-workflows/metrics"
 )
 
 type ActivityWorker struct {
@@ -108,6 +110,14 @@ func (aw *ActivityWorker) runDispatcher(ctx context.Context) {
 }
 
 func (aw *ActivityWorker) handleTask(ctx context.Context, task *task.Activity) {
+	a := task.Event.Attributes.(*history.ActivityScheduledAttributes)
+	ametrics := aw.backend.Metrics().WithTags(metrics.Tags{metrickeys.ActivityName: a.Name})
+
+	// Record how long this task was in the queue
+	scheduledAt := task.Event.Timestamp
+	timeInQueue := time.Since(scheduledAt)
+	ametrics.Distribution(metrickeys.ActivityTaskDelay, metrics.Tags{}, float64(timeInQueue/time.Millisecond))
+
 	// Start heartbeat while activity is running
 	heartbeatCtx, cancelHeartbeat := context.WithCancel(ctx)
 	go func(ctx context.Context) {
@@ -126,6 +136,9 @@ func (aw *ActivityWorker) handleTask(ctx context.Context, task *task.Activity) {
 		}
 	}(heartbeatCtx)
 
+	timer := metrics.Timer(ametrics, metrickeys.ActivityTaskProcessed, metrics.Tags{})
+	defer timer.Stop()
+
 	result, err := aw.activityTaskExecutor.ExecuteActivity(ctx, task)
 
 	cancelHeartbeat()

internal/worker/workflow.go

@@ -9,10 +9,12 @@ import (
 	"github.com/benbjohnson/clock"
 	"github.com/cschleiden/go-workflows/backend"
 	"github.com/cschleiden/go-workflows/internal/core"
+	"github.com/cschleiden/go-workflows/internal/metrickeys"
 	"github.com/cschleiden/go-workflows/internal/task"
 	"github.com/cschleiden/go-workflows/internal/workflow"
 	"github.com/cschleiden/go-workflows/internal/workflow/cache"
 	"github.com/cschleiden/go-workflows/log"
+	"github.com/cschleiden/go-workflows/metrics"
 )
 
 type WorkflowWorker struct {
@@ -36,7 +38,7 @@ func NewWorkflowWorker(backend backend.Backend, registry *workflow.Registry, opt
 	if options.WorkflowExecutorCache != nil {
 		c = options.WorkflowExecutorCache
 	} else {
-		c = cache.NewWorkflowExecutorLRUCache(options.WorkflowExecutorCacheSize, options.WorkflowExecutorCacheTTL)
+		c = cache.NewWorkflowExecutorLRUCache(backend.Metrics(), options.WorkflowExecutorCacheSize, options.WorkflowExecutorCacheTTL)
 	}
 
 	return &WorkflowWorker{
@@ -123,6 +125,14 @@ 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
+	timeInQueue := time.Since(scheduledAt)
+	ww.backend.Metrics().Distribution(metrickeys.WorkflowTaskDelay, metrics.Tags{}, float64(timeInQueue/time.Millisecond))
+
+	timer := metrics.Timer(ww.backend.Metrics(), metrickeys.WorkflowTaskProcessed, metrics.Tags{})
+	defer timer.Stop()
+
 	result, err := ww.handleTask(ctx, t)
 	if err != nil {
 		ww.logger.Panic("could not handle workflow task", "error", err)
@@ -131,8 +141,12 @@ func (ww *WorkflowWorker) handle(ctx context.Context, t *task.Workflow) {
 	state := core.WorkflowInstanceStateActive
 	if result.Completed {
 		state = core.WorkflowInstanceStateFinished
+
+		ww.backend.Metrics().Counter(metrickeys.WorkflowInstanceFinished, metrics.Tags{}, 1)
 	}
 
+	ww.backend.Metrics().Counter(metrickeys.ActivityTaskScheduled, metrics.Tags{}, int64(len(result.ActivityEvents)))
+
 	if err := ww.backend.CompleteWorkflowTask(
 		ctx, t, t.WorkflowInstance, state, result.Executed, result.ActivityEvents, result.TimerEvents, result.WorkflowEvents); err != nil {
 		ww.logger.Panic("could not complete workflow task", "error", err)

internal/workflow/cache/cache.go

@@ -5,26 +5,41 @@ import (
 	"time"
 
 	"github.com/cschleiden/go-workflows/internal/core"
+	"github.com/cschleiden/go-workflows/internal/metrickeys"
 	"github.com/cschleiden/go-workflows/internal/workflow"
+	"github.com/cschleiden/go-workflows/metrics"
 	"github.com/jellydator/ttlcache/v3"
 )
 
 type LruCache struct {
-	c *ttlcache.Cache[string, workflow.WorkflowExecutor]
+	mc metrics.Client
+	c  *ttlcache.Cache[string, workflow.WorkflowExecutor]
 }
 
-func NewWorkflowExecutorLRUCache(size int, expiration time.Duration) workflow.ExecutorCache {
+func NewWorkflowExecutorLRUCache(mc metrics.Client, size int, expiration time.Duration) workflow.ExecutorCache {
 	c := ttlcache.New(
 		ttlcache.WithCapacity[string, workflow.WorkflowExecutor](uint64(size)),
 		ttlcache.WithTTL[string, workflow.WorkflowExecutor](expiration),
 	)
 
 	c.OnEviction(func(ctx context.Context, er ttlcache.EvictionReason, i *ttlcache.Item[string, workflow.WorkflowExecutor]) {
+		// Close the executor to allow it to clean up resources.
 		i.Value().Close()
+
+		reason := ""
+		switch er {
+		case ttlcache.EvictionReasonExpired:
+			reason = "expired"
+		case ttlcache.EvictionReasonCapacityReached:
+			reason = "capacity"
+		}
+
+		mc.Counter(metrickeys.WorkflowInstanceCacheEviction, metrics.Tags{metrickeys.EvictionReason: reason}, 1)
 	})
 
 	return &LruCache{
-		c: c,
+		mc: mc,
+		c:  c,
 	}
 }
 
@@ -40,6 +55,8 @@ func (lc *LruCache) Get(ctx context.Context, instance *core.WorkflowInstance) (w
 func (lc *LruCache) Store(ctx context.Context, instance *core.WorkflowInstance, executor workflow.WorkflowExecutor) error {
 	lc.c.Set(getKey(instance), executor, ttlcache.DefaultTTL)
 
+	lc.mc.Gauge(metrickeys.WorkflowInstanceCacheSize, metrics.Tags{}, int64(lc.c.Len()))
+
 	return nil
 }
 

internal/workflow/cache/cache_test.go

@@ -11,14 +11,15 @@ import (
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/logger"
+	"github.com/cschleiden/go-workflows/internal/metrics"
 	wf "github.com/cschleiden/go-workflows/internal/workflow"
 	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel/trace"
 )
 
 func Test_Cache_StoreAndGet(t *testing.T) {
-	c := NewWorkflowExecutorLRUCache(1, time.Second*10)
+	c := NewWorkflowExecutorLRUCache(metrics.NewNoopMetricsClient(), 1, time.Second*10)
 
 	r := wf.NewRegistry()
 	r.RegisterWorkflow(workflowWithActivity)
@@ -51,7 +52,9 @@ func Test_Cache_StoreAndGet(t *testing.T) {
 }
 
 func Test_Cache_Evict(t *testing.T) {
-	c := NewWorkflowExecutorLRUCache(128,
+	c := NewWorkflowExecutorLRUCache(
+		metrics.NewNoopMetricsClient(),
+		128,
 		1, // Should evict immediately
 	)
 

internal/workflow/executor.go

@@ -101,7 +101,7 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 
 	if t.WorkflowInstanceState == core.WorkflowInstanceStateFinished {
 		// This should never happen. For now, log information and then panic.
-		logger.Debug("Received workflow task for finished workflow instance, discarding events")
+		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 {

metrics/metrics.go

@@ -5,10 +5,12 @@ import "time"
 type Tags map[string]string
 
 type Client interface {
-	Counter(name string, tags Tags, value float64)
+	Counter(name string, tags Tags, value int64)
 
 	Distribution(name string, tags Tags, value float64)
 
+	Gauge(name string, tags Tags, value int64)
+
 	Timing(name string, tags Tags, duration time.Duration)
 
 	WithTags(tags Tags) Client

metrics/timer.go

@@ -0,0 +1,27 @@
+package metrics
+
+import (
+	"time"
+)
+
+type timer struct {
+	client Client
+	start  time.Time
+	name   string
+	tags   Tags
+}
+
+func Timer(client Client, name string, tags Tags) *timer {
+	return &timer{
+		client: client,
+		start:  time.Now(),
+		name:   name,
+		tags:   tags,
+	}
+}
+
+// Stop the timer and send the elapsed time as milliseconds as a distribution metric
+func (t *timer) Stop() {
+	elapsed := time.Since(t.start)
+	t.client.Distribution(t.name, t.tags, float64(elapsed/time.Millisecond))
+}