go/opentelemetry-go
1
0
Fork 0
mirror of https://github.com/open-telemetry/opentelemetry-go.git synced 2026-06-03 18:35:08 +02:00
Code Issues Releases Activity

Move Aggregation/Temporality selection to the Exporter interface (#3260)

Browse Source 
* Add Aggregation/Temporality to Exporter iface

* Use Exporter selectors in periodic reader

* Move selector opts to just manual reader

* Simplify periodic reader ref to Exporter selectors

* Fix the periodic reader tests

* Add Aggregation/Temporality method to stdoutmetric

* Add Temporality/Aggregation to otlpmetric exp

* Add Temporality/Aggregation to http/grpc otlp clients

* Add oconf tests for selector opts

* Add tests to stdoutmetric for opts

* Correct comment subject

* Add changes to changelog

* Fix otest test client
This commit is contained in:
Tyler Yahn
2022-11-01 07:56:18 -07:00
committed by GitHub
parent 8a390acc34
commit 48a05478e2
19 changed files with 436 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sdk/metric/exporter.go
+8
View File
@@ -18,7 +18,9 @@ import (
"context"
"fmt"
"go.opentelemetry.io/otel/sdk/metric/aggregation"
"go.opentelemetry.io/otel/sdk/metric/metricdata"
"go.opentelemetry.io/otel/sdk/metric/view"
)
// ErrExporterShutdown is returned if Export or Shutdown are called after an
@@ -28,6 +30,12 @@ var ErrExporterShutdown = fmt.Errorf("exporter is shutdown")
// Exporter handles the delivery of metric data to external receivers. This is
// the final component in the metric push pipeline.
type Exporter interface {
// Temporality returns the Temporality to use for an instrument kind.
Temporality(view.InstrumentKind) metricdata.Temporality
// Aggregation returns the Aggregation to use for an instrument kind.
Aggregation(view.InstrumentKind) aggregation.Aggregation
// Export serializes and transmits metric data to a receiver.
//
// This is called synchronously, there is no concurrency safety
sdk/metric/manual_reader.go
+50
View File
@@ -129,3 +129,53 @@ func newManualReaderConfig(opts []ManualReaderOption) manualReaderConfig {
type ManualReaderOption interface {
applyManual(manualReaderConfig) manualReaderConfig
}
// WithTemporalitySelector sets the TemporalitySelector a reader will use to
// determine the Temporality of an instrument based on its kind. If this
// option is not used, the reader will use the DefaultTemporalitySelector.
func WithTemporalitySelector(selector TemporalitySelector) ManualReaderOption {
return temporalitySelectorOption{selector: selector}
}
type temporalitySelectorOption struct {
selector func(instrument view.InstrumentKind) metricdata.Temporality
}
// applyManual returns a manualReaderConfig with option applied.
func (t temporalitySelectorOption) applyManual(mrc manualReaderConfig) manualReaderConfig {
mrc.temporalitySelector = t.selector
return mrc
}
// WithAggregationSelector sets the AggregationSelector a reader will use to
// determine the aggregation to use for an instrument based on its kind. If
// this option is not used, the reader will use the DefaultAggregationSelector
// or the aggregation explicitly passed for a view matching an instrument.
func WithAggregationSelector(selector AggregationSelector) ManualReaderOption {
// Deep copy and validate before using.
wrapped := func(ik view.InstrumentKind) aggregation.Aggregation {
a := selector(ik)
cpA := a.Copy()
if err := cpA.Err(); err != nil {
cpA = DefaultAggregationSelector(ik)
global.Error(
err, "using default aggregation instead",
"aggregation", a,
"replacement", cpA,
)
}
return cpA
}
return aggregationSelectorOption{selector: wrapped}
}
type aggregationSelectorOption struct {
selector AggregationSelector
}
// applyManual returns a manualReaderConfig with option applied.
func (t aggregationSelectorOption) applyManual(c manualReaderConfig) manualReaderConfig {
c.aggregationSelector = t.selector
return c
}
sdk/metric/periodic_reader.go
+6 -16
View File
@@ -36,20 +36,16 @@ const (
// periodicReaderConfig contains configuration options for a PeriodicReader.
type periodicReaderConfig struct {
interval time.Duration
timeout time.Duration
temporalitySelector TemporalitySelector
aggregationSelector AggregationSelector
interval time.Duration
timeout time.Duration
}
// newPeriodicReaderConfig returns a periodicReaderConfig configured with
// options.
func newPeriodicReaderConfig(options []PeriodicReaderOption) periodicReaderConfig {
c := periodicReaderConfig{
interval: defaultInterval,
timeout: defaultTimeout,
temporalitySelector: DefaultTemporalitySelector,
aggregationSelector: DefaultAggregationSelector,
interval: defaultInterval,
timeout: defaultTimeout,
}
for _, o := range options {
c = o.applyPeriodic(c)
@@ -118,9 +114,6 @@ func NewPeriodicReader(exporter Exporter, options ...PeriodicReaderOption) Reade
flushCh: make(chan chan error),
cancel: cancel,
done: make(chan struct{}),
temporalitySelector: conf.temporalitySelector,
aggregationSelector: conf.aggregationSelector,
}
go func() {
@@ -140,9 +133,6 @@ type periodicReader struct {
exporter Exporter
flushCh chan chan error
temporalitySelector TemporalitySelector
aggregationSelector AggregationSelector
done chan struct{}
cancel context.CancelFunc
shutdownOnce sync.Once
@@ -187,12 +177,12 @@ func (r *periodicReader) register(p producer) {
// temporality reports the Temporality for the instrument kind provided.
func (r *periodicReader) temporality(kind view.InstrumentKind) metricdata.Temporality {
return r.temporalitySelector(kind)
return r.exporter.Temporality(kind)
}
// aggregation returns what Aggregation to use for kind.
func (r *periodicReader) aggregation(kind view.InstrumentKind) aggregation.Aggregation { // nolint:revive // import-shadow for method scoped by type.
return r.aggregationSelector(kind)
return r.exporter.Aggregation(kind)
}
// collectAndExport gather all metric data related to the periodicReader r from
sdk/metric/periodic_reader_test.go
+29 -16
View File
@@ -23,6 +23,7 @@ import (
"github.com/stretchr/testify/suite"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/sdk/metric/aggregation"
"go.opentelemetry.io/otel/sdk/metric/metricdata"
"go.opentelemetry.io/otel/sdk/metric/view"
)
@@ -54,13 +55,29 @@ func TestWithInterval(t *testing.T) {
}
type fnExporter struct {
exportFunc func(context.Context, metricdata.ResourceMetrics) error
flushFunc func(context.Context) error
shutdownFunc func(context.Context) error
temporalityFunc TemporalitySelector
aggregationFunc AggregationSelector
exportFunc func(context.Context, metricdata.ResourceMetrics) error
flushFunc func(context.Context) error
shutdownFunc func(context.Context) error
}
var _ Exporter = (*fnExporter)(nil)
func (e *fnExporter) Temporality(k view.InstrumentKind) metricdata.Temporality {
if e.temporalityFunc != nil {
return e.temporalityFunc(k)
}
return DefaultTemporalitySelector(k)
}
func (e *fnExporter) Aggregation(k view.InstrumentKind) aggregation.Aggregation {
if e.aggregationFunc != nil {
return e.aggregationFunc(k)
}
return DefaultAggregationSelector(k)
}
func (e *fnExporter) Export(ctx context.Context, m metricdata.ResourceMetrics) error {
if e.exportFunc != nil {
return e.exportFunc(ctx, m)
@@ -230,29 +247,25 @@ func BenchmarkPeriodicReader(b *testing.B) {
func TestPeriodiclReaderTemporality(t *testing.T) {
tests := []struct {
name string
options []PeriodicReaderOption
name string
exporter *fnExporter
// Currently only testing constant temporality. This should be expanded
// if we put more advanced selection in the SDK
wantTemporality metricdata.Temporality
}{
{
name: "default",
exporter: new(fnExporter),
wantTemporality: metricdata.CumulativeTemporality,
},
{
name: "delta",
options: []PeriodicReaderOption{
WithTemporalitySelector(deltaTemporalitySelector),
},
name: "delta",
exporter: &fnExporter{temporalityFunc: deltaTemporalitySelector},
wantTemporality: metricdata.DeltaTemporality,
},
{
name: "repeats overwrite",
options: []PeriodicReaderOption{
WithTemporalitySelector(deltaTemporalitySelector),
WithTemporalitySelector(cumulativeTemporalitySelector),
},
name: "cumulative",
exporter: &fnExporter{temporalityFunc: cumulativeTemporalitySelector},
wantTemporality: metricdata.CumulativeTemporality,
},
}
@@ -260,8 +273,8 @@ func TestPeriodiclReaderTemporality(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
var undefinedInstrument view.InstrumentKind
rdr := NewPeriodicReader(new(fnExporter), tt.options...)
assert.Equal(t, tt.wantTemporality, rdr.temporality(undefinedInstrument))
rdr := NewPeriodicReader(tt.exporter)
assert.Equal(t, tt.wantTemporality.String(), rdr.temporality(undefinedInstrument).String())
})
}
}
sdk/metric/reader.go
-70
View File
@@ -18,7 +18,6 @@ import (
"context"
"fmt"
"go.opentelemetry.io/otel/internal/global"
"go.opentelemetry.io/otel/sdk/metric/aggregation"
"go.opentelemetry.io/otel/sdk/metric/metricdata"
"go.opentelemetry.io/otel/sdk/metric/view"
@@ -108,13 +107,6 @@ func (p shutdownProducer) produce(context.Context) (metricdata.ResourceMetrics,
return metricdata.ResourceMetrics{}, ErrReaderShutdown
}
// ReaderOption applies a configuration option value to either a ManualReader or
// a PeriodicReader.
type ReaderOption interface {
ManualReaderOption
PeriodicReaderOption
}
// TemporalitySelector selects the temporality to use based on the InstrumentKind.
type TemporalitySelector func(view.InstrumentKind) metricdata.Temporality
@@ -125,29 +117,6 @@ func DefaultTemporalitySelector(view.InstrumentKind) metricdata.Temporality {
return metricdata.CumulativeTemporality
}
// WithTemporalitySelector sets the TemporalitySelector a reader will use to
// determine the Temporality of an instrument based on its kind. If this
// option is not used, the reader will use the DefaultTemporalitySelector.
func WithTemporalitySelector(selector TemporalitySelector) ReaderOption {
return temporalitySelectorOption{selector: selector}
}
type temporalitySelectorOption struct {
selector func(instrument view.InstrumentKind) metricdata.Temporality
}
// applyManual returns a manualReaderConfig with option applied.
func (t temporalitySelectorOption) applyManual(mrc manualReaderConfig) manualReaderConfig {
mrc.temporalitySelector = t.selector
return mrc
}
// applyPeriodic returns a periodicReaderConfig with option applied.
func (t temporalitySelectorOption) applyPeriodic(prc periodicReaderConfig) periodicReaderConfig {
prc.temporalitySelector = t.selector
return prc
}
// AggregationSelector selects the aggregation and the parameters to use for
// that aggregation based on the InstrumentKind.
type AggregationSelector func(view.InstrumentKind) aggregation.Aggregation
@@ -172,42 +141,3 @@ func DefaultAggregationSelector(ik view.InstrumentKind) aggregation.Aggregation
}
panic("unknown instrument kind")
}
// WithAggregationSelector sets the AggregationSelector a reader will use to
// determine the aggregation to use for an instrument based on its kind. If
// this option is not used, the reader will use the DefaultAggregationSelector
// or the aggregation explicitly passed for a view matching an instrument.
func WithAggregationSelector(selector AggregationSelector) ReaderOption {
// Deep copy and validate before using.
wrapped := func(ik view.InstrumentKind) aggregation.Aggregation {
a := selector(ik)
cpA := a.Copy()
if err := cpA.Err(); err != nil {
cpA = DefaultAggregationSelector(ik)
global.Error(
err, "using default aggregation instead",
"aggregation", a,
"replacement", cpA,
)
}
return cpA
}
return aggregationSelectorOption{selector: wrapped}
}
type aggregationSelectorOption struct {
selector AggregationSelector
}
// applyManual returns a manualReaderConfig with option applied.
func (t aggregationSelectorOption) applyManual(c manualReaderConfig) manualReaderConfig {
c.aggregationSelector = t.selector
return c
}
// applyPeriodic returns a periodicReaderConfig with option applied.
func (t aggregationSelectorOption) applyPeriodic(c periodicReaderConfig) periodicReaderConfig {
c.aggregationSelector = t.selector
return c
}