Skip to content

Commit 1ee3f46

Browse files
dashpoledashpole
committed
implement WithExplicitBucketBoundaries option in the metric SDK
1 parent 6980c9c commit 1ee3f46

File tree

5 files changed

+229
-34
lines changed

5 files changed

+229
-34
lines changed

‎sdk/metric/meter.go

+50-4
Original file line numberDiff line numberDiff line change
@@ -97,9 +97,8 @@ func (m *meter) Int64UpDownCounter(name string, options ...metric.Int64UpDownCou
9797
// distribution of int64 measurements during a computational operation.
9898
func (m *meter) Int64Histogram(name string, options ...metric.Int64HistogramOption) (metric.Int64Histogram, error) {
9999
cfg := metric.NewInt64HistogramConfig(options...)
100-
const kind = InstrumentKindHistogram
101100
p := int64InstProvider{m}
102-
i, err := p.lookup(kind, name, cfg.Description(), cfg.Unit())
101+
i, err := p.lookupHistogram(name, cfg)
103102
if err != nil {
104103
return i, err
105104
}
@@ -190,9 +189,8 @@ func (m *meter) Float64UpDownCounter(name string, options ...metric.Float64UpDow
190189
// distribution of float64 measurements during a computational operation.
191190
func (m *meter) Float64Histogram(name string, options ...metric.Float64HistogramOption) (metric.Float64Histogram, error) {
192191
cfg := metric.NewFloat64HistogramConfig(options...)
193-
const kind = InstrumentKindHistogram
194192
p := float64InstProvider{m}
195-
i, err := p.lookup(kind, name, cfg.Description(), cfg.Unit())
193+
i, err := p.lookupHistogram(name, cfg)
196194
if err != nil {
197195
return i, err
198196
}
@@ -456,12 +454,36 @@ func (p int64InstProvider) aggs(kind InstrumentKind, name, desc, u string) ([]ag
456454
return p.int64Resolver.Aggregators(inst)
457455
}
458456

457+
func (p int64InstProvider) histogramAggs(name string, cfg metric.Int64HistogramConfig) ([]aggregate.Measure[int64], error) {
458+
boundaries := cfg.ExplicitBucketBoundaries()
459+
aggError := AggregationExplicitBucketHistogram{Boundaries: boundaries}.err()
460+
if aggError != nil {
461+
// if boundaries are invalid, ignore them
462+
boundaries = nil
463+
}
464+
inst := Instrument{
465+
Name: name,
466+
Description: cfg.Description(),
467+
Unit: cfg.Unit(),
468+
Kind: InstrumentKindHistogram,
469+
Scope: p.scope,
470+
}
471+
measures, err := p.int64Resolver.HistogramAggregators(inst, boundaries)
472+
return measures, errors.Join(aggError, err)
473+
}
474+
459475
// lookup returns the resolved instrumentImpl.
460476
func (p int64InstProvider) lookup(kind InstrumentKind, name, desc, u string) (*int64Inst, error) {
461477
aggs, err := p.aggs(kind, name, desc, u)
462478
return &int64Inst{measures: aggs}, err
463479
}
464480

481+
// lookupHistogram returns the resolved instrumentImpl.
482+
func (p int64InstProvider) lookupHistogram(name string, cfg metric.Int64HistogramConfig) (*int64Inst, error) {
483+
aggs, err := p.histogramAggs(name, cfg)
484+
return &int64Inst{measures: aggs}, err
485+
}
486+
465487
// float64InstProvider provides float64 OpenTelemetry instruments.
466488
type float64InstProvider struct{ *meter }
467489

@@ -476,12 +498,36 @@ func (p float64InstProvider) aggs(kind InstrumentKind, name, desc, u string) ([]
476498
return p.float64Resolver.Aggregators(inst)
477499
}
478500

501+
func (p float64InstProvider) histogramAggs(name string, cfg metric.Float64HistogramConfig) ([]aggregate.Measure[float64], error) {
502+
boundaries := cfg.ExplicitBucketBoundaries()
503+
aggError := AggregationExplicitBucketHistogram{Boundaries: boundaries}.err()
504+
if aggError != nil {
505+
// if boundaries are invalid, ignore them
506+
boundaries = nil
507+
}
508+
inst := Instrument{
509+
Name: name,
510+
Description: cfg.Description(),
511+
Unit: cfg.Unit(),
512+
Kind: InstrumentKindHistogram,
513+
Scope: p.scope,
514+
}
515+
measures, err := p.float64Resolver.HistogramAggregators(inst, boundaries)
516+
return measures, errors.Join(aggError, err)
517+
}
518+
479519
// lookup returns the resolved instrumentImpl.
480520
func (p float64InstProvider) lookup(kind InstrumentKind, name, desc, u string) (*float64Inst, error) {
481521
aggs, err := p.aggs(kind, name, desc, u)
482522
return &float64Inst{measures: aggs}, err
483523
}
484524

525+
// lookupHistogram returns the resolved instrumentImpl.
526+
func (p float64InstProvider) lookupHistogram(name string, cfg metric.Float64HistogramConfig) (*float64Inst, error) {
527+
aggs, err := p.histogramAggs(name, cfg)
528+
return &float64Inst{measures: aggs}, err
529+
}
530+
485531
type int64ObservProvider struct{ *meter }
486532

487533
func (p int64ObservProvider) lookup(kind InstrumentKind, name, desc, u string) (int64Observable, error) {

‎sdk/metric/meter_test.go

+84
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ package metric
1616

1717
import (
1818
"context"
19+
"errors"
1920
"fmt"
2021
"strings"
2122
"sync"
@@ -550,6 +551,17 @@ func TestMeterCreatesInstrumentsValidations(t *testing.T) {
550551

551552
wantErr: fmt.Errorf("%w: _: must start with a letter", ErrInstrumentName),
552553
},
554+
{
555+
name: "Int64Histogram with invalid buckets",
556+
557+
fn: func(t *testing.T, m metric.Meter) error {
558+
i, err := m.Int64Histogram("histogram", metric.WithExplicitBucketBoundaries(-1, 1, -5))
559+
assert.NotNil(t, i)
560+
return err
561+
},
562+
563+
wantErr: errors.Join(fmt.Errorf("%w: non-monotonic boundaries: %v", errHist, []float64{-1, 1, -5})),
564+
},
553565
{
554566
name: "Int64ObservableCounter with no validation issues",
555567

@@ -670,6 +682,17 @@ func TestMeterCreatesInstrumentsValidations(t *testing.T) {
670682

671683
wantErr: fmt.Errorf("%w: _: must start with a letter", ErrInstrumentName),
672684
},
685+
{
686+
name: "Float64Histogram with invalid buckets",
687+
688+
fn: func(t *testing.T, m metric.Meter) error {
689+
i, err := m.Float64Histogram("histogram", metric.WithExplicitBucketBoundaries(-1, 1, -5))
690+
assert.NotNil(t, i)
691+
return err
692+
},
693+
694+
wantErr: errors.Join(fmt.Errorf("%w: non-monotonic boundaries: %v", errHist, []float64{-1, 1, -5})),
695+
},
673696
{
674697
name: "Float64ObservableCounter with no validation issues",
675698

@@ -1967,3 +1990,64 @@ func TestMalformedSelectors(t *testing.T) {
19671990
})
19681991
}
19691992
}
1993+
1994+
func TestHistogramBucketPrecedenceOrdering(t *testing.T) {
1995+
defaultBuckets := []float64{0, 5, 10, 25, 50, 75, 100, 250, 500, 750, 1000, 2500, 5000, 7500, 10000}
1996+
aggregationSelector := func(InstrumentKind) Aggregation {
1997+
return AggregationExplicitBucketHistogram{Boundaries: []float64{0, 1, 2, 3, 4, 5}}
1998+
}
1999+
for _, tt := range []struct {
2000+
desc string
2001+
reader Reader
2002+
views []View
2003+
histogramOpts []metric.Float64HistogramOption
2004+
expectedBucketBoundaries []float64
2005+
}{
2006+
{
2007+
desc: "default",
2008+
reader: NewManualReader(),
2009+
expectedBucketBoundaries: defaultBuckets,
2010+
},
2011+
{
2012+
desc: "custom reader aggregation overrides default",
2013+
reader: NewManualReader(WithAggregationSelector(aggregationSelector)),
2014+
expectedBucketBoundaries: []float64{0, 1, 2, 3, 4, 5},
2015+
},
2016+
{
2017+
desc: "overridden by histogram option",
2018+
reader: NewManualReader(WithAggregationSelector(aggregationSelector)),
2019+
histogramOpts: []metric.Float64HistogramOption{
2020+
metric.WithExplicitBucketBoundaries(0, 2, 4, 6, 8, 10),
2021+
},
2022+
expectedBucketBoundaries: []float64{0, 2, 4, 6, 8, 10},
2023+
},
2024+
{
2025+
desc: "overridden by view",
2026+
reader: NewManualReader(WithAggregationSelector(aggregationSelector)),
2027+
histogramOpts: []metric.Float64HistogramOption{
2028+
metric.WithExplicitBucketBoundaries(0, 2, 4, 6, 8, 10),
2029+
},
2030+
views: []View{NewView(Instrument{Name: "*"}, Stream{
2031+
Aggregation: AggregationExplicitBucketHistogram{Boundaries: []float64{0, 3, 6, 9, 12, 15}},
2032+
})},
2033+
expectedBucketBoundaries: []float64{0, 3, 6, 9, 12, 15},
2034+
},
2035+
} {
2036+
t.Run(tt.desc, func(t *testing.T) {
2037+
meter := NewMeterProvider(WithView(tt.views...), WithReader(tt.reader)).Meter("TestHistogramBucketPrecedenceOrdering")
2038+
sfHistogram, err := meter.Float64Histogram("sync.float64.histogram", tt.histogramOpts...)
2039+
require.NoError(t, err)
2040+
sfHistogram.Record(context.Background(), 1)
2041+
var rm metricdata.ResourceMetrics
2042+
err = tt.reader.Collect(context.Background(), &rm)
2043+
require.NoError(t, err)
2044+
require.Len(t, rm.ScopeMetrics, 1)
2045+
require.Len(t, rm.ScopeMetrics[0].Metrics, 1)
2046+
gotHist, ok := rm.ScopeMetrics[0].Metrics[0].Data.(metricdata.Histogram[float64])
2047+
require.True(t, ok)
2048+
require.Len(t, gotHist.DataPoints, 1)
2049+
assert.Equal(t, tt.expectedBucketBoundaries, gotHist.DataPoints[0].Bounds)
2050+
})
2051+
}
2052+
2053+
}

‎sdk/metric/pipeline.go

+56-25
Original file line numberDiff line numberDiff line change
@@ -231,7 +231,7 @@ func newInserter[N int64 | float64](p *pipeline, vc *cache[string, instID]) *ins
231231
//
232232
// If an instrument is determined to use a Drop aggregation, that instrument is
233233
// not inserted nor returned.
234-
func (i *inserter[N]) Instrument(inst Instrument) ([]aggregate.Measure[N], error) {
234+
func (i *inserter[N]) Instrument(inst Instrument, readerAggregation Aggregation) ([]aggregate.Measure[N], error) {
235235
var (
236236
matched bool
237237
measures []aggregate.Measure[N]
@@ -245,8 +245,7 @@ func (i *inserter[N]) Instrument(inst Instrument) ([]aggregate.Measure[N], error
245245
continue
246246
}
247247
matched = true
248-
249-
in, id, err := i.cachedAggregator(inst.Scope, inst.Kind, stream)
248+
in, id, err := i.cachedAggregator(inst.Scope, inst.Kind, stream, readerAggregation)
250249
if err != nil {
251250
errs.append(err)
252251
}
@@ -271,7 +270,7 @@ func (i *inserter[N]) Instrument(inst Instrument) ([]aggregate.Measure[N], error
271270
Description: inst.Description,
272271
Unit: inst.Unit,
273272
}
274-
in, _, err := i.cachedAggregator(inst.Scope, inst.Kind, stream)
273+
in, _, err := i.cachedAggregator(inst.Scope, inst.Kind, stream, readerAggregation)
275274
if err != nil {
276275
errs.append(err)
277276
}
@@ -291,6 +290,31 @@ type aggVal[N int64 | float64] struct {
291290
Err error
292291
}
293292

293+
// readerDefaultAggregation returns the default aggregation for the instrument
294+
// kind based on the reader's aggregation preferences. This is used unless the
295+
// aggregation is overridden with a view.
296+
func (i *inserter[N]) readerDefaultAggregation(kind InstrumentKind) Aggregation {
297+
aggregation := i.pipeline.reader.aggregation(kind)
298+
switch aggregation.(type) {
299+
case nil, AggregationDefault:
300+
// If the reader returns default or nil use the default selector.
301+
aggregation = DefaultAggregationSelector(kind)
302+
default:
303+
// Deep copy and validate before using.
304+
aggregation = aggregation.copy()
305+
if err := aggregation.err(); err != nil {
306+
orig := aggregation
307+
aggregation = DefaultAggregationSelector(kind)
308+
global.Error(
309+
err, "using default aggregation instead",
310+
"aggregation", orig,
311+
"replacement", aggregation,
312+
)
313+
}
314+
}
315+
return aggregation
316+
}
317+
294318
// cachedAggregator returns the appropriate aggregate input and output
295319
// functions for an instrument configuration. If the exact instrument has been
296320
// created within the inst.Scope, those aggregate function instances will be
@@ -305,29 +329,14 @@ type aggVal[N int64 | float64] struct {
305329
//
306330
// If the instrument defines an unknown or incompatible aggregation, an error
307331
// is returned.
308-
func (i *inserter[N]) cachedAggregator(scope instrumentation.Scope, kind InstrumentKind, stream Stream) (meas aggregate.Measure[N], aggID uint64, err error) {
332+
func (i *inserter[N]) cachedAggregator(scope instrumentation.Scope, kind InstrumentKind, stream Stream, readerAggregation Aggregation) (meas aggregate.Measure[N], aggID uint64, err error) {
309333
switch stream.Aggregation.(type) {
310334
case nil:
311-
// Undefined, nil, means to use the default from the reader.
312-
stream.Aggregation = i.pipeline.reader.aggregation(kind)
313-
switch stream.Aggregation.(type) {
314-
case nil, AggregationDefault:
315-
// If the reader returns default or nil use the default selector.
316-
stream.Aggregation = DefaultAggregationSelector(kind)
317-
default:
318-
// Deep copy and validate before using.
319-
stream.Aggregation = stream.Aggregation.copy()
320-
if err := stream.Aggregation.err(); err != nil {
321-
orig := stream.Aggregation
322-
stream.Aggregation = DefaultAggregationSelector(kind)
323-
global.Error(
324-
err, "using default aggregation instead",
325-
"aggregation", orig,
326-
"replacement", stream.Aggregation,
327-
)
328-
}
329-
}
335+
// The aggregation was not overridden with a view. Use the aggregation
336+
// provided by the reader.
337+
stream.Aggregation = readerAggregation
330338
case AggregationDefault:
339+
// The view explicitly requested the default aggregation.
331340
stream.Aggregation = DefaultAggregationSelector(kind)
332341
}
333342

@@ -596,7 +605,29 @@ func (r resolver[N]) Aggregators(id Instrument) ([]aggregate.Measure[N], error)
596605

597606
errs := &multierror{}
598607
for _, i := range r.inserters {
599-
in, err := i.Instrument(id)
608+
in, err := i.Instrument(id, i.readerDefaultAggregation(id.Kind))
609+
if err != nil {
610+
errs.append(err)
611+
}
612+
measures = append(measures, in...)
613+
}
614+
return measures, errs.errorOrNil()
615+
}
616+
617+
// HistogramAggregators returns the histogram Aggregators that must be updated by the instrument
618+
// defined by key. If boundaries were provided on instrument instantiation, those take precedence
619+
// over boundaries provided by the reader.
620+
func (r resolver[N]) HistogramAggregators(id Instrument, boundaries []float64) ([]aggregate.Measure[N], error) {
621+
var measures []aggregate.Measure[N]
622+
623+
errs := &multierror{}
624+
for _, i := range r.inserters {
625+
agg := i.readerDefaultAggregation(id.Kind)
626+
if histAgg, ok := agg.(AggregationExplicitBucketHistogram); ok && len(boundaries) > 0 {
627+
histAgg.Boundaries = boundaries
628+
agg = histAgg
629+
}
630+
in, err := i.Instrument(id, agg)
600631
if err != nil {
601632
errs.append(err)
602633
}

0 commit comments

Comments
 (0)