FEATURE: gauges

Gauges let you measure the current value of a number that can go up or down over time. They typically have last-win semantics in terms of multiple values being submitted in periods. One common use-case for gauges is periodically sampling the current amount of memory used or the number of active goroutines.

Author: cypriss

README.md

@@ -97,9 +97,9 @@ There are five types of completion statuses:
 * **ValidationError** - Your code was fine, but the user passed in bad inputs, and so the job wasn't completed successfully.
 * **Junk** - The job wasn't completed successfully, but not really because of an Error or ValidationError. For instance, maybe there's just a 404 (not found) or 401 (unauthorized) request to your app. This status code might not apply to all apps.
 
-### Events, Timings, and Errors
+### Events, Timings, Gauges, and Errors
 
-Within jobs, you can emit events, timings, and errors. The first argument of each of these methods is supposed to be a *key*. Camel case with dots is good because it works with other metrics stores like StatsD. Each method has a basic version as well as a version that accepts keys/values.
+Within jobs, you can emit events, timings, gauges, and errors. The first argument of each of these methods is supposed to be a *key*. Camel case with dots is good because it works with other metrics stores like StatsD. Each method has a basic version as well as a version that accepts keys/values.
 
 #### Events
 
@@ -140,7 +140,25 @@ job.TimingKv("fetch_user", time.Since(startTime).Nanoseconds(),
 ```
 
 * For the StatsD sink, we'll send it to StatsD as a timing.
-* The JSON polling will compute a summary of your timings: min, max, avg, stddev, count, sum.
+* The JSON polling sink will compute a summary of your timings: min, max, avg, stddev, count, sum.
+
+#### Gauges
+
+```go
+// Gauges:
+job.Gauge("num_goroutines", numRunningGoroutines()) 
+
+// Timings also support keys/values:
+job.GaugeKv("num_goroutines", numRunningGoroutines(),
+	health.Kvs{"dispatcher": dispatcherStatus()})
+```
+
+* For the WriterSink, a timing is just like logging to a file:
+```
+[2014-12-17T20:36:24.136663759Z]: job:/api/v2/user_stories event:num_goroutines gauge:17 kvs:[request-id:F8a8bQOWmRpO6ky dispatcher:running]
+```
+
+* For the StatsD sink, we'll send it to StatsD as a gauge.
 
 #### Errors
 
@@ -228,6 +246,7 @@ type Sink interface {
 	EmitEvent(job string, event string, kvs map[string]string)
 	EmitEventErr(job string, event string, err error, kvs map[string]string)
 	EmitTiming(job string, event string, nanoseconds int64, kvs map[string]string)
+	EmitGauge(job string, event string, value float64, kvs map[string]string)
 	EmitComplete(job string, status CompletionStatus, nanoseconds int64, kvs map[string]string)
 }
 ```
@@ -361,7 +380,6 @@ Request for contributions:
 
 health core:
 
-* A gauge metric type
 * A way to do fine-grained histograms with variable binning.
 
 healthd & healthtop

aggregator.go

@@ -40,6 +40,8 @@ AGGREGATE_LOOP:
 				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 == cmdKindGauge {
+				agg.EmitGauge(cmd.Job, cmd.Event, cmd.Value)
 			} else if cmd.Kind == cmdKindComplete {
 				agg.EmitComplete(cmd.Job, cmd.Status, cmd.Nanos)
 			}
@@ -104,6 +106,14 @@ func (a *aggregator) EmitTiming(job string, event string, nanos int64) {
 	intAgg.SerialNumber++
 }
 
+func (a *aggregator) EmitGauge(job string, event string, value float64) {
+	intAgg := a.getIntervalAggregation()
+	intAgg.Gauges[event] = value
+	jobAgg := intAgg.getJobAggregation(job)
+	jobAgg.Gauges[event] = value
+	intAgg.SerialNumber++
+}
+
 func (a *aggregator) EmitComplete(job string, status CompletionStatus, nanos int64) {
 	intAgg := a.getIntervalAggregation()
 	jobAgg := intAgg.getJobAggregation(job)

aggregator_test.go

@@ -164,6 +164,43 @@ func TestEmitTiming(t *testing.T) {
 	assert.EqualValues(t, 9, tAgg.NanosMax)
 }
 
+func TestEmitGauge(t *testing.T) {
+	setNowMock("2011-09-09T23:36:13Z")
+	defer resetNowMock()
+	a := newAggregator(time.Minute, time.Minute*5)
+	a.EmitGauge("foo", "bar", 100)
+
+	assert.Equal(t, 1, len(a.intervalAggregations))
+
+	intAgg := a.intervalAggregations[0]
+	assert.NotNil(t, intAgg.Gauges)
+	assert.EqualValues(t, 1, intAgg.SerialNumber)
+	v, ok := intAgg.Gauges["bar"]
+	assert.True(t, ok)
+	assert.Equal(t, 100.0, v)
+
+	assert.NotNil(t, intAgg.Jobs)
+	jobAgg := intAgg.Jobs["foo"]
+	assert.NotNil(t, jobAgg)
+	assert.NotNil(t, jobAgg.Gauges)
+	v, ok = intAgg.Gauges["bar"]
+	assert.True(t, ok)
+	assert.Equal(t, 100.0, v)
+
+	// Another gauge:
+	a.EmitGauge("baz", "bar", 3.14) // note: diff job
+
+	intAgg = a.intervalAggregations[0]
+	v, ok = intAgg.Gauges["bar"]
+	assert.True(t, ok)
+	assert.Equal(t, 3.14, v)
+
+	jobAgg = intAgg.Jobs["baz"]
+	v, ok = intAgg.Gauges["bar"]
+	assert.True(t, ok)
+	assert.Equal(t, 3.14, v)
+}
+
 func TestEmitComplete(t *testing.T) {
 	setNowMock("2011-09-09T23:36:13Z")
 	defer resetNowMock()

error_test.go

@@ -32,7 +32,8 @@ func (s *testSink) EmitEventErr(job string, event string, inputErr error, kvs ma
 		last.WasRaw = true
 	}
 }
-func (s *testSink) EmitTiming(job string, event string, nanos int64, kvs map[string]string) {}
+func (s *testSink) EmitTiming(job string, event string, nanos int64, kvs map[string]string)  {}
+func (s *testSink) EmitGauge(job string, event string, value float64, kvs map[string]string) {}
 func (s *testSink) EmitComplete(job string, status CompletionStatus, nanos int64, kvs map[string]string) {
 }
 

health.go

@@ -15,6 +15,8 @@ type EventReceiver interface {
 	EventErrKv(eventName string, err error, kvs map[string]string) error
 	Timing(eventName string, nanoseconds int64)
 	TimingKv(eventName string, nanoseconds int64, kvs map[string]string)
+	Gauge(eventName, value float64)
+	GaugeKv(eventName, value float64, kvs map[string]string)
 }
 
 type Stream struct {
@@ -57,6 +59,7 @@ type Sink interface {
 	EmitEventErr(job string, event string, err error, kvs map[string]string)
 	EmitTiming(job string, event string, nanoseconds int64, kvs map[string]string)
 	EmitComplete(job string, status CompletionStatus, nanoseconds int64, kvs map[string]string)
+	EmitGauge(job string, event string, value float64, kvs map[string]string)
 }
 
 func NewStream() *Stream {
@@ -146,6 +149,20 @@ func (j *Job) TimingKv(eventName string, nanoseconds int64, kvs map[string]strin
 	}
 }
 
+func (j *Job) Gauge(eventName string, value float64) {
+	allKvs := j.mergedKeyValues(nil)
+	for _, sink := range j.Stream.Sinks {
+		sink.EmitGauge(j.JobName, eventName, value, allKvs)
+	}
+}
+
+func (j *Job) GaugeKv(eventName string, value float64, kvs map[string]string) {
+	allKvs := j.mergedKeyValues(kvs)
+	for _, sink := range j.Stream.Sinks {
+		sink.EmitGauge(j.JobName, eventName, value, allKvs)
+	}
+}
+
 func (j *Job) Complete(status CompletionStatus) {
 	allKvs := j.mergedKeyValues(nil)
 	for _, sink := range j.Stream.Sinks {

interval_aggregation.go

@@ -24,6 +24,7 @@ type IntervalAggregation struct {
 
 type aggregationMaps struct {
 	Timers    map[string]*TimerAggregation `json:"timers"`
+	Gauges    map[string]float64           `json:"gauges"`
 	Events    map[string]int64             `json:"events"`
 	EventErrs map[string]*ErrorCounter     `json:"event_errs"`
 }
@@ -67,6 +68,7 @@ func NewIntervalAggregation(intervalStart time.Time) *IntervalAggregation {
 
 func (am *aggregationMaps) initAggregationMaps() {
 	am.Timers = make(map[string]*TimerAggregation)
+	am.Gauges = make(map[string]float64)
 	am.Events = make(map[string]int64)
 	am.EventErrs = make(map[string]*ErrorCounter)
 }

interval_aggregation_clone.go

@@ -24,6 +24,10 @@ func (am *aggregationMaps) Clone() *aggregationMaps {
 		dup.Events[k] = v
 	}
 
+	for k, v := range am.Gauges {
+		dup.Gauges[k] = v
+	}
+
 	for k, v := range am.Timers {
 		dup.Timers[k] = v.Clone()
 	}

interval_aggregation_clone_test.go

@@ -27,6 +27,7 @@ func TestClone(t *testing.T) {
 	a.EmitEvent("foo", "bar")
 	a.EmitTiming("foo", "bar", 100)
 	a.EmitEventErr("foo", "bar", fmt.Errorf("hi"))
+	a.EmitGauge("foo", "bar", 3.14)
 	a.EmitComplete("foo", Error, 99)
 
 	assert.Equal(t, 301, len(intAgg.Jobs))
@@ -51,11 +52,15 @@ func assertAggregationData(t *testing.T, intAgg *IntervalAggregation) {
 	assert.Equal(t, 300, len(intAgg.Jobs))
 	assert.Equal(t, 1200, len(intAgg.Events))
 	assert.Equal(t, 1200, len(intAgg.Timers))
+	assert.Equal(t, 1200, len(intAgg.Gauges))
 	assert.Equal(t, 1200, len(intAgg.EventErrs))
 
 	// Spot-check events:
 	assert.EqualValues(t, 1, intAgg.Events["event0"])
 
+	// Spot check gauges:
+	assert.EqualValues(t, 3.14, intAgg.Gauges["gauge0"])
+
 	// Spot-check timings:
 	assert.EqualValues(t, 1, intAgg.Timers["timing0"].Count)
 	assert.EqualValues(t, 12, intAgg.Timers["timing0"].NanosSum)
@@ -70,6 +75,8 @@ func assertAggregationData(t *testing.T, intAgg *IntervalAggregation) {
 	assert.EqualValues(t, 0, job.CountError)
 	assert.EqualValues(t, 1, job.Events["event0"])
 	assert.EqualValues(t, 0, job.Events["event4"])
+	assert.EqualValues(t, 3.14, job.Gauges["gauge0"])
+	assert.EqualValues(t, 0.0, job.Gauges["gauge4"])
 	assert.EqualValues(t, 1, job.Timers["timing0"].Count)
 	assert.EqualValues(t, 12, job.Timers["timing0"].NanosSum)
 	assert.EqualValues(t, 1, job.EventErrs["err0"].Count)
@@ -90,6 +97,7 @@ func aggregatorWithData() *aggregator {
 	// We want 300 jobs
 	// Each job will have 5 events, but we want 1200 events total
 	// Each job will have 5 timers, but we want 1200 timers total
+	// Each job will have 5 gauges, but we want 1200 gauges total
 	// Each job will have 5 errs, but we want 1200 errs total
 	// Given this 300/1200 dichotomy,
 	//  - the first job will have 4 events, the next job 4 events, etc.
@@ -109,6 +117,11 @@ func aggregatorWithData() *aggregator {
 		timings = append(timings, fmt.Sprintf("timing%d", i))
 	}
 
+	gauges := []string{}
+	for i := 0; i < 1200; i++ {
+		gauges = append(gauges, fmt.Sprintf("gauge%d", i))
+	}
+
 	eventErrs := []eventErr{}
 	for i := 0; i < 1200; i++ {
 		eventErrs = append(eventErrs, eventErr{
@@ -141,6 +154,14 @@ func aggregatorWithData() *aggregator {
 		}
 	}
 
+	cur = 0
+	for _, j := range jobs {
+		for i := 0; i < 4; i++ {
+			a.EmitGauge(j, gauges[cur], 3.14)
+			cur++
+		}
+	}
+
 	for _, j := range jobs {
 		a.EmitComplete(j, Success, 12)
 	}

interval_aggregation_merge.go

@@ -43,6 +43,10 @@ func (intoAm *aggregationMaps) merge(fromAm *aggregationMaps) {
 		intoAm.Events[k] += v
 	}
 
+	for k, v := range fromAm.Gauges {
+		intoAm.Gauges[k] = v
+	}
+
 	for k, v := range fromAm.Timers {
 		if existingTimer, ok := intoAm.Timers[k]; ok {
 			existingTimer.merge(v)

interval_aggregation_merge_test.go

@@ -30,17 +30,25 @@ func TestMerge(t *testing.T) {
 	a2 := aggregatorWithData()
 	intAgg2 := a2.intervalAggregations[0]
 
+	// Modify a gauge:
+	a2.EmitGauge("job0", "gauge1", 5.5)
+
 	intAgg.Merge(intAgg2)
 
 	// same number of events:
 	assert.Equal(t, 300, len(intAgg.Jobs))
 	assert.Equal(t, 1200, len(intAgg.Events))
 	assert.Equal(t, 1200, len(intAgg.Timers))
+	assert.Equal(t, 1200, len(intAgg.Gauges))
 	assert.Equal(t, 1200, len(intAgg.EventErrs))
 
 	// Spot-check events:
 	assert.EqualValues(t, 2, intAgg.Events["event0"])
 
+	// Spot-check gauges:
+	assert.EqualValues(t, 3.14, intAgg.Gauges["gauge0"])
+	assert.EqualValues(t, 5.5, intAgg.Gauges["gauge1"]) // 5.5 takes precedence over 3.14 (argument to merge takes precedence.)
+
 	// Spot-check timings:
 	assert.EqualValues(t, 2, intAgg.Timers["timing0"].Count)
 	assert.EqualValues(t, 24, intAgg.Timers["timing0"].NanosSum)
@@ -55,6 +63,7 @@ func TestMerge(t *testing.T) {
 	assert.EqualValues(t, 0, job.CountError)
 	assert.EqualValues(t, 2, job.Events["event0"])
 	assert.EqualValues(t, 0, job.Events["event4"])
+	assert.EqualValues(t, 3.14, job.Gauges["gauge0"])
 	assert.EqualValues(t, 2, job.Timers["timing0"].Count)
 	assert.EqualValues(t, 24, job.Timers["timing0"].NanosSum)
 	assert.EqualValues(t, 2, job.EventErrs["err0"].Count)