Add the periodic reader

sdk/metric/exporter.go

@@ -0,0 +1,58 @@
+// Copyright The 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 metric // import "go.opentelemetry.io/otel/sdk/metric"
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/otel/sdk/metric/export"
+)
+
+// ErrExporterShutdown is returned if Export or Shutdown are called after an
+// Exporter has been Shutdown.
+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 {
+	// Export serializes and transmits metric data to a receiver.
+	//
+	// This is called synchronously, there is no concurrency safety
+	// requirement. Because of this, it is critical that all timeouts and
+	// cancellations of the passed context be honored.
+	//
+	// All retry logic must be contained in this function. The SDK does not
+	// implement any retry logic. All errors returned by this function are
+	// considered unrecoverable and will be reported to a configured error
+	// Handler.
+	Export(context.Context, export.Metrics) error
+
+	// ForceFlush flushes any metric data held by an exporter.
+	//
+	// The deadline or cancellation of the passed context must be honored. An
+	// appropriate error should be returned in these situations.
+	ForceFlush(context.Context) error
+
+	// Shutdown flushes all metric data held by an exporter and releases any
+	// held computational resources.
+	//
+	// The deadline or cancellation of the passed context must be honored. An
+	// appropriate error should be returned in these situations.
+	//
+	// After Shutdown is called, calls to Export will perform no operation and
+	// instead will return an error indicating the shutdown state.
+	Shutdown(context.Context) error
+}

sdk/metric/go.mod

@@ -4,6 +4,7 @@ go 1.16
 
 require (
 	github.com/stretchr/testify v1.7.1
+	go.opentelemetry.io/otel v1.6.3
 	go.opentelemetry.io/otel/metric v0.0.0-00010101000000-000000000000
 )
 

sdk/metric/go.sum

@@ -1,7 +1,9 @@
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/logr v1.2.3 h1:2DntVwHkVopvECVRSlL5PSo9eG+cAkDCuckLubN+rq0=
 github.com/go-logr/logr v1.2.3/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o=
 github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=

sdk/metric/manual_reader.go

@@ -16,7 +16,6 @@ package metric // import "go.opentelemetry.io/otel/sdk/metric"
 
 import (
 	"context"
-	"fmt"
 	"sync"
 
 	"go.opentelemetry.io/otel/sdk/metric/export"
@@ -75,11 +74,3 @@ func (mr *manualReader) Collect(ctx context.Context) (export.Metrics, error) {
 	}
 	return mr.producer.produce(ctx)
 }
-
-// ErrReaderNotRegistered is returned if Collect or Shutdown are called before
-// the reader is registered with a MeterProvider
-var ErrReaderNotRegistered = fmt.Errorf("reader is not registered")
-
-// ErrReaderShutdown is returned if Collect or Shutdown are called after a
-// reader has been Shutdown once.
-var ErrReaderShutdown = fmt.Errorf("reader is shutdown")

sdk/metric/manual_reader_test.go

@@ -15,60 +15,9 @@
 package metric // import "go.opentelemetry.io/otel/sdk/metric/reader"
 
 import (
-	"context"
 	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"go.opentelemetry.io/otel/sdk/metric/export"
 )
 
-func TestManualReaderNotRegistered(t *testing.T) {
-	rdr := &manualReader{}
-
-	_, err := rdr.Collect(context.Background())
-	require.ErrorIs(t, err, ErrReaderNotRegistered)
-}
-
-type testProducer struct{}
-
-var testMetrics = export.Metrics{
-	// TODO: test with actual data.
-}
-
-func (p testProducer) produce(context.Context) (export.Metrics, error) {
-	return testMetrics, nil
-}
-
-func TestManualReaderProducer(t *testing.T) {
-	rdr := &manualReader{}
-	rdr.register(testProducer{})
-
-	m, err := rdr.Collect(context.Background())
-	assert.NoError(t, err)
-	assert.Equal(t, testMetrics, m)
-}
-
-func TestManualReaderCollectAfterShutdown(t *testing.T) {
-	rdr := &manualReader{}
-	rdr.register(testProducer{})
-	err := rdr.Shutdown(context.Background())
-	require.NoError(t, err)
-
-	m, err := rdr.Collect(context.Background())
-	assert.ErrorIs(t, err, ErrReaderShutdown)
-	assert.Equal(t, export.Metrics{}, m)
-}
-
-func TestManualReaderShutdown(t *testing.T) {
-	rdr := &manualReader{}
-	rdr.register(testProducer{})
-
-	err := rdr.Shutdown(context.Background())
-	require.NoError(t, err)
-
-	err = rdr.Shutdown(context.Background())
-	assert.ErrorIs(t, err, ErrReaderShutdown)
-
+func TestManualReader(t *testing.T) {
+	testReaderHarness(t, func() Reader { return NewManualReader() })
 }

sdk/metric/periodic_reader.go

@@ -0,0 +1,217 @@
+// Copyright The 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 metric // import "go.opentelemetry.io/otel/sdk/metric"
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/sdk/metric/export"
+)
+
+const (
+	defaultTimeout  = time.Millisecond * 30000
+	defaultInterval = time.Millisecond * 60000
+)
+
+// periodicReaderConfig contains configuration options for a PeriodicReader.
+type periodicReaderConfig struct {
+	interval time.Duration
+	timeout  time.Duration
+}
+
+func newPeriodicReaderConfig(options []PeriodicReaderOption) periodicReaderConfig {
+	c := periodicReaderConfig{
+		interval: defaultInterval,
+		timeout:  defaultTimeout,
+	}
+	for _, o := range options {
+		c = o.apply(c)
+	}
+	return c
+}
+
+// PeriodicReaderOption applies a configuration option value to a PeriodicReader.
+type PeriodicReaderOption interface {
+	apply(periodicReaderConfig) periodicReaderConfig
+}
+
+// periodicReaderOptionFunc applies a set of options to a periodicReaderConfig.
+type periodicReaderOptionFunc func(periodicReaderConfig) periodicReaderConfig
+
+// apply returns a periodicReaderConfig with option(s) applied.
+func (o periodicReaderOptionFunc) apply(conf periodicReaderConfig) periodicReaderConfig {
+	return o(conf)
+}
+
+// WithTimeout configures the time a PeriodicReader waits for an export to
+// complete before canceling it.
+//
+// If this option is not used or d is less than or equal to zero, 30 seconds
+// is used as the default.
+func WithTimeout(d time.Duration) PeriodicReaderOption {
+	return periodicReaderOptionFunc(func(conf periodicReaderConfig) periodicReaderConfig {
+		if d <= 0 {
+			return conf
+		}
+		conf.timeout = d
+		return conf
+	})
+}
+
+// WithInterval configures the intervening time between exports for a
+// PeriodicReader.
+//
+// If this option is not used or d is less than or equal to zero, 60 seconds
+// is used as the default.
+func WithInterval(d time.Duration) PeriodicReaderOption {
+	return periodicReaderOptionFunc(func(conf periodicReaderConfig) periodicReaderConfig {
+		if d <= 0 {
+			return conf
+		}
+		conf.interval = d
+		return conf
+	})
+}
+
+// NewPeriodicReader returns a Reader that collects and exports metric data to
+// the exporter at the defined interval.
+//
+// The Collect method of the returned Reader continues to gather and return
+// metric data to the user. It will not automatically send that data to the
+// exporter.
+func NewPeriodicReader(exporter Exporter, options ...PeriodicReaderOption) Reader {
+	conf := newPeriodicReaderConfig(options)
+	ctx, cancel := context.WithCancel(context.Background())
+	r := &periodicReader{
+		timeout:  conf.timeout,
+		exporter: exporter,
+		cancel:   cancel,
+	}
+
+	r.wg.Add(1)
+	go func() {
+		defer r.wg.Done()
+		r.run(ctx, conf.interval)
+	}()
+
+	return r
+}
+
+// periodicReader is a Reader that continuously collects and exports metric
+// data at a set interval.
+type periodicReader struct {
+	producer atomic.Value
+
+	timeout  time.Duration
+	exporter Exporter
+
+	wg           sync.WaitGroup
+	cancel       context.CancelFunc
+	shutdownOnce sync.Once
+}
+
+// newTicker allows testing override.
+var newTicker = time.NewTicker
+
+// run continuously collects and exports metric data for the specified
+// interval. This will run until ctx is canceled or times out.
+func (r *periodicReader) run(ctx context.Context, interval time.Duration) {
+	ticker := newTicker(interval)
+	defer func() { ticker.Stop() }()
+
+	for {
+		select {
+		case <-ticker.C:
+			m, err := r.Collect(ctx)
+			if err == nil {
+				c, cancel := context.WithTimeout(ctx, r.timeout)
+				err = r.exporter.Export(c, m)
+				cancel()
+			}
+			if err != nil {
+				otel.Handle(err)
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// register registers p as the producer of this reader.
+func (r *periodicReader) register(p producer) {
+	r.producer.Store(produceHolder{produce: p.produce})
+}
+
+// Collect gathers and returns all metric data related to the Reader from
+// the SDK. An error is returned if this is called after Shutdown.
+func (r *periodicReader) Collect(ctx context.Context) (export.Metrics, error) {
+	p := r.producer.Load()
+	if p == nil {
+		return export.Metrics{}, ErrReaderNotRegistered
+	}
+
+	ph, ok := p.(produceHolder)
+	if !ok {
+		// The atomic.Value is entirely in the periodicReader's control so
+		// this should never happen. In the unforeseen case that this does
+		// happen, return an error instead of panicking so a users code does
+		// not halt in the processes.
+		err := fmt.Errorf("periodic reader: invalid producer: %T", p)
+		return export.Metrics{}, err
+	}
+	return ph.produce(ctx)
+}
+
+// ForceFlush flushes the Exporter.
+func (r *periodicReader) ForceFlush(ctx context.Context) error {
+	return r.exporter.ForceFlush(ctx)
+}
+
+// Shutdown stops the export pipeline.
+func (r *periodicReader) Shutdown(ctx context.Context) error {
+	err := ErrReaderShutdown
+	r.shutdownOnce.Do(func() {
+		// Stop the run loop.
+		r.cancel()
+		r.wg.Wait()
+
+		// Any future call to Collect will now return ErrReaderShutdown.
+		r.producer.Store(produceHolder{
+			produce: shutdownProducer{}.produce,
+		})
+
+		err = r.exporter.Shutdown(ctx)
+	})
+	return err
+}
+
+// produceHolder is used as an atomic.Value to wrap the non-concrete producer
+// type.
+type produceHolder struct {
+	produce func(context.Context) (export.Metrics, error)
+}
+
+// shutdownProducer produces an ErrReaderShutdown error always.
+type shutdownProducer struct{}
+
+// produce returns an ErrReaderShutdown error.
+func (p shutdownProducer) produce(context.Context) (export.Metrics, error) {
+	return export.Metrics{}, ErrReaderShutdown
+}