Report job error when worker fails to unmarshal job config
@@ -176,6 +176,7 @@ func (q *AMQPJobQueue) Jobs(ctx gocontext.Context) (outChan <-chan Job, err erro
startAttributes: &backend.StartAttributes{},
stateUpdatePool: q.stateUpdatePool,
withLogSharding: q.withLogSharding,
+ logWriterChan: logWriterChannel,
}
startAttrs := &jobPayloadStartAttrs{Config: &backend.StartAttributes{}}
@@ -193,11 +194,16 @@ func (q *AMQPJobQueue) Jobs(ctx gocontext.Context) (outChan <-chan Job, err erro
err = json.Unmarshal(delivery.Body, &startAttrs)
if err != nil {
- logger.WithField("err", err).Error("start attributes JSON parse error, attempting to ack+drop delivery")
+ logger.WithField("err", err).
+ WithField("json", string(delivery.Body)).
+ Error("start attributes JSON parse error, attempting to ack+drop delivery")
+
err := delivery.Ack(false)
if err != nil {
logger.WithField("err", err).WithField("delivery", delivery).Error("couldn't ack+drop delivery")
}
+
+ buildJob.Error(ctx, "An error occurred while parsing the job config.")
continue
}
@@ -217,7 +223,6 @@ func (q *AMQPJobQueue) Jobs(ctx gocontext.Context) (outChan <-chan Job, err erro
buildJob.startAttributes.Warmer = buildJob.payload.Warmer
buildJob.startAttributes.SetDefaults(q.DefaultLanguage, q.DefaultDist, q.DefaultArch, q.DefaultGroup, q.DefaultOS, VMTypeDefault, VMConfigDefault)
buildJob.conn = q.conn
- buildJob.logWriterChan = logWriterChannel
buildJob.delivery = delivery
buildJob.stateCount = buildJob.payload.Meta.StateUpdateCount
