Add support for reader.

metric/metricexport/export.go

@@ -0,0 +1,26 @@
+// Copyright 2019, OpenCensus 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 metricexport
+
+import (
+	"context"
+
+	"go.opencensus.io/metric/metricdata"
+)
+
+// Exporter is an interface that exporters implement to export the metric data.
+type Exporter interface {
+	ExportMetrics(ctx context.Context, data []*metricdata.Metric) error
+}

metric/metricexport/reader.go

@@ -0,0 +1,187 @@
+// Copyright 2019, OpenCensus 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 metricexport
+
+import (
+	"fmt"
+	"time"
+
+	"context"
+	"go.opencensus.io/metric/metricdata"
+	"go.opencensus.io/metric/metricproducer"
+	"go.opencensus.io/trace"
+	"sync"
+)
+
+var (
+	defaultSampler             = trace.ProbabilitySampler(0.0001)
+	errReportingIntervalTooLow = fmt.Errorf("reporting interval less than %d", minimumReportingDuration)
+	errAlreadyStarted          = fmt.Errorf("already started")
+	errIntervalReaderNil       = fmt.Errorf("interval reader is nil")
+	errExporterNil             = fmt.Errorf("exporter is nil")
+	errReaderNil               = fmt.Errorf("reader is nil")
+)
+
+const (
+	defaultReportingDuration = 60 * time.Second
+	minimumReportingDuration = 1 * time.Second
+	defaultSpanName          = "ExportMetrics"
+)
+
+// ReaderOptions contains options pertaining to metrics reader.
+type ReaderOptions struct {
+	// SpanName is the name used for span created to export metrics.
+	SpanName string
+}
+
+// Reader reads metrics from all producers registered
+// with producer manager and exports those metrics using provided
+// exporter.
+type Reader struct {
+	sampler trace.Sampler
+
+	spanName string
+}
+
+// IntervalReader periodically reads metrics from all producers registered
+// with producer manager and exports those metrics using provided
+// exporter. Call Reader.Stop() to stop the reader.
+type IntervalReader struct {
+	// ReportingInterval it the time duration between two consecutive
+	// metrics reporting. defaultReportingDuration  is used if it is not set.
+	// It cannot be set lower than minimumReportingDuration.
+	ReportingInterval time.Duration
+
+	exporter   Exporter
+	timer      *time.Ticker
+	quit, done chan bool
+	mu         sync.RWMutex
+	reader     *Reader
+}
+
+// ReaderOption apply changes to ReaderOptions.
+type ReaderOption func(*ReaderOptions)
+
+// WithSpanName makes new reader to use given span name when exporting metrics.
+func WithSpanName(spanName string) ReaderOption {
+	return func(o *ReaderOptions) {
+		o.SpanName = spanName
+	}
+}
+
+// NewReader returns a reader configured with specified options.
+func NewReader(o ...ReaderOption) *Reader {
+	var opts ReaderOptions
+	for _, op := range o {
+		op(&opts)
+	}
+	reader := &Reader{defaultSampler, defaultSpanName}
+	if opts.SpanName != "" {
+		reader.spanName = opts.SpanName
+	}
+	return reader
+}
+
+// NewIntervalReader creates a reader. Once started it periodically
+// reads metrics from all producers and exports them using provided exporter.
+func NewIntervalReader(reader *Reader, exporter Exporter) (*IntervalReader, error) {
+	if exporter == nil {
+		return nil, errExporterNil
+	}
+	if reader == nil {
+		return nil, errReaderNil
+	}
+
+	r := &IntervalReader{
+		exporter: exporter,
+		reader:   reader,
+	}
+	return r, nil
+}
+
+// Start starts the IntervalReader which periodically reads metrics from all
+// producers registered with global producer manager. If the reporting interval
+// is not set prior to calling this function then default reporting interval
+// is used.
+func (ir *IntervalReader) Start() error {
+	if ir == nil {
+		return errIntervalReaderNil
+	}
+	ir.mu.Lock()
+	defer ir.mu.Unlock()
+	var reportingInterval = defaultReportingDuration
+	if ir.ReportingInterval != 0 {
+		if ir.ReportingInterval < minimumReportingDuration {
+			return errReportingIntervalTooLow
+		}
+		reportingInterval = ir.ReportingInterval
+	}
+
+	if ir.done != nil {
+		return errAlreadyStarted
+	}
+	ir.timer = time.NewTicker(reportingInterval)
+	ir.quit = make(chan bool)
+	ir.done = make(chan bool)
+
+	go ir.startInternal()
+	return nil
+}
+
+func (ir *IntervalReader) startInternal() {
+	for {
+		select {
+		case <-ir.timer.C:
+			ir.reader.ReadAndExport(ir.exporter)
+		case <-ir.quit:
+			ir.timer.Stop()
+			ir.done <- true
+			return
+		}
+	}
+}
+
+// Stop stops the reader from reading and exporting metrics.
+// Additional call to Stop are no-ops.
+func (ir *IntervalReader) Stop() {
+	if ir == nil {
+		return
+	}
+	ir.mu.Lock()
+	defer ir.mu.Unlock()
+	if ir.quit == nil {
+		return
+	}
+	ir.quit <- true
+	<-ir.done
+	close(ir.quit)
+	close(ir.done)
+	ir.quit = nil
+}
+
+// ReadAndExport reads metrics from all producer registered with
+// producer manager and then exports them using provided exporter.
+func (r *Reader) ReadAndExport(exporter Exporter) {
+	ctx, span := trace.StartSpan(context.Background(), r.spanName, trace.WithSampler(r.sampler))
+	defer span.End()
+	producers := metricproducer.GlobalManager().GetAll()
+	data := []*metricdata.Metric{}
+	for _, producer := range producers {
+		data = append(data, producer.Read()...)
+	}
+	// TODO: [rghetia] add metrics for errors.
+	exporter.ExportMetrics(ctx, data)
+}