Merge pull request #297 from cschleiden/redis-diag

Track newly created instances correctly

Author: cschleiden

backend/redis/diagnostics_test.go

@@ -0,0 +1,44 @@
+package redis
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cschleiden/go-workflows/client"
+	"github.com/cschleiden/go-workflows/diag"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_Diag_GetWorkflowInstances(t *testing.T) {
+	if testing.Short() {
+		t.Skip()
+	}
+
+	rclient := getClient()
+	setup := getCreateBackend(rclient)
+
+	b := setup()
+
+	t.Cleanup(func() {
+		b.Close()
+	})
+
+	bd := b.(diag.Backend)
+
+	ctx := context.Background()
+	instances, err := bd.GetWorkflowInstances(ctx, "", "", 5)
+	require.NoError(t, err)
+	require.Len(t, instances, 0)
+
+	c := client.New(b)
+
+	_, err = c.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{
+		InstanceID: "ex1",
+	}, "some-workflow")
+	require.NoError(t, err)
+
+	instances, err = bd.GetWorkflowInstances(ctx, "", "", 5)
+	require.NoError(t, err)
+	require.Len(t, instances, 1)
+	require.Equal(t, "ex1", instances[0].Instance.InstanceID)
+}

backend/redis/instance.go

@@ -48,6 +48,7 @@ func (rb *redisBackend) CreateWorkflowInstance(ctx context.Context, instance *wo
 		pendingEventsKey(instance),
 		payloadKey(instance),
 		instancesActive(),
+		instancesByCreation(),
 		keyInfo.SetKey,
 		keyInfo.StreamKey,
 	},
@@ -57,6 +58,7 @@ func (rb *redisBackend) CreateWorkflowInstance(ctx context.Context, instance *wo
 		event.ID,
 		eventData,
 		payloadData,
+		time.Now().UTC().Unix(),
 	).Result()
 
 	if err != nil {

backend/redis/keys.go

@@ -24,11 +24,13 @@ func instanceKeyFromSegment(segment string) string {
 }
 
 // instancesByCreation returns the key for the ZSET that contains all instances sorted by creation date. The score is the
-// creation time. Used for listing all workflow instances in the diagnostics UI.
+// creation time as a unix timestamp. Used for listing all workflow instances in the diagnostics UI.
 func instancesByCreation() string {
 	return "instances-by-creation"
 }
 
+// instancesActive returns the key for the SET that contains all active instances. Used for reporting active workflow
+// instances in stats.
 func instancesActive() string {
 	return "instances-active"
 }

backend/redis/scripts/complete_workflow_task.lua

@@ -21,6 +21,7 @@ local pendingEventsKey = getKey()
 local payloadHashKey = getKey()
 local futureEventZSetKey = getKey()
 local activeInstancesKey = getKey()
+local instancesByCreation = getKey()
 
 local workflowSetKey = getKey()
 local workflowStreamKey = getKey()
@@ -59,6 +60,7 @@ redis.call("XDEL", pendingEventsKey, lastPendingEventMessageId)
 
 -- Update instance state
 local now = getArgv()
+local nowUnix = tonumber(getArgv())
 local state = tonumber(getArgv())
 
 -- State constants
@@ -164,6 +166,7 @@ for i = 1, otherWorkflowInstances do
 
             -- Track active instance
             redis.call("SADD", activeInstancesKey, targetInstanceSegment)
+            redis.call("ZADD", instancesByCreation, nowUnix, targetInstanceSegment)
         end
     end
 

backend/redis/scripts/create_workflow_instance.lua

@@ -19,6 +19,7 @@ local pendingEventsKey = getKey()
 local payloadHashKey = getKey()
 
 local instancesActiveKey = getKey()
+local instancesByCreation = getKey()
 
 local workflowSetKey = getKey()
 local workflowStreamKey = getKey()
@@ -50,6 +51,9 @@ redis.call("XADD", pendingEventsKey, "*", "event", eventData)
 local payload = getArgv()
 redis.pcall("HSETNX", payloadHashKey, eventId, payload)
 
+local creationTimestamp = tonumber(getArgv())
+redis.call("ZADD", instancesByCreation, creationTimestamp, instanceSegment)
+
 -- queue workflow task
 local added = redis.call("SADD", workflowSetKey, instanceSegment)
 if added == 1 then

backend/redis/workflow.go

@@ -110,6 +110,7 @@ func (rb *redisBackend) CompleteWorkflowTask(
 		payloadKey(instance),
 		futureEventsKey(),
 		instancesActive(),
+		instancesByCreation(),
 		queueKeys.SetKey,
 		queueKeys.StreamKey,
 	)
@@ -137,8 +138,17 @@ func (rb *redisBackend) CompleteWorkflowTask(
 	args = append(args, lastPendingEventMessageID)
 
 	// Update instance state and update active execution
-	nowStr := time.Now().Format(time.RFC3339)
-	args = append(args, string(nowStr), int(state), int(core.WorkflowInstanceStateContinuedAsNew), int(core.WorkflowInstanceStateFinished))
+	now := time.Now().UTC()
+	nowStr := now.Format(time.RFC3339)
+	nowUnix := now.Unix()
+	args = append(
+		args,
+		string(nowStr),
+		nowUnix,
+		int(state),
+		int(core.WorkflowInstanceStateContinuedAsNew),
+		int(core.WorkflowInstanceStateFinished),
+	)
 	keys = append(keys, activeInstanceExecutionKey(instance.InstanceID))
 
 	// Remove canceled timers