Merge pull request #133 from cschleiden/metrics

Metrics

Author: cschleiden

backend/backend.go

@@ -8,6 +8,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/task"
 	"github.com/cschleiden/go-workflows/log"
+	"github.com/cschleiden/go-workflows/metrics"
 	"github.com/cschleiden/go-workflows/workflow"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -62,6 +63,9 @@ type Backend interface {
 	// Logger returns the configured logger for the backend
 	Logger() log.Logger
 
-	// Tracer returns th configured trace provider for the backend
+	// Tracer returns the configured trace provider for the backend
 	Tracer() trace.Tracer
+
+	// Metrics returns the configured metrics client for the backend
+	Metrics() metrics.Client
 }

backend/mock_Backend.go

@@ -13,8 +13,10 @@ import (
 	"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/log"
+	"github.com/cschleiden/go-workflows/metrics"
 	"github.com/cschleiden/go-workflows/workflow"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/google/uuid"
@@ -94,6 +96,10 @@ func (b *mysqlBackend) Tracer() trace.Tracer {
 	return b.options.TracerProvider.Tracer(backend.TracerName)
 }
 
+func (b *mysqlBackend) Metrics() metrics.Client {
+	return b.options.Metrics.WithTags(metrics.Tags{metrickeys.Backend: "mysql"})
+}
+
 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,

backend/mysql/mysql.go

@@ -4,13 +4,17 @@ import (
 	"time"
 
 	"github.com/cschleiden/go-workflows/internal/logger"
+	mi "github.com/cschleiden/go-workflows/internal/metrics"
 	"github.com/cschleiden/go-workflows/log"
+	"github.com/cschleiden/go-workflows/metrics"
 	"go.opentelemetry.io/otel/trace"
 )
 
 type Options struct {
 	Logger log.Logger
 
+	Metrics metrics.Client
+
 	TracerProvider trace.TracerProvider
 
 	StickyTimeout time.Duration
@@ -26,6 +30,7 @@ var DefaultOptions Options = Options{
 	ActivityLockTimeout: time.Minute * 2,
 
 	Logger:         logger.NewDefaultLogger(),
+	Metrics:        mi.NewNoopMetricsClient(),
 	TracerProvider: trace.NewNoopTracerProvider(),
 }
 
@@ -43,6 +48,12 @@ func WithLogger(logger log.Logger) BackendOption {
 	}
 }
 
+func WithMetrics(client metrics.Client) BackendOption {
+	return func(o *Options) {
+		o.Metrics = client
+	}
+}
+
 func WithTracerProvider(tp trace.TracerProvider) BackendOption {
 	return func(o *Options) {
 		o.TracerProvider = tp

backend/options.go

@@ -8,7 +8,9 @@ import (
 	"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/log"
+	"github.com/cschleiden/go-workflows/metrics"
 	"github.com/go-redis/redis/v8"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -106,6 +108,10 @@ func (rb *redisBackend) Logger() log.Logger {
 	return rb.options.Logger
 }
 
+func (rb *redisBackend) Metrics() metrics.Client {
+	return rb.options.Metrics.WithTags(metrics.Tags{metrickeys.Backend: "redis"})
+}
+
 func (rb *redisBackend) Tracer() trace.Tracer {
 	return rb.options.TracerProvider.Tracer(backend.TracerName)
 }

backend/redis/redis.go

@@ -13,8 +13,10 @@ import (
 	"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/log"
+	"github.com/cschleiden/go-workflows/metrics"
 	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/google/uuid"
 	"go.opentelemetry.io/otel/trace"
@@ -65,6 +67,10 @@ func (sb *sqliteBackend) Logger() log.Logger {
 	return sb.options.Logger
 }
 
+func (sb *sqliteBackend) Metrics() metrics.Client {
+	return sb.options.Metrics.WithTags(metrics.Tags{metrickeys.Backend: "sqlite"})
+}
+
 func (sb *sqliteBackend) Tracer() trace.Tracer {
 	return sb.options.TracerProvider.Tracer(backend.TracerName)
 }

backend/sqlite/sqlite.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
 }
 

client/client.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/metrickeys/metrickeys.go

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

internal/metrics/noop.go

@@ -10,16 +10,13 @@ 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 interface {
-	Start(context.Context) error
-	WaitForCompletion() error
-}
-
-type activityWorker struct {
+type ActivityWorker struct {
 	backend backend.Backend
 
 	options *Options
@@ -32,8 +29,8 @@ type activityWorker struct {
 	clock clock.Clock
 }
 
-func NewActivityWorker(backend backend.Backend, registry *workflow.Registry, clock clock.Clock, options *Options) ActivityWorker {
-	return &activityWorker{
+func NewActivityWorker(backend backend.Backend, registry *workflow.Registry, clock clock.Clock, options *Options) *ActivityWorker {
+	return &ActivityWorker{
 		backend: backend,
 
 		options: options,
@@ -47,7 +44,7 @@ func NewActivityWorker(backend backend.Backend, registry *workflow.Registry, clo
 	}
 }
 
-func (aw *activityWorker) Start(ctx context.Context) error {
+func (aw *ActivityWorker) Start(ctx context.Context) error {
 	for i := 0; i <= aw.options.ActivityPollers; i++ {
 		go aw.runPoll(ctx)
 	}
@@ -57,15 +54,15 @@ func (aw *activityWorker) Start(ctx context.Context) error {
 	return nil
 }
 
-func (aw *activityWorker) WaitForCompletion() error {
+func (aw *ActivityWorker) WaitForCompletion() error {
 	close(aw.activityTaskQueue)
 
 	aw.wg.Wait()
 
 	return nil
 }
 
-func (aw *activityWorker) runPoll(ctx context.Context) {
+func (aw *ActivityWorker) runPoll(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -84,7 +81,7 @@ func (aw *activityWorker) runPoll(ctx context.Context) {
 	}
 }
 
-func (aw *activityWorker) runDispatcher(ctx context.Context) {
+func (aw *ActivityWorker) runDispatcher(ctx context.Context) {
 	var sem chan struct{}
 	if aw.options.MaxParallelActivityTasks > 0 {
 		sem = make(chan struct{}, aw.options.MaxParallelActivityTasks)
@@ -112,7 +109,15 @@ func (aw *activityWorker) runDispatcher(ctx context.Context) {
 	}
 }
 
-func (aw *activityWorker) handleTask(ctx context.Context, task *task.Activity) {
+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) {
@@ -131,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()
@@ -161,7 +169,7 @@ func (aw *activityWorker) handleTask(ctx context.Context, task *task.Activity) {
 	}
 }
 
-func (aw *activityWorker) poll(ctx context.Context, timeout time.Duration) (*task.Activity, error) {
+func (aw *ActivityWorker) poll(ctx context.Context, timeout time.Duration) (*task.Activity, error) {
 	if timeout == 0 {
 		timeout = 30 * time.Second
 	}