1
0
mirror of https://github.com/open-telemetry/opentelemetry-go.git synced 2025-03-05 15:05:51 +02:00

Add Min() interface - rename MaxSumCount to MinMaxSumCount ()

* Add Min() interface - rename MaxSumCount to MinMaxSumCount

Fixes https://github.com/open-telemetry/opentelemetry-go/issues/319

* update stdout exporter to collect and output the minimum value
* update min and max atomically in Aggregator Update
* changed all references to maxsumcount to minmaxsumcount

* Address PR comments
This commit is contained in:
ET 2019-11-26 14:07:58 -08:00 committed by Joshua MacDonald
parent 5ec1f5c643
commit fef504d469
11 changed files with 126 additions and 50 deletions
api/core
exporter/metric/stdout
sdk
export/metric/aggregator
metric
aggregator
array
ddsketch
minmaxsumcount
benchmark_test.go
selector/simple

@ -49,6 +49,21 @@ func (k NumberKind) Minimum() Number {
}
}
// Maximum returns the maximum representable value
// for a given NumberKind
func (k NumberKind) Maximum() Number {
switch k {
case Int64NumberKind:
return NewInt64Number(math.MaxInt64)
case Float64NumberKind:
return NewFloat64Number(math.MaxFloat64)
case Uint64NumberKind:
return NewUint64Number(math.MaxUint64)
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.

@ -63,6 +63,7 @@ type expoBatch struct {
type expoLine struct {
Name string `json:"name"`
Min interface{} `json:"min,omitempty"`
Max interface{} `json:"max,omitempty"`
Sum interface{} `json:"sum,omitempty"`
Count interface{} `json:"count,omitempty"`
@ -122,15 +123,15 @@ func (e *Exporter) Export(_ context.Context, checkpointSet export.CheckpointSet)
}
}
if msc, ok := agg.(aggregator.MaxSumCount); ok {
if count, err := msc.Count(); err != nil {
if mmsc, ok := agg.(aggregator.MinMaxSumCount); ok {
if count, err := mmsc.Count(); err != nil {
aggError = err
expose.Count = "NaN"
} else {
expose.Count = count
}
if max, err := msc.Max(); err != nil {
if max, err := mmsc.Max(); err != nil {
if err == aggregator.ErrEmptyDataSet {
// This is a special case, indicates an aggregator that
// was checkpointed before its first value was set.
@ -143,6 +144,19 @@ func (e *Exporter) Export(_ context.Context, checkpointSet export.CheckpointSet)
expose.Max = max.AsInterface(kind)
}
if min, err := mmsc.Min(); 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.Min = "NaN"
} else {
expose.Min = min.AsInterface(kind)
}
if dist, ok := agg.(aggregator.Distribution); ok && len(e.options.Quantiles) != 0 {
summary := make([]expoQuantile, len(e.options.Quantiles))
expose.Quantiles = summary

@ -21,7 +21,7 @@ import (
"go.opentelemetry.io/otel/sdk/metric/aggregator/counter"
"go.opentelemetry.io/otel/sdk/metric/aggregator/ddsketch"
"go.opentelemetry.io/otel/sdk/metric/aggregator/gauge"
"go.opentelemetry.io/otel/sdk/metric/aggregator/maxsumcount"
"go.opentelemetry.io/otel/sdk/metric/aggregator/minmaxsumcount"
aggtest "go.opentelemetry.io/otel/sdk/metric/aggregator/test"
)
@ -156,13 +156,13 @@ func TestStdoutGaugeFormat(t *testing.T) {
require.Equal(t, `{"updates":[{"name":"test.name{A=B,C=D}","last":123.456}]}`, fix.Output())
}
func TestStdoutMaxSumCount(t *testing.T) {
func TestStdoutMinMaxSumCount(t *testing.T) {
fix := newFixture(t, stdout.Options{})
checkpointSet := test.NewCheckpointSet(sdk.NewDefaultLabelEncoder())
desc := export.NewDescriptor("test.name", export.MeasureKind, nil, "", "", core.Float64NumberKind, false)
magg := maxsumcount.New(desc)
magg := minmaxsumcount.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)
@ -171,7 +171,7 @@ func TestStdoutMaxSumCount(t *testing.T) {
fix.Export(checkpointSet)
require.Equal(t, `{"updates":[{"name":"test.name{A=B,C=D}","max":876.543,"sum":999.999,"count":2}]}`, fix.Output())
require.Equal(t, `{"updates":[{"name":"test.name{A=B,C=D}","min":123.456,"max":876.543,"sum":999.999,"count":2}]}`, fix.Output())
}
func TestStdoutMeasureFormat(t *testing.T) {
@ -198,6 +198,7 @@ func TestStdoutMeasureFormat(t *testing.T) {
"updates": [
{
"name": "test.name{A=B,C=D}",
"min": 0.5,
"max": 999.5,
"sum": 500000,
"count": 1000,
@ -223,8 +224,8 @@ func TestStdoutMeasureFormat(t *testing.T) {
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),
"ddsketch": ddsketch.New(ddsketch.NewDefaultConfig(), desc),
"minmaxsumcount": minmaxsumcount.New(desc),
} {
tc := tc
t.Run(name, func(t *testing.T) {

@ -37,6 +37,11 @@ type (
Count() (int64, error)
}
// Min returns the minimum value over the set of values that were aggregated.
Min interface {
Min() (core.Number, error)
}
// Max returns the maximum value over the set of values that were aggregated.
Max interface {
Max() (core.Number, error)
@ -58,17 +63,18 @@ type (
Points() ([]core.Number, error)
}
// MaxSumCount supports the Max, Sum, and Count interfaces.
MaxSumCount interface {
// MinMaxSumCount supports the Min, Max, Sum, and Count interfaces.
MinMaxSumCount interface {
Min
Max
Sum
Count
Max
}
// MaxSumCount supports the Max, Sum, Count, and Quantile
// Distribution supports the Min, Max, Sum, Count, and Quantile
// interfaces.
Distribution interface {
MaxSumCount
MinMaxSumCount
Quantile
}
)

@ -38,7 +38,7 @@ type (
)
var _ export.Aggregator = &Aggregator{}
var _ aggregator.MaxSumCount = &Aggregator{}
var _ aggregator.MinMaxSumCount = &Aggregator{}
var _ aggregator.Distribution = &Aggregator{}
var _ aggregator.Points = &Aggregator{}

@ -39,7 +39,7 @@ type Aggregator struct {
}
var _ export.Aggregator = &Aggregator{}
var _ aggregator.MaxSumCount = &Aggregator{}
var _ aggregator.MinMaxSumCount = &Aggregator{}
var _ aggregator.Distribution = &Aggregator{}
// New returns a new DDSketch aggregator.

@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package maxsumcount // import "go.opentelemetry.io/otel/sdk/metric/aggregator/maxsumcount"
package minmaxsumcount // import "go.opentelemetry.io/otel/sdk/metric/aggregator/minmaxsumcount"
import (
"context"
@ -34,17 +34,15 @@ type (
state struct {
count core.Number
sum core.Number
min core.Number
max core.Number
}
)
// TODO: The SDK specification says this type should support Min
// values, see #319.
var _ export.Aggregator = &Aggregator{}
var _ aggregator.MaxSumCount = &Aggregator{}
var _ aggregator.MinMaxSumCount = &Aggregator{}
// New returns a new measure aggregator for computing max, sum, and
// New returns a new measure aggregator for computing min, max, sum, and
// count. It does not compute quantile information other than Max.
//
// Note that this aggregator maintains each value using independent
@ -54,12 +52,12 @@ var _ aggregator.MaxSumCount = &Aggregator{}
func New(desc *export.Descriptor) *Aggregator {
return &Aggregator{
kind: desc.NumberKind(),
current: unsetMaxSumCount(desc.NumberKind()),
current: unsetMinMaxSumCount(desc.NumberKind()),
}
}
func unsetMaxSumCount(kind core.NumberKind) state {
return state{max: kind.Minimum()}
func unsetMinMaxSumCount(kind core.NumberKind) state {
return state{min: kind.Maximum(), max: kind.Minimum()}
}
// Sum returns the sum of values in the checkpoint.
@ -72,9 +70,23 @@ func (c *Aggregator) Count() (int64, error) {
return int64(c.checkpoint.count.AsUint64()), nil
}
// Min returns the minimum value in the checkpoint.
// The error value aggregator.ErrEmptyDataSet will be returned if
// (due to a race condition) the checkpoint was set prior to
// current.min being computed in Update().
//
// Note: If a measure's recorded values for a given checkpoint are
// all equal to NumberKind.Maximum(), Min() will return ErrEmptyDataSet
func (c *Aggregator) Min() (core.Number, error) {
if c.checkpoint.min == c.kind.Maximum() {
return core.Number(0), aggregator.ErrEmptyDataSet
}
return c.checkpoint.min, nil
}
// 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
// (due to a race condition) the checkpoint was set prior to
// current.max being computed in Update().
//
// Note: If a measure's recorded values for a given checkpoint are
@ -88,7 +100,7 @@ func (c *Aggregator) Max() (core.Number, error) {
// 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
// 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
@ -104,6 +116,7 @@ func (c *Aggregator) Checkpoint(ctx context.Context, desc *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(c.kind.Minimum())
c.checkpoint.min = c.current.min.SwapNumberAtomic(c.kind.Maximum())
}
// Update adds the recorded measurement to the current data set.
@ -113,6 +126,16 @@ func (c *Aggregator) Update(_ context.Context, number core.Number, desc *export.
c.current.count.AddUint64Atomic(1)
c.current.sum.AddNumberAtomic(kind, number)
for {
current := c.current.min.AsNumberAtomic()
if number.CompareNumber(kind, current) >= 0 {
break
}
if c.current.min.CompareAndSwapNumber(current, number) {
break
}
}
for {
current := c.current.max.AsNumberAtomic()
@ -136,6 +159,9 @@ func (c *Aggregator) Merge(oa export.Aggregator, desc *export.Descriptor) error
c.checkpoint.sum.AddNumber(desc.NumberKind(), o.checkpoint.sum)
c.checkpoint.count.AddNumber(core.Uint64NumberKind, o.checkpoint.count)
if c.checkpoint.min.CompareNumber(desc.NumberKind(), o.checkpoint.min) > 0 {
c.checkpoint.min.SetNumber(o.checkpoint.min)
}
if c.checkpoint.max.CompareNumber(desc.NumberKind(), o.checkpoint.max) < 0 {
c.checkpoint.max.SetNumber(o.checkpoint.max)
}

@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package maxsumcount
package minmaxsumcount
import (
"context"
@ -59,26 +59,26 @@ var (
}
)
func TestMaxSumCountAbsolute(t *testing.T) {
func TestMinMaxSumCountAbsolute(t *testing.T) {
test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
maxSumCount(t, profile, positiveOnly)
minMaxSumCount(t, profile, positiveOnly)
})
}
func TestMaxSumCountNegativeOnly(t *testing.T) {
func TestMinMaxSumCountNegativeOnly(t *testing.T) {
test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
maxSumCount(t, profile, negativeOnly)
minMaxSumCount(t, profile, negativeOnly)
})
}
func TestMaxSumCountPositiveAndNegative(t *testing.T) {
func TestMinMaxSumCountPositiveAndNegative(t *testing.T) {
test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
maxSumCount(t, profile, positiveAndNegative)
minMaxSumCount(t, profile, positiveAndNegative)
})
}
// Validates max, sum and count for a given profile and policy
func maxSumCount(t *testing.T, profile test.Profile, policy policy) {
// Validates min, max, sum and count for a given profile and policy
func minMaxSumCount(t *testing.T, profile test.Profile, policy policy) {
ctx := context.Background()
descriptor := test.NewAggregatorTest(export.MeasureKind, profile.NumberKind, !policy.absolute)
@ -108,6 +108,13 @@ func maxSumCount(t *testing.T, profile test.Profile, policy policy) {
require.Equal(t, all.Count(), count, "Same count -"+policy.name)
require.Nil(t, err)
min, err := agg.Min()
require.Nil(t, err)
require.Equal(t,
all.Min(),
min,
"Same min -"+policy.name)
max, err := agg.Max()
require.Nil(t, err)
require.Equal(t,
@ -116,7 +123,7 @@ func maxSumCount(t *testing.T, profile test.Profile, policy policy) {
"Same max -"+policy.name)
}
func TestMaxSumCountMerge(t *testing.T) {
func TestMinMaxSumCountMerge(t *testing.T) {
ctx := context.Background()
test.RunProfiles(t, func(t *testing.T, profile test.Profile) {
@ -157,6 +164,13 @@ func TestMaxSumCountMerge(t *testing.T) {
require.Equal(t, all.Count(), count, "Same count - absolute")
require.Nil(t, err)
min, err := agg1.Min()
require.Nil(t, err)
require.Equal(t,
all.Min(),
min,
"Same min - absolute")
max, err := agg1.Max()
require.Nil(t, err)
require.Equal(t,

@ -29,7 +29,7 @@ import (
"go.opentelemetry.io/otel/sdk/metric/aggregator/counter"
"go.opentelemetry.io/otel/sdk/metric/aggregator/ddsketch"
"go.opentelemetry.io/otel/sdk/metric/aggregator/gauge"
"go.opentelemetry.io/otel/sdk/metric/aggregator/maxsumcount"
"go.opentelemetry.io/otel/sdk/metric/aggregator/minmaxsumcount"
)
type benchFixture struct {
@ -53,8 +53,8 @@ func (*benchFixture) AggregatorFor(descriptor *export.Descriptor) export.Aggrega
case export.GaugeKind:
return gauge.New()
case export.MeasureKind:
if strings.HasSuffix(descriptor.Name(), "maxsumcount") {
return maxsumcount.New(descriptor)
if strings.HasSuffix(descriptor.Name(), "minmaxsumcount") {
return minmaxsumcount.New(descriptor)
} else if strings.HasSuffix(descriptor.Name(), "ddsketch") {
return ddsketch.New(ddsketch.NewDefaultConfig(), descriptor)
} else if strings.HasSuffix(descriptor.Name(), "array") {
@ -360,19 +360,19 @@ func benchmarkFloat64MeasureHandleAdd(b *testing.B, name string) {
// MaxSumCount
func BenchmarkInt64MaxSumCountAdd(b *testing.B) {
benchmarkInt64MeasureAdd(b, "int64.maxsumcount")
benchmarkInt64MeasureAdd(b, "int64.minmaxsumcount")
}
func BenchmarkInt64MaxSumCountHandleAdd(b *testing.B) {
benchmarkInt64MeasureHandleAdd(b, "int64.maxsumcount")
benchmarkInt64MeasureHandleAdd(b, "int64.minmaxsumcount")
}
func BenchmarkFloat64MaxSumCountAdd(b *testing.B) {
benchmarkFloat64MeasureAdd(b, "float64.maxsumcount")
benchmarkFloat64MeasureAdd(b, "float64.minmaxsumcount")
}
func BenchmarkFloat64MaxSumCountHandleAdd(b *testing.B) {
benchmarkFloat64MeasureHandleAdd(b, "float64.maxsumcount")
benchmarkFloat64MeasureHandleAdd(b, "float64.minmaxsumcount")
}
// DDSketch

@ -20,7 +20,7 @@ import (
"go.opentelemetry.io/otel/sdk/metric/aggregator/counter"
"go.opentelemetry.io/otel/sdk/metric/aggregator/ddsketch"
"go.opentelemetry.io/otel/sdk/metric/aggregator/gauge"
"go.opentelemetry.io/otel/sdk/metric/aggregator/maxsumcount"
"go.opentelemetry.io/otel/sdk/metric/aggregator/minmaxsumcount"
)
type (
@ -38,9 +38,9 @@ var (
)
// NewWithInexpensiveMeasure returns a simple aggregation selector
// that uses counter, gauge, and maxsumcount aggregators for the three
// that uses counter, gauge, and minmaxsumcount aggregators for the three
// kinds of metric. This selector is faster and uses less memory than
// the others because maxsumcount does not aggregate quantile
// the others because minmaxsumcount does not aggregate quantile
// information.
func NewWithInexpensiveMeasure() export.AggregationSelector {
return selectorInexpensive{}
@ -71,7 +71,7 @@ func (selectorInexpensive) AggregatorFor(descriptor *export.Descriptor) export.A
case export.GaugeKind:
return gauge.New()
case export.MeasureKind:
return maxsumcount.New(descriptor)
return minmaxsumcount.New(descriptor)
default:
return counter.New()
}

@ -25,7 +25,7 @@ import (
"go.opentelemetry.io/otel/sdk/metric/aggregator/counter"
"go.opentelemetry.io/otel/sdk/metric/aggregator/ddsketch"
"go.opentelemetry.io/otel/sdk/metric/aggregator/gauge"
"go.opentelemetry.io/otel/sdk/metric/aggregator/maxsumcount"
"go.opentelemetry.io/otel/sdk/metric/aggregator/minmaxsumcount"
"go.opentelemetry.io/otel/sdk/metric/selector/simple"
)
@ -39,7 +39,7 @@ func TestInexpensiveMeasure(t *testing.T) {
inex := simple.NewWithInexpensiveMeasure()
require.NotPanics(t, func() { _ = inex.AggregatorFor(testGaugeDesc).(*gauge.Aggregator) })
require.NotPanics(t, func() { _ = inex.AggregatorFor(testCounterDesc).(*counter.Aggregator) })
require.NotPanics(t, func() { _ = inex.AggregatorFor(testMeasureDesc).(*maxsumcount.Aggregator) })
require.NotPanics(t, func() { _ = inex.AggregatorFor(testMeasureDesc).(*minmaxsumcount.Aggregator) })
}
func TestSketchMeasure(t *testing.T) {