Manager; convert QueryJobTaskSummaries and SummarizeJobStatuses to sqlc
Ref: #104305
This commit is contained in:
parent
cda0b916fb
commit
35313477a0
@ -1129,8 +1129,7 @@ func convertSqlcTask(task sqlc.Task, jobUUID string, workerUUID string) (*Task,
|
|||||||
}
|
}
|
||||||
|
|
||||||
if err := json.Unmarshal(task.Commands, &dbTask.Commands); err != nil {
|
if err := json.Unmarshal(task.Commands, &dbTask.Commands); err != nil {
|
||||||
return nil, taskError(err, fmt.Sprintf("task %s of job %s has invalid commands: %v",
|
return nil, taskError(err, "task %s of job %s has invalid commands: %v", task.UUID, jobUUID, err)
|
||||||
task.UUID, jobUUID, err))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return &dbTask, nil
|
return &dbTask, nil
|
||||||
|
@ -14,14 +14,32 @@ func (db *DB) QueryJobTaskSummaries(ctx context.Context, jobUUID string) ([]*Tas
|
|||||||
logger := log.Ctx(ctx)
|
logger := log.Ctx(ctx)
|
||||||
logger.Debug().Str("job", jobUUID).Msg("querying task summaries")
|
logger.Debug().Str("job", jobUUID).Msg("querying task summaries")
|
||||||
|
|
||||||
var result []*Task
|
queries := db.queries()
|
||||||
tx := db.gormDB.WithContext(ctx).Model(&Task{}).
|
sqlcPartialTasks, err := queries.QueryJobTaskSummaries(ctx, jobUUID)
|
||||||
Select("tasks.id", "tasks.uuid", "tasks.name", "tasks.priority", "tasks.status", "tasks.type", "tasks.updated_at").
|
if err != nil {
|
||||||
Joins("left join jobs on jobs.id = tasks.job_id").
|
return nil, err
|
||||||
Where("jobs.uuid=?", jobUUID).
|
}
|
||||||
Scan(&result)
|
|
||||||
|
|
||||||
return result, tx.Error
|
// Convert to partial GORM tasks.
|
||||||
|
gormTasks := make([]*Task, len(sqlcPartialTasks))
|
||||||
|
for index, task := range sqlcPartialTasks {
|
||||||
|
gormTask := Task{
|
||||||
|
Model: Model{
|
||||||
|
ID: uint(task.ID),
|
||||||
|
UpdatedAt: task.UpdatedAt.Time,
|
||||||
|
},
|
||||||
|
|
||||||
|
UUID: task.UUID,
|
||||||
|
Name: task.Name,
|
||||||
|
Type: task.Type,
|
||||||
|
Priority: int(task.Priority),
|
||||||
|
Status: api.TaskStatus(task.Status),
|
||||||
|
JobUUID: jobUUID,
|
||||||
|
}
|
||||||
|
gormTasks[index] = &gormTask
|
||||||
|
}
|
||||||
|
|
||||||
|
return gormTasks, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// JobStatusCount is a mapping from job status to the number of jobs in that status.
|
// JobStatusCount is a mapping from job status to the number of jobs in that status.
|
||||||
@ -31,24 +49,16 @@ func (db *DB) SummarizeJobStatuses(ctx context.Context) (JobStatusCount, error)
|
|||||||
logger := log.Ctx(ctx)
|
logger := log.Ctx(ctx)
|
||||||
logger.Debug().Msg("database: summarizing job statuses")
|
logger.Debug().Msg("database: summarizing job statuses")
|
||||||
|
|
||||||
// Query the database using a data structure that's easy to handle in GORM.
|
queries := db.queries()
|
||||||
type queryResult struct {
|
result, err := queries.SummarizeJobStatuses(ctx)
|
||||||
Status api.JobStatus
|
if err != nil {
|
||||||
StatusCount int
|
return nil, jobError(err, "summarizing job statuses")
|
||||||
}
|
|
||||||
result := []*queryResult{}
|
|
||||||
tx := db.gormDB.WithContext(ctx).Model(&Job{}).
|
|
||||||
Select("status as Status", "count(id) as StatusCount").
|
|
||||||
Group("status").
|
|
||||||
Scan(&result)
|
|
||||||
if tx.Error != nil {
|
|
||||||
return nil, jobError(tx.Error, "summarizing job statuses")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Convert the array-of-structs to a map that's easier to handle by the caller.
|
// Convert the array-of-structs to a map that's easier to handle by the caller.
|
||||||
statusCounts := make(JobStatusCount)
|
statusCounts := make(JobStatusCount)
|
||||||
for _, singleStatusCount := range result {
|
for _, singleStatusCount := range result {
|
||||||
statusCounts[singleStatusCount.Status] = singleStatusCount.StatusCount
|
statusCounts[api.JobStatus(singleStatusCount.Status)] = int(singleStatusCount.StatusCount)
|
||||||
}
|
}
|
||||||
|
|
||||||
return statusCounts, nil
|
return statusCounts, nil
|
||||||
|
@ -306,3 +306,14 @@ WHERE
|
|||||||
TF.worker_id = @worker_id
|
TF.worker_id = @worker_id
|
||||||
AND T.job_id = @job_id
|
AND T.job_id = @job_id
|
||||||
AND T.type = @task_type;
|
AND T.type = @task_type;
|
||||||
|
|
||||||
|
|
||||||
|
-- name: QueryJobTaskSummaries :many
|
||||||
|
SELECT tasks.id, tasks.uuid, tasks.name, tasks.priority, tasks.status, tasks.type, tasks.updated_at
|
||||||
|
FROM tasks
|
||||||
|
LEFT JOIN jobs ON jobs.id = tasks.job_id
|
||||||
|
WHERE jobs.uuid=@job_uuid;
|
||||||
|
|
||||||
|
-- name: SummarizeJobStatuses :many
|
||||||
|
SELECT status, count(id) as status_count FROM jobs
|
||||||
|
GROUP BY status;
|
||||||
|
@ -916,6 +916,54 @@ func (q *Queries) JobCountTasksInStatus(ctx context.Context, arg JobCountTasksIn
|
|||||||
return num_tasks, err
|
return num_tasks, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const queryJobTaskSummaries = `-- name: QueryJobTaskSummaries :many
|
||||||
|
SELECT tasks.id, tasks.uuid, tasks.name, tasks.priority, tasks.status, tasks.type, tasks.updated_at
|
||||||
|
FROM tasks
|
||||||
|
LEFT JOIN jobs ON jobs.id = tasks.job_id
|
||||||
|
WHERE jobs.uuid=?1
|
||||||
|
`
|
||||||
|
|
||||||
|
type QueryJobTaskSummariesRow struct {
|
||||||
|
ID int64
|
||||||
|
UUID string
|
||||||
|
Name string
|
||||||
|
Priority int64
|
||||||
|
Status string
|
||||||
|
Type string
|
||||||
|
UpdatedAt sql.NullTime
|
||||||
|
}
|
||||||
|
|
||||||
|
func (q *Queries) QueryJobTaskSummaries(ctx context.Context, jobUuid string) ([]QueryJobTaskSummariesRow, error) {
|
||||||
|
rows, err := q.db.QueryContext(ctx, queryJobTaskSummaries, jobUuid)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var items []QueryJobTaskSummariesRow
|
||||||
|
for rows.Next() {
|
||||||
|
var i QueryJobTaskSummariesRow
|
||||||
|
if err := rows.Scan(
|
||||||
|
&i.ID,
|
||||||
|
&i.UUID,
|
||||||
|
&i.Name,
|
||||||
|
&i.Priority,
|
||||||
|
&i.Status,
|
||||||
|
&i.Type,
|
||||||
|
&i.UpdatedAt,
|
||||||
|
); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
items = append(items, i)
|
||||||
|
}
|
||||||
|
if err := rows.Close(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return items, nil
|
||||||
|
}
|
||||||
|
|
||||||
const removeFromJobBlocklist = `-- name: RemoveFromJobBlocklist :exec
|
const removeFromJobBlocklist = `-- name: RemoveFromJobBlocklist :exec
|
||||||
DELETE FROM job_blocks
|
DELETE FROM job_blocks
|
||||||
WHERE
|
WHERE
|
||||||
@ -1068,6 +1116,39 @@ func (q *Queries) StoreTaskDependency(ctx context.Context, arg StoreTaskDependen
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const summarizeJobStatuses = `-- name: SummarizeJobStatuses :many
|
||||||
|
SELECT status, count(id) as status_count FROM jobs
|
||||||
|
GROUP BY status
|
||||||
|
`
|
||||||
|
|
||||||
|
type SummarizeJobStatusesRow struct {
|
||||||
|
Status string
|
||||||
|
StatusCount int64
|
||||||
|
}
|
||||||
|
|
||||||
|
func (q *Queries) SummarizeJobStatuses(ctx context.Context) ([]SummarizeJobStatusesRow, error) {
|
||||||
|
rows, err := q.db.QueryContext(ctx, summarizeJobStatuses)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var items []SummarizeJobStatusesRow
|
||||||
|
for rows.Next() {
|
||||||
|
var i SummarizeJobStatusesRow
|
||||||
|
if err := rows.Scan(&i.Status, &i.StatusCount); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
items = append(items, i)
|
||||||
|
}
|
||||||
|
if err := rows.Close(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return items, nil
|
||||||
|
}
|
||||||
|
|
||||||
const taskAssignToWorker = `-- name: TaskAssignToWorker :exec
|
const taskAssignToWorker = `-- name: TaskAssignToWorker :exec
|
||||||
UPDATE tasks SET
|
UPDATE tasks SET
|
||||||
updated_at = ?1,
|
updated_at = ?1,
|
||||||
|
Loading…
x
Reference in New Issue
Block a user