[aggregator] Use NaN instead of math.MaxFloat64 and add sentinel guar…

…ds (#2929)

src/aggregator/aggregation/gauge.go

@@ -27,10 +27,6 @@ import (
 	"github.com/m3db/m3/src/metrics/aggregation"
 )
 
-const (
-	minFloat64 = -math.MaxFloat64
-)
-
 // Gauge aggregates gauge values.
 type Gauge struct {
 	Options
@@ -48,13 +44,18 @@ type Gauge struct {
 func NewGauge(opts Options) Gauge {
 	return Gauge{
 		Options: opts,
-		max:     minFloat64,
-		min:     math.MaxFloat64,
+		max:     math.NaN(),
+		min:     math.NaN(),
 	}
 }
 
 // Update updates the gauge value.
 func (g *Gauge) Update(timestamp time.Time, value float64) {
+	// If this is not a valid value, drop it.
+	if math.IsNaN(value) {
+		return
+	}
+
 	if g.lastAt.IsZero() || timestamp.After(g.lastAt) {
 		// NB(r): Only set the last value if this value arrives
 		// after the wall clock timestamp of previous values, not
@@ -67,10 +68,10 @@ func (g *Gauge) Update(timestamp time.Time, value float64) {
 
 	g.sum += value
 	g.count++
-	if g.max < value {
+	if math.IsNaN(g.max) || g.max < value {
 		g.max = value
 	}
-	if g.min > value {
+	if math.IsNaN(g.min) || g.min > value {
 		g.min = value
 	}
 
@@ -108,10 +109,20 @@ func (g *Gauge) Stdev() float64 {
 }
 
 // Min returns the minimum gauge value.
-func (g *Gauge) Min() float64 { return g.min }
+func (g *Gauge) Min() float64 {
+	if math.IsNaN(g.min) {
+		return 0.0
+	}
+	return g.min
+}
 
 // Max returns the maximum gauge value.
-func (g *Gauge) Max() float64 { return g.max }
+func (g *Gauge) Max() float64 {
+	if math.IsNaN(g.max) {
+		return 0.0
+	}
+	return g.max
+}
 
 // ValueOf returns the value for the aggregation type.
 func (g *Gauge) ValueOf(aggType aggregation.Type) float64 {

src/aggregator/aggregation/gauge_test.go

@@ -42,6 +42,13 @@ func TestGaugeDefaultAggregationType(t *testing.T) {
 	require.Equal(t, 100.0, g.ValueOf(aggregation.Count))
 	require.Equal(t, 50.5, g.ValueOf(aggregation.Mean))
 	require.Equal(t, 0.0, g.ValueOf(aggregation.SumSq))
+
+	g = NewGauge(NewOptions(instrument.NewOptions()))
+	require.Equal(t, 0.0, g.Last())
+	require.Equal(t, 0.0, g.ValueOf(aggregation.Last))
+	require.Equal(t, 0.0, g.ValueOf(aggregation.Count))
+	require.Equal(t, 0.0, g.ValueOf(aggregation.Mean))
+	require.Equal(t, 0.0, g.ValueOf(aggregation.SumSq))
 }
 
 func TestGaugeCustomAggregationType(t *testing.T) {
@@ -80,6 +87,33 @@ func TestGaugeCustomAggregationType(t *testing.T) {
 			require.False(t, aggType.IsValidForGauge())
 		}
 	}
+
+	g = NewGauge(opts)
+	require.Equal(t, 0.0, g.Last())
+	for aggType := range aggregation.ValidTypes {
+		v := g.ValueOf(aggType)
+		switch aggType {
+		case aggregation.Last:
+			require.Equal(t, 0.0, v)
+		case aggregation.Min:
+			require.Equal(t, 0.0, v)
+		case aggregation.Max:
+			require.Equal(t, 0.0, v)
+		case aggregation.Mean:
+			require.Equal(t, 0.0, v)
+		case aggregation.Count:
+			require.Equal(t, 0.0, v)
+		case aggregation.Sum:
+			require.Equal(t, 0.0, v)
+		case aggregation.SumSq:
+			require.Equal(t, 0.0, v)
+		case aggregation.Stdev:
+			require.InDelta(t, 0.0, v, 0.0)
+		default:
+			require.Equal(t, 0.0, v)
+			require.False(t, aggType.IsValidForGauge())
+		}
+	}
 }
 
 func TestGaugeLastOutOfOrderValues(t *testing.T) {