Repository: census-ecosystem/opencensus-go-exporter-stackdriver
Pull request #191 (2019-08-30 17:07:53)

Revert "Remove support for exporting ViewData (#188)"
This reverts commit 4b7f30648a4f2cf529cecb4274f016b9cc30df7e.

#188 has caused some issues to those who still use the old interface. It should be better to follow the standard deprecation policy (https://github.com/census-instrumentation/opencensus-go#deprecation-policy) before we remove it.


Patch:
diff --git a/stackdriver.go b/stackdriver.go
index 6e766bd..a15972a 100644
--- a/stackdriver.go
+++ b/stackdriver.go
@@ -341,6 +341,13 @@ func NewExporter(o Options) (*Exporter, error) {
 	}, nil
 }
 
+// ExportView exports to the Stackdriver Monitoring if view data
+// has one or more rows.
+// Deprecated: use ExportMetrics and StartMetricsExporter instead.
+func (e *Exporter) ExportView(vd *view.Data) {
+	e.statsExporter.ExportView(vd)
+}
+
 // ExportMetricsProto exports OpenCensus Metrics Proto to Stackdriver Monitoring synchronously,
 // without de-duping or adding proto metrics to the bundler.
 func (e *Exporter) ExportMetricsProto(ctx context.Context, node *commonpb.Node, rsc *resourcepb.Resource, metrics []*metricspb.Metric) error {
diff --git a/stats.go b/stats.go
index 09c5b3a..2528907 100644
--- a/stats.go
+++ b/stats.go
@@ -23,15 +23,25 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	opencensus "go.opencensus.io"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+	"go.opencensus.io/trace"
 
 	monitoring "cloud.google.com/go/monitoring/apiv3"
+	"github.com/golang/protobuf/ptypes/timestamp"
 	"go.opencensus.io/metric/metricdata"
 	"go.opencensus.io/metric/metricexport"
 	"google.golang.org/api/option"
 	"google.golang.org/api/support/bundler"
+	distributionpb "google.golang.org/genproto/googleapis/api/distribution"
+	labelpb "google.golang.org/genproto/googleapis/api/label"
 	"google.golang.org/genproto/googleapis/api/metric"
+	metricpb "google.golang.org/genproto/googleapis/api/metric"
+	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
 	monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
 )
 
@@ -49,7 +59,11 @@ var userAgent = fmt.Sprintf("opencensus-go %s; stackdriver-exporter %s", opencen
 type statsExporter struct {
 	o Options
 
-	metricsBundler *bundler.Bundler
+	viewDataBundler *bundler.Bundler
+	metricsBundler  *bundler.Bundler
+
+	createdViewsMu sync.Mutex
+	createdViews   map[string]*metricpb.MetricDescriptor // Views already created remotely
 
 	protoMu                sync.Mutex
 	protoMetricDescriptors map[string]bool // Saves the metric descriptors that were already created remotely
@@ -88,6 +102,7 @@ func newStatsExporter(o Options) (*statsExporter, error) {
 	e := &statsExporter{
 		c:                      client,
 		o:                      o,
+		createdViews:           make(map[string]*metricpb.MetricDescriptor),
 		protoMetricDescriptors: make(map[string]bool),
 		metricDescriptors:      make(map[string]bool),
 	}
@@ -107,14 +122,20 @@ func newStatsExporter(o Options) (*statsExporter, error) {
 		e.defaultLabels[sanitize(key)] = label
 	}
 
+	e.viewDataBundler = bundler.NewBundler((*view.Data)(nil), func(bundle interface{}) {
+		vds := bundle.([]*view.Data)
+		e.handleUpload(vds...)
+	})
 	e.metricsBundler = bundler.NewBundler((*metricdata.Metric)(nil), func(bundle interface{}) {
 		metrics := bundle.([]*metricdata.Metric)
 		e.handleMetricsUpload(metrics)
 	})
 	if delayThreshold := e.o.BundleDelayThreshold; delayThreshold > 0 {
+		e.viewDataBundler.DelayThreshold = delayThreshold
 		e.metricsBundler.DelayThreshold = delayThreshold
 	}
 	if countThreshold := e.o.BundleCountThreshold; countThreshold > 0 {
+		e.viewDataBundler.BundleCountThreshold = countThreshold
 		e.metricsBundler.BundleCountThreshold = countThreshold
 	}
 	return e, nil
@@ -134,6 +155,33 @@ func (e *statsExporter) stopMetricsReader() {
 	}
 }
 
+func (e *statsExporter) getMonitoredResource(v *view.View, tags []tag.Tag) ([]tag.Tag, *monitoredrespb.MonitoredResource) {
+	resource := e.o.Resource
+	if resource == nil {
+		resource = &monitoredrespb.MonitoredResource{
+			Type: "global",
+		}
+	}
+	return tags, resource
+}
+
+// ExportView exports to the Stackdriver Monitoring if view data
+// has one or more rows.
+func (e *statsExporter) ExportView(vd *view.Data) {
+	if len(vd.Rows) == 0 {
+		return
+	}
+	err := e.viewDataBundler.Add(vd, 1)
+	switch err {
+	case nil:
+		return
+	case bundler.ErrOverflow:
+		e.o.handleError(errors.New("failed to upload: buffer full"))
+	default:
+		e.o.handleError(err)
+	}
+}
+
 // getTaskValue returns a task label value in the format of
 // "go-<pid>@<hostname>".
 func getTaskValue() string {
@@ -144,14 +192,200 @@ func getTaskValue() string {
 	return "go-" + strconv.Itoa(os.Getpid()) + "@" + hostname
 }
 
+// handleUpload handles uploading a slice
+// of Data, as well as error handling.
+func (e *statsExporter) handleUpload(vds ...*view.Data) {
+	if err := e.uploadStats(vds); err != nil {
+		e.o.handleError(err)
+	}
+}
+
 // Flush waits for exported view data and metrics to be uploaded.
 //
 // This is useful if your program is ending and you do not
 // want to lose data that hasn't yet been exported.
 func (e *statsExporter) Flush() {
+	e.viewDataBundler.Flush()
 	e.metricsBundler.Flush()
 }
 
+func (e *statsExporter) uploadStats(vds []*view.Data) error {
+	ctx, cancel := e.o.newContextWithTimeout()
+	defer cancel()
+	ctx, span := trace.StartSpan(
+		ctx,
+		"contrib.go.opencensus.io/exporter/stackdriver.uploadStats",
+		trace.WithSampler(trace.NeverSample()),
+	)
+	defer span.End()
+
+	for _, vd := range vds {
+		if err := e.createMeasure(ctx, vd.View); err != nil {
+			span.SetStatus(trace.Status{Code: 2, Message: err.Error()})
+			return err
+		}
+	}
+	for _, req := range e.makeReq(vds, maxTimeSeriesPerUpload) {
+		if err := createTimeSeries(ctx, e.c, req); err != nil {
+			span.SetStatus(trace.Status{Code: 2, Message: err.Error()})
+			// TODO(jbd): Don't fail fast here, batch errors?
+			return err
+		}
+	}
+	return nil
+}
+
+func (e *statsExporter) makeReq(vds []*view.Data, limit int) []*monitoringpb.CreateTimeSeriesRequest {
+	var reqs []*monitoringpb.CreateTimeSeriesRequest
+
+	var allTimeSeries []*monitoringpb.TimeSeries
+	for _, vd := range vds {
+		for _, row := range vd.Rows {
+			tags, resource := e.getMonitoredResource(vd.View, append([]tag.Tag(nil), row.Tags...))
+			ts := &monitoringpb.TimeSeries{
+				Metric: &metricpb.Metric{
+					Type:   e.metricType(vd.View),
+					Labels: newLabels(e.defaultLabels, tags),
+				},
+				Resource: resource,
+				Points:   []*monitoringpb.Point{newPoint(vd.View, row, vd.Start, vd.End)},
+			}
+			allTimeSeries = append(allTimeSeries, ts)
+		}
+	}
+
+	var timeSeries []*monitoringpb.TimeSeries
+	for _, ts := range allTimeSeries {
+		timeSeries = append(timeSeries, ts)
+		if len(timeSeries) == limit {
+			ctsreql := e.combineTimeSeriesToCreateTimeSeriesRequest(timeSeries)
+			reqs = append(reqs, ctsreql...)
+			timeSeries = timeSeries[:0]
+		}
+	}
+
+	if len(timeSeries) > 0 {
+		ctsreql := e.combineTimeSeriesToCreateTimeSeriesRequest(timeSeries)
+		reqs = append(reqs, ctsreql...)
+	}
+	return reqs
+}
+
+func (e *statsExporter) viewToMetricDescriptor(ctx context.Context, v *view.View) (*metricpb.MetricDescriptor, error) {
+	m := v.Measure
+	agg := v.Aggregation
+	viewName := v.Name
+
+	metricType := e.metricType(v)
+	var valueType metricpb.MetricDescriptor_ValueType
+	unit := m.Unit()
+	// Default metric Kind
+	metricKind := metricpb.MetricDescriptor_CUMULATIVE
+
+	switch agg.Type {
+	case view.AggTypeCount:
+		valueType = metricpb.MetricDescriptor_INT64
+		// If the aggregation type is count, which counts the number of recorded measurements, the unit must be "1",
+		// because this view does not apply to the recorded values.
+		unit = stats.UnitDimensionless
+	case view.AggTypeSum:
+		switch m.(type) {
+		case *stats.Int64Measure:
+			valueType = metricpb.MetricDescriptor_INT64
+		case *stats.Float64Measure:
+			valueType = metricpb.MetricDescriptor_DOUBLE
+		}
+	case view.AggTypeDistribution:
+		valueType = metricpb.MetricDescriptor_DISTRIBUTION
+	case view.AggTypeLastValue:
+		metricKind = metricpb.MetricDescriptor_GAUGE
+		switch m.(type) {
+		case *stats.Int64Measure:
+			valueType = metricpb.MetricDescriptor_INT64
+		case *stats.Float64Measure:
+			valueType = metricpb.MetricDescriptor_DOUBLE
+		}
+	default:
+		return nil, fmt.Errorf("unsupported aggregation type: %s", agg.Type.String())
+	}
+
+	var displayName string
+	if e.o.GetMetricDisplayName == nil {
+		displayName = e.displayName(viewName)
+	} else {
+		displayName = e.o.GetMetricDisplayName(v)
+	}
+
+	res := &metricpb.MetricDescriptor{
+		Name:        fmt.Sprintf("projects/%s/metricDescriptors/%s", e.o.ProjectID, metricType),
+		DisplayName: displayName,
+		Description: v.Description,
+		Unit:        unit,
+		Type:        metricType,
+		MetricKind:  metricKind,
+		ValueType:   valueType,
+		Labels:      newLabelDescriptors(e.defaultLabels, v.TagKeys),
+	}
+	return res, nil
+}
+
+func (e *statsExporter) viewToCreateMetricDescriptorRequest(ctx context.Context, v *view.View) (*monitoringpb.CreateMetricDescriptorRequest, error) {
+	inMD, err := e.viewToMetricDescriptor(ctx, v)
+	if err != nil {
+		return nil, err
+	}
+
+	cmrdesc := &monitoringpb.CreateMetricDescriptorRequest{
+		Name:             fmt.Sprintf("projects/%s", e.o.ProjectID),
+		MetricDescriptor: inMD,
+	}
+	return cmrdesc, nil
+}
+
+// createMeasure creates a MetricDescriptor for the given view data in Stackdriver Monitoring.
+// An error will be returned if there is already a metric descriptor created with the same name
+// but it has a different aggregation or keys.
+func (e *statsExporter) createMeasure(ctx context.Context, v *view.View) error {
+	e.createdViewsMu.Lock()
+	defer e.createdViewsMu.Unlock()
+
+	viewName := v.Name
+
+	if md, ok := e.createdViews[viewName]; ok {
+		// [TODO:rghetia] Temporary fix for https://github.com/census-ecosystem/opencensus-go-exporter-stackdriver/issues/76#issuecomment-459459091
+		if builtinMetric(md.Type) {
+			return nil
+		}
+		return e.equalMeasureAggTagKeys(md, v.Measure, v.Aggregation, v.TagKeys)
+	}
+
+	inMD, err := e.viewToMetricDescriptor(ctx, v)
+	if err != nil {
+		return err
+	}
+
+	var dmd *metric.MetricDescriptor
+	if builtinMetric(inMD.Type) {
+		gmrdesc := &monitoringpb.GetMetricDescriptorRequest{
+			Name: inMD.Name,
+		}
+		dmd, err = getMetricDescriptor(ctx, e.c, gmrdesc)
+	} else {
+		cmrdesc := &monitoringpb.CreateMetricDescriptorRequest{
+			Name:             fmt.Sprintf("projects/%s", e.o.ProjectID),
+			MetricDescriptor: inMD,
+		}
+		dmd, err = createMetricDescriptor(ctx, e.c, cmrdesc)
+	}
+	if err != nil {
+		return err
+	}
+
+	// Now cache the metric descriptor
+	e.createdViews[viewName] = dmd
+	return err
+}
+
 func (e *statsExporter) displayName(suffix string) string {
 	displayNamePrefix := defaultDisplayNamePrefix
 	if e.o.MetricPrefix != "" {
@@ -160,6 +394,98 @@ func (e *statsExporter) displayName(suffix string) string {
 	return path.Join(displayNamePrefix, suffix)
 }
 
+func newPoint(v *view.View, row *view.Row, start, end time.Time) *monitoringpb.Point {
+	switch v.Aggregation.Type {
+	case view.AggTypeLastValue:
+		return newGaugePoint(v, row, end)
+	default:
+		return newCumulativePoint(v, row, start, end)
+	}
+}
+
+func newCumulativePoint(v *view.View, row *view.Row, start, end time.Time) *monitoringpb.Point {
+	return &monitoringpb.Point{
+		Interval: &monitoringpb.TimeInterval{
+			StartTime: &timestamp.Timestamp{
+				Seconds: start.Unix(),
+				Nanos:   int32(start.Nanosecond()),
+			},
+			EndTime: &timestamp.Timestamp{
+				Seconds: end.Unix(),
+				Nanos:   int32(end.Nanosecond()),
+			},
+		},
+		Value: newTypedValue(v, row),
+	}
+}
+
+func newGaugePoint(v *view.View, row *view.Row, end time.Time) *monitoringpb.Point {
+	gaugeTime := &timestamp.Timestamp{
+		Seconds: end.Unix(),
+		Nanos:   int32(end.Nanosecond()),
+	}
+	return &monitoringpb.Point{
+		Interval: &monitoringpb.TimeInterval{
+			EndTime: gaugeTime,
+		},
+		Value: newTypedValue(v, row),
+	}
+}
+
+func newTypedValue(vd *view.View, r *view.Row) *monitoringpb.TypedValue {
+	switch v := r.Data.(type) {
+	case *view.CountData:
+		return &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+			Int64Value: v.Value,
+		}}
+	case *view.SumData:
+		switch vd.Measure.(type) {
+		case *stats.Int64Measure:
+			return &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+				Int64Value: int64(v.Value),
+			}}
+		case *stats.Float64Measure:
+			return &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_DoubleValue{
+				DoubleValue: v.Value,
+			}}
+		}
+	case *view.DistributionData:
+		insertZeroBound := shouldInsertZeroBound(vd.Aggregation.Buckets...)
+		return &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_DistributionValue{
+			DistributionValue: &distributionpb.Distribution{
+				Count:                 v.Count,
+				Mean:                  v.Mean,
+				SumOfSquaredDeviation: v.SumOfSquaredDev,
+				// TODO(songya): uncomment this once Stackdriver supports min/max.
+				// Range: &distributionpb.Distribution_Range{
+				// 	Min: v.Min,
+				// 	Max: v.Max,
+				// },
+				BucketOptions: &distributionpb.Distribution_BucketOptions{
+					Options: &distributionpb.Distribution_BucketOptions_ExplicitBuckets{
+						ExplicitBuckets: &distributionpb.Distribution_BucketOptions_Explicit{
+							Bounds: addZeroBoundOnCondition(insertZeroBound, vd.Aggregation.Buckets...),
+						},
+					},
+				},
+				BucketCounts: addZeroBucketCountOnCondition(insertZeroBound, v.CountPerBucket...),
+			},
+		}}
+	case *view.LastValueData:
+		switch vd.Measure.(type) {
+		case *stats.Int64Measure:
+			return &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+				Int64Value: int64(v.Value),
+			}}
+		case *stats.Float64Measure:
+			return &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_DoubleValue{
+				DoubleValue: v.Value,
+			}}
+		}
+	}
+	return nil
+}
+
 func shouldInsertZeroBound(bounds ...float64) bool {
 	if len(bounds) > 0 && bounds[0] != 0.0 {
 		return true
@@ -181,6 +507,89 @@ func addZeroBoundOnCondition(insert bool, bounds ...float64) []float64 {
 	return bounds
 }
 
+func (e *statsExporter) metricType(v *view.View) string {
+	if formatter := e.o.GetMetricType; formatter != nil {
+		return formatter(v)
+	}
+	return path.Join("custom.googleapis.com", "opencensus", v.Name)
+}
+
+func newLabels(defaults map[string]labelValue, tags []tag.Tag) map[string]string {
+	labels := make(map[string]string)
+	for k, lbl := range defaults {
+		labels[sanitize(k)] = lbl.val
+	}
+	for _, tag := range tags {
+		labels[sanitize(tag.Key.Name())] = tag.Value
+	}
+	return labels
+}
+
+func newLabelDescriptors(defaults map[string]labelValue, keys []tag.Key) []*labelpb.LabelDescriptor {
+	labelDescriptors := make([]*labelpb.LabelDescriptor, 0, len(keys)+len(defaults))
+	for key, lbl := range defaults {
+		labelDescriptors = append(labelDescriptors, &labelpb.LabelDescriptor{
+			Key:         sanitize(key),
+			Description: lbl.desc,
+			ValueType:   labelpb.LabelDescriptor_STRING,
+		})
+	}
+	for _, key := range keys {
+		labelDescriptors = append(labelDescriptors, &labelpb.LabelDescriptor{
+			Key:       sanitize(key.Name()),
+			ValueType: labelpb.LabelDescriptor_STRING, // We only use string tags
+		})
+	}
+	return labelDescriptors
+}
+
+func (e *statsExporter) equalMeasureAggTagKeys(md *metricpb.MetricDescriptor, m stats.Measure, agg *view.Aggregation, keys []tag.Key) error {
+	var aggTypeMatch bool
+	switch md.ValueType {
+	case metricpb.MetricDescriptor_INT64:
+		if _, ok := m.(*stats.Int64Measure); !(ok || agg.Type == view.AggTypeCount) {
+			return fmt.Errorf("stackdriver metric descriptor was not created as int64")
+		}
+		aggTypeMatch = agg.Type == view.AggTypeCount || agg.Type == view.AggTypeSum || agg.Type == view.AggTypeLastValue
+	case metricpb.MetricDescriptor_DOUBLE:
+		if _, ok := m.(*stats.Float64Measure); !ok {
+			return fmt.Errorf("stackdriver metric descriptor was not created as double")
+		}
+		aggTypeMatch = agg.Type == view.AggTypeSum || agg.Type == view.AggTypeLastValue
+	case metricpb.MetricDescriptor_DISTRIBUTION:
+		aggTypeMatch = agg.Type == view.AggTypeDistribution
+	}
+
+	if !aggTypeMatch {
+		return fmt.Errorf("stackdriver metric descriptor was not created with aggregation type %T", agg.Type)
+	}
+
+	labels := make(map[string]struct{}, len(keys)+len(e.defaultLabels))
+	for _, k := range keys {
+		labels[sanitize(k.Name())] = struct{}{}
+	}
+	for k := range e.defaultLabels {
+		labels[sanitize(k)] = struct{}{}
+	}
+
+	for _, k := range md.Labels {
+		if _, ok := labels[k.Key]; !ok {
+			return fmt.Errorf("stackdriver metric descriptor %q was not created with label %q", md.Type, k)
+		}
+		delete(labels, k.Key)
+	}
+
+	if len(labels) > 0 {
+		extra := make([]string, 0, len(labels))
+		for k := range labels {
+			extra = append(extra, k)
+		}
+		return fmt.Errorf("stackdriver metric descriptor %q contains unexpected labels: %s", md.Type, strings.Join(extra, ", "))
+	}
+
+	return nil
+}
+
 var createMetricDescriptor = func(ctx context.Context, c *monitoring.MetricClient, mdr *monitoringpb.CreateMetricDescriptorRequest) (*metric.MetricDescriptor, error) {
 	return c.CreateMetricDescriptor(ctx, mdr)
 }


Test patch:
diff --git a/example_test.go b/example_test.go
index 7c6d304..d8e581d 100644
--- a/example_test.go
+++ b/example_test.go
@@ -34,6 +34,11 @@ func Example_defaults() {
 		log.Fatal(err)
 	}
 
+	// Export to Stackdriver Monitoring.
+	if err = exporter.StartMetricsExporter(); err != nil {
+		log.Fatal(err)
+	}
+
 	// Subscribe views to see stats in Stackdriver Monitoring.
 	if err := view.Register(
 		ochttp.ClientLatencyView,
@@ -73,7 +78,7 @@ func Example_gKE() {
 		zone = "unknown"
 	}
 
-	_, err = stackdriver.NewExporter(stackdriver.Options{
+	exporter, err := stackdriver.NewExporter(stackdriver.Options{
 		ProjectID: "google-project-id",
 		// Set a MonitoredResource that represents a GKE container.
 		Resource: &monitoredres.MonitoredResource{
@@ -100,4 +105,9 @@ func Example_gKE() {
 	if err != nil {
 		log.Fatal(err)
 	}
+
+	// Register so that views are exported.
+	if err = exporter.StartMetricsExporter(); err != nil {
+		log.Fatal(err)
+	}
 }
diff --git a/stackdriver_test.go b/stackdriver_test.go
index 6d13429..3238489 100644
--- a/stackdriver_test.go
+++ b/stackdriver_test.go
@@ -25,6 +25,7 @@ import (
 
 	"contrib.go.opencensus.io/exporter/stackdriver/internal/testpb"
 	"go.opencensus.io/plugin/ochttp"
+	"go.opencensus.io/stats/view"
 	"go.opencensus.io/trace"
 	"golang.org/x/net/context/ctxhttp"
 )
@@ -47,6 +48,8 @@ func TestExport(t *testing.T) {
 
 	trace.RegisterExporter(exporter)
 	defer trace.UnregisterExporter(exporter)
+	view.RegisterExporter(exporter)
+	defer view.UnregisterExporter(exporter)
 
 	trace.ApplyConfig(trace.Config{DefaultSampler: trace.AlwaysSample()})
 
@@ -110,6 +113,8 @@ func TestGRPC(t *testing.T) {
 
 	trace.RegisterExporter(exporter)
 	defer trace.UnregisterExporter(exporter)
+	view.RegisterExporter(exporter)
+	defer view.UnregisterExporter(exporter)
 
 	trace.ApplyConfig(trace.Config{DefaultSampler: trace.AlwaysSample()})
 
diff --git a/stats_test.go b/stats_test.go
index 1547aa3..9596262 100644
--- a/stats_test.go
+++ b/stats_test.go
@@ -15,9 +15,26 @@
 package stackdriver
 
 import (
+	"context"
+	"fmt"
 	"testing"
+	"time"
 
+	"contrib.go.opencensus.io/exporter/stackdriver/monitoredresource"
+
+	"cloud.google.com/go/monitoring/apiv3"
+	"github.com/golang/protobuf/ptypes/timestamp"
+	"github.com/google/go-cmp/cmp"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
 	"google.golang.org/api/option"
+	"google.golang.org/genproto/googleapis/api/distribution"
+	"google.golang.org/genproto/googleapis/api/label"
+	"google.golang.org/genproto/googleapis/api/metric"
+	metricpb "google.golang.org/genproto/googleapis/api/metric"
+	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
+	monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
 	"google.golang.org/grpc"
 )
 
@@ -35,3 +52,1306 @@ func TestRejectBlankProjectID(t *testing.T) {
 		}
 	}
 }
+
+func TestExporter_makeReq(t *testing.T) {
+	m := stats.Float64("test-measure", "measure desc", "unit")
+
+	key, err := tag.NewKey("test_key")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	v := &view.View{
+		Name:        "example.com/views/testview",
+		Description: "desc",
+		TagKeys:     []tag.Key{key},
+		Measure:     m,
+		Aggregation: view.Count(),
+	}
+
+	lastValueView := &view.View{
+		Name:        "lasttestview",
+		Description: "desc",
+		TagKeys:     []tag.Key{key},
+		Measure:     m,
+		Aggregation: view.LastValue(),
+	}
+
+	distView := &view.View{
+		Name:        "distview",
+		Description: "desc",
+		Measure:     m,
+		Aggregation: view.Distribution(2, 4, 7),
+	}
+
+	start := time.Now()
+	end := start.Add(time.Minute)
+	count1 := &view.CountData{Value: 10}
+	count2 := &view.CountData{Value: 16}
+	sum1 := &view.SumData{Value: 5.5}
+	sum2 := &view.SumData{Value: -11.1}
+	last1 := view.LastValueData{Value: 100}
+	last2 := view.LastValueData{Value: 200}
+	taskValue := getTaskValue()
+
+	tests := []struct {
+		name   string
+		projID string
+		vd     *view.Data
+		want   []*monitoringpb.CreateTimeSeriesRequest
+		opts   Options
+	}{
+		{
+			name:   "count agg + timeline",
+			projID: "proj-id",
+			vd:     newTestViewData(v, start, end, count1, count2),
+			want: []*monitoringpb.CreateTimeSeriesRequest{
+				{
+					Name: monitoring.MetricProjectPath("proj-id"),
+					TimeSeries: []*monitoringpb.TimeSeries{
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/example.com/views/testview",
+								Labels: map[string]string{
+									"test_key":        "test-value-1",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: &monitoredrespb.MonitoredResource{
+								Type: "global",
+							},
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+										Int64Value: 10,
+									}},
+								},
+							},
+						},
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/example.com/views/testview",
+								Labels: map[string]string{
+									"test_key":        "test-value-2",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: &monitoredrespb.MonitoredResource{
+								Type: "global",
+							},
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+										Int64Value: 16,
+									}},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name:   "metric type formatter",
+			projID: "proj-id",
+			vd:     newTestViewData(v, start, end, sum1, sum2),
+			opts: Options{
+				GetMetricType: func(v *view.View) string {
+					return fmt.Sprintf("external.googleapis.com/%s", v.Name)
+				},
+			},
+			want: []*monitoringpb.CreateTimeSeriesRequest{
+				{
+					Name: monitoring.MetricProjectPath("proj-id"),
+					TimeSeries: []*monitoringpb.TimeSeries{
+						{
+							Metric: &metricpb.Metric{
+								Type: "external.googleapis.com/example.com/views/testview",
+								Labels: map[string]string{
+									"test_key":        "test-value-1",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: &monitoredrespb.MonitoredResource{
+								Type: "global",
+							},
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_DoubleValue{
+										DoubleValue: 5.5,
+									}},
+								},
+							},
+						},
+						{
+							Metric: &metricpb.Metric{
+								Type: "external.googleapis.com/example.com/views/testview",
+								Labels: map[string]string{
+									"test_key":        "test-value-2",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: &monitoredrespb.MonitoredResource{
+								Type: "global",
+							},
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_DoubleValue{
+										DoubleValue: -11.1,
+									}},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name:   "sum agg + timeline",
+			projID: "proj-id",
+			vd:     newTestViewData(v, start, end, sum1, sum2),
+			want: []*monitoringpb.CreateTimeSeriesRequest{
+				{
+					Name: monitoring.MetricProjectPath("proj-id"),
+					TimeSeries: []*monitoringpb.TimeSeries{
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/example.com/views/testview",
+								Labels: map[string]string{
+									"test_key":        "test-value-1",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: &monitoredrespb.MonitoredResource{
+								Type: "global",
+							},
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_DoubleValue{
+										DoubleValue: 5.5,
+									}},
+								},
+							},
+						},
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/example.com/views/testview",
+								Labels: map[string]string{
+									"test_key":        "test-value-2",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: &monitoredrespb.MonitoredResource{
+								Type: "global",
+							},
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_DoubleValue{
+										DoubleValue: -11.1,
+									}},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name:   "last value agg",
+			projID: "proj-id",
+			vd:     newTestViewData(lastValueView, start, end, &last1, &last2),
+			want: []*monitoringpb.CreateTimeSeriesRequest{
+				{
+					Name: monitoring.MetricProjectPath("proj-id"),
+					TimeSeries: []*monitoringpb.TimeSeries{
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/lasttestview",
+								Labels: map[string]string{
+									"test_key":        "test-value-1",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: &monitoredrespb.MonitoredResource{
+								Type: "global",
+							},
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_DoubleValue{
+										DoubleValue: 100,
+									}},
+								},
+							},
+						},
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/lasttestview",
+								Labels: map[string]string{
+									"test_key":        "test-value-2",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: &monitoredrespb.MonitoredResource{
+								Type: "global",
+							},
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_DoubleValue{
+										DoubleValue: 200,
+									}},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name:   "dist agg + time window - without zero bucket",
+			projID: "proj-id",
+			vd:     newTestDistViewData(distView, start, end),
+			want: []*monitoringpb.CreateTimeSeriesRequest{{
+				Name: monitoring.MetricProjectPath("proj-id"),
+				TimeSeries: []*monitoringpb.TimeSeries{
+					{
+						Metric: &metricpb.Metric{
+							Type: "custom.googleapis.com/opencensus/distview",
+							Labels: map[string]string{
+								opencensusTaskKey: taskValue,
+							},
+						},
+						Resource: &monitoredrespb.MonitoredResource{
+							Type: "global",
+						},
+						Points: []*monitoringpb.Point{
+							{
+								Interval: &monitoringpb.TimeInterval{
+									StartTime: &timestamp.Timestamp{
+										Seconds: start.Unix(),
+										Nanos:   int32(start.Nanosecond()),
+									},
+									EndTime: &timestamp.Timestamp{
+										Seconds: end.Unix(),
+										Nanos:   int32(end.Nanosecond()),
+									},
+								},
+								Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_DistributionValue{
+									DistributionValue: &distribution.Distribution{
+										Count:                 5,
+										Mean:                  3.0,
+										SumOfSquaredDeviation: 1.5,
+										BucketOptions: &distribution.Distribution_BucketOptions{
+											Options: &distribution.Distribution_BucketOptions_ExplicitBuckets{
+												ExplicitBuckets: &distribution.Distribution_BucketOptions_Explicit{
+													Bounds: []float64{0.0, 2.0, 4.0, 7.0}}}},
+										BucketCounts: []int64{0, 2, 2, 1}},
+								}},
+							},
+						},
+					},
+				},
+			}},
+		},
+		{
+			name:   "dist agg + time window + zero bucket",
+			projID: "proj-id",
+			vd:     newTestDistViewData(distView, start, end),
+			want: []*monitoringpb.CreateTimeSeriesRequest{{
+				Name: monitoring.MetricProjectPath("proj-id"),
+				TimeSeries: []*monitoringpb.TimeSeries{
+					{
+						Metric: &metricpb.Metric{
+							Type: "custom.googleapis.com/opencensus/distview",
+							Labels: map[string]string{
+								opencensusTaskKey: taskValue,
+							},
+						},
+						Resource: &monitoredrespb.MonitoredResource{
+							Type: "global",
+						},
+						Points: []*monitoringpb.Point{
+							{
+								Interval: &monitoringpb.TimeInterval{
+									StartTime: &timestamp.Timestamp{
+										Seconds: start.Unix(),
+										Nanos:   int32(start.Nanosecond()),
+									},
+									EndTime: &timestamp.Timestamp{
+										Seconds: end.Unix(),
+										Nanos:   int32(end.Nanosecond()),
+									},
+								},
+								Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_DistributionValue{
+									DistributionValue: &distribution.Distribution{
+										Count:                 5,
+										Mean:                  3.0,
+										SumOfSquaredDeviation: 1.5,
+										BucketOptions: &distribution.Distribution_BucketOptions{
+											Options: &distribution.Distribution_BucketOptions_ExplicitBuckets{
+												ExplicitBuckets: &distribution.Distribution_BucketOptions_Explicit{
+													Bounds: []float64{0.0, 2.0, 4.0, 7.0}}}},
+										BucketCounts: []int64{0, 2, 2, 1}},
+								}},
+							},
+						},
+					},
+				},
+			}},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			opts := tt.opts
+			opts.ProjectID = tt.projID
+			opts.MonitoringClientOptions = authOptions
+			e, err := newStatsExporter(opts)
+			if err != nil {
+				t.Fatal(err)
+			}
+			resps := e.makeReq([]*view.Data{tt.vd}, maxTimeSeriesPerUpload)
+			if got, want := len(resps), len(tt.want); got != want {
+				t.Fatalf("%v: Exporter.makeReq() returned %d responses; want %d", tt.name, got, want)
+			}
+			if len(tt.want) == 0 {
+				return
+			}
+			if diff := cmp.Diff(resps, tt.want); diff != "" {
+				t.Errorf("Values differ -got +want: %s", diff)
+			}
+		})
+	}
+}
+
+func TestExporter_makeReq_batching(t *testing.T) {
+	m := stats.Float64("test-measure/makeReq_batching", "measure desc", "unit")
+
+	key, err := tag.NewKey("test_key")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	v := &view.View{
+		Name:        "view",
+		Description: "desc",
+		TagKeys:     []tag.Key{key},
+		Measure:     m,
+		Aggregation: view.Count(),
+	}
+
+	tests := []struct {
+		name      string
+		iter      int
+		limit     int
+		wantReqs  int
+		wantTotal int
+	}{
+		{
+			name:      "4 vds; 3 limit",
+			iter:      2,
+			limit:     3,
+			wantReqs:  3,
+			wantTotal: 4,
+		},
+		{
+			name:      "4 vds; 4 limit",
+			iter:      2,
+			limit:     4,
+			wantReqs:  2,
+			wantTotal: 4,
+		},
+		{
+			name:      "4 vds; 5 limit",
+			iter:      2,
+			limit:     5,
+			wantReqs:  2,
+			wantTotal: 4,
+		},
+	}
+
+	count1 := &view.CountData{Value: 10}
+	count2 := &view.CountData{Value: 16}
+
+	for _, tt := range tests {
+		var vds []*view.Data
+		for i := 0; i < tt.iter; i++ {
+			vds = append(vds, newTestViewData(v, time.Now(), time.Now(), count1, count2))
+		}
+
+		e, err := newStatsExporter(testOptions)
+		if err != nil {
+			t.Fatal(err)
+		}
+		resps := e.makeReq(vds, tt.limit)
+		if len(resps) != tt.wantReqs {
+			t.Errorf("%v:\ngot %d:: %v;\n\nwant %d requests\n\n", tt.name, len(resps), resps, tt.wantReqs)
+		}
+
+		var total int
+		for _, resp := range resps {
+			total += len(resp.TimeSeries)
+		}
+		if got, want := total, tt.wantTotal; got != want {
+			t.Errorf("%v: len(resps[...].TimeSeries) = %d; want %d", tt.name, got, want)
+		}
+	}
+}
+
+func TestEqualAggWindowTagKeys(t *testing.T) {
+	key1, _ := tag.NewKey("test-key-one")
+	key2, _ := tag.NewKey("test-key-two")
+	tests := []struct {
+		name    string
+		md      *metricpb.MetricDescriptor
+		m       stats.Measure
+		agg     *view.Aggregation
+		keys    []tag.Key
+		wantErr bool
+	}{
+		{
+			name: "count agg with in64 measure",
+			md: &metricpb.MetricDescriptor{
+				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
+				ValueType:  metricpb.MetricDescriptor_INT64,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
+			},
+			m:       stats.Int64("name", "", ""),
+			agg:     view.Count(),
+			wantErr: false,
+		},
+		{
+			name: "count agg with double measure",
+			md: &metricpb.MetricDescriptor{
+				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
+				ValueType:  metricpb.MetricDescriptor_INT64,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
+			},
+			m:       stats.Float64("name", "", ""),
+			agg:     view.Count(),
+			wantErr: false,
+		},
+		{
+			name: "sum agg double",
+			md: &metricpb.MetricDescriptor{
+				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
+				ValueType:  metricpb.MetricDescriptor_DOUBLE,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
+			},
+			m:       stats.Float64("name", "", ""),
+			agg:     view.Sum(),
+			wantErr: false,
+		},
+		{
+			name: "sum agg int64",
+			md: &metricpb.MetricDescriptor{
+				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
+				ValueType:  metricpb.MetricDescriptor_INT64,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
+			},
+			m:       stats.Int64("name", "", ""),
+			agg:     view.Sum(),
+			wantErr: false,
+		},
+		{
+			name: "last value agg double",
+			md: &metricpb.MetricDescriptor{
+				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
+				ValueType:  metricpb.MetricDescriptor_DOUBLE,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
+			},
+			m:       stats.Float64("name", "", ""),
+			agg:     view.LastValue(),
+			wantErr: false,
+		},
+		{
+			name: "last value agg int64",
+			md: &metricpb.MetricDescriptor{
+				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
+				ValueType:  metricpb.MetricDescriptor_INT64,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
+			},
+			m:       stats.Int64("name", "", ""),
+			agg:     view.LastValue(),
+			wantErr: false,
+		},
+		{
+			name: "distribution - mismatch",
+			md: &metricpb.MetricDescriptor{
+				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
+				ValueType:  metricpb.MetricDescriptor_DISTRIBUTION,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
+			},
+			m:       stats.Int64("name", "", ""),
+			agg:     view.Count(),
+			wantErr: true,
+		},
+		{
+			name: "last value - measure mismatch",
+			md: &metricpb.MetricDescriptor{
+				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
+				ValueType:  metricpb.MetricDescriptor_INT64,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
+			},
+			m:       stats.Float64("name", "", ""),
+			agg:     view.LastValue(),
+			wantErr: true,
+		},
+		{
+			name: "distribution agg with keys",
+			md: &metricpb.MetricDescriptor{
+				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
+				ValueType:  metricpb.MetricDescriptor_DISTRIBUTION,
+				Labels: []*label.LabelDescriptor{
+					{Key: "test_key_one"},
+					{Key: "test_key_two"},
+					{Key: opencensusTaskKey},
+				},
+			},
+			m:       stats.Int64("name", "", ""),
+			agg:     view.Distribution(),
+			keys:    []tag.Key{key1, key2},
+			wantErr: false,
+		},
+		{
+			name: "distribution agg with keys -- mismatch",
+			md: &metricpb.MetricDescriptor{
+				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
+				ValueType:  metricpb.MetricDescriptor_DISTRIBUTION,
+			},
+			m:       stats.Int64("name", "", ""),
+			agg:     view.Distribution(),
+			keys:    []tag.Key{key1, key2},
+			wantErr: true,
+		},
+		{
+			name: "count agg with pointers",
+			md: &metricpb.MetricDescriptor{
+				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
+				ValueType:  metricpb.MetricDescriptor_INT64,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
+			},
+			m:       stats.Int64("name", "", ""),
+			agg:     view.Count(),
+			wantErr: false,
+		},
+	}
+	e, err := newStatsExporter(testOptions)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := e.equalMeasureAggTagKeys(tt.md, tt.m, tt.agg, tt.keys)
+			if err != nil && !tt.wantErr {
+				t.Errorf("equalAggTagKeys() = %q; want no error", err)
+			}
+			if err == nil && tt.wantErr {
+				t.Errorf("equalAggTagKeys() = %q; want error", err)
+			}
+
+		})
+	}
+}
+
+func TestExporter_createMeasure(t *testing.T) {
+	oldCreateMetricDescriptor := createMetricDescriptor
+
+	defer func() {
+		createMetricDescriptor = oldCreateMetricDescriptor
+	}()
+
+	key, _ := tag.NewKey("test-key-one")
+	m := stats.Float64("test-measure/TestExporter_createMeasure", "measure desc", stats.UnitMilliseconds)
+
+	v := &view.View{
+		Name:        "test_view_sum",
+		Description: "view_description",
+		TagKeys:     []tag.Key{key},
+		Measure:     m,
+		Aggregation: view.Sum(),
+	}
+
+	data := &view.CountData{Value: 0}
+	vd := newTestViewData(v, time.Now(), time.Now(), data, data)
+
+	var customLabels Labels
+	customLabels.Set("pid", "1234", "Local process identifier")
+	customLabels.Set("hostname", "test.example.com", "Local hostname")
+	customLabels.Set("a/b/c/host-name", "test.example.com", "Local hostname")
+
+	tests := []struct {
+		name string
+		opts Options
+	}{
+		{
+			name: "default",
+		},
+		{
+			name: "no default labels",
+			opts: Options{DefaultMonitoringLabels: &Labels{}},
+		},
+		{
+			name: "custom default labels",
+			opts: Options{DefaultMonitoringLabels: &customLabels},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			opts := tt.opts
+			opts.MonitoringClientOptions = authOptions
+			opts.ProjectID = "test_project"
+			e, err := newStatsExporter(opts)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			var createCalls int
+			createMetricDescriptor = func(ctx context.Context, c *monitoring.MetricClient, mdr *monitoringpb.CreateMetricDescriptorRequest) (*metric.MetricDescriptor, error) {
+				createCalls++
+				if got, want := mdr.MetricDescriptor.Name, "projects/test_project/metricDescriptors/custom.googleapis.com/opencensus/test_view_sum"; got != want {
+					t.Errorf("MetricDescriptor.Name = %q; want %q", got, want)
+				}
+				if got, want := mdr.MetricDescriptor.Type, "custom.googleapis.com/opencensus/test_view_sum"; got != want {
+					t.Errorf("MetricDescriptor.Type = %q; want %q", got, want)
+				}
+				if got, want := mdr.MetricDescriptor.ValueType, metricpb.MetricDescriptor_DOUBLE; got != want {
+					t.Errorf("MetricDescriptor.ValueType = %q; want %q", got, want)
+				}
+				if got, want := mdr.MetricDescriptor.MetricKind, metricpb.MetricDescriptor_CUMULATIVE; got != want {
+					t.Errorf("MetricDescriptor.MetricKind = %q; want %q", got, want)
+				}
+				if got, want := mdr.MetricDescriptor.Description, "view_description"; got != want {
+					t.Errorf("MetricDescriptor.Description = %q; want %q", got, want)
+				}
+				if got, want := mdr.MetricDescriptor.DisplayName, "OpenCensus/test_view_sum"; got != want {
+					t.Errorf("MetricDescriptor.DisplayName = %q; want %q", got, want)
+				}
+				if got, want := mdr.MetricDescriptor.Unit, stats.UnitMilliseconds; got != want {
+					t.Errorf("MetricDescriptor.Unit = %q; want %q", got, want)
+				}
+				return &metric.MetricDescriptor{
+					DisplayName: "OpenCensus/test_view_sum",
+					Description: "view_description",
+					Unit:        stats.UnitMilliseconds,
+					Type:        "custom.googleapis.com/opencensus/test_view_sum",
+					MetricKind:  metricpb.MetricDescriptor_CUMULATIVE,
+					ValueType:   metricpb.MetricDescriptor_DOUBLE,
+					Labels:      newLabelDescriptors(e.defaultLabels, vd.View.TagKeys),
+				}, nil
+			}
+
+			ctx := context.Background()
+			if err := e.createMeasure(ctx, vd.View); err != nil {
+				t.Errorf("Exporter.createMeasure() error = %v", err)
+			}
+			if err := e.createMeasure(ctx, vd.View); err != nil {
+				t.Errorf("Exporter.createMeasure() error = %v", err)
+			}
+			if count := createCalls; count != 1 {
+				t.Errorf("createMetricDescriptor needs to be called for once; called %v times", count)
+			}
+			if count := len(e.createdViews); count != 1 {
+				t.Errorf("len(e.createdViews) = %v; want 1", count)
+			}
+		})
+	}
+}
+
+func TestExporter_createMeasure_CountAggregation(t *testing.T) {
+	oldCreateMetricDescriptor := createMetricDescriptor
+
+	defer func() {
+		createMetricDescriptor = oldCreateMetricDescriptor
+	}()
+
+	key, _ := tag.NewKey("test-key-one")
+	m := stats.Float64("test-measure/TestExporter_createMeasure", "measure desc", stats.UnitMilliseconds)
+
+	v := &view.View{
+		Name:        "test_view_count",
+		Description: "view_description",
+		TagKeys:     []tag.Key{key},
+		Measure:     m,
+		Aggregation: view.Count(),
+	}
+
+	data := &view.CountData{Value: 0}
+	vd := newTestViewData(v, time.Now(), time.Now(), data, data)
+
+	e := &statsExporter{
+		createdViews: make(map[string]*metricpb.MetricDescriptor),
+		o:            Options{ProjectID: "test_project"},
+	}
+
+	createMetricDescriptor = func(ctx context.Context, c *monitoring.MetricClient, mdr *monitoringpb.CreateMetricDescriptorRequest) (*metric.MetricDescriptor, error) {
+		if got, want := mdr.MetricDescriptor.Name, "projects/test_project/metricDescriptors/custom.googleapis.com/opencensus/test_view_count"; got != want {
+			t.Errorf("MetricDescriptor.Name = %q; want %q", got, want)
+		}
+		if got, want := mdr.MetricDescriptor.Type, "custom.googleapis.com/opencensus/test_view_count"; got != want {
+			t.Errorf("MetricDescriptor.Type = %q; want %q", got, want)
+		}
+		if got, want := mdr.MetricDescriptor.ValueType, metricpb.MetricDescriptor_INT64; got != want {
+			t.Errorf("MetricDescriptor.ValueType = %q; want %q", got, want)
+		}
+		if got, want := mdr.MetricDescriptor.MetricKind, metricpb.MetricDescriptor_CUMULATIVE; got != want {
+			t.Errorf("MetricDescriptor.MetricKind = %q; want %q", got, want)
+		}
+		if got, want := mdr.MetricDescriptor.Description, "view_description"; got != want {
+			t.Errorf("MetricDescriptor.Description = %q; want %q", got, want)
+		}
+		if got, want := mdr.MetricDescriptor.DisplayName, "OpenCensus/test_view_count"; got != want {
+			t.Errorf("MetricDescriptor.DisplayName = %q; want %q", got, want)
+		}
+		if got, want := mdr.MetricDescriptor.Unit, stats.UnitDimensionless; got != want {
+			t.Errorf("MetricDescriptor.Unit = %q; want %q", got, want)
+		}
+		return &metric.MetricDescriptor{
+			DisplayName: "OpenCensus/test_view_sum",
+			Description: "view_description",
+			Unit:        stats.UnitDimensionless,
+			Type:        "custom.googleapis.com/opencensus/test_view_count",
+			MetricKind:  metricpb.MetricDescriptor_CUMULATIVE,
+			ValueType:   metricpb.MetricDescriptor_INT64,
+			Labels:      newLabelDescriptors(nil, vd.View.TagKeys),
+		}, nil
+	}
+	ctx := context.Background()
+	if err := e.createMeasure(ctx, vd.View); err != nil {
+		t.Errorf("Exporter.createMeasure() error = %v", err)
+	}
+}
+
+func TestExporter_makeReq_withCustomMonitoredResource(t *testing.T) {
+	m := stats.Float64("test-measure/TestExporter_makeReq_withCustomMonitoredResource", "measure desc", "unit")
+
+	key, err := tag.NewKey("test_key")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	v := &view.View{
+		Name:        "testview",
+		Description: "desc",
+		TagKeys:     []tag.Key{key},
+		Measure:     m,
+		Aggregation: view.Count(),
+	}
+	if err := view.Register(v); err != nil {
+		t.Fatal(err)
+	}
+	defer view.Unregister(v)
+
+	start := time.Now()
+	end := start.Add(time.Minute)
+	count1 := &view.CountData{Value: 10}
+	count2 := &view.CountData{Value: 16}
+	taskValue := getTaskValue()
+
+	resource := &monitoredrespb.MonitoredResource{
+		Type: "gce_instance",
+		Labels: map[string]string{
+			"project_id":  "proj-id",
+			"instance_id": "instance",
+			"zone":        "us-west-1a",
+		},
+	}
+
+	gceInst := &monitoredresource.GCEInstance{
+		ProjectID:  "proj-id",
+		InstanceID: "instance",
+		Zone:       "us-west-1a",
+	}
+
+	tests := []struct {
+		name string
+		opts Options
+		vd   *view.Data
+		want []*monitoringpb.CreateTimeSeriesRequest
+	}{
+		{
+			name: "count agg timeline",
+			opts: Options{Resource: resource},
+			vd:   newTestViewData(v, start, end, count1, count2),
+			want: []*monitoringpb.CreateTimeSeriesRequest{
+				{
+					Name: monitoring.MetricProjectPath("proj-id"),
+					TimeSeries: []*monitoringpb.TimeSeries{
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/testview",
+								Labels: map[string]string{
+									"test_key":        "test-value-1",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: resource,
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+										Int64Value: 10,
+									}},
+								},
+							},
+						},
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/testview",
+								Labels: map[string]string{
+									"test_key":        "test-value-2",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: resource,
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+										Int64Value: 16,
+									}},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "with MonitoredResource and labels",
+			opts: func() Options {
+				var labels Labels
+				labels.Set("pid", "1234", "Process identifier")
+				return Options{
+					MonitoredResource:       gceInst,
+					DefaultMonitoringLabels: &labels,
+				}
+			}(),
+			vd: newTestViewData(v, start, end, count1, count2),
+			want: []*monitoringpb.CreateTimeSeriesRequest{
+				{
+					Name: monitoring.MetricProjectPath("proj-id"),
+					TimeSeries: []*monitoringpb.TimeSeries{
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/testview",
+								Labels: map[string]string{
+									"test_key": "test-value-1",
+									"pid":      "1234",
+								},
+							},
+							Resource: resource,
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+										Int64Value: 10,
+									}},
+								},
+							},
+						},
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/testview",
+								Labels: map[string]string{
+									"test_key": "test-value-2",
+									"pid":      "1234",
+								},
+							},
+							Resource: resource,
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+										Int64Value: 16,
+									}},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "custom default monitoring labels",
+			opts: func() Options {
+				var labels Labels
+				labels.Set("pid", "1234", "Process identifier")
+				return Options{
+					Resource:                resource,
+					DefaultMonitoringLabels: &labels,
+				}
+			}(),
+			vd: newTestViewData(v, start, end, count1, count2),
+			want: []*monitoringpb.CreateTimeSeriesRequest{
+				{
+					Name: monitoring.MetricProjectPath("proj-id"),
+					TimeSeries: []*monitoringpb.TimeSeries{
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/testview",
+								Labels: map[string]string{
+									"test_key": "test-value-1",
+									"pid":      "1234",
+								},
+							},
+							Resource: resource,
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+										Int64Value: 10,
+									}},
+								},
+							},
+						},
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/testview",
+								Labels: map[string]string{
+									"test_key": "test-value-2",
+									"pid":      "1234",
+								},
+							},
+							Resource: resource,
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+										Int64Value: 16,
+									}},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "count agg timeline",
+			opts: Options{Resource: resource},
+			vd:   newTestViewData(v, start, end, count1, count2),
+			want: []*monitoringpb.CreateTimeSeriesRequest{
+				{
+					Name: monitoring.MetricProjectPath("proj-id"),
+					TimeSeries: []*monitoringpb.TimeSeries{
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/testview",
+								Labels: map[string]string{
+									"test_key":        "test-value-1",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: resource,
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+										Int64Value: 10,
+									}},
+								},
+							},
+						},
+						{
+							Metric: &metricpb.Metric{
+								Type: "custom.googleapis.com/opencensus/testview",
+								Labels: map[string]string{
+									"test_key":        "test-value-2",
+									opencensusTaskKey: taskValue,
+								},
+							},
+							Resource: resource,
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: &timestamp.Timestamp{
+											Seconds: start.Unix(),
+											Nanos:   int32(start.Nanosecond()),
+										},
+										EndTime: &timestamp.Timestamp{
+											Seconds: end.Unix(),
+											Nanos:   int32(end.Nanosecond()),
+										},
+									},
+									Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+										Int64Value: 16,
+									}},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			opts := tt.opts
+			opts.MonitoringClientOptions = authOptions
+			opts.ProjectID = "proj-id"
+			e, err := NewExporter(opts)
+			if err != nil {
+				t.Fatal(err)
+			}
+			resps := e.statsExporter.makeReq([]*view.Data{tt.vd}, maxTimeSeriesPerUpload)
+			if got, want := len(resps), len(tt.want); got != want {
+				t.Fatalf("%v: Exporter.makeReq() returned %d responses; want %d", tt.name, got, want)
+			}
+			if len(tt.want) == 0 {
+				return
+			}
+			if diff := cmp.Diff(resps, tt.want); diff != "" {
+				t.Errorf("Requests differ, -got +want: %s", diff)
+			}
+		})
+	}
+}
+
+func TestExporter_customContext(t *testing.T) {
+	oldCreateMetricDescriptor := createMetricDescriptor
+	oldCreateTimeSeries := createTimeSeries
+
+	defer func() {
+		createMetricDescriptor = oldCreateMetricDescriptor
+		createTimeSeries = oldCreateTimeSeries
+	}()
+
+	var timedOut = 0
+	createMetricDescriptor = func(ctx context.Context, c *monitoring.MetricClient, mdr *monitoringpb.CreateMetricDescriptorRequest) (*metric.MetricDescriptor, error) {
+		select {
+		case <-time.After(1 * time.Second):
+			fmt.Println("createMetricDescriptor did not time out")
+		case <-ctx.Done():
+			timedOut++
+		}
+		return &metric.MetricDescriptor{}, nil
+	}
+	createTimeSeries = func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error {
+		select {
+		case <-time.After(1 * time.Second):
+			fmt.Println("createTimeSeries did not time out")
+		case <-ctx.Done():
+			timedOut++
+		}
+		return nil
+	}
+
+	v := &view.View{
+		Name:        "test_view_count",
+		Description: "view_description",
+		Measure:     stats.Float64("test-measure/TestExporter_createMeasure", "measure desc", stats.UnitMilliseconds),
+		Aggregation: view.Count(),
+	}
+
+	data := &view.CountData{Value: 0}
+	vd := newTestViewData(v, time.Now(), time.Now(), data, data)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+	defer cancel()
+	e := &statsExporter{
+		createdViews: make(map[string]*metricpb.MetricDescriptor),
+		o:            Options{ProjectID: "test_project", Context: ctx},
+	}
+	if err := e.uploadStats([]*view.Data{vd}); err != nil {
+		t.Errorf("Exporter.uploadStats() error = %v", err)
+	}
+	if ctx.Err() != context.DeadlineExceeded {
+		t.Errorf("expected context to time out; got %v", ctx.Err())
+	}
+	if timedOut != 2 {
+		t.Errorf("expected two functions to time out; got %d", timedOut)
+	}
+}
+
+func newTestViewData(v *view.View, start, end time.Time, data1, data2 view.AggregationData) *view.Data {
+	key, _ := tag.NewKey("test-key")
+	tag1 := tag.Tag{Key: key, Value: "test-value-1"}
+	tag2 := tag.Tag{Key: key, Value: "test-value-2"}
+	return &view.Data{
+		View: v,
+		Rows: []*view.Row{
+			{
+				Tags: []tag.Tag{tag1},
+				Data: data1,
+			},
+			{
+				Tags: []tag.Tag{tag2},
+				Data: data2,
+			},
+		},
+		Start: start,
+		End:   end,
+	}
+}
+
+func newTestDistViewData(v *view.View, start, end time.Time) *view.Data {
+	return &view.Data{
+		View: v,
+		Rows: []*view.Row{
+			{Data: &view.DistributionData{
+				Count:           5,
+				Min:             1,
+				Max:             7,
+				Mean:            3,
+				SumOfSquaredDev: 1.5,
+				CountPerBucket:  []int64{2, 2, 1},
+			}},
+		},
+		Start: start,
+		End:   end,
+	}
+}
+
+func newTestDistViewDataWithZeroBucket(v *view.View, start, end time.Time) *view.Data {
+	return &view.Data{
+		View: v,
+		Rows: []*view.Row{
+			{Data: &view.DistributionData{
+				Count:           5,
+				Min:             1,
+				Max:             7,
+				Mean:            3,
+				SumOfSquaredDev: 1.5,
+				CountPerBucket:  []int64{0, 2, 2, 1},
+			}},
+		},
+		Start: start,
+		End:   end,
+	}
+}