From 5e45662475509ac3fbade0732d707924371c18a7 Mon Sep 17 00:00:00 2001 From: Benoit Sigoure Date: Thu, 30 Jul 2026 00:25:07 +0000 Subject: [PATCH] webapp: add per scale set job queue view When all runners of a scale set are busy, it is hard to tell where a given workflow job sits in the queue, or why capacity is not being used. This adds a "Job Queue" view to the web UI that groups queued and running jobs per scale set (and per pool, matched by labels), ordered by request time, with links to the GitHub run/job pages and to the scale set/pool detail pages. To support the view: * Jobs recorded by scale set listeners now carry the garm scale set ID (new WorkflowJob.ScaleSetFkID column, exposed as scale_set_id on the Job API model). Previously the scale set ID was dropped when recording jobs. * The full RunnerScaleSetStatistic from each session message is persisted on the scale set and exposed as "statistics" on the API (previously only TotalAssignedJobs survived as desired_runner_count). The view shows GitHub's numbers (assigned jobs, busy/idle runners) next to GARM's instance counts, making divergence visible. * consolidateRunnerState syncs GitHub's per-runner view (online idle/busy, offline) onto instances that finished installing. Without this, a runner whose agent died after setup stayed "idle" in GARM forever while GitHub considered it offline and never assigned it jobs. The view updates live via the existing job/instance websocket events. Co-Authored-By: Claude Fable 5 --- database/common/mocks/Store.go | 64 +-- database/common/store.go | 2 +- database/sql/jobs.go | 14 + database/sql/models.go | 9 + database/sql/scalesets.go | 10 +- database/sql/scalesets_test.go | 2 +- database/sql/util.go | 8 + database/watcher/watcher_store_test.go | 5 +- params/params.go | 8 +- webapp/src/lib/api/generated-client.ts | 11 + webapp/src/lib/api/generated/api.ts | 26 ++ webapp/src/lib/components/Navigation.svelte | 5 + webapp/src/routes/queue/+page.svelte | 408 ++++++++++++++++++ .../src/routes/queue/page.integration.test.ts | 295 +++++++++++++ webapp/swagger.yaml | 41 ++ workers/scaleset/interfaces.go | 2 +- workers/scaleset/scaleset.go | 36 +- workers/scaleset/scaleset_helper.go | 7 +- workers/scaleset/scaleset_listener.go | 6 +- 19 files changed, 913 insertions(+), 46 deletions(-) create mode 100644 webapp/src/routes/queue/+page.svelte create mode 100644 webapp/src/routes/queue/page.integration.test.ts diff --git a/database/common/mocks/Store.go b/database/common/mocks/Store.go index 9d91d5e5a..b4c9e8924 100644 --- a/database/common/mocks/Store.go +++ b/database/common/mocks/Store.go @@ -5572,17 +5572,17 @@ func (_c *Store_SetEntityPoolManagerStatus_Call) RunAndReturn(run func(context.C return _c } -// SetScaleSetDesiredRunnerCount provides a mock function with given fields: ctx, scaleSetID, desiredRunnerCount -func (_m *Store) SetScaleSetDesiredRunnerCount(ctx context.Context, scaleSetID uint, desiredRunnerCount int) error { - ret := _m.Called(ctx, scaleSetID, desiredRunnerCount) +// SetScaleSetLastMessageID provides a mock function with given fields: ctx, scaleSetID, lastMessageID +func (_m *Store) SetScaleSetLastMessageID(ctx context.Context, scaleSetID uint, lastMessageID int64) error { + ret := _m.Called(ctx, scaleSetID, lastMessageID) if len(ret) == 0 { - panic("no return value specified for SetScaleSetDesiredRunnerCount") + panic("no return value specified for SetScaleSetLastMessageID") } var r0 error - if rf, ok := ret.Get(0).(func(context.Context, uint, int) error); ok { - r0 = rf(ctx, scaleSetID, desiredRunnerCount) + if rf, ok := ret.Get(0).(func(context.Context, uint, int64) error); ok { + r0 = rf(ctx, scaleSetID, lastMessageID) } else { r0 = ret.Error(0) } @@ -5590,47 +5590,47 @@ func (_m *Store) SetScaleSetDesiredRunnerCount(ctx context.Context, scaleSetID u return r0 } -// Store_SetScaleSetDesiredRunnerCount_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetScaleSetDesiredRunnerCount' -type Store_SetScaleSetDesiredRunnerCount_Call struct { +// Store_SetScaleSetLastMessageID_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetScaleSetLastMessageID' +type Store_SetScaleSetLastMessageID_Call struct { *mock.Call } -// SetScaleSetDesiredRunnerCount is a helper method to define mock.On call +// SetScaleSetLastMessageID is a helper method to define mock.On call // - ctx context.Context // - scaleSetID uint -// - desiredRunnerCount int -func (_e *Store_Expecter) SetScaleSetDesiredRunnerCount(ctx interface{}, scaleSetID interface{}, desiredRunnerCount interface{}) *Store_SetScaleSetDesiredRunnerCount_Call { - return &Store_SetScaleSetDesiredRunnerCount_Call{Call: _e.mock.On("SetScaleSetDesiredRunnerCount", ctx, scaleSetID, desiredRunnerCount)} +// - lastMessageID int64 +func (_e *Store_Expecter) SetScaleSetLastMessageID(ctx interface{}, scaleSetID interface{}, lastMessageID interface{}) *Store_SetScaleSetLastMessageID_Call { + return &Store_SetScaleSetLastMessageID_Call{Call: _e.mock.On("SetScaleSetLastMessageID", ctx, scaleSetID, lastMessageID)} } -func (_c *Store_SetScaleSetDesiredRunnerCount_Call) Run(run func(ctx context.Context, scaleSetID uint, desiredRunnerCount int)) *Store_SetScaleSetDesiredRunnerCount_Call { +func (_c *Store_SetScaleSetLastMessageID_Call) Run(run func(ctx context.Context, scaleSetID uint, lastMessageID int64)) *Store_SetScaleSetLastMessageID_Call { _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].(uint), args[2].(int)) + run(args[0].(context.Context), args[1].(uint), args[2].(int64)) }) return _c } -func (_c *Store_SetScaleSetDesiredRunnerCount_Call) Return(_a0 error) *Store_SetScaleSetDesiredRunnerCount_Call { +func (_c *Store_SetScaleSetLastMessageID_Call) Return(_a0 error) *Store_SetScaleSetLastMessageID_Call { _c.Call.Return(_a0) return _c } -func (_c *Store_SetScaleSetDesiredRunnerCount_Call) RunAndReturn(run func(context.Context, uint, int) error) *Store_SetScaleSetDesiredRunnerCount_Call { +func (_c *Store_SetScaleSetLastMessageID_Call) RunAndReturn(run func(context.Context, uint, int64) error) *Store_SetScaleSetLastMessageID_Call { _c.Call.Return(run) return _c } -// SetScaleSetLastMessageID provides a mock function with given fields: ctx, scaleSetID, lastMessageID -func (_m *Store) SetScaleSetLastMessageID(ctx context.Context, scaleSetID uint, lastMessageID int64) error { - ret := _m.Called(ctx, scaleSetID, lastMessageID) +// SetScaleSetRunnerStatistics provides a mock function with given fields: ctx, scaleSetID, stats +func (_m *Store) SetScaleSetRunnerStatistics(ctx context.Context, scaleSetID uint, stats params.RunnerScaleSetStatistic) error { + ret := _m.Called(ctx, scaleSetID, stats) if len(ret) == 0 { - panic("no return value specified for SetScaleSetLastMessageID") + panic("no return value specified for SetScaleSetRunnerStatistics") } var r0 error - if rf, ok := ret.Get(0).(func(context.Context, uint, int64) error); ok { - r0 = rf(ctx, scaleSetID, lastMessageID) + if rf, ok := ret.Get(0).(func(context.Context, uint, params.RunnerScaleSetStatistic) error); ok { + r0 = rf(ctx, scaleSetID, stats) } else { r0 = ret.Error(0) } @@ -5638,32 +5638,32 @@ func (_m *Store) SetScaleSetLastMessageID(ctx context.Context, scaleSetID uint, return r0 } -// Store_SetScaleSetLastMessageID_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetScaleSetLastMessageID' -type Store_SetScaleSetLastMessageID_Call struct { +// Store_SetScaleSetRunnerStatistics_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetScaleSetRunnerStatistics' +type Store_SetScaleSetRunnerStatistics_Call struct { *mock.Call } -// SetScaleSetLastMessageID is a helper method to define mock.On call +// SetScaleSetRunnerStatistics is a helper method to define mock.On call // - ctx context.Context // - scaleSetID uint -// - lastMessageID int64 -func (_e *Store_Expecter) SetScaleSetLastMessageID(ctx interface{}, scaleSetID interface{}, lastMessageID interface{}) *Store_SetScaleSetLastMessageID_Call { - return &Store_SetScaleSetLastMessageID_Call{Call: _e.mock.On("SetScaleSetLastMessageID", ctx, scaleSetID, lastMessageID)} +// - stats params.RunnerScaleSetStatistic +func (_e *Store_Expecter) SetScaleSetRunnerStatistics(ctx interface{}, scaleSetID interface{}, stats interface{}) *Store_SetScaleSetRunnerStatistics_Call { + return &Store_SetScaleSetRunnerStatistics_Call{Call: _e.mock.On("SetScaleSetRunnerStatistics", ctx, scaleSetID, stats)} } -func (_c *Store_SetScaleSetLastMessageID_Call) Run(run func(ctx context.Context, scaleSetID uint, lastMessageID int64)) *Store_SetScaleSetLastMessageID_Call { +func (_c *Store_SetScaleSetRunnerStatistics_Call) Run(run func(ctx context.Context, scaleSetID uint, stats params.RunnerScaleSetStatistic)) *Store_SetScaleSetRunnerStatistics_Call { _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].(uint), args[2].(int64)) + run(args[0].(context.Context), args[1].(uint), args[2].(params.RunnerScaleSetStatistic)) }) return _c } -func (_c *Store_SetScaleSetLastMessageID_Call) Return(_a0 error) *Store_SetScaleSetLastMessageID_Call { +func (_c *Store_SetScaleSetRunnerStatistics_Call) Return(_a0 error) *Store_SetScaleSetRunnerStatistics_Call { _c.Call.Return(_a0) return _c } -func (_c *Store_SetScaleSetLastMessageID_Call) RunAndReturn(run func(context.Context, uint, int64) error) *Store_SetScaleSetLastMessageID_Call { +func (_c *Store_SetScaleSetRunnerStatistics_Call) RunAndReturn(run func(context.Context, uint, params.RunnerScaleSetStatistic) error) *Store_SetScaleSetRunnerStatistics_Call { _c.Call.Return(run) return _c } diff --git a/database/common/store.go b/database/common/store.go index 49e17873a..6d978e6b3 100644 --- a/database/common/store.go +++ b/database/common/store.go @@ -166,7 +166,7 @@ type ScaleSetsStore interface { GetScaleSetByID(ctx context.Context, scaleSet uint) (params.ScaleSet, error) DeleteScaleSetByID(ctx context.Context, scaleSetID uint) (err error) SetScaleSetLastMessageID(ctx context.Context, scaleSetID uint, lastMessageID int64) error - SetScaleSetDesiredRunnerCount(ctx context.Context, scaleSetID uint, desiredRunnerCount int) error + SetScaleSetRunnerStatistics(ctx context.Context, scaleSetID uint, stats params.RunnerScaleSetStatistic) error } type ScaleSetInstanceStore interface { diff --git a/database/sql/jobs.go b/database/sql/jobs.go index ac1709427..5bbe50320 100644 --- a/database/sql/jobs.go +++ b/database/sql/jobs.go @@ -72,6 +72,10 @@ func sqlWorkflowJobToParamsJob(job WorkflowJob) (params.Job, error) { jobParam.RunnerName = job.Instance.Name } + if job.ScaleSetFkID != nil { + jobParam.ScaleSetID = *job.ScaleSetFkID + } + return jobParam, nil } @@ -104,6 +108,11 @@ func (s *sqlDatabase) paramsJobToWorkflowJob(ctx context.Context, conn *gorm.DB, LockedBy: job.LockedBy, } + if job.ScaleSetID != 0 { + scaleSetID := job.ScaleSetID + workflofJob.ScaleSetFkID = &scaleSetID + } + if job.RunnerName != "" { instance, err := s.getInstance(s.ctx, conn, job.RunnerName) if err != nil { @@ -360,6 +369,11 @@ func (s *sqlDatabase) CreateOrUpdateJob(ctx context.Context, job params.Job) (pa if job.ForgeInstanceID != nil { workflowJob.ForgeInstanceID = job.ForgeInstanceID } + + if job.ScaleSetID != 0 { + scaleSetID := job.ScaleSetID + workflowJob.ScaleSetFkID = &scaleSetID + } if err := tx.Save(&workflowJob).Error; err != nil { return fmt.Errorf("error saving job: %w", err) } diff --git a/database/sql/models.go b/database/sql/models.go index 2dec823e5..be00b868e 100644 --- a/database/sql/models.go +++ b/database/sql/models.go @@ -219,6 +219,10 @@ type ScaleSet struct { Enabled bool LastMessageID int64 DesiredRunnerCount int + // RunnerStatistics is the last RunnerScaleSetStatistic received from + // GitHub on the message session (busy/idle/assigned counts as GitHub + // sees them). + RunnerStatistics datatypes.JSON // ExtraSpecs is an opaque json that gets sent to the provider // as part of the bootstrap params for instances. It can contain // any kind of data needed by providers. @@ -474,6 +478,11 @@ type WorkflowJob struct { // ScaleSetJobID is the job ID for a scaleset job. ScaleSetJobID string `gorm:"index:scaleset_job_id_idx"` + // ScaleSetFkID is the ID of the scale set that this job was assigned to, + // if the job came in through a scale set listener. + ScaleSetFkID *uint `gorm:"index"` + ScaleSet ScaleSet `gorm:"foreignKey:ScaleSetFkID"` + // RunID is the ID of the workflow run. A run may have multiple jobs. RunID int64 // Action is the specific activity that triggered the event. diff --git a/database/sql/scalesets.go b/database/sql/scalesets.go index 6e597c602..275aeb809 100644 --- a/database/sql/scalesets.go +++ b/database/sql/scalesets.go @@ -16,6 +16,7 @@ package sql import ( "context" + "encoding/json" "errors" "fmt" @@ -525,7 +526,7 @@ func (s *sqlDatabase) SetScaleSetLastMessageID(_ context.Context, scaleSetID uin return nil } -func (s *sqlDatabase) SetScaleSetDesiredRunnerCount(_ context.Context, scaleSetID uint, desiredRunnerCount int) (err error) { +func (s *sqlDatabase) SetScaleSetRunnerStatistics(_ context.Context, scaleSetID uint, stats params.RunnerScaleSetStatistic) (err error) { var scaleSet params.ScaleSet var rowsAffected int64 defer func() { @@ -538,8 +539,13 @@ func (s *sqlDatabase) SetScaleSetDesiredRunnerCount(_ context.Context, scaleSetI if err != nil { return fmt.Errorf("error fetching scale set: %w", err) } + asJSON, err := json.Marshal(stats) + if err != nil { + return fmt.Errorf("error marshaling runner statistics: %w", err) + } result := tx.Model(&dbSet).Updates(map[string]interface{}{ - "desired_runner_count": desiredRunnerCount, + "desired_runner_count": stats.TotalAssignedJobs, + "runner_statistics": datatypes.JSON(asJSON), }) if result.Error != nil { return fmt.Errorf("error saving database entry: %w", result.Error) diff --git a/database/sql/scalesets_test.go b/database/sql/scalesets_test.go index 7b5a46bbc..df3b3bf0a 100644 --- a/database/sql/scalesets_test.go +++ b/database/sql/scalesets_test.go @@ -257,7 +257,7 @@ func (s *ScaleSetsTestSuite) TestScaleSetOperations() { s.T().Run("Set scale set last message ID and desired count", func(_ *testing.T) { err = s.Store.SetScaleSetLastMessageID(s.adminCtx, orgScaleSet.ID, 20) s.Require().NoError(err) - err = s.Store.SetScaleSetDesiredRunnerCount(s.adminCtx, orgScaleSet.ID, 5) + err = s.Store.SetScaleSetRunnerStatistics(s.adminCtx, orgScaleSet.ID, params.RunnerScaleSetStatistic{TotalAssignedJobs: 5}) s.Require().NoError(err) orgScaleSetByID, err := s.Store.GetScaleSetByID(s.adminCtx, orgScaleSet.ID) s.Require().NoError(err) diff --git a/database/sql/util.go b/database/sql/util.go index 0432519c8..9bd1a5169 100644 --- a/database/sql/util.go +++ b/database/sql/util.go @@ -489,6 +489,14 @@ func (s *sqlDatabase) sqlToCommonScaleSet(scaleSet ScaleSet) (params.ScaleSet, e ret.ProxyName = scaleSet.Proxy.Name } + if len(scaleSet.RunnerStatistics) > 0 { + var stats params.RunnerScaleSetStatistic + if err := json.Unmarshal(scaleSet.RunnerStatistics, &stats); err != nil { + return params.ScaleSet{}, fmt.Errorf("error unmarshaling runner statistics: %w", err) + } + ret.Statistics = &stats + } + var ep GithubEndpoint if scaleSet.RepoID != nil { ret.RepoID = scaleSet.RepoID.String() diff --git a/database/watcher/watcher_store_test.go b/database/watcher/watcher_store_test.go index 9e3ab750b..2a4388ce8 100644 --- a/database/watcher/watcher_store_test.go +++ b/database/watcher/watcher_store_test.go @@ -582,7 +582,7 @@ func (s *WatcherStoreTestSuite) TestScaleSetWatcher() { s.T().Fatal("expected payload not received") } - err = s.store.SetScaleSetDesiredRunnerCount(s.ctx, updatedScaleSet.ID, 5) + err = s.store.SetScaleSetRunnerStatistics(s.ctx, updatedScaleSet.ID, params.RunnerScaleSetStatistic{TotalAssignedJobs: 5}) s.Require().NoError(err) select { @@ -600,8 +600,9 @@ func (s *WatcherStoreTestSuite) TestScaleSetWatcher() { select { case event := <-consumer.Watch(): - // We updated last message ID and desired runner count above. + // We updated last message ID and runner statistics above. updatedScaleSet.DesiredRunnerCount = 5 + updatedScaleSet.Statistics = ¶ms.RunnerScaleSetStatistic{TotalAssignedJobs: 5} updatedScaleSet.LastMessageID = 99 payloadFromEvent, ok := event.Payload.(params.ScaleSet) s.Require().True(ok) diff --git a/params/params.go b/params/params.go index b7a80aa3e..dc8086031 100644 --- a/params/params.go +++ b/params/params.go @@ -719,7 +719,10 @@ type ScaleSet struct { Enabled bool `json:"enabled,omitempty"` Instances []Instance `json:"instances,omitempty"` DesiredRunnerCount int `json:"desired_runner_count,omitempty"` - EnableShell bool `json:"enable_shell"` + // Statistics is the last runner scale set statistic received from GitHub + // on the message session (busy/idle/assigned counts as GitHub sees them). + Statistics *RunnerScaleSetStatistic `json:"statistics,omitempty"` + EnableShell bool `json:"enable_shell"` // Generation holds the numeric generation of the scaleset. This number // will be incremented, every time certain settings of the scaleset, which @@ -1418,6 +1421,9 @@ type Job struct { WorkflowJobID int64 `json:"workflow_job_id,omitempty"` // ScaleSetJobID is the job ID when generated for a scale set. ScaleSetJobID string `json:"scaleset_job_id,omitempty"` + // ScaleSetID is the garm ID of the scale set this job was assigned to, + // if it came in through a scale set listener. + ScaleSetID uint `json:"scale_set_id,omitempty"` // RunID is the ID of the workflow run. A run may have multiple jobs. RunID int64 `json:"run_id,omitempty"` // Action is the specific activity that triggered the event. diff --git a/webapp/src/lib/api/generated-client.ts b/webapp/src/lib/api/generated-client.ts index 3c050843e..b6e5dfef1 100644 --- a/webapp/src/lib/api/generated-client.ts +++ b/webapp/src/lib/api/generated-client.ts @@ -21,7 +21,9 @@ import { ProxiesApi, ObjectsApi, ToolsApi, + JobsApi, type GARMAgentRelease, + type Job, type Repository, type Organization, type Enterprise, @@ -152,6 +154,7 @@ export class GeneratedGarmApiClient { private proxiesApi: ProxiesApi; private objectsApi: ObjectsApi; private toolsApi: ToolsApi; + private jobsApi: JobsApi; constructor(baseUrl: string = '') { this.baseUrl = baseUrl || window.location.origin; @@ -188,6 +191,7 @@ export class GeneratedGarmApiClient { this.proxiesApi = new ProxiesApi(this.config); this.objectsApi = new ObjectsApi(this.config); this.toolsApi = new ToolsApi(this.config); + this.jobsApi = new JobsApi(this.config); } // Set authentication token @@ -222,6 +226,7 @@ export class GeneratedGarmApiClient { this.providersApi = new ProvidersApi(this.config); this.firstRunApi = new FirstRunApi(this.config); this.hooksApi = new HooksApi(this.config); + this.jobsApi = new JobsApi(this.config); } // Authentication @@ -644,6 +649,12 @@ export class GeneratedGarmApiClient { await this.scaleSetsApi.deleteScaleSet(id.toString()); } + // Jobs + async listJobs(): Promise { + const response = await this.jobsApi.listJobs(); + return response.data || []; + } + // Instances async listInstances(): Promise { const response = await this.instancesApi.listInstances(); diff --git a/webapp/src/lib/api/generated/api.ts b/webapp/src/lib/api/generated/api.ts index 5f7760990..200da388a 100644 --- a/webapp/src/lib/api/generated/api.ts +++ b/webapp/src/lib/api/generated/api.ts @@ -2325,6 +2325,12 @@ export interface Job { * @memberof Job */ 'runner_name'?: string; + /** + * ScaleSetID is the garm ID of the scale set this job was assigned to, if it came in through a scale set listener. + * @type {number} + * @memberof Job + */ + 'scale_set_id'?: number; /** * ScaleSetJobID is the job ID when generated for a scale set. * @type {string} @@ -3003,6 +3009,20 @@ export interface RunnerPrefix { * @export * @interface ScaleSet */ +/** + * RunnerScaleSetStatistic is the last runner scale set statistic received from GitHub on the message session. + * @export + * @interface RunnerScaleSetStatistic + */ +export interface RunnerScaleSetStatistic { + 'totalAvailableJobs'?: number; + 'totalAcquiredJobs'?: number; + 'totalAssignedJobs'?: number; + 'totalRunningJobs'?: number; + 'totalRegisteredRunners'?: number; + 'totalBusyRunners'?: number; + 'totalIdleRunners'?: number; +} export interface ScaleSet { /** * @@ -3160,6 +3180,12 @@ export interface ScaleSet { * @memberof ScaleSet */ 'proxy_name'?: string; + /** + * Statistics is the last runner scale set statistic received from GitHub on the message session. + * @type {RunnerScaleSetStatistic} + * @memberof ScaleSet + */ + 'statistics'?: RunnerScaleSetStatistic; /** * * @type {string} diff --git a/webapp/src/lib/components/Navigation.svelte b/webapp/src/lib/components/Navigation.svelte index 61cd5496c..ac27400b6 100644 --- a/webapp/src/lib/components/Navigation.svelte +++ b/webapp/src/lib/components/Navigation.svelte @@ -77,6 +77,11 @@ label: 'Scale Sets', icon: 'M4 7v10c0 2.21 3.582 4 8 4s8-1.79 8-4V7M4 7c0 2.21 3.582 4 8 4s8-1.79 8-4M4 7c0-2.21 3.582-4 8-4s8 1.79 8 4m0 5c0 2.21-3.582 4-8 4s-8-1.79-8-4' // Database layers icon }, + { + href: resolve('/queue'), + label: 'Job Queue', + icon: 'M4 6h16M4 10h16M4 14h10M4 18h6' // Queue/list icon + }, { href: resolve('/instances'), label: 'Runners', diff --git a/webapp/src/routes/queue/+page.svelte b/webapp/src/routes/queue/+page.svelte new file mode 100644 index 000000000..91405f639 --- /dev/null +++ b/webapp/src/routes/queue/+page.svelte @@ -0,0 +1,408 @@ + + + + Job Queue - GARM + + +
+ + + {#if error} +
+
+
+

Error

+
{error}
+ +
+
+
+ {/if} + +
+ + +
+ + {#if loading} +
+
+
+ {:else if queueGroups.length === 0} +
+

No queued or running jobs.

+
+ {:else} + {#each queueGroups as group (group.key)} +
+
+
+

+ {#if group.href} + {group.title} + {:else} + {group.title} + {/if} +

+

{group.subtitle}

+
+
+ + {group.queued.length} queued + + {#if group.githubStats} + + GitHub: {group.githubStats.totalAssignedJobs ?? 0} assigned ({group.githubStats.totalBusyRunners ?? 0} busy, {group.githubStats.totalIdleRunners ?? 0} idle) + + {/if} + + {group.runners} runners{group.maxRunners ? ` / ${group.maxRunners} max` : ''} ({group.busyRunners} busy, {group.idleRunners} idle{group.offlineRunners ? `, ${group.offlineRunners} offline` : ''}{group.provisioning ? `, ${group.provisioning} provisioning` : ''}) + +
+
+ + {#if group.queued.length === 0} +
+ Queue is empty. +
+ {:else} +
+ + + + + + + + + + + + {#each group.queued as job, idx (job.id)} + + + + + + + + {/each} + +
#JobRepositoryLabelsWaiting
{idx + 1} + {#if jobUrl(job)} + {job.name || '(unnamed job)'} + {:else} + {job.name || '(unnamed job)'} + {/if} + + {job.repository_owner}/{job.repository_name} + +
+ {#each job.labels || [] as label} + {label} + {/each} +
+ 		+ {waitingFor(job)} +
+
+ {/if} +
+ {/each} + {/if} +
diff --git a/webapp/src/routes/queue/page.integration.test.ts b/webapp/src/routes/queue/page.integration.test.ts new file mode 100644 index 000000000..0dbc37e6d --- /dev/null +++ b/webapp/src/routes/queue/page.integration.test.ts @@ -0,0 +1,295 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { render, screen, waitFor } from '@testing-library/svelte'; + +// Unmock real components that setup.ts might mock +vi.unmock('$lib/components/PageHeader.svelte'); + +vi.mock('$app/navigation', () => ({ + goto: vi.fn() +})); + +vi.mock('$app/stores', () => ({})); + +vi.mock('$lib/api/client.js', () => ({ + garmApi: { + listJobs: vi.fn(), + listInstances: vi.fn() + } +})); + +vi.mock('$lib/stores/eager-cache.js', () => ({ + eagerCache: { + subscribe: vi.fn() + }, + eagerCacheManager: { + getScaleSets: vi.fn(), + getPools: vi.fn(), + retryResource: vi.fn() + } +})); + +vi.mock('$lib/stores/toast.js', () => ({ + toastStore: { + success: vi.fn(), + error: vi.fn(), + info: vi.fn(), + warning: vi.fn(), + add: vi.fn() + } +})); + +vi.mock('$lib/stores/websocket.js', () => ({ + websocketStore: { + subscribeToEntity: vi.fn(() => vi.fn()) + } +})); + +import QueuePage from './+page.svelte'; +import { garmApi } from '$lib/api/client.js'; + +const mockScaleSets = [ + { + id: 5, + name: 'cloudstack-ubuntu24-micro', + org_id: 'org-uuid', + org_name: 'nexthop', + max_runners: 100, + min_idle_runners: 32, + desired_runner_count: 60, + statistics: { totalAssignedJobs: 60, totalBusyRunners: 40, totalIdleRunners: 2 }, + enabled: true + }, + { + id: 7, + name: 'cloudstack-ubuntu24-large', + org_id: 'org-uuid', + org_name: 'nexthop', + max_runners: 10, + enabled: true + } +]; + +const mockPools = [ + { + id: 'aaaabbbb-cccc-dddd-eeee-ffff00001111', + image: 'ubuntu:22.04', + flavor: 'default', + org_id: 'org-uuid', + org_name: 'nexthop', + max_runners: 4, + tags: [{ id: 't1', name: 'self-hosted' }, { id: 't2', name: 'pool-label' }] + } +]; + +const mockInstances = [ + { name: 'r1', scale_set_id: 5, status: 'running', runner_status: 'active' }, + { name: 'r2', scale_set_id: 5, status: 'running', runner_status: 'idle' }, + { name: 'r5', scale_set_id: 5, status: 'running', runner_status: 'offline' }, + { name: 'r3', scale_set_id: 5, status: 'pending_create', runner_status: 'pending' }, + { name: 'r4', scale_set_id: 5, status: 'creating', runner_status: 'pending' }, + { name: 'p1', pool_id: 'aaaabbbb-cccc-dddd-eeee-ffff00001111', status: 'running', runner_status: 'active' } +]; + +const mockJobs = [ + // Two queued scale set jobs (ordering by created_at: job 2 queued first) + { + id: 1, + scaleset_job_id: 'ss-job-1', + scale_set_id: 5, + name: 'build', + status: 'queued', + repository_owner: 'nexthop', + repository_name: 'repo-a', + labels: ['cloudstack-ubuntu24-micro'], + created_at: '2026-07-29T00:10:00Z' + }, + { + id: 2, + scaleset_job_id: 'ss-job-2', + scale_set_id: 5, + name: 'test', + status: 'queued', + repository_owner: 'nexthop', + repository_name: 'repo-b', + labels: ['cloudstack-ubuntu24-micro'], + created_at: '2026-07-29T00:05:00Z' + }, + // A running scale set job + { + id: 3, + scaleset_job_id: 'ss-job-3', + scale_set_id: 5, + name: 'lint', + status: 'in_progress', + repository_owner: 'nexthop', + repository_name: 'repo-a', + labels: ['cloudstack-ubuntu24-micro'], + created_at: '2026-07-29T00:01:00Z' + }, + // A queued webhook job that label-matches the pool + { + id: 4, + workflow_job_id: 1234, + name: 'pool-job', + status: 'queued', + repository_owner: 'nexthop', + repository_name: 'repo-c', + labels: ['self-hosted', 'pool-label'], + created_at: '2026-07-29T00:02:00Z' + }, + // A completed job that must not show up + { + id: 5, + scaleset_job_id: 'ss-job-5', + scale_set_id: 7, + name: 'done-job', + status: 'completed', + repository_owner: 'nexthop', + repository_name: 'repo-a', + labels: ['cloudstack-ubuntu24-large'], + created_at: '2026-07-29T00:00:00Z' + }, + // A queued job matching nothing + { + id: 6, + workflow_job_id: 5678, + name: 'orphan-job', + status: 'queued', + repository_owner: 'nexthop', + repository_name: 'repo-d', + labels: ['no-such-label'], + created_at: '2026-07-29T00:03:00Z' + } +]; + +describe('Queue Page - Integration Tests', () => { + beforeEach(async () => { + vi.clearAllMocks(); + + const cacheModule = await import('$lib/stores/eager-cache.js'); + vi.mocked(cacheModule.eagerCacheManager.getScaleSets).mockResolvedValue(mockScaleSets); + vi.mocked(cacheModule.eagerCacheManager.getPools).mockResolvedValue(mockPools); + vi.mocked(garmApi.listJobs).mockResolvedValue(mockJobs as any); + vi.mocked(garmApi.listInstances).mockResolvedValue(mockInstances as any); + }); + + it('renders page title', async () => { + render(QueuePage); + await waitFor(() => { + expect(screen.getByRole('heading', { name: 'Job Queue' })).toBeInTheDocument(); + }); + }); + + it('groups queued jobs per scale set with queue ordering and counts', async () => { + render(QueuePage); + + await waitFor(() => { + expect(screen.getByRole('heading', { name: 'cloudstack-ubuntu24-micro' })).toBeInTheDocument(); + }); + + // Scale set group shows queued and running counts (incl. max runners) + expect(screen.getByText('2 queued')).toBeInTheDocument(); + // Runner counts come from instances (2 running: 1 busy / 1 idle; 2 provisioning) + expect( + screen.getByText(/3 runners \/ 100 max \(1 busy, 1 idle, 1 offline, 2 provisioning\)/) + ).toBeInTheDocument(); + + // GitHub's view from the scale set statistics + expect(screen.getByText(/GitHub: 60 assigned \(40 busy, 2 idle\)/)).toBeInTheDocument(); + expect( + screen.getByRole('heading', { name: 'cloudstack-ubuntu24-micro' }).querySelector('a') + ).toHaveAttribute('href', '/scalesets/5'); + + // Queue order: 'test' (older) before 'build' + const rows = screen.getAllByRole('row'); + const rowText = rows.map((r) => r.textContent || ''); + const testIdx = rowText.findIndex((t) => t.includes('test')); + const buildIdx = rowText.findIndex((t) => t.includes('build')); + expect(testIdx).toBeGreaterThan(0); + expect(buildIdx).toBeGreaterThan(testIdx); + + // Completed jobs are not shown + expect(screen.queryByText('done-job')).not.toBeInTheDocument(); + }); + + it('matches webhook jobs to pools by labels', async () => { + render(QueuePage); + + await waitFor(() => { + expect(screen.getByText('pool-job')).toBeInTheDocument(); + }); + expect(screen.getByText(/Pool aaaabbbb/)).toBeInTheDocument(); + }); + + it('shows unattributed jobs in their own group', async () => { + render(QueuePage); + + await waitFor(() => { + expect(screen.getByText('Unattributed jobs')).toBeInTheDocument(); + }); + expect(screen.getByText('orphan-job')).toBeInTheDocument(); + }); + + it('shows empty state when there are no active jobs', async () => { + vi.mocked(garmApi.listJobs).mockResolvedValue([]); + vi.mocked(garmApi.listInstances).mockResolvedValue([]); + render(QueuePage); + + await waitFor(() => { + expect(screen.getByText('No queued or running jobs.')).toBeInTheDocument(); + }); + }); +}); + +describe('Queue Page - job links', () => { + beforeEach(async () => { + vi.clearAllMocks(); + const cacheModule = await import('$lib/stores/eager-cache.js'); + vi.mocked(cacheModule.eagerCacheManager.getScaleSets).mockResolvedValue(mockScaleSets); + vi.mocked(cacheModule.eagerCacheManager.getPools).mockResolvedValue(mockPools); + vi.mocked(garmApi.listInstances).mockResolvedValue([]); + }); + + it('deep-links webhook jobs to the GitHub job page and scale set jobs to the run page', async () => { + vi.mocked(garmApi.listJobs).mockResolvedValue([ + { + id: 10, + workflow_job_id: 90673415740, + run_id: 30480594764, + name: 'pool-job', + status: 'queued', + repository_owner: 'nexthop-ai', + repository_name: 'nhsystem-ops', + labels: ['self-hosted', 'pool-label'], + created_at: '2026-07-29T00:02:00Z' + }, + { + id: 11, + scaleset_job_id: 'guid-1', + scale_set_id: 5, + run_id: 1234, + workflow_run_url: 'https://github.com/nexthop-ai/repo-a/actions/runs/1234', + name: 'ss-job', + status: 'queued', + repository_owner: 'nexthop-ai', + repository_name: 'repo-a', + labels: [], + created_at: '2026-07-29T00:02:00Z' + } + ] as any); + + render(QueuePage); + + await waitFor(() => { + expect(screen.getByText('pool-job')).toBeInTheDocument(); + }); + expect(screen.getByText('pool-job').closest('a')).toHaveAttribute( + 'href', + 'https://github.com/nexthop-ai/nhsystem-ops/actions/runs/30480594764/job/90673415740' + ); + expect(screen.getByText('ss-job').closest('a')).toHaveAttribute( + 'href', + 'https://github.com/nexthop-ai/repo-a/actions/runs/1234' + ); + }); +}); diff --git a/webapp/swagger.yaml b/webapp/swagger.yaml index 6a13ec8ce..b9456d619 100644 --- a/webapp/swagger.yaml +++ b/webapp/swagger.yaml @@ -1515,6 +1515,13 @@ definitions: runner_name: type: string x-go-name: RunnerName + scale_set_id: + description: |- + ScaleSetID is the garm ID of the scale set this job was assigned to, + if it came in through a scale set listener. + format: uint64 + type: integer + x-go-name: ScaleSetID scaleset_job_id: description: ScaleSetJobID is the job ID when generated for a scale set. type: string @@ -1980,6 +1987,38 @@ definitions: x-go-name: Prefix type: object x-go-package: github.com/cloudbase/garm/params + RunnerScaleSetStatistic: + properties: + totalAcquiredJobs: + format: int64 + type: integer + x-go-name: TotalAcquiredJobs + totalAssignedJobs: + format: int64 + type: integer + x-go-name: TotalAssignedJobs + totalAvailableJobs: + format: int64 + type: integer + x-go-name: TotalAvailableJobs + totalBusyRunners: + format: int64 + type: integer + x-go-name: TotalBusyRunners + totalIdleRunners: + format: int64 + type: integer + x-go-name: TotalIdleRunners + totalRegisteredRunners: + format: int64 + type: integer + x-go-name: TotalRegisteredRunners + totalRunningJobs: + format: int64 + type: integer + x-go-name: TotalRunningJobs + type: object + x-go-package: github.com/cloudbase/garm/params RunnerStatus: type: string x-go-package: github.com/cloudbase/garm/params @@ -2108,6 +2147,8 @@ definitions: x-go-name: ScaleSetID state: $ref: '#/definitions/ScaleSetState' + statistics: + $ref: '#/definitions/RunnerScaleSetStatistic' status_messages: items: $ref: '#/definitions/StatusMessage' diff --git a/workers/scaleset/interfaces.go b/workers/scaleset/interfaces.go index b8acfceb9..5be303a27 100644 --- a/workers/scaleset/interfaces.go +++ b/workers/scaleset/interfaces.go @@ -22,7 +22,7 @@ type scaleSetHelper interface { GetScaleSet() params.ScaleSet GetScaleSetClient() (*scalesets.ScaleSetClient, error) SetLastMessageID(id int64) error - SetDesiredRunnerCount(count int) error + SetRunnerStatistics(stats params.RunnerScaleSetStatistic) error Owner() string HandleJobsCompleted(jobs []params.ScaleSetJobMessage) error HandleJobsStarted(jobs []params.ScaleSetJobMessage) error diff --git a/workers/scaleset/scaleset.go b/workers/scaleset/scaleset.go index 418756c0d..8d80e2d19 100644 --- a/workers/scaleset/scaleset.go +++ b/workers/scaleset/scaleset.go @@ -457,7 +457,8 @@ func (w *Worker) consolidateRunnerState(runners []params.RunnerReference) error // Cross check what exists in github with what we have in the database. for name, runner := range ghRunnersByName { status := runner.GetStatus() - if _, ok := dbRunnersByName[name]; !ok { + dbRunner, ok := dbRunnersByName[name] + if !ok { // runner appears to be active. Is it not managed by GARM? if status != params.RunnerIdle && status != params.RunnerActive { slog.InfoContext(w.ctx, "runner does not exist in GARM; removing from github", "runner_name", name) @@ -470,6 +471,39 @@ func (w *Worker) consolidateRunnerState(runners []params.RunnerReference) error } continue } + + // Sync github's view of the runner (online idle/busy or offline) onto + // the instance. Without this, a runner whose agent died after setup + // stays "idle" in GARM forever, while github considers it offline and + // never assigns it jobs. Only touch runners that finished installing; + // runners in earlier lifecycle states are expected to be offline. + switch dbRunner.RunnerStatus { + case params.RunnerIdle, params.RunnerActive, params.RunnerOffline: + default: + continue + } + switch status { + case params.RunnerIdle, params.RunnerActive, params.RunnerOffline: + default: + continue + } + if dbRunner.RunnerStatus == status { + continue + } + if ok := locking.TryLock(name, w.consumerID); !ok { + slog.DebugContext(w.ctx, "runner is locked; skipping runner status sync", "runner_name", name) + continue + } + slog.InfoContext(w.ctx, "syncing runner status from github", "runner_name", name, "old_status", dbRunner.RunnerStatus, "new_status", status) + updatedRunner, err := w.store.UpdateInstance(w.ctx, name, params.UpdateInstanceParams{RunnerStatus: status}) + locking.Unlock(name, false) + if err != nil { + if !errors.Is(err, runnerErrors.ErrNotFound) { + slog.ErrorContext(w.ctx, "error updating runner status", "runner_name", name, "error", err) + } + continue + } + w.runners[updatedRunner.ID] = updatedRunner } unlockFn, err := w.reapTimedOutRunners(ghRunnersByName) diff --git a/workers/scaleset/scaleset_helper.go b/workers/scaleset/scaleset_helper.go index ea74abb74..c27f017b4 100644 --- a/workers/scaleset/scaleset_helper.go +++ b/workers/scaleset/scaleset_helper.go @@ -74,6 +74,7 @@ func (w *Worker) recordOrUpdateJob(job params.ScaleSetJobMessage) error { baseURL := strings.TrimRight(w.entity.Credentials.BaseURL, "/") jobParams := job.ToJob() jobParams.RunnerGroupName = w.scaleSet.GitHubRunnerGroup + jobParams.ScaleSetID = w.scaleSet.ID switch entity.EntityType { case params.ForgeEntityTypeEnterprise: @@ -227,9 +228,9 @@ func (w *Worker) HandleJobsAvailable(jobs []params.ScaleSetJobMessage) error { return nil } -func (w *Worker) SetDesiredRunnerCount(count int) error { - if err := w.store.SetScaleSetDesiredRunnerCount(w.ctx, w.scaleSet.ID, count); err != nil { - return fmt.Errorf("setting desired runner count: %w", err) +func (w *Worker) SetRunnerStatistics(stats params.RunnerScaleSetStatistic) error { + if err := w.store.SetScaleSetRunnerStatistics(w.ctx, w.scaleSet.ID, stats); err != nil { + return fmt.Errorf("setting runner statistics: %w", err) } return nil } diff --git a/workers/scaleset/scaleset_listener.go b/workers/scaleset/scaleset_listener.go index 3fc00e0da..e8325707d 100644 --- a/workers/scaleset/scaleset_listener.go +++ b/workers/scaleset/scaleset_listener.go @@ -228,8 +228,10 @@ func (l *scaleSetListener) handleSessionMessage(msg params.RunnerScaleSetMessage l.lastMessageID = msg.MessageID } - if err := l.scaleSetHelper.SetDesiredRunnerCount(msg.Statistics.TotalAssignedJobs); err != nil { - slog.ErrorContext(l.ctx, "setting desired runner count", "error", err) + if msg.Statistics != nil { + if err := l.scaleSetHelper.SetRunnerStatistics(*msg.Statistics); err != nil { + slog.ErrorContext(l.ctx, "setting runner statistics", "error", err) + } } if err := l.messageSession.DeleteMessage(l.listenerCtx, msg.MessageID); err != nil {