DevFW-CICD/garm
15
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

Handle new jobID for scale sets

Browse source 
There seems to be a change in the scale set message. It now includes
a jobID and sets the runner request ID to 0. This change adds separate
job ID fields for workflow jobs and scaleset jobs.

Signed-off-by: Gabriel Adrian Samfira <gsamfira@cloudbasesolutions.com>
This commit is contained in:
Gabriel Adrian Samfira 2025-07-18 07:51:50 +00:00
parent d26973da2a
commit a984782fd7
8 changed files with 107 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

48
database/sql/jobs.go
View file

@ -41,6 +41,8 @@ func sqlWorkflowJobToParamsJob(job WorkflowJob) (params.Job, error) {
jobParam := params.Job{
ID: job.ID,
WorkflowJobID: job.WorkflowJobID,
ScaleSetJobID: job.ScaleSetJobID,
RunID: job.RunID,
Action: job.Action,
Status: job.Status,
@ -75,7 +77,8 @@ func (s *sqlDatabase) paramsJobToWorkflowJob(ctx context.Context, job params.Job
}
workflofJob := WorkflowJob{
ID: job.ID,
ScaleSetJobID: job.ScaleSetJobID,
WorkflowJobID: job.ID,
RunID: job.RunID,
Action: job.Action,
Status: job.Status,
@ -109,14 +112,27 @@ func (s *sqlDatabase) paramsJobToWorkflowJob(ctx context.Context, job params.Job
}
func (s *sqlDatabase) DeleteJob(_ context.Context, jobID int64) (err error) {
var workflowJob WorkflowJob
q := s.conn.Where("workflow_job_id = ?", jobID).Preload("Instance").First(&workflowJob)
if q.Error != nil {
if errors.Is(q.Error, gorm.ErrRecordNotFound) {
return nil
}
return errors.Wrap(q.Error, "fetching job")
}
removedJob, err := sqlWorkflowJobToParamsJob(workflowJob)
if err != nil {
return errors.Wrap(err, "converting job")
}
defer func() {
if err == nil {
if notifyErr := s.sendNotify(common.JobEntityType, common.DeleteOperation, params.Job{ID: jobID}); notifyErr != nil {
if notifyErr := s.sendNotify(common.JobEntityType, common.DeleteOperation, removedJob); notifyErr != nil {
slog.With(slog.Any("error", notifyErr)).Error("failed to send notify")
}
}
}()
q := s.conn.Delete(&WorkflowJob{}, jobID)
q = s.conn.Delete(&workflowJob)
if q.Error != nil {
if errors.Is(q.Error, gorm.ErrRecordNotFound) {
return nil
@ -132,7 +148,7 @@ func (s *sqlDatabase) LockJob(_ context.Context, jobID int64, entityID string) e
return errors.Wrap(err, "parsing entity id")
}
var workflowJob WorkflowJob
q := s.conn.Clauses(clause.Locking{Strength: "UPDATE"}).Preload("Instance").Where("id = ?", jobID).First(&workflowJob)
q := s.conn.Preload("Instance").Where("id = ?", jobID).First(&workflowJob)
if q.Error != nil {
if errors.Is(q.Error, gorm.ErrRecordNotFound) {
@ -167,7 +183,7 @@ func (s *sqlDatabase) LockJob(_ context.Context, jobID int64, entityID string) e
func (s *sqlDatabase) BreakLockJobIsQueued(_ context.Context, jobID int64) (err error) {
var workflowJob WorkflowJob
q := s.conn.Clauses(clause.Locking{Strength: "UPDATE"}).Preload("Instance").Where("id = ? and status = ?", jobID, params.JobStatusQueued).First(&workflowJob)
q := s.conn.Clauses(clause.Locking{Strength: "UPDATE"}).Preload("Instance").Where("workflow_job_id = ? and status = ?", jobID, params.JobStatusQueued).First(&workflowJob)
if q.Error != nil {
if errors.Is(q.Error, gorm.ErrRecordNotFound) {
@ -195,7 +211,7 @@ func (s *sqlDatabase) BreakLockJobIsQueued(_ context.Context, jobID int64) (err
func (s *sqlDatabase) UnlockJob(_ context.Context, jobID int64, entityID string) error {
var workflowJob WorkflowJob
q := s.conn.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", jobID).First(&workflowJob)
q := s.conn.Clauses(clause.Locking{Strength: "UPDATE"}).Where("workflow_job_id = ?", jobID).First(&workflowJob)
if q.Error != nil {
if errors.Is(q.Error, gorm.ErrRecordNotFound) {
@ -229,7 +245,14 @@ func (s *sqlDatabase) UnlockJob(_ context.Context, jobID int64, entityID string)
func (s *sqlDatabase) CreateOrUpdateJob(ctx context.Context, job params.Job) (params.Job, error) {
var workflowJob WorkflowJob
var err error
q := s.conn.Clauses(clause.Locking{Strength: "UPDATE"}).Preload("Instance").Where("id = ?", job.ID).First(&workflowJob)
searchField := "workflow_job_id = ?"
var searchVal any = job.ID
if job.ScaleSetJobID != "" {
searchField = "scale_set_job_id = ?"
searchVal = job.ScaleSetJobID
}
q := s.conn.Preload("Instance").Where(searchField, searchVal).First(&workflowJob)
if q.Error != nil {
if !errors.Is(q.Error, gorm.ErrRecordNotFound) {
@ -249,6 +272,9 @@ func (s *sqlDatabase) CreateOrUpdateJob(ctx context.Context, job params.Job) (pa
workflowJob.GithubRunnerID = job.GithubRunnerID
workflowJob.RunnerGroupID = job.RunnerGroupID
workflowJob.RunnerGroupName = job.RunnerGroupName
if job.RunID != 0 && workflowJob.RunID == 0 {
workflowJob.RunID = job.RunID
}
if job.LockedBy != uuid.Nil {
workflowJob.LockedBy = job.LockedBy
@ -327,7 +353,11 @@ func (s *sqlDatabase) ListEntityJobsByStatus(_ context.Context, entityType param
}
var jobs []WorkflowJob
query := s.conn.Model(&WorkflowJob{}).Preload("Instance").Where("status = ?", status)
query := s.conn.
Model(&WorkflowJob{}).
Preload("Instance").
Where("status = ?", status).
Where("workflow_job_id > 0")
switch entityType {
case params.ForgeEntityTypeOrganization:
@ -381,7 +411,7 @@ func (s *sqlDatabase) ListAllJobs(_ context.Context) ([]params.Job, error) {
// GetJobByID gets a job by id.
func (s *sqlDatabase) GetJobByID(_ context.Context, jobID int64) (params.Job, error) {
var job WorkflowJob
query := s.conn.Model(&WorkflowJob{}).Preload("Instance").Where("id = ?", jobID)
query := s.conn.Model(&WorkflowJob{}).Preload("Instance").Where("workflow_job_id = ?", jobID)
if err := query.First(&job); err.Error != nil {
if errors.Is(err.Error, gorm.ErrRecordNotFound) {

6
database/sql/models.go
View file

@ -319,6 +319,12 @@ type User struct {
type WorkflowJob struct {
// ID is the ID of the job.
ID int64 `gorm:"index"`
// WorkflowJobID is the ID of the workflow job.
WorkflowJobID int64 `gorm:"index:workflow_job_id_idx"`
// ScaleSetJobID is the job ID for a scaleset job.
ScaleSetJobID string `gorm:"index:scaleset_job_id_idx"`
// 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.

29
database/sql/sql.go
View file

@ -374,6 +374,22 @@ func (s *sqlDatabase) migrateCredentialsToDB() (err error) {
return nil
}
func (s *sqlDatabase) migrateWorkflow() error {
if s.conn.Migrator().HasTable(&WorkflowJob{}) {
if s.conn.Migrator().HasColumn(&WorkflowJob{}, "runner_name") {
// Remove jobs that are not in "queued" status. We really only care about queued jobs. Once they transition
// to something else, we don't really consume them anyway.
if err := s.conn.Exec("delete from workflow_jobs where status is not 'queued'").Error; err != nil {
return errors.Wrap(err, "updating workflow_jobs")
}
if err := s.conn.Migrator().DropColumn(&WorkflowJob{}, "runner_name"); err != nil {
return errors.Wrap(err, "updating workflow_jobs")
}
}
}
return nil
}
func (s *sqlDatabase) migrateDB() error {
if s.conn.Migrator().HasIndex(&Organization{}, "idx_organizations_name") {
if err := s.conn.Migrator().DropIndex(&Organization{}, "idx_organizations_name"); err != nil {
@ -405,17 +421,8 @@ func (s *sqlDatabase) migrateDB() error {
}
}
if s.conn.Migrator().HasTable(&WorkflowJob{}) {
if s.conn.Migrator().HasColumn(&WorkflowJob{}, "runner_name") {
// Remove jobs that are not in "queued" status. We really only care about queued jobs. Once they transition
// to something else, we don't really consume them anyway.
if err := s.conn.Exec("delete from workflow_jobs where status is not 'queued'").Error; err != nil {
return errors.Wrap(err, "updating workflow_jobs")
}
if err := s.conn.Migrator().DropColumn(&WorkflowJob{}, "runner_name"); err != nil {
return errors.Wrap(err, "updating workflow_jobs")
}
}
if err := s.migrateWorkflow(); err != nil {
return errors.Wrap(err, "migrating workflows")
}
if s.conn.Migrator().HasTable(&GithubEndpoint{}) {