Add errors to dropped overdue jobs.

- Added AddErrors() to job interface.

scheduler/job.go

@@ -23,6 +23,7 @@ import (
 	"bytes"
 	"encoding/gob"
 	"fmt"
+	"sync"
 	"time"
 
 	log "github.com/Sirupsen/logrus"
@@ -80,6 +81,7 @@ func (j *qj) Promise() Promise {
 // the scheduler.  Job encompasses all
 // all job types -- collect, process, and publish.
 type job interface {
+	AddErrors(errs ...error)
 	Errors() []error
 	StartTime() time.Time
 	Deadline() time.Time
@@ -90,6 +92,8 @@ type job interface {
 type jobType int
 
 type coreJob struct {
+	sync.Mutex
+
 	jtype     jobType
 	deadline  time.Time
 	starttime time.Time
@@ -117,6 +121,12 @@ func (c *coreJob) Type() jobType {
 	return c.jtype
 }
 
+func (c *coreJob) AddErrors(errs ...error) {
+	c.Lock()
+	defer c.Unlock()
+	c.errors = append(c.errors, errs...)
+}
+
 func (c *coreJob) Errors() []error {
 	return c.errors
 }
@@ -202,7 +212,7 @@ func (c *collectorJob) Run() {
 				"error":    e,
 			}).Error("collector run error")
 		}
-		c.errors = errs
+		c.AddErrors(errs...)
 	}
 }
 
@@ -273,7 +283,7 @@ func (p *processJob) Run() {
 						"error":          e.Error(),
 					}).Error("error with processor job")
 				}
-				p.errors = append(p.errors, errs...)
+				p.AddErrors(errs...)
 			}
 			p.content = content
 		default:
@@ -366,7 +376,7 @@ func (p *publisherJob) Run() {
 						"error":          e.Error(),
 					}).Error("error with publisher job")
 				}
-				p.errors = append(p.errors, errs...)
+				p.AddErrors(errs...)
 			}
 		default:
 			log.WithFields(log.Fields{
@@ -397,7 +407,7 @@ func (p *publisherJob) Run() {
 						"error":          e.Error(),
 					}).Error("error with publisher job")
 				}
-				p.errors = append(p.errors, errs...)
+				p.AddErrors(errs...)
 			}
 		}
 	default:

scheduler/job_test.go

@@ -20,6 +20,7 @@ limitations under the License.
 package scheduler
 
 import (
+	"errors"
 	"testing"
 	"time"
 
@@ -75,6 +76,21 @@ func TestCollectorJob(t *testing.T) {
 			So(cj.Errors(), ShouldResemble, []error{})
 		})
 	})
+	Convey("AddErrors()", t, func() {
+		Convey("it should append errors to the job", func() {
+			cj := newCollectorJob([]core.RequestedMetric{}, defaultDeadline, &mockCollector{}, cdt)
+			So(cj.Errors(), ShouldResemble, []error{})
+
+			e1 := errors.New("1")
+			e2 := errors.New("2")
+			e3 := errors.New("3")
+
+			cj.AddErrors(e1)
+			So(cj.Errors(), ShouldResemble, []error{e1})
+			cj.AddErrors(e2, e3)
+			So(cj.Errors(), ShouldResemble, []error{e1, e2, e3})
+		})
+	})
 	Convey("Run()", t, func() {
 		Convey("it should complete without errors", func() {
 			cj := newCollectorJob([]core.RequestedMetric{}, defaultDeadline, &mockCollector{}, cdt)

scheduler/work_manager_test.go

@@ -72,6 +72,11 @@ func newMultiSyncMockJob(n int) *mockJob {
 	}
 }
 
+func (mj *mockJob) AddErrors(errs ...error) {
+	mj.Lock()
+	defer mj.Unlock()
+	mj.errors = append(mj.errors, errs...)
+}
 func (mj *mockJob) Errors() []error      { return mj.errors }
 func (mj *mockJob) StartTime() time.Time { return mj.starttime }
 func (mj *mockJob) Deadline() time.Time  { return mj.deadline }

scheduler/worker.go

@@ -20,6 +20,8 @@ limitations under the License.
 package scheduler
 
 import (
+	"errors"
+
 	"github.com/intelsdi-x/snap/pkg/chrono"
 	"github.com/pborman/uuid"
 )
@@ -48,11 +50,12 @@ func (w *worker) start() {
 			// assert that deadline is not exceeded
 			if chrono.Chrono.Now().Before(q.Job().Deadline()) {
 				q.Job().Run()
+			} else {
+				// the deadline was exceeded and this job will not run
+				q.Job().AddErrors(errors.New("Worker refused to run overdue job."))
 			}
 
-			// mark the job complete for one of two reasons:
-			// - this job was just run
-			// - the deadline was exceeded and this job will not run
+			// mark the job complete
 			q.Promise().Complete(q.Job().Errors())
 
 		// the single kill-channel -- used when resizing worker pools

scheduler/worker_test.go

@@ -57,8 +57,9 @@ func TestWorker(t *testing.T) {
 		chrono.Chrono.Forward(1500 * time.Millisecond)
 		qj := newQueuedJob(mj)
 		rcv <- qj
-		qj.Promise().Await()
-		So(mj.worked, ShouldEqual, false)
+		errors := qj.Promise().Await()
+		So(errors, ShouldNotBeEmpty)
+		So(mj.worked, ShouldBeFalse)
 	})
 	Convey("stops the worker if kamikaze chan is closed", t, func() {
 		workerKillChan = make(chan struct{})