Merge pull request #279 from cschleiden/redis-metrics

Add pending workflow tasks metric to stats interface

Author: cschleiden

backend/mysql/stats.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"time"
 
 	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/core"
 )
 
 func (b *mysqlBackend) GetStats(ctx context.Context) (*backend.Stats, error) {
@@ -19,6 +21,7 @@ func (b *mysqlBackend) GetStats(ctx context.Context) (*backend.Stats, error) {
 	}
 	defer tx.Rollback()
 
+	// Get active instances
 	row := tx.QueryRowContext(
 		ctx,
 		"SELECT COUNT(*) FROM instances i WHERE i.completed_at IS NULL",
@@ -34,6 +37,34 @@ func (b *mysqlBackend) GetStats(ctx context.Context) (*backend.Stats, error) {
 
 	s.ActiveWorkflowInstances = activeInstances
 
+	// Get workflow instances ready to be picked up
+	now := time.Now()
+	row = tx.QueryRowContext(
+		ctx,
+		`SELECT COUNT(*)
+			FROM instances i
+			INNER JOIN pending_events pe ON i.instance_id = pe.instance_id
+			WHERE
+				state = ? AND i.completed_at IS NULL
+				AND (pe.visible_at IS NULL OR pe.visible_at <= ?)
+				AND (i.locked_until IS NULL OR i.locked_until < ?)
+			LIMIT 1
+			FOR UPDATE OF i SKIP LOCKED`,
+		core.WorkflowInstanceStateActive,
+		now, // event.visible_at
+		now, // locked_until
+	)
+	if err := row.Err(); err != nil {
+		return nil, fmt.Errorf("failed to query active instances: %w", err)
+	}
+
+	var pendingInstances int64
+	if err := row.Scan(&pendingInstances); err != nil {
+		return nil, fmt.Errorf("failed to scan active instances: %w", err)
+	}
+
+	s.PendingWorkflowTasks = pendingInstances
+
 	// Get pending activities
 	row = tx.QueryRowContext(
 		ctx,

backend/redis/stats.go

@@ -20,6 +20,14 @@ func (rb *redisBackend) GetStats(ctx context.Context) (*backend.Stats, error) {
 
 	s.ActiveWorkflowInstances = activeInstances
 
+	// get pending workflow tasks
+	pendingWorkflows, err := rb.workflowQueue.Size(ctx, rb.rdb)
+	if err != nil {
+		return nil, fmt.Errorf("getting active workflows: %w", err)
+	}
+
+	s.PendingWorkflowTasks = pendingWorkflows
+
 	// get pending activities
 	pendingActivities, err := rb.activityQueue.Size(ctx, rb.rdb)
 	if err != nil {

backend/sqlite/stats.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"time"
 
 	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/core"
 )
 
 func (b *sqliteBackend) GetStats(ctx context.Context) (*backend.Stats, error) {
@@ -34,6 +36,35 @@ func (b *sqliteBackend) GetStats(ctx context.Context) (*backend.Stats, error) {
 
 	s.ActiveWorkflowInstances = activeInstances
 
+	// Get workflow instances ready to be picked up
+	now := time.Now()
+	row = tx.QueryRowContext(
+		ctx,
+		`SELECT COUNT(*) FROM instances i
+			WHERE
+				(locked_until IS NULL OR locked_until < ?)
+				AND state = ? AND i.completed_at IS NULL
+				AND EXISTS (
+					SELECT 1
+						FROM pending_events
+						WHERE instance_id = i.id AND execution_id = i.execution_id AND (visible_at IS NULL OR visible_at <= ?)
+				)
+			LIMIT 1`,
+		now,                              // locked_until
+		core.WorkflowInstanceStateActive, // state
+		now,                              // pending_event.visible_at
+	)
+	if err := row.Err(); err != nil {
+		return nil, fmt.Errorf("failed to query active instances: %w", err)
+	}
+
+	var pendingInstances int64
+	if err := row.Scan(&pendingInstances); err != nil {
+		return nil, fmt.Errorf("failed to scan active instances: %w", err)
+	}
+
+	s.PendingWorkflowTasks = pendingInstances
+
 	// Get pending activities
 	row = tx.QueryRowContext(
 		ctx,

backend/stats.go

@@ -3,5 +3,11 @@ package backend
 type Stats struct {
 	ActiveWorkflowInstances int64
 
+	// PendingWorkflowTasks are the number of workflow tasks that are currently in the queue,
+	// waiting to be processed by a worker
+	PendingWorkflowTasks int64
+
+	// PendingActivities are the number of activities that are currently in the queue,
+	// waiting to be processed by a worker
 	PendingActivities int64
 }

backend/test/e2e_stats.go

@@ -15,12 +15,12 @@ var e2eStatsTests = []backendTest{
 	{
 		name: "Stats_ActiveInstance",
 		f: func(t *testing.T, ctx context.Context, c *client.Client, w *worker.Worker, b TestBackend) {
-			as := make(chan bool, 1)
-			af := make(chan bool, 1)
+			activityRunning := make(chan bool, 1)
+			activityBlocked := make(chan bool, 1)
 
 			a := func(ctx context.Context) error {
-				as <- true
-				<-af
+				activityRunning <- true
+				<-activityBlocked
 
 				return nil
 			}
@@ -31,28 +31,44 @@ var e2eStatsTests = []backendTest{
 
 				return true, nil
 			}
-			register(t, ctx, w, []interface{}{wf}, []interface{}{a})
+
+			require.NoError(t, w.RegisterWorkflow(wf))
+			require.NoError(t, w.RegisterActivity(a))
 
 			s, err := b.GetStats(ctx)
 			require.NoError(t, err)
 			require.Equal(t, int64(0), s.ActiveWorkflowInstances)
+			require.Equal(t, int64(0), s.PendingWorkflowTasks)
 			require.Equal(t, int64(0), s.PendingActivities)
 
 			wfi := runWorkflow(t, ctx, c, wf)
 
 			s, err = b.GetStats(ctx)
 			require.NoError(t, err)
 			require.Equal(t, int64(1), s.ActiveWorkflowInstances)
+			require.Equal(t, int64(1), s.PendingWorkflowTasks)
+			require.Equal(t, int64(0), s.PendingActivities)
+
+			// Start worker
+			require.NoError(t, w.Start(ctx))
+
+			// Wait until the activity is running
+			<-activityRunning
 
-			<-as
+			s, err = b.GetStats(ctx)
+			require.NoError(t, err)
+			require.Equal(t, int64(0), s.PendingWorkflowTasks)
+			require.Equal(t, int64(1), s.ActiveWorkflowInstances)
 
 			s, err = b.GetStats(ctx)
 			require.NoError(t, err)
 			require.Equal(t, int64(1), s.ActiveWorkflowInstances)
 			require.Equal(t, int64(1), s.PendingActivities)
 
-			af <- true
+			// Let the activity finish
+			activityBlocked <- true
 
+			// Let the workflow finish
 			err = c.SignalWorkflow(ctx, wfi.InstanceID, "test-signal", nil)
 			require.NoError(t, err)