histogram aggregator draft

sdk/export/metric/aggregator/aggregator.go

@@ -63,6 +63,18 @@ type (
 		Points() ([]core.Number, error)
 	}
 
+	// Quantile returns an exact or estimated quantile over the
+	// set of values that were aggregated.
+	Histogram interface {
+		Buckets() (HistogramValue, error)
+	}
+
+	HistogramValue struct {
+		Buckets []core.Number
+		Count   core.Number
+		Sum     core.Number
+	}
+
 	// MinMaxSumCount supports the Min, Max, Sum, and Count interfaces.
 	MinMaxSumCount interface {
 		Min

sdk/metric/aggregator/histogram/histogram.go

@@ -0,0 +1,123 @@
+// Copyright 2019, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package histogram // import "go.opentelemetry.io/otel/sdk/metric/aggregator/histogram"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/api/core"
+	export "go.opentelemetry.io/otel/sdk/export/metric"
+	"go.opentelemetry.io/otel/sdk/export/metric/aggregator"
+)
+
+type (
+	// Aggregator aggregates measure events, keeping only the le,
+	// sum, and count.
+	Aggregator struct {
+		current    aggregator.HistogramValue
+		checkpoint aggregator.HistogramValue
+		bounds     []float64
+		kind       core.NumberKind
+	}
+)
+
+var _ export.Aggregator = &Aggregator{}
+var _ aggregator.Sum = &Aggregator{}
+var _ aggregator.Count = &Aggregator{}
+var _ aggregator.Histogram = &Aggregator{}
+
+// New returns a new measure aggregator for computing count, sum and buckets.
+//
+// Note that this aggregator maintains each value using independent
+// atomic operations, which introduces the possibility that
+// checkpoints are inconsistent.
+func New(desc *export.Descriptor, bounds []float64) *Aggregator {
+	return &Aggregator{
+		kind: desc.NumberKind(),
+		current: aggregator.HistogramValue{
+			Buckets: make([]core.Number, len(bounds)+1),
+		},
+		bounds: bounds,
+	}
+}
+
+// Count returns the number of values in the checkpoint.
+func (c *Aggregator) Sum() (core.Number, error) {
+	return c.checkpoint.Sum, nil
+}
+
+// Count returns the number of values in the checkpoint.
+func (c *Aggregator) Count() (int64, error) {
+	return int64(c.checkpoint.Count.AsUint64()), nil
+}
+
+func (c *Aggregator) Buckets() (aggregator.HistogramValue, error) {
+	return c.checkpoint, nil
+}
+
+// Checkpoint saves the current bucket and resets the current bucket to
+// the empty set.  Since no locks are taken, there is a chance that
+// the independent Min, Max, Sum, and Count are not consistent with each
+// other.
+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.
+	//
+	// This aggregator is intended to trade this correctness for
+	// speed.
+	//
+	// Therefore, atomically swap fields independently, knowing
+	// that individually the three parts of this aggregation could
+	// be spread across multiple collections in rare cases.
+
+	c.checkpoint.Count.SetUint64(c.current.Count.SwapUint64Atomic(0))
+	c.checkpoint.Sum = c.current.Sum.SwapNumberAtomic(core.Number(0))
+	c.checkpoint.Buckets = c.current.Buckets
+	c.current.Buckets = make([]core.Number, len(c.bounds)+1)
+}
+
+// Update adds the recorded measurement to the current data set.
+func (c *Aggregator) Update(_ context.Context, number core.Number, desc *export.Descriptor) error {
+	kind := desc.NumberKind()
+
+	c.current.Count.AddUint64Atomic(1)
+	c.current.Sum.AddNumberAtomic(kind, number)
+
+	for i, boundary := range c.bounds {
+		if number.CoerceToFloat64(kind) <= boundary {
+			c.current.Buckets[i].AddUint64Atomic(1)
+			return nil
+		}
+	}
+
+	c.current.Buckets[len(c.bounds)].AddUint64Atomic(1)
+	return nil
+}
+
+// Merge combines two data sets into one.
+func (c *Aggregator) Merge(oa export.Aggregator, desc *export.Descriptor) error {
+	o, _ := oa.(*Aggregator)
+	if o == nil {
+		return aggregator.NewInconsistentMergeError(c, oa)
+	}
+
+	c.checkpoint.Sum.AddNumber(desc.NumberKind(), o.checkpoint.Sum)
+	c.checkpoint.Count.AddNumber(core.Uint64NumberKind, o.checkpoint.Count)
+
+	for i := 0; i < len(c.current.Buckets); i++ {
+		c.checkpoint.Buckets[i].AddNumber(desc.NumberKind(), o.checkpoint.Buckets[i])
+	}
+	return nil
+}

sdk/metric/aggregator/histogram/histogram_test.go

@@ -0,0 +1,180 @@
+// Copyright 2019, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package histogram
+
+import (
+	"context"
+	"fmt"
+	"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/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 TestHistogramAbsolute(t *testing.T) {
+	test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
+		histogram(t, profile, positiveOnly)
+	})
+}
+
+func TestHistogramNegativeOnly(t *testing.T) {
+	test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
+		histogram(t, profile, negativeOnly)
+	})
+}
+
+func TestHistogramPositiveAndNegative(t *testing.T) {
+	test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
+		histogram(t, profile, positiveAndNegative)
+	})
+}
+
+// Validates count, sum and buckets for a given profile and policy
+func histogram(t *testing.T, profile test.Profile, policy policy) {
+	ctx := context.Background()
+	descriptor := test.NewAggregatorTest(export.MeasureKind, profile.NumberKind, !policy.absolute)
+
+	agg := New(descriptor, []float64{250, 500, 700})
+
+	all := test.NewNumbers(profile.NumberKind)
+
+	for i := 0; i < count; i++ {
+		x := profile.Random(policy.sign())
+		all.Append(x)
+		test.CheckedUpdate(t, agg, x, descriptor)
+	}
+
+	agg.Checkpoint(ctx, descriptor)
+
+	all.Sort()
+
+	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)
+
+	for _, p := range all.Points() {
+		fmt.Print(p.Emit(profile.NumberKind), " ")
+	}
+	fmt.Println()
+	fmt.Println(agg.checkpoint)
+}
+
+func TestHistogramMerge(t *testing.T) {
+	ctx := context.Background()
+
+	test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
+		descriptor := test.NewAggregatorTest(export.MeasureKind, profile.NumberKind, false)
+
+		agg1 := New(descriptor, []float64{250, 500, 700})
+		agg2 := New(descriptor, []float64{250, 500, 700})
+
+		all := test.NewNumbers(profile.NumberKind)
+
+		for i := 0; i < count; i++ {
+			x := profile.Random(+1)
+			all.Append(x)
+			test.CheckedUpdate(t, agg1, x, descriptor)
+		}
+		for i := 0; i < count; i++ {
+			x := profile.Random(+1)
+			all.Append(x)
+			test.CheckedUpdate(t, agg2, x, descriptor)
+		}
+
+		agg1.Checkpoint(ctx, descriptor)
+		agg2.Checkpoint(ctx, descriptor)
+
+		test.CheckedMerge(t, agg1, agg2, descriptor)
+
+		all.Sort()
+
+		asum, err := agg1.Sum()
+		require.InEpsilon(t,
+			all.Sum().CoerceToFloat64(profile.NumberKind),
+			asum.CoerceToFloat64(profile.NumberKind),
+			0.000000001,
+			"Same sum - absolute")
+		require.Nil(t, err)
+
+		count, err := agg1.Count()
+		require.Equal(t, all.Count(), count, "Same count - absolute")
+		require.Nil(t, err)
+
+	})
+}
+
+func TestHistogramNotSet(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, []float64{250, 500, 700})
+		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)
+
+	})
+}