diff --git a/runs/repository/impl/action.go b/runs/repository/impl/action.go index 4abc9b4d69..664d8a755d 100644 --- a/runs/repository/impl/action.go +++ b/runs/repository/impl/action.go @@ -44,6 +44,26 @@ const ( defaultListenerMinReconnectInterval = 10 * time.Second ) +// terminalPhases are the phases an action ends in. FAILED and TIMED_OUT can +// still move on when the action is retried (see UpdateActionPhase). +var terminalPhases = []int32{ + int32(common.ActionPhase_ACTION_PHASE_SUCCEEDED), + int32(common.ActionPhase_ACTION_PHASE_FAILED), + int32(common.ActionPhase_ACTION_PHASE_ABORTED), + int32(common.ActionPhase_ACTION_PHASE_TIMED_OUT), + int32(common.ActionPhase_ACTION_PHASE_RECOVERED), +} + +// activePhases is every ActionPhase not in terminalPhases. +var activePhases = []int32{ + int32(common.ActionPhase_ACTION_PHASE_UNSPECIFIED), + int32(common.ActionPhase_ACTION_PHASE_QUEUED), + int32(common.ActionPhase_ACTION_PHASE_WAITING_FOR_RESOURCES), + int32(common.ActionPhase_ACTION_PHASE_INITIALIZING), + int32(common.ActionPhase_ACTION_PHASE_RUNNING), + int32(common.ActionPhase_ACTION_PHASE_PAUSED), +} + type notificationBuffer string // NotificationConfig configures the pending notification queues and delivery retries. @@ -556,13 +576,6 @@ func (r *actionRepo) UpdateActionPhase( int32(common.ActionPhase_ACTION_PHASE_FAILED), int32(common.ActionPhase_ACTION_PHASE_TIMED_OUT), } - terminalPhases := []int32{ - int32(common.ActionPhase_ACTION_PHASE_SUCCEEDED), - int32(common.ActionPhase_ACTION_PHASE_FAILED), - int32(common.ActionPhase_ACTION_PHASE_ABORTED), - int32(common.ActionPhase_ACTION_PHASE_TIMED_OUT), - int32(common.ActionPhase_ACTION_PHASE_RECOVERED), - } var queryBuilder strings.Builder var args []interface{} @@ -1224,6 +1237,28 @@ func (r *actionRepo) ListRootActions(ctx context.Context, project, domain string return actions, nil } +// CountActiveActionsByPhase counts the actions in an active phase, split by phase +// and by whether the action is a root (a run). +func (r *actionRepo) CountActiveActionsByPhase(ctx context.Context) (map[interfaces.PhaseKey]int64, error) { + var rows []struct { + Phase int32 `db:"phase"` + IsRoot bool `db:"is_root"` + Count int64 `db:"count"` + } + err := sqlx.SelectContext(ctx, r.db, &rows, + `SELECT phase, parent_action_name IS NULL AS is_root, COUNT(*) AS count + FROM actions WHERE phase = ANY($1) GROUP BY 1, 2`, + pq.Array(activePhases)) + if err != nil { + return nil, fmt.Errorf("failed to count actions by phase: %w", err) + } + counts := make(map[interfaces.PhaseKey]int64, len(rows)) + for _, row := range rows { + counts[interfaces.PhaseKey{Phase: row.Phase, IsRoot: row.IsRoot}] = row.Count + } + return counts, nil +} + // startNotifyLoop acquires a dedicated DB connection for NOTIFY commands and // starts processing in the background. Returns an error if the initial // connection cannot be established. diff --git a/runs/repository/impl/action_test.go b/runs/repository/impl/action_test.go index 712418a3e7..f5ab98b6db 100644 --- a/runs/repository/impl/action_test.go +++ b/runs/repository/impl/action_test.go @@ -545,6 +545,82 @@ func TestListRuns_HasPausedActionFilter(t *testing.T) { assert.False(t, runs[0].ParentActionName.Valid, "only the root action should be returned") } +func TestCountActiveActionsByPhase(t *testing.T) { + db := setupActionDB(t) + actionRepo, err := NewActionRepo(db, testDbConfig, testNotificationConfig) + require.NoError(t, err) + ctx := context.Background() + + t.Run("empty table", func(t *testing.T) { + counts, err := actionRepo.CountActiveActionsByPhase(ctx) + require.NoError(t, err) + assert.Empty(t, counts) + }) + + t.Run("counts only active actions", func(t *testing.T) { + root := func(run string, phase common.ActionPhase) *models.Action { + return &models.Action{Project: "proj1", Domain: "domain1", RunName: run, Name: rootActionName, + Phase: int32(phase)} + } + child := func(run, name string, phase common.ActionPhase) *models.Action { + return &models.Action{Project: "proj1", Domain: "domain1", RunName: run, Name: name, + ParentActionName: sql.NullString{String: rootActionName, Valid: true}, + Phase: int32(phase)} + } + for _, a := range []*models.Action{ + // r1: a live run with children in live and terminal phases. + root("r1", common.ActionPhase_ACTION_PHASE_RUNNING), + child("r1", "c1", common.ActionPhase_ACTION_PHASE_QUEUED), + child("r1", "c2", common.ActionPhase_ACTION_PHASE_WAITING_FOR_RESOURCES), + child("r1", "c3", common.ActionPhase_ACTION_PHASE_INITIALIZING), + child("r1", "c4", common.ActionPhase_ACTION_PHASE_RUNNING), + child("r1", "c5", common.ActionPhase_ACTION_PHASE_SUCCEEDED), + child("r1", "c6", common.ActionPhase_ACTION_PHASE_FAILED), + // r2: a paused run is still live. + root("r2", common.ActionPhase_ACTION_PHASE_PAUSED), + // r3-r7: one run in each terminal phase; none are counted. + root("r3", common.ActionPhase_ACTION_PHASE_SUCCEEDED), + child("r3", "c1", common.ActionPhase_ACTION_PHASE_SUCCEEDED), + root("r4", common.ActionPhase_ACTION_PHASE_FAILED), + root("r5", common.ActionPhase_ACTION_PHASE_ABORTED), + child("r5", "c1", common.ActionPhase_ACTION_PHASE_ABORTED), + root("r6", common.ActionPhase_ACTION_PHASE_TIMED_OUT), + root("r7", common.ActionPhase_ACTION_PHASE_RECOVERED), + } { + _, err := actionRepo.CreateAction(ctx, a, false) + require.NoError(t, err) + } + + counts, err := actionRepo.CountActiveActionsByPhase(ctx) + require.NoError(t, err) + key := func(phase common.ActionPhase, isRoot bool) interfaces.PhaseKey { + return interfaces.PhaseKey{Phase: int32(phase), IsRoot: isRoot} + } + assert.Equal(t, map[interfaces.PhaseKey]int64{ + // Runs (root actions). + key(common.ActionPhase_ACTION_PHASE_RUNNING, true): 1, // r1 + key(common.ActionPhase_ACTION_PHASE_PAUSED, true): 1, // r2 + + // Child actions. + key(common.ActionPhase_ACTION_PHASE_QUEUED, false): 1, // r1/c1 + key(common.ActionPhase_ACTION_PHASE_WAITING_FOR_RESOURCES, false): 1, // r1/c2 + key(common.ActionPhase_ACTION_PHASE_INITIALIZING, false): 1, // r1/c3 + key(common.ActionPhase_ACTION_PHASE_RUNNING, false): 1, // r1/c4 + }, counts) + + // The gauge totals: runs are the root rows, actions are all rows. + var runs, actions int64 + for k, n := range counts { + if k.IsRoot { + runs += n + } + actions += n + } + assert.Equal(t, int64(2), runs, "active runs: r1, r2") + assert.Equal(t, int64(6), actions, "active actions: r1, r2, r1/c1-c4") + }) +} + // TestListRuns_SearchFilter runs the console's run search predicate against a real // database: one term must match either the run name or the task name, case-insensitively, // and EQUAL on a full run name must select exactly that run. diff --git a/runs/repository/interfaces/action.go b/runs/repository/interfaces/action.go index 0672d6cb0f..814b7d8bd6 100644 --- a/runs/repository/interfaces/action.go +++ b/runs/repository/interfaces/action.go @@ -54,4 +54,14 @@ type ActionRepo interface { // Aggregation operations ListRootActions(ctx context.Context, project, domain string, startDate, endDate *time.Time, limit int) ([]*models.Action, error) + + // CountActiveActionsByPhase counts the actions in an active phase, split by + // phase and by whether the action is a root (a run). + CountActiveActionsByPhase(ctx context.Context) (counts map[PhaseKey]int64, err error) +} + +// PhaseKey groups the counts returned by ActionRepo.CountActiveActionsByPhase. +type PhaseKey struct { + Phase int32 + IsRoot bool } diff --git a/runs/repository/mocks/mocks.go b/runs/repository/mocks/mocks.go index 032b0b913b..a608eb9cfb 100644 --- a/runs/repository/mocks/mocks.go +++ b/runs/repository/mocks/mocks.go @@ -237,6 +237,68 @@ func (_c *ActionRepo_ClearAbortRequest_Call) RunAndReturn(run func(ctx context.C return _c } +// CountActiveActionsByPhase provides a mock function for the type ActionRepo +func (_mock *ActionRepo) CountActiveActionsByPhase(ctx context.Context) (map[interfaces.PhaseKey]int64, error) { + ret := _mock.Called(ctx) + + if len(ret) == 0 { + panic("no return value specified for CountActiveActionsByPhase") + } + + var r0 map[interfaces.PhaseKey]int64 + var r1 error + if returnFunc, ok := ret.Get(0).(func(context.Context) (map[interfaces.PhaseKey]int64, error)); ok { + return returnFunc(ctx) + } + if returnFunc, ok := ret.Get(0).(func(context.Context) map[interfaces.PhaseKey]int64); ok { + r0 = returnFunc(ctx) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(map[interfaces.PhaseKey]int64) + } + } + if returnFunc, ok := ret.Get(1).(func(context.Context) error); ok { + r1 = returnFunc(ctx) + } else { + r1 = ret.Error(1) + } + return r0, r1 +} + +// ActionRepo_CountActiveActionsByPhase_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'CountActiveActionsByPhase' +type ActionRepo_CountActiveActionsByPhase_Call struct { + *mock.Call +} + +// CountActiveActionsByPhase is a helper method to define mock.On call +// - ctx context.Context +func (_e *ActionRepo_Expecter) CountActiveActionsByPhase(ctx interface{}) *ActionRepo_CountActiveActionsByPhase_Call { + return &ActionRepo_CountActiveActionsByPhase_Call{Call: _e.mock.On("CountActiveActionsByPhase", ctx)} +} + +func (_c *ActionRepo_CountActiveActionsByPhase_Call) Run(run func(ctx context.Context)) *ActionRepo_CountActiveActionsByPhase_Call { + _c.Call.Run(func(args mock.Arguments) { + var arg0 context.Context + if args[0] != nil { + arg0 = args[0].(context.Context) + } + run( + arg0, + ) + }) + return _c +} + +func (_c *ActionRepo_CountActiveActionsByPhase_Call) Return(counts map[interfaces.PhaseKey]int64, err error) *ActionRepo_CountActiveActionsByPhase_Call { + _c.Call.Return(counts, err) + return _c +} + +func (_c *ActionRepo_CountActiveActionsByPhase_Call) RunAndReturn(run func(ctx context.Context) (map[interfaces.PhaseKey]int64, error)) *ActionRepo_CountActiveActionsByPhase_Call { + _c.Call.Return(run) + return _c +} + // CreateAction provides a mock function for the type ActionRepo func (_mock *ActionRepo) CreateAction(ctx context.Context, action *models.Action, updateTriggeredAt bool) (*models.Action, error) { ret := _mock.Called(ctx, action, updateTriggeredAt)