Add metrics client

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"
+	mi "github.com/cschleiden/go-workflows/internal/metrics"
 	"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{mi.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"
+	mi "github.com/cschleiden/go-workflows/internal/metrics"
 	"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{mi.Backend: "mysql"})
+}
+
 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"
+	mi "github.com/cschleiden/go-workflows/internal/metrics"
 	"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{mi.Backend: "mysql"})
+}
+
 func (sb *sqliteBackend) Tracer() trace.Tracer {
 	return sb.options.TracerProvider.Tracer(backend.TracerName)
 }

backend/sqlite/sqlite.go

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

internal/metrics/noop.go

@@ -0,0 +1,15 @@
+package metrics
+
+import "time"
+
+type Tags map[string]string
+
+type Client interface {
+	Counter(name string, tags Tags, value float64)
+
+	Distribution(name string, tags Tags, value float64)
+
+	Timing(name string, tags Tags, duration time.Duration)
+
+	WithTags(tags Tags) Client
+}

metrics/metrics.go

@@ -2,6 +2,7 @@ package worker
 
 import (
 	"context"
+	"fmt"
 	"sync"
 
 	"github.com/benbjohnson/clock"
@@ -45,8 +46,8 @@ type worker struct {
 
 	registry *workflowinternal.Registry
 
-	workflowWorker internal.WorkflowWorker
-	activityWorker internal.ActivityWorker
+	workflowWorker *internal.WorkflowWorker
+	activityWorker *internal.ActivityWorker
 
 	workflows  map[string]interface{}
 	activities map[string]interface{}
@@ -85,8 +86,13 @@ func New(backend backend.Backend, options *Options) Worker {
 }
 
 func (w *worker) Start(ctx context.Context) error {
-	w.workflowWorker.Start(ctx)
-	w.activityWorker.Start(ctx)
+	if err := w.workflowWorker.Start(ctx); err != nil {
+		return fmt.Errorf("starting workflow worker: %w", err)
+	}
+
+	if err := w.activityWorker.Start(ctx); err != nil {
+		return fmt.Errorf("starting activity worker: %w", err)
+	}
 
 	return nil
 }