Skip to content

webapp: add per scale set job queue view #836

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
benoit-nexthop wants to merge 1 commit into
base: main
Choose a base branch
from
Open

webapp: add per scale set job queue view #836

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 32 additions & 32 deletions database/common/mocks/Store.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion database/common/store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
14 changes: 14 additions & 0 deletions database/sql/jobs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
}
Expand Down
9 changes: 9 additions & 0 deletions database/sql/models.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

@gabriel-samfira gabriel-samfira Aug 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The model changes will require a migration to be added. We switched to gormigrate to do away with the custom code in sql.go (and the convoluted conditionals we used to have).

A file in database/sql/migrations/0007_scaleset_job_queue.go with something like:

package migrations

import (
	"github.com/go-gormigrate/gormigrate/v2"
	"gorm.io/datatypes"
	"gorm.io/gorm"
)

// scaleSet0007 adds the cached RunnerScaleSetStatistic received from
// GitHub on the message session.
// This is just a stub, AutoMigrate ignores everything else.
type scaleSet0007 struct {
	RunnerStatistics datatypes.JSON
}

func (scaleSet0007) TableName() string {
	return "scale_sets"
}

// workflowJob0007 adds the scale set attribution for jobs recorded by
// scale set listeners.
type workflowJob0007 struct {
	ScaleSetFkID *uint `gorm:"index"`
}

func (workflowJob0007) TableName() string { return "workflow_jobs" }

func init() {
	Register(&gormigrate.Migration{
		ID: "0007_scaleset_job_queue",
		Migrate: func(tx *gorm.DB) error {
			if err := tx.AutoMigrate(&scaleSet0007{}); err != nil {
				return err
			}
			return tx.AutoMigrate(&workflowJob0007{})
		},
	})
}

At the end, your migrations table should look like this:

sqlite> select * from migrations;
SCHEMA_INIT
0001_baseline
0002_case_insensitive_indexes
0002_lower_indexes
0003_forge_instances
0004_garm_agent_version
0005_allow_insecure_garm_agent
0006_proxies
0007_scaleset_job_queue
sqlite>

// 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.
Expand Down Expand Up @@ -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"`

@gabriel-samfira gabriel-samfira Aug 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this needs a:

constraint:OnDelete:SET NULL

Otherwise we'd get a foreign key constraint error when trying to remove a scaleset if a job is associated with it. We have guards to not allow a scaleset to be removed if it has a runner, but a runner may not be spun up instantly as a job is recorded.


// 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.
Expand Down
10 changes: 8 additions & 2 deletions database/sql/scalesets.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ package sql

import (
"context"
"encoding/json"
"errors"
"fmt"

Expand Down Expand Up @@ -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() {
Expand All @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion database/sql/scalesets_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
8 changes: 8 additions & 0 deletions database/sql/util.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

@gabriel-samfira gabriel-samfira Aug 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we ever run into a corrupted stat, listing scalesets will fail. This should probably be logged as an error, but not bail here. See:

garm/database/sql/util.go

Lines 81 to 88 in 2edcc2f

if len(instance.Capabilities) > 0 {
var caps params.AgentCapabilities
if err := json.Unmarshal(instance.Capabilities, &caps); err == nil {
ret.Capabilities = caps
} else {
slog.ErrorContext(s.ctx, "failed to unmarshal capabilities", "instance_name", instance.Name, "error", err)
}
}

as a reference.

}
ret.Statistics = &stats
}

var ep GithubEndpoint
if scaleSet.RepoID != nil {
ret.RepoID = scaleSet.RepoID.String()
Expand Down
5 changes: 3 additions & 2 deletions database/watcher/watcher_store_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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 = &params.RunnerScaleSetStatistic{TotalAssignedJobs: 5}
updatedScaleSet.LastMessageID = 99
payloadFromEvent, ok := event.Payload.(params.ScaleSet)
s.Require().True(ok)
Expand Down
8 changes: 7 additions & 1 deletion params/params.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
11 changes: 11 additions & 0 deletions webapp/src/lib/api/generated-client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,9 @@ import {
ProxiesApi,
ObjectsApi,
ToolsApi,
JobsApi,
type GARMAgentRelease,
type Job,
type Repository,
type Organization,
type Enterprise,
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -644,6 +649,12 @@ export class GeneratedGarmApiClient {
await this.scaleSetsApi.deleteScaleSet(id.toString());
}

// Jobs
async listJobs(): Promise<Job[]> {
const response = await this.jobsApi.listJobs();

@gabriel-samfira gabriel-samfira Aug 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This might have the potential to overload the DB. If the jobs table stays small, this should be fine, but if we ever end up with many jobs, it might impact performance. Worth keeping track of.

return response.data || [];
}

// Instances
async listInstances(): Promise<Instance[]> {
const response = await this.instancesApi.listInstances();
Expand Down
26 changes: 26 additions & 0 deletions webapp/src/lib/api/generated/api.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2325,6 +2325,12 @@ export interface Job {
* @memberof Job
*/
'runner_name'?: string;
/**

@gabriel-samfira gabriel-samfira Aug 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hmm. Did you use:

make generate

to update this file, or was it manually edited? This file is meant to be generated from the swagger definitions.

Have a look here:

https://openapi-generator.tech/docs/installation/

and here:

https://github.com/cloudbase/garm/blob/main/webapp/DEV_SETUP.md

* 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}
Expand Down Expand Up @@ -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 {
/**
*
Expand Down Expand Up @@ -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}
Expand Down
Loading