FEATURE: health revamp

1. Added a 'JsonPollingSink'. This sink is an in-memory aggregator. It exposes these aggregations via HTTP.
2. Added healthd. This is a daemon that polls the JsonPollingSink's and aggregates the results.
3. Added healthtop. This is a CLI utility that queries a healthd instance and displays top jobs and hosts.

Author: cypriss

README.txt

@@ -0,0 +1,160 @@
+package health
+
+import (
+	"time"
+)
+
+type aggregator struct {
+	// How long is each aggregation interval. Eg, 1 minute
+	intervalDuration time.Duration
+
+	// Retain controls how many metrics interval we keep. Eg, 5 minutes
+	retain time.Duration
+
+	// maxIntervals is the maximum length of intervals.
+	// It is retain / interval.
+	maxIntervals int
+
+	// intervals is a slice of the retained intervals
+	intervalAggregations []*IntervalAggregation
+}
+
+func startAggregator(intervalDuration time.Duration, retain time.Duration, sink *JsonPollingSink) {
+	cmdChan := sink.cmdChan
+	doneChan := sink.doneChan
+	intervalsChanChan := sink.intervalsChanChan
+	ticker := time.Tick(1 * time.Second)
+
+	agg := newAggregator(intervalDuration, retain)
+
+	for {
+		select {
+		case <-doneChan:
+			break
+		case cmd := <-cmdChan:
+			if cmd.Kind == cmdKindEvent {
+				agg.EmitEvent(cmd.Job, cmd.Event)
+			} else if cmd.Kind == cmdKindEventErr {
+				agg.EmitEventErr(cmd.Job, cmd.Event, cmd.Err)
+			} else if cmd.Kind == cmdKindTiming {
+				agg.EmitTiming(cmd.Job, cmd.Event, cmd.Nanos)
+			} else if cmd.Kind == cmdKindComplete {
+				agg.EmitComplete(cmd.Job, cmd.Status, cmd.Nanos)
+			}
+		case <-ticker:
+			agg.getIntervalAggregation() // this has the side effect of sliding the interval window if necessary.
+		case intervalsChan := <-intervalsChanChan:
+			intervalsChan <- agg.memorySafeIntervals()
+		}
+	}
+}
+
+func newAggregator(intervalDuration time.Duration, retain time.Duration) *aggregator {
+	maxIntervals := int(retain / intervalDuration)
+	return &aggregator{
+		intervalDuration:     intervalDuration,
+		retain:               retain,
+		maxIntervals:         maxIntervals,
+		intervalAggregations: make([]*IntervalAggregation, 0, maxIntervals),
+	}
+}
+
+func (a *aggregator) memorySafeIntervals() []*IntervalAggregation {
+	ret := make([]*IntervalAggregation, 0, len(a.intervalAggregations))
+	curAgg := a.getIntervalAggregation()
+
+	for _, intAgg := range a.intervalAggregations {
+		if intAgg == curAgg {
+			ret = append(ret, intAgg.Clone())
+		} else {
+			ret = append(ret, intAgg)
+		}
+	}
+
+	return ret
+}
+
+func (a *aggregator) EmitEvent(job string, event string) {
+	intAgg := a.getIntervalAggregation()
+	intAgg.Events[event] = intAgg.Events[event] + 1
+	jobAgg := intAgg.getJobAggregation(job)
+	jobAgg.Events[event] = jobAgg.Events[event] + 1
+	intAgg.SerialNumber++
+}
+
+func (a *aggregator) EmitEventErr(job string, event string, inputErr error) {
+	intAgg := a.getIntervalAggregation()
+	errc := intAgg.getCounterErrs(event)
+	errc.incrementAndAddError(inputErr)
+	jobAgg := intAgg.getJobAggregation(job)
+	jerrc := jobAgg.getCounterErrs(event)
+	jerrc.incrementAndAddError(inputErr)
+	intAgg.SerialNumber++
+}
+
+func (a *aggregator) EmitTiming(job string, event string, nanos int64) {
+	intAgg := a.getIntervalAggregation()
+	t := intAgg.getTimers(event)
+	t.ingest(nanos)
+	jobAgg := intAgg.getJobAggregation(job)
+	jt := jobAgg.getTimers(event)
+	jt.ingest(nanos)
+	intAgg.SerialNumber++
+}
+
+func (a *aggregator) EmitComplete(job string, status CompletionStatus, nanos int64) {
+	intAgg := a.getIntervalAggregation()
+	jobAgg := intAgg.getJobAggregation(job)
+	jobAgg.ingest(status, nanos)
+	intAgg.SerialNumber++
+}
+
+func (a *aggregator) getIntervalAggregation() *IntervalAggregation {
+	intervalStart := now().Truncate(a.intervalDuration)
+
+	n := len(a.intervalAggregations)
+	if n > 0 && a.intervalAggregations[n-1].IntervalStart == intervalStart {
+		return a.intervalAggregations[n-1]
+	}
+
+	return a.createIntervalAggregation(intervalStart)
+}
+
+func (a *aggregator) createIntervalAggregation(interval time.Time) *IntervalAggregation {
+	// Make new interval:
+	current := NewIntervalAggregation(interval)
+
+	// If we've reached our max intervals, and we're going to shift everything down, then set the last one
+	n := len(a.intervalAggregations)
+	if n == a.maxIntervals {
+		for i := 1; i < n; i++ {
+			a.intervalAggregations[i-1] = a.intervalAggregations[i]
+		}
+		a.intervalAggregations[n-1] = current
+	} else {
+		a.intervalAggregations = append(a.intervalAggregations, current)
+	}
+
+	return current
+}
+
+var nowMock time.Time
+
+func now() time.Time {
+	if nowMock.IsZero() {
+		return time.Now()
+	}
+	return nowMock
+}
+
+func setNowMock(t string) {
+	var err error
+	nowMock, err = time.Parse(time.RFC3339, t)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func resetNowMock() {
+	nowMock = time.Time{}
+}