metrics_test.go

package workflow_test

import (
	"context"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/luno/jettison/errors"
	"github.com/luno/jettison/jtest"
	"github.com/prometheus/client_golang/prometheus/testutil"
	clock_testing "k8s.io/utils/clock/testing"

	"github.com/luno/workflow"
	"github.com/luno/workflow/adapters/memrecordstore"
	"github.com/luno/workflow/adapters/memrolescheduler"
	"github.com/luno/workflow/adapters/memstreamer"
	"github.com/luno/workflow/adapters/memtimeoutstore"
	"github.com/luno/workflow/internal/metrics"
)

func runWorkflow(t *testing.T) *workflow.Workflow[string, status] {
	b := workflow.NewBuilder[string, status]("example")
	b.AddStep(StatusStart,
		func(ctx context.Context, r *workflow.Record[string, status]) (status, error) {
			return 0, nil
		},
		StatusMiddle,
	).WithOptions(
		workflow.PollingFrequency(time.Millisecond * 100),
	)

	clock := newClock()
	streamer := memstreamer.New(memstreamer.WithClock(clock))
	recordStore := memrecordstore.New(memrecordstore.WithClock(clock))
	w := b.Build(
		streamer,
		recordStore,
		memtimeoutstore.New(),
		memrolescheduler.New(),
		workflow.WithClock(clock),
		workflow.WithOutboxConfig(
			workflow.WithOutboxParallelCount(2),
		),
	)

	ctx := context.Background()

	uid, err := uuid.NewUUID()
	jtest.RequireNil(t, err)

	runID := uid.String()

	var s string
	payload, err := workflow.Marshal(&s)
	jtest.RequireNil(t, err)

	err = update(ctx, recordStore, &workflow.WireRecord{
		WorkflowName: "example",
		ForeignID:    "29384723984732",
		RunID:        runID,
		Status:       int(StatusStart),
		RunState:     workflow.RunStateRunning,
		Object:       payload,
		CreatedAt:    clock.Now(),
	})
	jtest.RequireNil(t, err)

	// 1 hour = 3600 seconds
	clock.Step(time.Hour)

	w.Run(ctx)
	t.Cleanup(w.Stop)

	return w
}

func TestMetricProcessLag(t *testing.T) {
	metrics.ConsumerLag.Reset()

	runWorkflow(t)

	time.Sleep(time.Millisecond * 500)

	expected := `
# HELP workflow_process_lag_seconds lag between now and the current event timestamp in seconds
# TYPE workflow_process_lag_seconds gauge
workflow_process_lag_seconds{process_name="outbox-consumer-2-of-2",workflow_name="example"} 3600
workflow_process_lag_seconds{process_name="start-consumer-1-of-1",workflow_name="example"} 0
`

	err := testutil.CollectAndCompare(metrics.ConsumerLag, strings.NewReader(expected))
	jtest.RequireNil(t, err)

	metrics.ConsumerLag.Reset()
}

func TestMetricProcessLagAlert(t *testing.T) {
	metrics.ConsumerLagAlert.Reset()

	runWorkflow(t)

	time.Sleep(time.Millisecond * 750)

	expected := `
# HELP workflow_process_lag_alert Whether or not the consumer lag crosses its alert threshold
# TYPE workflow_process_lag_alert gauge
workflow_process_lag_alert{process_name="start-consumer-1-of-1",workflow_name="example"} 0
workflow_process_lag_alert{process_name="outbox-consumer-2-of-2",workflow_name="example"} 1
`

	err := testutil.CollectAndCompare(metrics.ConsumerLagAlert, strings.NewReader(expected))
	jtest.RequireNil(t, err)

	metrics.ConsumerLagAlert.Reset()
}

func TestMetricProcessStates(t *testing.T) {
	metrics.ProcessStates.Reset()

	w := runWorkflow(t)

	time.Sleep(time.Millisecond * 500)

	expected := `
# HELP workflow_process_states The current states of all the processes
# TYPE workflow_process_states gauge
workflow_process_states{process_name="example-delete-consumer",workflow_name="example"} 2
workflow_process_states{process_name="outbox-consumer-1-of-2",workflow_name="example"} 2
workflow_process_states{process_name="outbox-consumer-2-of-2",workflow_name="example"} 2
workflow_process_states{process_name="start-consumer-1-of-1",workflow_name="example"} 2
`

	err := testutil.CollectAndCompare(metrics.ProcessStates, strings.NewReader(expected))
	jtest.RequireNil(t, err)

	w.Stop()

	// Ensure that the metrics are updated to false when stopping the process
	expected = `
# HELP workflow_process_states The current states of all the processes
# TYPE workflow_process_states gauge
workflow_process_states{process_name="example-delete-consumer",workflow_name="example"} 1
workflow_process_states{process_name="start-consumer-1-of-1",workflow_name="example"} 1
workflow_process_states{process_name="outbox-consumer-1-of-2",workflow_name="example"} 1
workflow_process_states{process_name="outbox-consumer-2-of-2",workflow_name="example"} 1
`

	err = testutil.CollectAndCompare(metrics.ProcessStates, strings.NewReader(expected))
	jtest.RequireNil(t, err)

	metrics.ProcessStates.Reset()
}

type mockScheduler struct {
	mu    sync.Mutex
	allow bool
}

func (m *mockScheduler) Await(ctx context.Context, role string) (context.Context, context.CancelFunc, error) {
	if ctx.Err() != nil {
		return nil, nil, ctx.Err()
	}

	for {
		m.mu.Lock()
		if m.allow {
			m.mu.Unlock()
			break
		}
		m.mu.Unlock()

		if ctx.Err() != nil {
			return nil, nil, ctx.Err()
		}

		time.Sleep(time.Millisecond * 10)
	}

	ctx, cancel := context.WithCancel(ctx)
	return ctx, cancel, nil
}

var _ workflow.RoleScheduler = (*mockScheduler)(nil)

func TestMetricProcessIdleState(t *testing.T) {
	metrics.ProcessStates.Reset()

	b := workflow.NewBuilder[string, status]("example")
	b.AddStep(StatusStart,
		func(ctx context.Context, r *workflow.Record[string, status]) (status, error) {
			return 0, nil
		}, StatusMiddle,
	).WithOptions(
		workflow.PollingFrequency(time.Millisecond * 100),
	)
	b.AddStep(StatusMiddle,
		func(ctx context.Context, r *workflow.Record[string, status]) (status, error) {
			return 0, nil
		}, StatusEnd,
	).WithOptions(
		workflow.PollingFrequency(time.Millisecond * 100),
	)

	clock := newClock()
	streamer := memstreamer.New(memstreamer.WithClock(clock))
	recordStore := memrecordstore.New()
	scheduler := &mockScheduler{}
	wf := b.Build(
		streamer,
		recordStore,
		memtimeoutstore.New(),
		scheduler,
		workflow.WithClock(clock),
	)

	ctx := context.Background()

	wf.Run(ctx)
	t.Cleanup(wf.Stop)

	time.Sleep(time.Millisecond * 250)

	// Ensure that the metrics are updated to idle before obtaining the role
	expected := `
# HELP workflow_process_states The current states of all the processes
# TYPE workflow_process_states gauge
workflow_process_states{process_name="example-delete-consumer",workflow_name="example"} 3
workflow_process_states{process_name="start-consumer-1-of-1", workflow_name="example"} 3
workflow_process_states{process_name="middle-consumer-1-of-1", workflow_name="example"} 3
workflow_process_states{process_name="outbox-consumer-1-of-1",workflow_name="example"} 3
`

	err := testutil.CollectAndCompare(metrics.ProcessStates, strings.NewReader(expected))
	jtest.RequireNil(t, err)

	scheduler.mu.Lock()
	scheduler.allow = true
	scheduler.mu.Unlock()

	time.Sleep(time.Millisecond * 50)

	// Ensure that the metrics are updated to running obtaining the role
	expected = `
# HELP workflow_process_states The current states of all the processes
# TYPE workflow_process_states gauge
workflow_process_states{process_name="example-delete-consumer",workflow_name="example"} 2
workflow_process_states{process_name="middle-consumer-1-of-1",workflow_name="example"} 2
workflow_process_states{process_name="start-consumer-1-of-1",workflow_name="example"} 2
workflow_process_states{process_name="outbox-consumer-1-of-1",workflow_name="example"} 2
`

	err = testutil.CollectAndCompare(metrics.ProcessStates, strings.NewReader(expected))
	jtest.RequireNil(t, err)

	wf.Stop()

	// Ensure that the metrics are updated to shut down when processes are shutdown
	expected = `
# HELP workflow_process_states The current states of all the processes
# TYPE workflow_process_states gauge
workflow_process_states{process_name="example-delete-consumer",workflow_name="example"} 1
workflow_process_states{process_name="middle-consumer-1-of-1",workflow_name="example"} 1
workflow_process_states{process_name="start-consumer-1-of-1",workflow_name="example"} 1
workflow_process_states{process_name="outbox-consumer-1-of-1",workflow_name="example"} 1
`

	err = testutil.CollectAndCompare(metrics.ProcessStates, strings.NewReader(expected))
	jtest.RequireNil(t, err)

	metrics.ProcessStates.Reset()
}

func TestMetricProcessLatency(t *testing.T) {
	metrics.ProcessLatency.Reset()

	runWorkflow(t)

	time.Sleep(time.Millisecond * 500)

	expected := `
# HELP workflow_process_latency_seconds Event loop latency in seconds
# TYPE workflow_process_latency_seconds histogram
workflow_process_latency_seconds_bucket{process_name="outbox-consumer-2-of-2",workflow_name="example",le="0.01"} 1
workflow_process_latency_seconds_bucket{process_name="outbox-consumer-2-of-2",workflow_name="example",le="0.1"} 1
workflow_process_latency_seconds_bucket{process_name="outbox-consumer-2-of-2",workflow_name="example",le="1"} 1
workflow_process_latency_seconds_bucket{process_name="outbox-consumer-2-of-2",workflow_name="example",le="5"} 1
workflow_process_latency_seconds_bucket{process_name="outbox-consumer-2-of-2",workflow_name="example",le="10"} 1
workflow_process_latency_seconds_bucket{process_name="outbox-consumer-2-of-2",workflow_name="example",le="60"} 1
workflow_process_latency_seconds_bucket{process_name="outbox-consumer-2-of-2",workflow_name="example",le="300"} 1
workflow_process_latency_seconds_bucket{process_name="outbox-consumer-2-of-2",workflow_name="example",le="+Inf"} 1
workflow_process_latency_seconds_sum{process_name="outbox-consumer-2-of-2",workflow_name="example"} 0
workflow_process_latency_seconds_count{process_name="outbox-consumer-2-of-2",workflow_name="example"} 1
workflow_process_latency_seconds_bucket{process_name="start-consumer-1-of-1",workflow_name="example",le="0.01"} 1
workflow_process_latency_seconds_bucket{process_name="start-consumer-1-of-1",workflow_name="example",le="0.1"} 1
workflow_process_latency_seconds_bucket{process_name="start-consumer-1-of-1",workflow_name="example",le="1"} 1
workflow_process_latency_seconds_bucket{process_name="start-consumer-1-of-1",workflow_name="example",le="5"} 1
workflow_process_latency_seconds_bucket{process_name="start-consumer-1-of-1",workflow_name="example",le="10"} 1
workflow_process_latency_seconds_bucket{process_name="start-consumer-1-of-1",workflow_name="example",le="60"} 1
workflow_process_latency_seconds_bucket{process_name="start-consumer-1-of-1",workflow_name="example",le="300"} 1
workflow_process_latency_seconds_bucket{process_name="start-consumer-1-of-1",workflow_name="example",le="+Inf"} 1
workflow_process_latency_seconds_sum{process_name="start-consumer-1-of-1",workflow_name="example"} 0
workflow_process_latency_seconds_count{process_name="start-consumer-1-of-1",workflow_name="example"} 1
`

	err := testutil.CollectAndCompare(metrics.ProcessLatency, strings.NewReader(expected))
	jtest.RequireNil(t, err)

	metrics.ProcessLatency.Reset()
}

func TestMetricProcessErrors(t *testing.T) {
	metrics.ProcessErrors.Reset()

	b := workflow.NewBuilder[string, status]("example")
	b.AddStep(StatusStart,
		func(ctx context.Context, r *workflow.Record[string, status]) (status, error) {
			return 0, errors.New("mock error")
		}, StatusMiddle,
	).WithOptions(
		workflow.PollingFrequency(time.Millisecond * 100),
	)

	clock := newClock()
	streamer := memstreamer.New(memstreamer.WithClock(clock))
	recordStore := memrecordstore.New()
	wf := b.Build(
		streamer,
		recordStore,
		memtimeoutstore.New(),
		memrolescheduler.New(),
		workflow.WithClock(clock),
	)

	ctx := context.Background()

	uid, err := uuid.NewUUID()
	jtest.RequireNil(t, err)

	runID := uid.String()

	var s string
	payload, err := workflow.Marshal(&s)
	jtest.RequireNil(t, err)

	err = update(ctx, recordStore, &workflow.WireRecord{
		WorkflowName: "example",
		ForeignID:    "29384723984732",
		RunID:        runID,
		Status:       int(StatusStart),
		RunState:     workflow.RunStateRunning,
		Object:       payload,
		CreatedAt:    clock.Now(),
	})
	jtest.RequireNil(t, err)

	wf.Run(ctx)
	t.Cleanup(wf.Stop)

	time.Sleep(time.Millisecond * 100)

	expected := `
# HELP workflow_process_error_count Number of errors processing events
# TYPE workflow_process_error_count counter
workflow_process_error_count{process_name="start-consumer-1-of-1",workflow_name="example"} 1
`

	err = testutil.CollectAndCompare(metrics.ProcessErrors, strings.NewReader(expected))
	jtest.RequireNil(t, err)

	metrics.ProcessErrors.Reset()
}

func TestRunStateChanges(t *testing.T) {
	metrics.RunStateChanges.Reset()

	b := workflow.NewBuilder[string, status]("example")
	b.AddStep(StatusStart,
		func(ctx context.Context, r *workflow.Record[string, status]) (status, error) {
			return StatusMiddle, nil
		}, StatusMiddle,
	).WithOptions(
		workflow.PollingFrequency(time.Millisecond * 10),
	)
	b.AddStep(StatusMiddle,
		func(ctx context.Context, r *workflow.Record[string, status]) (status, error) {
			return StatusEnd, nil
		}, StatusEnd,
	).WithOptions(
		workflow.PollingFrequency(time.Millisecond * 10),
	)

	w := b.Build(
		memstreamer.New(),
		memrecordstore.New(),
		memtimeoutstore.New(),
		memrolescheduler.New(),
		workflow.WithOutboxConfig(workflow.WithOutboxPollingFrequency(time.Millisecond)),
	)

	ctx := context.Background()
	w.Run(ctx)
	t.Cleanup(w.Stop)

	_, err := w.Trigger(ctx, "983467934", StatusStart)
	jtest.RequireNil(t, err)

	time.Sleep(time.Millisecond * 500)

	expected := `
# HELP workflow_run_state_changes The number of workflow run state changes going from state to a new state
# TYPE workflow_run_state_changes counter
workflow_run_state_changes{current_run_state="Initiated",previous_run_state="Unknown",workflow_name="example"} 1
workflow_run_state_changes{current_run_state="Running",previous_run_state="Running",workflow_name="example"} 1
workflow_run_state_changes{current_run_state="Completed",previous_run_state="Running",workflow_name="example"} 1
`

	err = testutil.CollectAndCompare(metrics.RunStateChanges, strings.NewReader(expected))
	jtest.RequireNil(t, err)

	metrics.RunStateChanges.Reset()
}

func TestMetricProcessSkippedEvents(t *testing.T) {
	metrics.ProcessSkippedEvents.Reset()

	b := workflow.NewBuilder[string, status]("example")
	b.AddStep(StatusStart,
		func(ctx context.Context, r *workflow.Record[string, status]) (status, error) {
			return 0, nil
		}, StatusMiddle,
	).WithOptions(
		workflow.PollingFrequency(time.Millisecond * 10),
	)

	w := b.Build(
		memstreamer.New(),
		memrecordstore.New(),
		memtimeoutstore.New(),
		memrolescheduler.New(),
		workflow.WithOutboxConfig(workflow.WithOutboxPollingFrequency(time.Millisecond)),
	)

	ctx := context.Background()
	w.Run(ctx)
	t.Cleanup(w.Stop)

	_, err := w.Trigger(ctx, "9834679343", StatusStart)
	jtest.RequireNil(t, err)

	_, err = w.Trigger(ctx, "2349839483", StatusStart)
	jtest.RequireNil(t, err)

	_, err = w.Trigger(ctx, "7548702398", StatusStart)
	jtest.RequireNil(t, err)

	time.Sleep(time.Millisecond * 500)

	expected := `
# HELP workflow_process_skipped_events_count Number of events skipped by consumer
# TYPE workflow_process_skipped_events_count counter
workflow_process_skipped_events_count{process_name="start-consumer-1-of-1",reason="next value specified skip",workflow_name="example"} 3
`

	err = testutil.CollectAndCompare(metrics.ProcessSkippedEvents, strings.NewReader(expected))
	jtest.RequireNil(t, err)

	metrics.ProcessSkippedEvents.Reset()
}

func update(ctx context.Context, store workflow.RecordStore, wr *workflow.WireRecord) error {
	return store.Store(ctx, wr, func(recordID int64) (workflow.OutboxEventData, error) {
		// Record ID would not have been set if it is a new record. Assign the recordID that the Store provides
		wr.ID = recordID
		return workflow.WireRecordToOutboxEventData(*wr, workflow.RunStateUnknown)
	})
}

func newClock() *clock_testing.FakeClock {
	nw := time.Now()
	now := time.Date(nw.Year(), nw.Month(), nw.Day(), nw.Hour(), 0, 0, 0, time.UTC)
	return clock_testing.NewFakeClock(now)
}