Implement support for NonAbsolute Measurement MaxSumCount

api/core/number.go

@@ -34,6 +34,21 @@ const (
 	Uint64NumberKind
 )
 
+// Minimum returns the minimum representable value
+// for a given NumberKind
+func (k NumberKind) Minimum() Number {
+	switch k {
+	case Int64NumberKind:
+		return NewInt64Number(math.MinInt64)
+	case Float64NumberKind:
+		return NewFloat64Number(-1. * math.MaxFloat64)
+	case Uint64NumberKind:
+		return NewUint64Number(0)
+	default:
+		return Number(0)
+	}
+}
+
 // Number represents either an integral or a floating point value. It
 // needs to be accompanied with a source of NumberKind that describes
 // the actual type of the value stored within Number.

exporter/metric/stdout/stdout.go

@@ -130,13 +130,13 @@ func (e *Exporter) Export(_ context.Context, checkpointSet export.CheckpointSet)
 				expose.Count = count
 			}
 
-			// TODO: Should tolerate ErrEmptyDataSet here,
-			// just like ErrNoLastValue below, since
-			// there's a race condition between creating
-			// the Aggregator and updating the first
-			// value.
-
 			if max, err := msc.Max(); err != nil {
+				if err == aggregator.ErrEmptyDataSet {
+					// This is a special case, indicates an aggregator that
+					// was checkpointed before its first value was set.
+					return
+				}
+
 				aggError = err
 				expose.Max = "NaN"
 			} else {

exporter/metric/stdout/stdout_test.go

@@ -162,7 +162,7 @@ func TestStdoutMaxSumCount(t *testing.T) {
 	checkpointSet := test.NewCheckpointSet(sdk.NewDefaultLabelEncoder())
 
 	desc := export.NewDescriptor("test.name", export.MeasureKind, nil, "", "", core.Float64NumberKind, false)
-	magg := maxsumcount.New()
+	magg := maxsumcount.New(desc)
 	aggtest.CheckedUpdate(fix.t, magg, core.NewFloat64Number(123.456), desc)
 	aggtest.CheckedUpdate(fix.t, magg, core.NewFloat64Number(876.543), desc)
 	magg.Checkpoint(fix.ctx, desc)
@@ -220,23 +220,30 @@ func TestStdoutMeasureFormat(t *testing.T) {
 }`, fix.Output())
 }
 
-func TestStdoutAggError(t *testing.T) {
-	fix := newFixture(t, stdout.Options{})
+func TestStdoutEmptyDataSet(t *testing.T) {
+	desc := export.NewDescriptor("test.name", export.MeasureKind, nil, "", "", core.Float64NumberKind, false)
+	for name, tc := range map[string]export.Aggregator{
+		"ddsketch":    ddsketch.New(ddsketch.NewDefaultConfig(), desc),
+		"maxsumcount": maxsumcount.New(desc),
+	} {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
 
-	checkpointSet := test.NewCheckpointSet(sdk.NewDefaultLabelEncoder())
+			fix := newFixture(t, stdout.Options{})
 
-	desc := export.NewDescriptor("test.name", export.MeasureKind, nil, "", "", core.Float64NumberKind, false)
-	magg := ddsketch.New(ddsketch.NewDefaultConfig(), desc)
-	magg.Checkpoint(fix.ctx, desc)
+			checkpointSet := test.NewCheckpointSet(sdk.NewDefaultLabelEncoder())
 
-	checkpointSet.Add(desc, magg)
+			magg := tc
+			magg.Checkpoint(fix.ctx, desc)
 
-	err := fix.exporter.Export(fix.ctx, checkpointSet)
+			checkpointSet.Add(desc, magg)
 
-	// An error is returned and NaN values are printed.
-	require.Error(t, err)
-	require.Equal(t, aggregator.ErrEmptyDataSet, err)
-	require.Equal(t, `{"updates":[{"name":"test.name","max":"NaN","sum":0,"count":0,"quantiles":[{"q":0.5,"v":"NaN"},{"q":0.9,"v":"NaN"},{"q":0.99,"v":"NaN"}]}]}`, fix.Output())
+			fix.Export(checkpointSet)
+
+			require.Equal(t, `{"updates":null}`, fix.Output())
+		})
+	}
 }
 
 func TestStdoutGaugeNotSet(t *testing.T) {

sdk/metric/aggregator/ddsketch/ddsketch.go

@@ -75,7 +75,7 @@ func (c *Aggregator) Max() (core.Number, error) {
 	return c.Quantile(1)
 }
 
-// Min returns the mininum value in the checkpoint.
+// Min returns the minimum value in the checkpoint.
 func (c *Aggregator) Min() (core.Number, error) {
 	return c.Quantile(0)
 }

sdk/metric/aggregator/maxsumcount/msc.go

@@ -28,6 +28,7 @@ type (
 	Aggregator struct {
 		current    state
 		checkpoint state
+		kind       core.NumberKind
 	}
 
 	state struct {
@@ -50,8 +51,15 @@ var _ aggregator.MaxSumCount = &Aggregator{}
 // atomic operations, which introduces the possibility that
 // checkpoints are inconsistent.  For greater consistency and lower
 // performance, consider using Array or DDSketch aggregators.
-func New() *Aggregator {
-	return &Aggregator{}
+func New(desc *export.Descriptor) *Aggregator {
+	return &Aggregator{
+		kind:    desc.NumberKind(),
+		current: unsetMaxSumCount(desc.NumberKind()),
+	}
+}
+
+func unsetMaxSumCount(kind core.NumberKind) state {
+	return state{max: kind.Minimum()}
 }
 
 // Sum returns the sum of values in the checkpoint.
@@ -65,15 +73,24 @@ func (c *Aggregator) Count() (int64, error) {
 }
 
 // Max returns the maximum value in the checkpoint.
+// The error value aggregator.ErrEmptyDataSet will be returned if
+// (due to a race condition) the checkpoint was set prior to the
+// current.max being computed in Update().
+//
+// Note: If a measure's recorded values for a given checkpoint are
+// all equal to NumberKind.Minimum(), Max() will return ErrEmptyDataSet
 func (c *Aggregator) Max() (core.Number, error) {
+	if c.checkpoint.max == c.kind.Minimum() {
+		return core.Number(0), aggregator.ErrEmptyDataSet
+	}
 	return c.checkpoint.max, nil
 }
 
 // Checkpoint saves the current state and resets the current state to
 // the empty set.  Since no locks are taken, there is a chance that
 // the independent Max, Sum, and Count are not consistent with each
 // other.
-func (c *Aggregator) Checkpoint(ctx context.Context, _ *export.Descriptor) {
+func (c *Aggregator) Checkpoint(ctx context.Context, desc *export.Descriptor) {
 	// N.B. There is no atomic operation that can update all three
 	// values at once without a memory allocation.
 	//
@@ -86,7 +103,7 @@ func (c *Aggregator) Checkpoint(ctx context.Context, _ *export.Descriptor) {
 
 	c.checkpoint.count.SetUint64(c.current.count.SwapUint64Atomic(0))
 	c.checkpoint.sum = c.current.sum.SwapNumberAtomic(core.Number(0))
-	c.checkpoint.max = c.current.max.SwapNumberAtomic(core.Number(0))
+	c.checkpoint.max = c.current.max.SwapNumberAtomic(c.kind.Minimum())
 }
 
 // Update adds the recorded measurement to the current data set.

sdk/metric/aggregator/maxsumcount/msc_test.go

@@ -16,55 +16,104 @@ package maxsumcount
 
 import (
 	"context"
+	"math"
+	"math/rand"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 
+	"go.opentelemetry.io/otel/api/core"
 	export "go.opentelemetry.io/otel/sdk/export/metric"
+	"go.opentelemetry.io/otel/sdk/export/metric/aggregator"
 	"go.opentelemetry.io/otel/sdk/metric/aggregator/test"
 )
 
 const count = 100
 
+type policy struct {
+	name     string
+	absolute bool
+	sign     func() int
+}
+
+var (
+	positiveOnly = policy{
+		name:     "absolute",
+		absolute: true,
+		sign:     func() int { return +1 },
+	}
+	negativeOnly = policy{
+		name:     "negative",
+		absolute: false,
+		sign:     func() int { return -1 },
+	}
+	positiveAndNegative = policy{
+		name:     "positiveAndNegative",
+		absolute: false,
+		sign: func() int {
+			if rand.Uint32() > math.MaxUint32/2 {
+				return -1
+			}
+			return 1
+		},
+	}
+)
+
 func TestMaxSumCountAbsolute(t *testing.T) {
-	ctx := context.Background()
+	test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
+		maxSumCount(t, profile, positiveOnly)
+	})
+}
 
+func TestMaxSumCountNegativeOnly(t *testing.T) {
 	test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
-		record := test.NewAggregatorTest(export.MeasureKind, profile.NumberKind, false)
+		maxSumCount(t, profile, negativeOnly)
+	})
+}
 
-		agg := New()
+func TestMaxSumCountPositiveAndNegative(t *testing.T) {
+	test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
+		maxSumCount(t, profile, positiveAndNegative)
+	})
+}
 
-		all := test.NewNumbers(profile.NumberKind)
+// Validates max, sum and count for a given profile and policy
+func maxSumCount(t *testing.T, profile test.Profile, policy policy) {
+	ctx := context.Background()
+	descriptor := test.NewAggregatorTest(export.MeasureKind, profile.NumberKind, !policy.absolute)
 
-		for i := 0; i < count; i++ {
-			x := profile.Random(+1)
-			all.Append(x)
-			test.CheckedUpdate(t, agg, x, record)
-		}
+	agg := New(descriptor)
 
-		agg.Checkpoint(ctx, record)
+	all := test.NewNumbers(profile.NumberKind)
 
-		all.Sort()
+	for i := 0; i < count; i++ {
+		x := profile.Random(policy.sign())
+		all.Append(x)
+		test.CheckedUpdate(t, agg, x, descriptor)
+	}
 
-		asum, err := agg.Sum()
-		require.InEpsilon(t,
-			all.Sum().CoerceToFloat64(profile.NumberKind),
-			asum.CoerceToFloat64(profile.NumberKind),
-			0.000000001,
-			"Same sum - absolute")
-		require.Nil(t, err)
+	agg.Checkpoint(ctx, descriptor)
 
-		count, err := agg.Count()
-		require.Equal(t, all.Count(), count, "Same count - absolute")
-		require.Nil(t, err)
+	all.Sort()
 
-		max, err := agg.Max()
-		require.Nil(t, err)
-		require.Equal(t,
-			all.Max(),
-			max,
-			"Same max - absolute")
-	})
+	asum, err := agg.Sum()
+	require.InEpsilon(t,
+		all.Sum().CoerceToFloat64(profile.NumberKind),
+		asum.CoerceToFloat64(profile.NumberKind),
+		0.000000001,
+		"Same sum - "+policy.name)
+	require.Nil(t, err)
+
+	count, err := agg.Count()
+	require.Equal(t, all.Count(), count, "Same count -"+policy.name)
+	require.Nil(t, err)
+
+	max, err := agg.Max()
+	require.Nil(t, err)
+	require.Equal(t,
+		all.Max(),
+		max,
+		"Same max -"+policy.name)
 }
 
 func TestMaxSumCountMerge(t *testing.T) {
@@ -73,8 +122,8 @@ func TestMaxSumCountMerge(t *testing.T) {
 	test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
 		descriptor := test.NewAggregatorTest(export.MeasureKind, profile.NumberKind, false)
 
-		agg1 := New()
-		agg2 := New()
+		agg1 := New(descriptor)
+		agg2 := New(descriptor)
 
 		all := test.NewNumbers(profile.NumberKind)
 
@@ -116,3 +165,26 @@ func TestMaxSumCountMerge(t *testing.T) {
 			"Same max - absolute")
 	})
 }
+
+func TestMaxSumCountNotSet(t *testing.T) {
+	ctx := context.Background()
+
+	test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
+		descriptor := test.NewAggregatorTest(export.MeasureKind, profile.NumberKind, false)
+
+		agg := New(descriptor)
+		agg.Checkpoint(ctx, descriptor)
+
+		asum, err := agg.Sum()
+		require.Equal(t, core.Number(0), asum, "Empty checkpoint sum = 0")
+		require.Nil(t, err)
+
+		count, err := agg.Count()
+		require.Equal(t, int64(0), count, "Empty checkpoint count = 0")
+		require.Nil(t, err)
+
+		max, err := agg.Max()
+		require.Equal(t, aggregator.ErrEmptyDataSet, err)
+		require.Equal(t, core.Number(0), max)
+	})
+}

sdk/metric/benchmark_test.go

@@ -54,7 +54,7 @@ func (*benchFixture) AggregatorFor(descriptor *export.Descriptor) export.Aggrega
 		return gauge.New()
 	case export.MeasureKind:
 		if strings.HasSuffix(descriptor.Name(), "maxsumcount") {
-			return maxsumcount.New()
+			return maxsumcount.New(descriptor)
 		} else if strings.HasSuffix(descriptor.Name(), "ddsketch") {
 			return ddsketch.New(ddsketch.NewDefaultConfig(), descriptor)
 		} else if strings.HasSuffix(descriptor.Name(), "array") {

sdk/metric/selector/simple/simple.go

@@ -71,7 +71,7 @@ func (selectorInexpensive) AggregatorFor(descriptor *export.Descriptor) export.A
 	case export.GaugeKind:
 		return gauge.New()
 	case export.MeasureKind:
-		return maxsumcount.New()
+		return maxsumcount.New(descriptor)
 	default:
 		return counter.New()
 	}