Create a export.Checkpointer API; refactor metric export pipeline tes…

…t helpers (open-telemetry#1055)

* Add regexp filter in api/label, test

* Add regexp option to sdk.Config

* Return indistinct values only when keyRe != nil

* Filter in sdk

* Add an accumulator filter test

* SDK tests pass

* Precommit

* Undo set filters

* Backout related filter changes

* Add a new test

* Checkpoint

* Comments

* Comments in label.Set

* Lint

* Add Checkpointer

* Test refactor continues

* Refactor reducer test

* Checkpoint

* Update push_test

* Update pull controller

* Comment

* Remove pending PRs

* Remove exportertest pkg

* Revert basic changes

* Revert testing changes

* Restore processortest changes

* Precommit & comments

* Comments on pull semantics

* Comments

* Fix buggy test; incorrect expectation following error

* Finish this test

* Comments

* Apply suggestions from code review

Co-authored-by: Tyler Yahn <[email protected]>

Co-authored-by: Tyler Yahn <[email protected]>

example/otel-collector/main.go

@@ -31,6 +31,7 @@ import (
 	apitrace "go.opentelemetry.io/otel/api/trace"
 	"go.opentelemetry.io/otel/exporters/otlp"
 	"go.opentelemetry.io/otel/sdk/metric/controller/push"
+	"go.opentelemetry.io/otel/sdk/metric/processor/basic"
 	"go.opentelemetry.io/otel/sdk/metric/selector/simple"
 	"go.opentelemetry.io/otel/sdk/resource"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
@@ -64,7 +65,10 @@ func initProvider() (*otlp.Exporter, *push.Controller) {
 	handleErr(err, "failed to create trace provider")
 
 	pusher := push.New(
-		simple.NewWithExactDistribution(),
+		basic.New(
+			simple.NewWithExactDistribution(),
+			exp,
+		),
 		exp,
 		push.WithPeriod(2*time.Second),
 	)

exporters/metric/prometheus/prometheus.go

@@ -29,6 +29,7 @@ import (
 	export "go.opentelemetry.io/otel/sdk/export/metric"
 	"go.opentelemetry.io/otel/sdk/export/metric/aggregation"
 	"go.opentelemetry.io/otel/sdk/metric/controller/pull"
+	"go.opentelemetry.io/otel/sdk/metric/processor/basic"
 	"go.opentelemetry.io/otel/sdk/metric/selector/simple"
 )
 
@@ -144,8 +145,11 @@ func (e *Exporter) SetController(config Config, options ...pull.Option) {
 	defer e.lock.Unlock()
 
 	e.controller = pull.New(
-		simple.NewWithHistogramDistribution(config.DefaultHistogramBoundaries),
-		e,
+		basic.New(
+			simple.NewWithHistogramDistribution(config.DefaultHistogramBoundaries),
+			e,
+			basic.WithMemory(true),
+		),
 		options...,
 	)
 }

exporters/stdout/exporter.go

@@ -20,6 +20,7 @@ import (
 	"go.opentelemetry.io/otel/sdk/export/metric"
 	"go.opentelemetry.io/otel/sdk/export/trace"
 	"go.opentelemetry.io/otel/sdk/metric/controller/push"
+	"go.opentelemetry.io/otel/sdk/metric/processor/basic"
 	"go.opentelemetry.io/otel/sdk/metric/selector/simple"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 )
@@ -62,7 +63,10 @@ func NewExportPipeline(exportOpts []Option, pushOpts []push.Option) (apitrace.Pr
 	}
 
 	pusher := push.New(
-		simple.NewWithExactDistribution(),
+		basic.New(
+			simple.NewWithExactDistribution(),
+			exporter,
+		),
 		exporter,
 		pushOpts...,
 	)

sdk/export/metric/metric.go

@@ -97,6 +97,33 @@ type AggregatorSelector interface {
 	AggregatorFor(*metric.Descriptor, ...*Aggregator)
 }
 
+// Checkpointer is the interface used by a Controller to coordinate
+// the Processor with Accumulator(s) and Exporter(s).  The
+// StartCollection() and FinishCollection() methods start and finish a
+// collection interval.  Controllers call the Accumulator(s) during
+// collection to process Accumulations.
+type Checkpointer interface {
+	// Processor processes metric data for export.  The Process
+	// method is bracketed by StartCollection and FinishCollection
+	// calls.  The embedded AggregatorSelector can be called at
+	// any time.
+	Processor
+
+	// CheckpointSet returns the current data set.  This may be
+	// called before and after collection.  The
+	// implementation is required to return the same value
+	// throughout its lifetime, since CheckpointSet exposes a
+	// sync.Locker interface.  The caller is responsible for
+	// locking the CheckpointSet before initiating collection.
+	CheckpointSet() CheckpointSet
+
+	// StartCollection begins a collection interval.
+	StartCollection()
+
+	// FinishCollection ends a collection interval.
+	FinishCollection() error
+}
+
 // Aggregator implements a specific aggregation behavior, e.g., a
 // behavior to track a sequence of updates to an instrument.  Sum-only
 // instruments commonly use a simple Sum aggregator, but for the

sdk/export/metric/metrictest/test.go

@@ -35,6 +35,7 @@ type mapkey struct {
 }
 
 // CheckpointSet is useful for testing Exporters.
+// TODO(#872): Uses of this can be replaced by processortest.Output.
 type CheckpointSet struct {
 	sync.RWMutex
 	records  map[mapkey]export.Record

sdk/metric/controller/pull/pull.go

@@ -23,7 +23,6 @@ import (
 	export "go.opentelemetry.io/otel/sdk/export/metric"
 	sdk "go.opentelemetry.io/otel/sdk/metric"
 	controllerTime "go.opentelemetry.io/otel/sdk/metric/controller/time"
-	processor "go.opentelemetry.io/otel/sdk/metric/processor/basic"
 	"go.opentelemetry.io/otel/sdk/resource"
 )
 
@@ -35,45 +34,48 @@ const DefaultCachePeriod time.Duration = 10 * time.Second
 // *basic.Processor.  Use Provider() for obtaining Meters.  Use
 // Foreach() for accessing current records.
 type Controller struct {
-	accumulator *sdk.Accumulator
-	processor   *processor.Processor
-	provider    *registry.Provider
-	period      time.Duration
-	lastCollect time.Time
-	clock       controllerTime.Clock
-	checkpoint  export.CheckpointSet
+	accumulator  *sdk.Accumulator
+	checkpointer export.Checkpointer
+	provider     *registry.Provider
+	period       time.Duration
+	lastCollect  time.Time
+	clock        controllerTime.Clock
+	checkpoint   export.CheckpointSet
 }
 
-// New returns a *Controller configured with an aggregation selector and options.
-func New(aselector export.AggregatorSelector, eselector export.ExportKindSelector, options ...Option) *Controller {
+// New returns a *Controller configured with an export.Checkpointer.
+//
+// Pull controllers are typically used in an environment where there
+// are multiple readers.  It is common, therefore, when configuring a
+// basic Processor for use with this controller, to use a
+// CumulativeExport strategy and the basic.WithMemory(true) option,
+// which ensures that every CheckpointSet includes full state.
+func New(checkpointer export.Checkpointer, options ...Option) *Controller {
 	config := &Config{
 		Resource:    resource.Empty(),
 		CachePeriod: DefaultCachePeriod,
 	}
 	for _, opt := range options {
 		opt.Apply(config)
 	}
-	// This controller uses WithMemory() as a requirement to
-	// support multiple readers.
-	processor := processor.New(aselector, eselector, processor.WithMemory(true))
 	accum := sdk.NewAccumulator(
-		processor,
+		checkpointer,
 		sdk.WithResource(config.Resource),
 	)
 	return &Controller{
-		accumulator: accum,
-		processor:   processor,
-		provider:    registry.NewProvider(accum),
-		period:      config.CachePeriod,
-		checkpoint:  processor.CheckpointSet(),
-		clock:       controllerTime.RealClock{},
+		accumulator:  accum,
+		checkpointer: checkpointer,
+		provider:     registry.NewProvider(accum),
+		period:       config.CachePeriod,
+		checkpoint:   checkpointer.CheckpointSet(),
+		clock:        controllerTime.RealClock{},
 	}
 }
 
 // SetClock sets the clock used for caching.  For testing purposes.
 func (c *Controller) SetClock(clock controllerTime.Clock) {
-	c.processor.Lock()
-	defer c.processor.Unlock()
+	c.checkpointer.CheckpointSet().Lock()
+	defer c.checkpointer.CheckpointSet().Unlock()
 	c.clock = clock
 }
 
@@ -86,17 +88,17 @@ func (c *Controller) Provider() metric.Provider {
 // Foreach gives the caller read-locked access to the current
 // export.CheckpointSet.
 func (c *Controller) ForEach(ks export.ExportKindSelector, f func(export.Record) error) error {
-	c.processor.RLock()
-	defer c.processor.RUnlock()
+	c.checkpointer.CheckpointSet().RLock()
+	defer c.checkpointer.CheckpointSet().RUnlock()
 
 	return c.checkpoint.ForEach(ks, f)
 }
 
 // Collect requests a collection.  The collection will be skipped if
 // the last collection is aged less than the CachePeriod.
 func (c *Controller) Collect(ctx context.Context) error {
-	c.processor.Lock()
-	defer c.processor.Unlock()
+	c.checkpointer.CheckpointSet().Lock()
+	defer c.checkpointer.CheckpointSet().Unlock()
 
 	if c.period > 0 {
 		now := c.clock.Now()
@@ -108,9 +110,9 @@ func (c *Controller) Collect(ctx context.Context) error {
 		c.lastCollect = now
 	}
 
-	c.processor.StartCollection()
+	c.checkpointer.StartCollection()
 	c.accumulator.Collect(ctx)
-	err := c.processor.FinishCollection()
-	c.checkpoint = c.processor.CheckpointSet()
+	err := c.checkpointer.FinishCollection()
+	c.checkpoint = c.checkpointer.CheckpointSet()
 	return err
 }

sdk/metric/controller/pull/pull_test.go

@@ -28,14 +28,18 @@ import (
 	export "go.opentelemetry.io/otel/sdk/export/metric"
 	"go.opentelemetry.io/otel/sdk/metric/controller/controllertest"
 	"go.opentelemetry.io/otel/sdk/metric/controller/pull"
+	"go.opentelemetry.io/otel/sdk/metric/processor/basic"
 	"go.opentelemetry.io/otel/sdk/metric/processor/processortest"
 	selector "go.opentelemetry.io/otel/sdk/metric/selector/simple"
 )
 
 func TestPullNoCache(t *testing.T) {
 	puller := pull.New(
-		selector.NewWithExactDistribution(),
-		export.CumulativeExporter,
+		basic.New(
+			selector.NewWithExactDistribution(),
+			export.CumulativeExporter,
+			basic.WithMemory(true),
+		),
 		pull.WithCachePeriod(0),
 	)
 
@@ -66,8 +70,11 @@ func TestPullNoCache(t *testing.T) {
 
 func TestPullWithCache(t *testing.T) {
 	puller := pull.New(
-		selector.NewWithExactDistribution(),
-		export.CumulativeExporter,
+		basic.New(
+			selector.NewWithExactDistribution(),
+			export.CumulativeExporter,
+			basic.WithMemory(true),
+		),
 		pull.WithCachePeriod(time.Second),
 	)
 	mock := controllertest.NewMockClock()

sdk/metric/controller/push/push.go

@@ -25,31 +25,30 @@ import (
 	export "go.opentelemetry.io/otel/sdk/export/metric"
 	sdk "go.opentelemetry.io/otel/sdk/metric"
 	controllerTime "go.opentelemetry.io/otel/sdk/metric/controller/time"
-	"go.opentelemetry.io/otel/sdk/metric/processor/basic"
 )
 
 // DefaultPushPeriod is the default time interval between pushes.
 const DefaultPushPeriod = 10 * time.Second
 
 // Controller organizes a periodic push of metric data.
 type Controller struct {
-	lock        sync.Mutex
-	accumulator *sdk.Accumulator
-	provider    *registry.Provider
-	processor   *basic.Processor
-	exporter    export.Exporter
-	wg          sync.WaitGroup
-	ch          chan struct{}
-	period      time.Duration
-	timeout     time.Duration
-	clock       controllerTime.Clock
-	ticker      controllerTime.Ticker
+	lock         sync.Mutex
+	accumulator  *sdk.Accumulator
+	provider     *registry.Provider
+	checkpointer export.Checkpointer
+	exporter     export.Exporter
+	wg           sync.WaitGroup
+	ch           chan struct{}
+	period       time.Duration
+	timeout      time.Duration
+	clock        controllerTime.Clock
+	ticker       controllerTime.Ticker
 }
 
 // New constructs a Controller, an implementation of metric.Provider,
-// using the provided exporter and options to configure an SDK with
-// periodic collection.
-func New(selector export.AggregatorSelector, exporter export.Exporter, opts ...Option) *Controller {
+// using the provided checkpointer, exporter, and options to configure
+// an SDK with periodic collection.
+func New(checkpointer export.Checkpointer, exporter export.Exporter, opts ...Option) *Controller {
 	c := &Config{
 		Period: DefaultPushPeriod,
 	}
@@ -60,20 +59,19 @@ func New(selector export.AggregatorSelector, exporter export.Exporter, opts ...O
 		c.Timeout = c.Period
 	}
 
-	processor := basic.New(selector, exporter)
 	impl := sdk.NewAccumulator(
-		processor,
+		checkpointer,
 		sdk.WithResource(c.Resource),
 	)
 	return &Controller{
-		provider:    registry.NewProvider(impl),
-		accumulator: impl,
-		processor:   processor,
-		exporter:    exporter,
-		ch:          make(chan struct{}),
-		period:      c.Period,
-		timeout:     c.Timeout,
-		clock:       controllerTime.RealClock{},
+		provider:     registry.NewProvider(impl),
+		accumulator:  impl,
+		checkpointer: checkpointer,
+		exporter:     exporter,
+		ch:           make(chan struct{}),
+		period:       c.Period,
+		timeout:      c.Timeout,
+		clock:        controllerTime.RealClock{},
 	}
 }
 
@@ -139,16 +137,17 @@ func (c *Controller) tick() {
 	ctx, cancel := context.WithTimeout(context.Background(), c.timeout)
 	defer cancel()
 
-	c.processor.Lock()
-	defer c.processor.Unlock()
+	ckpt := c.checkpointer.CheckpointSet()
+	ckpt.Lock()
+	defer ckpt.Unlock()
 
-	c.processor.StartCollection()
+	c.checkpointer.StartCollection()
 	c.accumulator.Collect(ctx)
-	if err := c.processor.FinishCollection(); err != nil {
+	if err := c.checkpointer.FinishCollection(); err != nil {
 		global.Handle(err)
 	}
 
-	if err := c.exporter.Export(ctx, c.processor.CheckpointSet()); err != nil {
+	if err := c.exporter.Export(ctx, ckpt); err != nil {
 		global.Handle(err)
 	}
 }