Port AMQP job state and timestamp changes to HTTP job
@@ -26,6 +26,8 @@ type httpJob struct {
startAttributes *backend.StartAttributes
received time.Time
started time.Time
+ finished time.Time
+ stateCount uint
jobBoardURL *url.URL
site string
@@ -50,8 +52,15 @@ type httpJobPayload struct {
type httpJobStateUpdate struct {
CurrentState string `json:"cur"`
NewState string `json:"new"`
- ReceivedAt time.Time `json:"received,omitempty"`
- StartedAt time.Time `json:"started,omitempty"`
+ Queued *time.Time `json:"queued,omitempty"`
+ Received time.Time `json:"received,omitempty"`
+ Started time.Time `json:"started,omitempty"`
+ Finished time.Time `json:"finished,omitempty"`
+ Meta *httpJobStateUpdateMeta `json:"meta,omitempty"`
+}
+
+type httpJobStateUpdateMeta struct {
+ StateUpdateCount uint `json:"state_update_count,omitempty"`
}
func (j *httpJob) GoString() string {
@@ -90,12 +99,10 @@ func (j *httpJob) Requeue(ctx gocontext.Context) error {
metrics.Mark("worker.job.requeue")
- currentState := j.currentState()
-
j.received = time.Time{}
j.started = time.Time{}
- return j.sendStateUpdate(currentState, "created")
+ return j.sendStateUpdate(j.currentState(), "created")
}
func (j *httpJob) Received() error {
@@ -162,15 +169,13 @@ func (j *httpJob) Finish(ctx gocontext.Context, state FinishState) error {
return errors.Errorf("job board job delete request errored with status %d: %s", resp.StatusCode, errorResp.Error)
}
- finishedAt := time.Now()
- receivedAt := j.received
- if receivedAt.IsZero() {
- receivedAt = finishedAt
+ j.finished = time.Now()
+ if j.received.IsZero() {
+ j.received = j.finished
}
- startedAt := j.started
- if startedAt.IsZero() {
- startedAt = finishedAt
+ if j.started.IsZero() {
+ j.started = j.finished
}
return j.sendStateUpdate(j.currentState(), string(state))
@@ -198,12 +203,18 @@ func (j *httpJob) Generate(ctx gocontext.Context, job Job) ([]byte, error) {
return script, nil
}
-func (j *httpJob) sendStateUpdate(currentState, newState string) error {
+func (j *httpJob) sendStateUpdate(curState, newState string) error {
+ j.stateCount++
payload := &httpJobStateUpdate{
- CurrentState: currentState,
+ CurrentState: curState,
NewState: newState,
- ReceivedAt: j.received,
- StartedAt: j.started,
+ Queued: j.Payload().Job.QueuedAt,
+ Received: j.received,
+ Started: j.started,
+ Finished: j.finished,
+ Meta: &httpJobStateUpdateMeta{
+ StateUpdateCount: j.stateCount,
+ },
}
encodedPayload, err := json.Marshal(payload)
