Merge pull request #218 from cschleiden/backend-stats

Expose simple backend stats for active instances/activities

Author: cschleiden

backend/backend.go

@@ -68,6 +68,9 @@ type Backend interface {
 	// ExtendActivityTask extends the lock of an activity task
 	ExtendActivityTask(ctx context.Context, activityID string) error
 
+	// GetStats returns stats about the backend
+	GetStats(ctx context.Context) (*Stats, error)
+
 	// Logger returns the configured logger for the backend
 	Logger() log.Logger
 

backend/mock_Backend.go

@@ -0,0 +1,53 @@
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/cschleiden/go-workflows/backend"
+)
+
+func (b *mysqlBackend) GetStats(ctx context.Context) (*backend.Stats, error) {
+	s := &backend.Stats{}
+
+	tx, err := b.db.BeginTx(ctx, &sql.TxOptions{
+		Isolation: sql.LevelReadCommitted,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to start transaction: %w", err)
+	}
+	defer tx.Rollback()
+
+	row := tx.QueryRowContext(
+		ctx,
+		"SELECT COUNT(*) FROM instances i WHERE i.completed_at IS NULL",
+	)
+	if err := row.Err(); err != nil {
+		return nil, fmt.Errorf("failed to query active instances: %w", err)
+	}
+
+	var activeInstances int64
+	if err := row.Scan(&activeInstances); err != nil {
+		return nil, fmt.Errorf("failed to scan active instances: %w", err)
+	}
+
+	s.ActiveWorkflowInstances = activeInstances
+
+	// Get pending activities
+	row = tx.QueryRowContext(
+		ctx,
+		"SELECT COUNT(*) FROM activities")
+	if err := row.Err(); err != nil {
+		return nil, fmt.Errorf("failed to query active activities: %w", err)
+	}
+
+	var pendingActivities int64
+	if err := row.Scan(&pendingActivities); err != nil {
+		return nil, fmt.Errorf("failed to scan active activities: %w", err)
+	}
+
+	s.PendingActivities = pendingActivities
+
+	return s, nil
+}

backend/mysql/stats.go

@@ -160,6 +160,8 @@ func createInstanceP(ctx context.Context, p redis.Pipeliner, instance *core.Work
 		Score:  float64(createdAt.UnixMilli()),
 	})
 
+	p.SAdd(ctx, instancesActive(), instanceSegment(instance))
+
 	return nil
 }
 
@@ -173,6 +175,10 @@ func updateInstanceP(ctx context.Context, p redis.Pipeliner, instance *core.Work
 
 	p.Set(ctx, key, string(b), 0)
 
+	if state.State != core.WorkflowInstanceStateActive {
+		p.SRem(ctx, instancesActive(), instanceSegment(instance))
+	}
+
 	// CreatedAt does not change, so skip updating the instancesByCreation() ZSET
 
 	return nil

backend/redis/instance.go

@@ -29,6 +29,10 @@ func instancesByCreation() string {
 	return "instances-by-creation"
 }
 
+func instancesActive() string {
+	return "instances-active"
+}
+
 func instancesExpiring() string {
 	return "instances-expiring"
 }

backend/redis/keys.go

@@ -74,6 +74,10 @@ func (q *taskQueue[T]) Keys() KeyInfo {
 	}
 }
 
+func (q *taskQueue[T]) Size(ctx context.Context, rdb redis.UniversalClient) (int64, error) {
+	return rdb.XLen(ctx, q.streamKey).Result()
+}
+
 // KEYS[1] = set
 // KEYS[2] = stream
 // ARGV[1] = caller provided id of the task

backend/redis/queue.go

@@ -0,0 +1,32 @@
+package redis
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cschleiden/go-workflows/backend"
+)
+
+func (rb *redisBackend) GetStats(ctx context.Context) (*backend.Stats, error) {
+	var err error
+
+	s := &backend.Stats{}
+
+	// get workflow instances
+	activeInstances, err := rb.rdb.SCard(ctx, instancesActive()).Result()
+	if err != nil {
+		return nil, fmt.Errorf("getting active instances: %w", err)
+	}
+
+	s.ActiveWorkflowInstances = activeInstances
+
+	// get pending activities
+	pendingActivities, err := rb.activityQueue.Size(ctx, rb.rdb)
+	if err != nil {
+		return nil, fmt.Errorf("getting active activities: %w", err)
+	}
+
+	s.PendingActivities = pendingActivities
+
+	return s, nil
+}

backend/redis/stats.go

@@ -83,8 +83,8 @@ func (sb *sqliteBackend) Converter() converter.Converter {
 	return sb.options.Converter
 }
 
-func (b *sqliteBackend) ContextPropagators() []contextpropagation.ContextPropagator {
-	return b.options.ContextPropagators
+func (sb *sqliteBackend) ContextPropagators() []contextpropagation.ContextPropagator {
+	return sb.options.ContextPropagators
 }
 
 func (sb *sqliteBackend) CreateWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {

backend/sqlite/sqlite.go

@@ -0,0 +1,53 @@
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/cschleiden/go-workflows/backend"
+)
+
+func (b *sqliteBackend) GetStats(ctx context.Context) (*backend.Stats, error) {
+	s := &backend.Stats{}
+
+	tx, err := b.db.BeginTx(ctx, &sql.TxOptions{
+		Isolation: sql.LevelReadCommitted,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to start transaction: %w", err)
+	}
+	defer tx.Rollback()
+
+	row := tx.QueryRowContext(
+		ctx,
+		"SELECT COUNT(*) FROM instances i WHERE i.completed_at IS NULL",
+	)
+	if err := row.Err(); err != nil {
+		return nil, fmt.Errorf("failed to query active instances: %w", err)
+	}
+
+	var activeInstances int64
+	if err := row.Scan(&activeInstances); err != nil {
+		return nil, fmt.Errorf("failed to scan active instances: %w", err)
+	}
+
+	s.ActiveWorkflowInstances = activeInstances
+
+	// Get pending activities
+	row = tx.QueryRowContext(
+		ctx,
+		"SELECT COUNT(*) FROM activities")
+	if err := row.Err(); err != nil {
+		return nil, fmt.Errorf("failed to query active activities: %w", err)
+	}
+
+	var pendingActivities int64
+	if err := row.Scan(&pendingActivities); err != nil {
+		return nil, fmt.Errorf("failed to scan active activities: %w", err)
+	}
+
+	s.PendingActivities = pendingActivities
+
+	return s, nil
+}

backend/sqlite/stats.go

@@ -0,0 +1,7 @@
+package backend
+
+type Stats struct {
+	ActiveWorkflowInstances int64
+
+	PendingActivities int64
+}