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 {