particle-os/debian-forge-composer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

jobqueue: drop JobStatus type

Browse source 
The enum is redundant information that can be deduced from the job's
times: queuedAt, startedAt, and finishedAt. Not having it reduces the
potential for inconsistent state.
This commit is contained in:
Lars Karlitski 2020-05-10 10:16:00 +02:00 • committed by Tom Gundersen
parent e1805d5f62
commit 6773c01722
5 changed files with 70 additions and 101 deletions
Download patch file Download diff file Expand all files Collapse all files

21
internal/jobqueue/fsjobqueue/fsjobqueue.go
View file

@ -51,10 +51,9 @@ type job struct {
Dependencies []uuid.UUID `json:"dependencies"`
Result json.RawMessage `json:"result,omitempty"`
Status jobqueue.JobStatus `json:"status"`
QueuedAt time.Time `json:"queued_at,omitempty"`
StartedAt time.Time `json:"started_at,omitempty"`
FinishedAt time.Time `json:"finished_at,omitempty"`
QueuedAt time.Time `json:"queued_at,omitempty"`
StartedAt time.Time `json:"started_at,omitempty"`
FinishedAt time.Time `json:"finished_at,omitempty"`
}
// Create a new fsJobQueue object for `dir`. This object must have exclusive
@ -82,7 +81,7 @@ func New(dir string) (*fsJobQueue, error) {
return nil, err
}
// We only enqueue jobs that were previously pending.
if j.Status != jobqueue.JobPending {
if j.StartedAt.IsZero() {
continue
}
// Enqueue the job again if all dependencies have finished, or
@ -112,7 +111,6 @@ func (q *fsJobQueue) Enqueue(jobType string, args interface{}, dependencies []uu
Id: uuid.New(),
Type: jobType,
Dependencies: uniqueUUIDList(dependencies),
Status: jobqueue.JobPending,
QueuedAt: time.Now(),
}
@ -192,7 +190,6 @@ func (q *fsJobQueue) Dequeue(ctx context.Context, jobTypes []string, args interf
return uuid.Nil, fmt.Errorf("error unmarshaling arguments for job '%s': %v", j.Id, err)
}
j.Status = jobqueue.JobRunning
j.StartedAt = time.Now()
err = q.db.Write(id.String(), j)
@ -209,11 +206,10 @@ func (q *fsJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
return err
}
if j.Status != jobqueue.JobRunning {
if j.StartedAt.IsZero() || !j.FinishedAt.IsZero() {
return jobqueue.ErrNotRunning
}
j.Status = jobqueue.JobFinished
j.FinishedAt = time.Now()
j.Result, err = json.Marshal(result)
@ -247,7 +243,7 @@ func (q *fsJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
return nil
}
func (q *fsJobQueue) JobStatus(id uuid.UUID, result interface{}) (status jobqueue.JobStatus, queued, started, finished time.Time, err error) {
func (q *fsJobQueue) JobStatus(id uuid.UUID, result interface{}) (queued, started, finished time.Time, err error) {
var j *job
j, err = q.readJob(id)
@ -255,7 +251,7 @@ func (q *fsJobQueue) JobStatus(id uuid.UUID, result interface{}) (status jobqueu
return
}
if j.Status == jobqueue.JobFinished {
if !j.FinishedAt.IsZero() {
err = json.Unmarshal(j.Result, result)
if err != nil {
err = fmt.Errorf("error unmarshaling result for job '%s': %v", id, err)
@ -263,7 +259,6 @@ func (q *fsJobQueue) JobStatus(id uuid.UUID, result interface{}) (status jobqueu
}
}
status = j.Status
queued = j.QueuedAt
started = j.StartedAt
finished = j.FinishedAt
@ -279,7 +274,7 @@ func (q *fsJobQueue) countFinishedJobs(ids []uuid.UUID) (int, error) {
if err != nil {
return 0, err
}
if j.Status == jobqueue.JobFinished {
if !j.FinishedAt.IsZero() {
n += 1
}
}

24
internal/jobqueue/fsjobqueue/fsjobqueue_test.go
View file

@ -130,15 +130,19 @@ func TestDependencies(t *testing.T) {
require.ElementsMatch(t, []uuid.UUID{one, two}, r)
j := pushTestJob(t, q, "test", nil, []uuid.UUID{one, two})
status, _, _, _, err := q.JobStatus(j, nil)
queued, started, finished, err := q.JobStatus(j, nil)
require.NoError(t, err)
require.Equal(t, jobqueue.JobPending, status)
require.True(t, !queued.IsZero())
require.True(t, started.IsZero())
require.True(t, finished.IsZero())
require.Equal(t, j, finishNextTestJob(t, q, "test", testResult{}))
status, _, _, _, err = q.JobStatus(j, &testResult{})
queued, started, finished, err = q.JobStatus(j, &testResult{})
require.NoError(t, err)
require.Equal(t, jobqueue.JobFinished, status)
require.True(t, !queued.IsZero())
require.True(t, !started.IsZero())
require.True(t, !finished.IsZero())
})
t.Run("done-after-pushing-dependant", func(t *testing.T) {
@ -146,9 +150,11 @@ func TestDependencies(t *testing.T) {
two := pushTestJob(t, q, "test", nil, nil)
j := pushTestJob(t, q, "test", nil, []uuid.UUID{one, two})
status, _, _, _, err := q.JobStatus(j, nil)
queued, started, finished, err := q.JobStatus(j, nil)
require.NoError(t, err)
require.Equal(t, jobqueue.JobPending, status)
require.True(t, !queued.IsZero())
require.True(t, started.IsZero())
require.True(t, finished.IsZero())
r := []uuid.UUID{}
r = append(r, finishNextTestJob(t, q, "test", testResult{}))
@ -157,8 +163,10 @@ func TestDependencies(t *testing.T) {
require.Equal(t, j, finishNextTestJob(t, q, "test", testResult{}))
status, _, _, _, err = q.JobStatus(j, &testResult{})
queued, started, finished, err = q.JobStatus(j, &testResult{})
require.NoError(t, err)
require.Equal(t, jobqueue.JobFinished, status)
require.True(t, !queued.IsZero())
require.True(t, !started.IsZero())
require.True(t, !finished.IsZero())
})
}

56
internal/jobqueue/jobqueue.go
View file

@ -14,7 +14,6 @@ package jobqueue
import (
"context"
"encoding/json"
"errors"
"time"
@ -49,54 +48,13 @@ type JobQueue interface {
// job type and must be serializable to JSON.
FinishJob(id uuid.UUID, result interface{}) error
// Returns the current status of the job. If the job has already
// finished, its result will be returned in `result`. Also returns the
// time the job was
// queued - always valid
// started - valid when the job is running or has finished
// finished - valid when the job has finished
JobStatus(id uuid.UUID, result interface{}) (status JobStatus, queued, started, finished time.Time, err error)
}
type JobStatus int
const (
JobPending JobStatus = iota
JobRunning
JobFinished
)
func (s JobStatus) String() string {
switch s {
case JobPending:
return "pending"
case JobRunning:
return "running"
case JobFinished:
return "finished"
default:
return "<invalid>"
}
}
func (s JobStatus) MarshalJSON() ([]byte, error) {
return json.Marshal(s.String())
}
func (s *JobStatus) UnmarshalJSON(data []byte) error {
var str string
if err := json.Unmarshal(data, &str); err != nil {
return err
}
switch str {
case "pending":
*s = JobPending
case "running":
*s = JobRunning
case "finished":
*s = JobFinished
}
return nil
// Returns the current status of the job, in the form of three times:
// queued, started, and finished. `started` and `finished` might be the
// zero time (check with t.IsZero()), when the job is not running or
// finished, respectively.
//
// If the job is finished, its result will be returned in `result`.
JobStatus(id uuid.UUID, result interface{}) (queued, started, finished time.Time, err error)
}
var (

22
internal/jobqueue/testjobqueue/testjobqueue.go
View file

@ -30,7 +30,9 @@ type job struct {
Args json.RawMessage
Dependencies []uuid.UUID
Result json.RawMessage
Status jobqueue.JobStatus
QueuedAt time.Time
StartedAt time.Time
FinishedAt time.Time
}
func New() *testJobQueue {
@ -45,7 +47,7 @@ func (q *testJobQueue) Enqueue(jobType string, args interface{}, dependencies []
Id: uuid.New(),
Type: jobType,
Dependencies: uniqueUUIDList(dependencies),
Status: jobqueue.JobPending,
QueuedAt: time.Now(),
}
var err error
@ -92,7 +94,7 @@ func (q *testJobQueue) Dequeue(ctx context.Context, jobTypes []string, args inte
return uuid.Nil, err
}
j.Status = jobqueue.JobRunning
j.StartedAt = time.Now()
return j.Id, nil
}
@ -105,7 +107,7 @@ func (q *testJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
return jobqueue.ErrNotExist
}
if j.Status != jobqueue.JobRunning {
if j.StartedAt.IsZero() || !j.FinishedAt.IsZero() {
return jobqueue.ErrNotRunning
}
@ -115,7 +117,7 @@ func (q *testJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
return fmt.Errorf("error marshaling result: %v", err)
}
j.Status = jobqueue.JobFinished
j.FinishedAt = time.Now()
for _, depid := range q.dependants[id] {
dep := q.jobs[depid]
@ -132,7 +134,7 @@ func (q *testJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
return nil
}
func (q *testJobQueue) JobStatus(id uuid.UUID, result interface{}) (status jobqueue.JobStatus, queued, started, finished time.Time, err error) {
func (q *testJobQueue) JobStatus(id uuid.UUID, result interface{}) (queued, started, finished time.Time, err error) {
var j *job
j, exists := q.jobs[id]
@ -141,16 +143,16 @@ func (q *testJobQueue) JobStatus(id uuid.UUID, result interface{}) (status jobqu
return
}
if j.Status == jobqueue.JobFinished {
if !j.FinishedAt.IsZero() {
err = json.Unmarshal(j.Result, result)
if err != nil {
return
}
}
queued = time.Time{}
started = time.Time{}
finished = time.Time{}
queued = j.QueuedAt
started = j.StartedAt
finished = j.FinishedAt
return
}

48
internal/worker/server.go
View file

@ -80,25 +80,47 @@ func (s *Server) Enqueue(manifest *osbuild.Manifest, targets []*target.Target) (
func (s *Server) JobStatus(id uuid.UUID) (state common.ComposeState, queued, started, finished time.Time, err error) {
var result OSBuildJobResult
var status jobqueue.JobStatus
status, queued, started, finished, err = s.jobs.JobStatus(id, &result)
queued, started, finished, err = s.jobs.JobStatus(id, &result)
if err != nil {
return
}
state = composeStateFromJobStatus(status, result.OSBuildOutput)
if !finished.IsZero() {
if result.OSBuildOutput.Success {
state = common.CFinished
} else {
state = common.CFailed
}
} else if !started.IsZero() {
state = common.CRunning
} else {
state = common.CWaiting
}
return
}
func (s *Server) JobResult(id uuid.UUID) (common.ComposeState, *common.ComposeResult, error) {
var result OSBuildJobResult
status, _, _, _, err := s.jobs.JobStatus(id, &result)
_, started, finished, err := s.jobs.JobStatus(id, &result)
if err != nil {
return common.CWaiting, nil, err
}
return composeStateFromJobStatus(status, result.OSBuildOutput), result.OSBuildOutput, nil
state := common.CWaiting
if !finished.IsZero() {
if result.OSBuildOutput.Success {
state = common.CFinished
} else {
state = common.CFailed
}
} else if !started.IsZero() {
state = common.CRunning
}
return state, result.OSBuildOutput, nil
}
// jsonErrorf() is similar to http.Error(), but returns the message in a json
@ -216,19 +238,3 @@ func (s *Server) addJobImageHandler(writer http.ResponseWriter, request *http.Re
jsonErrorf(writer, http.StatusInternalServerError, "%v", err)
}
}
func composeStateFromJobStatus(status jobqueue.JobStatus, output *common.ComposeResult) common.ComposeState {
switch status {
case jobqueue.JobPending:
return common.CWaiting
case jobqueue.JobRunning:
return common.CRunning
case jobqueue.JobFinished:
if output.Success {
return common.CFinished
} else {
return common.CFailed
}
}
return common.CWaiting
}