Repository: census-instrumentation/opencensus-go
Pull request #252 (2017-12-14 01:35:02)

Add a opencensus_task label to Stackdriver exporter
A supplementary change to https://github.com/census-instrumentation/opencensus-go/pull/249.

**Background**: sometimes using a MonitoredResource is not enough to uniquely identify stats. For example, if there are multiple exporters running on the same GCE instance, they will all have the same MonitoredResource. If they export stats to the same Stackdriver metric, there will still be a race condition. To fix that, @jcd2 suggest to have a reserved `opencensus_task` label that can further separate the stats. The value of this label will be in the format "go-pid@hostname".

Equivalent changes for Java: https://github.com/census-instrumentation/opencensus-java/pull/885.


Patch:
diff --git a/README.md b/README.md
index ea679d6..0f12fba 100644
--- a/README.md
+++ b/README.md
@@ -204,7 +204,7 @@ Unregister view:
 
 [embedmd]:# (stats.go unregisterView)
 ```go
-if v.Unregister(); err != nil {
+if stats.UnregisterView(v); err != nil {
 	log.Fatal(err)
 }
 ```
diff --git a/exporter/stats/prometheus/prometheus.go b/exporter/stats/prometheus/prometheus.go
index ffab6c2..d1fc2a2 100644
--- a/exporter/stats/prometheus/prometheus.go
+++ b/exporter/stats/prometheus/prometheus.go
@@ -219,11 +219,11 @@ func (c *collector) toMetric(desc *prometheus.Desc, view *stats.View, row *stats
 
 	case stats.MeanAggregation:
 		data := row.Data.(*stats.MeanData)
-		return prometheus.NewConstMetric(desc, prometheus.GaugeValue, data.Mean, tagValues(row.Tags)...)
+		return prometheus.NewConstMetric(desc, prometheus.UntypedValue, data.Mean, tagValues(row.Tags)...)
 
 	case stats.SumAggregation:
 		data := row.Data.(*stats.SumData)
-		return prometheus.NewConstMetric(desc, prometheus.GaugeValue, float64(*data), tagValues(row.Tags)...)
+		return prometheus.NewConstMetric(desc, prometheus.UntypedValue, float64(*data), tagValues(row.Tags)...)
 
 	default:
 		return nil, fmt.Errorf("aggregation %T is not yet supported", view.Aggregation())
diff --git a/exporter/stats/stackdriver/stackdriver.go b/exporter/stats/stackdriver/stackdriver.go
index c762178..f2d2c7e 100644
--- a/exporter/stats/stackdriver/stackdriver.go
+++ b/exporter/stats/stackdriver/stackdriver.go
@@ -24,8 +24,10 @@ import (
 	"fmt"
 	"log"
 	"net/url"
+	"os"
 	"path"
 	"reflect"
+	"strconv"
 	"sync"
 	"time"
 
@@ -48,6 +50,8 @@ import (
 )
 
 const maxTimeSeriesPerUpload = 200
+const opencensusTaskKey = "opencensus_task"
+const opencensusTaskDescription = "Opencensus task identifier"
 
 // Exporter exports stats to the Stackdriver Monitoring.
 type Exporter struct {
@@ -57,7 +61,8 @@ type Exporter struct {
 	createdViewsMu sync.Mutex
 	createdViews   map[string]*metricpb.MetricDescriptor // Views already created remotely
 
-	c *monitoring.MetricClient
+	c         *monitoring.MetricClient
+	taskValue string
 }
 
 // Options contains options for configuring the exporter.
@@ -97,6 +102,7 @@ type Options struct {
 }
 
 // NewExporter returns an exporter that uploads stats data to Stackdriver Monitoring.
+// Only one Stackdriver exporter should be created per process.
 func NewExporter(o Options) (*Exporter, error) {
 	opts := append(o.ClientOptions, option.WithUserAgent(internal.UserAgent))
 	client, err := monitoring.NewMetricClient(context.Background(), opts...)
@@ -107,6 +113,7 @@ func NewExporter(o Options) (*Exporter, error) {
 		c:            client,
 		o:            o,
 		createdViews: make(map[string]*metricpb.MetricDescriptor),
+		taskValue:    getTaskValue(),
 	}
 	e.bundler = bundler.NewBundler((*stats.ViewData)(nil), func(bundle interface{}) {
 		vds := bundle.([]*stats.ViewData)
@@ -136,6 +143,16 @@ func (e *Exporter) Export(vd *stats.ViewData) {
 	}
 }
 
+// getTaskValue returns a task label value in the format of
+// "go-<pid>@<hostname>".
+func getTaskValue() string {
+	hostname, err := os.Hostname()
+	if err != nil {
+		hostname = "localhost"
+	}
+	return "go-" + strconv.Itoa(os.Getpid()) + "@" + hostname
+}
+
 // handleUpload handles uploading a slice
 // of ViewData, as well as error handling.
 func (e *Exporter) handleUpload(vds ...*stats.ViewData) {
@@ -203,7 +220,7 @@ func (e *Exporter) makeReq(vds []*stats.ViewData, limit int) []*monitoringpb.Cre
 			ts := &monitoringpb.TimeSeries{
 				Metric: &metricpb.Metric{
 					Type:   namespacedViewName(vd.View.Name(), false),
-					Labels: newLabels(row.Tags),
+					Labels: newLabels(row.Tags, e.taskValue),
 				},
 				Resource: resource,
 				Points:   []*monitoringpb.Point{newPoint(vd.View, row, vd.Start, vd.End)},
@@ -384,22 +401,29 @@ func namespacedViewName(v string, escaped bool) string {
 	return path.Join("custom.googleapis.com", p)
 }
 
-func newLabels(tags []tag.Tag) map[string]string {
+func newLabels(tags []tag.Tag, taskValue string) map[string]string {
 	labels := make(map[string]string)
 	for _, tag := range tags {
 		labels[internal.Sanitize(tag.Key.Name())] = tag.Value
 	}
+	labels[opencensusTaskKey] = taskValue
 	return labels
 }
 
 func newLabelDescriptors(keys []tag.Key) []*labelpb.LabelDescriptor {
-	labelDescriptors := make([]*labelpb.LabelDescriptor, len(keys))
+	labelDescriptors := make([]*labelpb.LabelDescriptor, len(keys)+1)
 	for i, key := range keys {
 		labelDescriptors[i] = &labelpb.LabelDescriptor{
 			Key:       internal.Sanitize(key.Name()),
 			ValueType: labelpb.LabelDescriptor_STRING, // We only use string tags
 		}
 	}
+	// Add a specific open census task id label.
+	labelDescriptors[len(keys)] = &labelpb.LabelDescriptor{
+		Key:         opencensusTaskKey,
+		ValueType:   labelpb.LabelDescriptor_STRING,
+		Description: opencensusTaskDescription,
+	}
 	return labelDescriptors
 }
 
@@ -439,14 +463,15 @@ func equalAggWindowTagKeys(md *metricpb.MetricDescriptor, agg stats.Aggregation,
 		return fmt.Errorf("stackdriver metric descriptor was not created with window type %T", w)
 	}
 
-	if len(md.Labels) != len(keys) {
+	if len(md.Labels) != len(keys)+1 {
 		return errors.New("stackdriver metric descriptor was not created with the view labels")
 	}
 
-	labels := make(map[string]struct{}, len(keys))
+	labels := make(map[string]struct{}, len(keys)+1)
 	for _, k := range keys {
 		labels[internal.Sanitize(k.Name())] = struct{}{}
 	}
+	labels[opencensusTaskKey] = struct{}{}
 
 	for _, k := range md.Labels {
 		if _, ok := labels[k.Key]; !ok {
diff --git a/internal/readme/stats.go b/internal/readme/stats.go
index 631716f..a3d7e9a 100644
--- a/internal/readme/stats.go
+++ b/internal/readme/stats.go
@@ -99,7 +99,7 @@ func statsExamples() {
 	_ = v
 
 	// START unregisterView
-	if v.Unregister(); err != nil {
+	if stats.UnregisterView(v); err != nil {
 		log.Fatal(err)
 	}
 	// END unregisterView
diff --git a/stats/aggregation.go b/stats/aggregation.go
index bff1728..fa9bf8e 100644
--- a/stats/aggregation.go
+++ b/stats/aggregation.go
@@ -91,15 +91,13 @@ func (a DistributionAggregation) newData() func() AggregationData {
 // aggregatorCumulative indicates that the aggregation occurs over all samples
 // seen since the view collection started.
 type aggregatorCumulative struct {
-	started time.Time
-	av      AggregationData
+	data AggregationData
 }
 
 // newAggregatorCumulative creates an aggregatorCumulative.
 func newAggregatorCumulative(now time.Time, newAggregationValue func() AggregationData) *aggregatorCumulative {
 	return &aggregatorCumulative{
-		started: now,
-		av:      newAggregationValue(),
+		data: newAggregationValue(),
 	}
 }
 
@@ -108,11 +106,11 @@ func (a *aggregatorCumulative) isAggregator() bool {
 }
 
 func (a *aggregatorCumulative) addSample(v interface{}, now time.Time) {
-	a.av.addSample(v)
+	a.data.addSample(v)
 }
 
 func (a *aggregatorCumulative) retrieveCollected(now time.Time) AggregationData {
-	return a.av
+	return a.data
 }
 
 // aggregatorInterval indicates that the aggregation occurs over a
@@ -178,7 +176,7 @@ func (a *aggregatorInterval) retrieveCollected(now time.Time) AggregationData {
 	for j := 1; j < len(a.entries); j++ {
 		oldestIdx = (oldestIdx + 1) % len(a.entries)
 		e = a.entries[oldestIdx]
-		ret.addToIt(e.av)
+		ret.addOther(e.av)
 	}
 	return ret
 }
diff --git a/stats/aggregation_data.go b/stats/aggregation_data.go
index 9f43ad7..240503c 100644
--- a/stats/aggregation_data.go
+++ b/stats/aggregation_data.go
@@ -23,12 +23,12 @@ import (
 // They are reported on the view data during exporting.
 // Mosts users won't directly access aggregration data.
 type AggregationData interface {
-	equal(other AggregationData) bool
-	isAggregate() bool
+	isAggregationData() bool
 	addSample(v interface{})
+	addOther(other AggregationData)
 	multiplyByFraction(fraction float64) AggregationData
-	addToIt(other AggregationData)
 	clear()
+	equal(other AggregationData) bool
 }
 
 const epsilon = 1e-9
@@ -44,7 +44,7 @@ func newCountData(v int64) *CountData {
 	return &tmp
 }
 
-func (a *CountData) isAggregate() bool { return true }
+func (a *CountData) isAggregationData() bool { return true }
 
 func (a *CountData) addSample(v interface{}) {
 	*a = *a + 1
@@ -52,10 +52,9 @@ func (a *CountData) addSample(v interface{}) {
 
 func (a *CountData) multiplyByFraction(fraction float64) AggregationData {
 	return newCountData(int64(float64(int64(*a))*fraction + 0.5)) // adding 0.5 because go runtime will take floor instead of rounding
-
 }
 
-func (a *CountData) addToIt(av AggregationData) {
+func (a *CountData) addOther(av AggregationData) {
 	other, ok := av.(*CountData)
 	if !ok {
 		return
@@ -87,7 +86,7 @@ func newSumData(v float64) *SumData {
 	return &tmp
 }
 
-func (a *SumData) isAggregate() bool { return true }
+func (a *SumData) isAggregationData() bool { return true }
 
 func (a *SumData) addSample(v interface{}) {
 	// Both float64 and int64 values will be cast to float64
@@ -107,7 +106,7 @@ func (a *SumData) multiplyByFraction(fraction float64) AggregationData {
 	return newSumData(float64(*a) * fraction)
 }
 
-func (a *SumData) addToIt(av AggregationData) {
+func (a *SumData) addOther(av AggregationData) {
 	other, ok := av.(*SumData)
 	if !ok {
 		return
@@ -146,7 +145,7 @@ func newMeanData(mean float64, count float64) *MeanData {
 // Sum returns the sum of all samples collected.
 func (a *MeanData) Sum() float64 { return a.Mean * float64(a.Count) }
 
-func (a *MeanData) isAggregate() bool { return true }
+func (a *MeanData) isAggregationData() bool { return true }
 
 func (a *MeanData) addSample(v interface{}) {
 	var f float64
@@ -172,7 +171,7 @@ func (a *MeanData) multiplyByFraction(fraction float64) AggregationData {
 	return newMeanData(a.Mean, a.Count*fraction)
 }
 
-func (a *MeanData) addToIt(av AggregationData) {
+func (a *MeanData) addOther(av AggregationData) {
 	other, ok := av.(*MeanData)
 	if !ok {
 		return
@@ -232,7 +231,7 @@ func (a *DistributionData) variance() float64 {
 	return a.SumOfSquaredDev / float64(a.Count-1)
 }
 
-func (a *DistributionData) isAggregate() bool { return true }
+func (a *DistributionData) isAggregationData() bool { return true }
 
 func (a *DistributionData) addSample(v interface{}) {
 	var f float64
@@ -299,7 +298,7 @@ func (a *DistributionData) multiplyByFraction(fraction float64) AggregationData
 	return ret
 }
 
-func (a *DistributionData) addToIt(av AggregationData) {
+func (a *DistributionData) addOther(av AggregationData) {
 	other, ok := av.(*DistributionData)
 	if !ok {
 		return
diff --git a/stats/view.go b/stats/view.go
index 2ecf2d7..22dca80 100644
--- a/stats/view.go
+++ b/stats/view.go
@@ -39,9 +39,6 @@ type View struct {
 	// Examples of measures are cpu:tickCount, diskio:time...
 	m Measure
 
-	// start is time when view collection was started originally.
-	start time.Time
-
 	subscribed uint32 // 1 if someone is subscribed and data need to be exported, use atomic to access
 
 	collector *collector
@@ -66,7 +63,6 @@ func NewView(name, description string, keys []tag.Key, measure Measure, agg Aggr
 		description: description,
 		tagKeys:     keys,
 		m:           measure,
-		start:       time.Time{},
 		collector:   &collector{make(map[string]aggregator), agg, window},
 	}, nil
 }
@@ -168,6 +164,5 @@ func (r *Row) Equal(other *Row) bool {
 	if r == other {
 		return true
 	}
-
 	return reflect.DeepEqual(r.Tags, other.Tags) && r.Data.equal(other.Data)
 }
diff --git a/stats/window.go b/stats/window.go
index 9700da7..9522944 100644
--- a/stats/window.go
+++ b/stats/window.go
@@ -28,7 +28,7 @@ type aggregator interface {
 // which the aggregation occurs.
 type Window interface {
 	isWindow()
-	newAggregator(now time.Time, aggregationValueConstructor func() AggregationData) aggregator
+	newAggregator(now time.Time, newAggregationData func() AggregationData) aggregator
 }
 
 // Cumulative is a window that indicates that the aggregation occurs
@@ -37,8 +37,8 @@ type Cumulative struct{}
 
 func (w Cumulative) isWindow() {}
 
-func (w Cumulative) newAggregator(now time.Time, aggregationValueConstructor func() AggregationData) aggregator {
-	return newAggregatorCumulative(now, aggregationValueConstructor)
+func (w Cumulative) newAggregator(now time.Time, newAggregationData func() AggregationData) aggregator {
+	return newAggregatorCumulative(now, newAggregationData)
 }
 
 // Interval is a window that indicates that the aggregation occurs over a sliding
@@ -50,6 +50,6 @@ type Interval struct {
 
 func (w Interval) isWindow() {}
 
-func (w Interval) newAggregator(now time.Time, aggregationValueConstructor func() AggregationData) aggregator {
-	return newAggregatorInterval(now, w.Duration, w.Intervals, aggregationValueConstructor)
+func (w Interval) newAggregator(now time.Time, newAggregationData func() AggregationData) aggregator {
+	return newAggregatorInterval(now, w.Duration, w.Intervals, newAggregationData)
 }
diff --git a/stats/worker.go b/stats/worker.go
index ad41fac..6184166 100644
--- a/stats/worker.go
+++ b/stats/worker.go
@@ -121,10 +121,10 @@ func RegisterView(v *View) error {
 	return <-req.err
 }
 
-// Unregister removes the previously registered view. It returns an error
+// UnregisterView removes the previously registered view. It returns an error
 // if the view wasn't registered. All data collected and not reported for the
 // corresponding view will be lost. The view is automatically be unsubscribed.
-func (v *View) Unregister() error {
+func UnregisterView(v *View) error {
 	req := &unregisterViewReq{
 		v:   v,
 		err: make(chan error),


Test patch:
diff --git a/exporter/stats/stackdriver/stackdriver_test.go b/exporter/stats/stackdriver/stackdriver_test.go
index bdf68a4..e8aa625 100644
--- a/exporter/stats/stackdriver/stackdriver_test.go
+++ b/exporter/stats/stackdriver/stackdriver_test.go
@@ -53,7 +53,7 @@ func TestExporter_makeReq(t *testing.T) {
 	if err := stats.RegisterView(cumView); err != nil {
 		t.Fatal(err)
 	}
-	defer cumView.Unregister()
+	defer stats.UnregisterView(cumView)
 
 	distView, err := stats.NewView("distview", "desc", nil, m, stats.DistributionAggregation([]float64{2, 4, 7}), stats.Interval{})
 	if err != nil {
@@ -62,7 +62,7 @@ func TestExporter_makeReq(t *testing.T) {
 	if err := stats.RegisterView(distView); err != nil {
 		t.Fatal(err)
 	}
-	defer distView.Unregister()
+	defer stats.UnregisterView(distView)
 
 	start := time.Now()
 	end := start.Add(time.Minute)
@@ -78,6 +78,7 @@ func TestExporter_makeReq(t *testing.T) {
 		Mean:  -7.7,
 		Count: 5,
 	}
+	taskValue := getTaskValue()
 
 	tests := []struct {
 		name   string
@@ -94,8 +95,11 @@ func TestExporter_makeReq(t *testing.T) {
 				TimeSeries: []*monitoringpb.TimeSeries{
 					{
 						Metric: &metricpb.Metric{
-							Type:   "custom.googleapis.com/opencensus/cumview",
-							Labels: map[string]string{"test_key": "test-value-1"},
+							Type: "custom.googleapis.com/opencensus/cumview",
+							Labels: map[string]string{
+								"test_key":        "test-value-1",
+								opencensusTaskKey: taskValue,
+							},
 						},
 						Resource: &monitoredrespb.MonitoredResource{
 							Type: "global",
@@ -120,8 +124,11 @@ func TestExporter_makeReq(t *testing.T) {
 					},
 					{
 						Metric: &metricpb.Metric{
-							Type:   "custom.googleapis.com/opencensus/cumview",
-							Labels: map[string]string{"test_key": "test-value-2"},
+							Type: "custom.googleapis.com/opencensus/cumview",
+							Labels: map[string]string{
+								"test_key":        "test-value-2",
+								opencensusTaskKey: taskValue,
+							},
 						},
 						Resource: &monitoredrespb.MonitoredResource{
 							Type: "global",
@@ -156,8 +163,11 @@ func TestExporter_makeReq(t *testing.T) {
 				TimeSeries: []*monitoringpb.TimeSeries{
 					{
 						Metric: &metricpb.Metric{
-							Type:   "custom.googleapis.com/opencensus/cumview",
-							Labels: map[string]string{"test_key": "test-value-1"},
+							Type: "custom.googleapis.com/opencensus/cumview",
+							Labels: map[string]string{
+								"test_key":        "test-value-1",
+								opencensusTaskKey: taskValue,
+							},
 						},
 						Resource: &monitoredrespb.MonitoredResource{
 							Type: "global",
@@ -182,8 +192,11 @@ func TestExporter_makeReq(t *testing.T) {
 					},
 					{
 						Metric: &metricpb.Metric{
-							Type:   "custom.googleapis.com/opencensus/cumview",
-							Labels: map[string]string{"test_key": "test-value-2"},
+							Type: "custom.googleapis.com/opencensus/cumview",
+							Labels: map[string]string{
+								"test_key":        "test-value-2",
+								opencensusTaskKey: taskValue,
+							},
 						},
 						Resource: &monitoredrespb.MonitoredResource{
 							Type: "global",
@@ -218,8 +231,11 @@ func TestExporter_makeReq(t *testing.T) {
 				TimeSeries: []*monitoringpb.TimeSeries{
 					{
 						Metric: &metricpb.Metric{
-							Type:   "custom.googleapis.com/opencensus/cumview",
-							Labels: map[string]string{"test_key": "test-value-1"},
+							Type: "custom.googleapis.com/opencensus/cumview",
+							Labels: map[string]string{
+								"test_key":        "test-value-1",
+								opencensusTaskKey: taskValue,
+							},
 						},
 						Resource: &monitoredrespb.MonitoredResource{
 							Type: "global",
@@ -256,8 +272,11 @@ func TestExporter_makeReq(t *testing.T) {
 					},
 					{
 						Metric: &metricpb.Metric{
-							Type:   "custom.googleapis.com/opencensus/cumview",
-							Labels: map[string]string{"test_key": "test-value-2"},
+							Type: "custom.googleapis.com/opencensus/cumview",
+							Labels: map[string]string{
+								"test_key":        "test-value-2",
+								opencensusTaskKey: taskValue,
+							},
 						},
 						Resource: &monitoredrespb.MonitoredResource{
 							Type: "global",
@@ -304,7 +323,10 @@ func TestExporter_makeReq(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			e := &Exporter{o: Options{ProjectID: tt.projID}}
+			e := &Exporter{
+				o:         Options{ProjectID: tt.projID},
+				taskValue: taskValue,
+			}
 			resps := e.makeReq([]*stats.ViewData{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)
@@ -338,7 +360,7 @@ func TestExporter_makeReq_batching(t *testing.T) {
 	if err := stats.RegisterView(view); err != nil {
 		t.Fatal(err)
 	}
-	defer view.Unregister()
+	defer stats.UnregisterView(view)
 
 	tests := []struct {
 		name      string
@@ -411,6 +433,7 @@ func TestEqualAggWindowTagKeys(t *testing.T) {
 			md: &metricpb.MetricDescriptor{
 				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
 				ValueType:  metricpb.MetricDescriptor_INT64,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
 			},
 			agg:     stats.CountAggregation{},
 			window:  stats.Cumulative{},
@@ -421,6 +444,7 @@ func TestEqualAggWindowTagKeys(t *testing.T) {
 			md: &metricpb.MetricDescriptor{
 				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
 				ValueType:  metricpb.MetricDescriptor_DOUBLE,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
 			},
 			agg:     stats.SumAggregation{},
 			window:  stats.Cumulative{},
@@ -431,6 +455,7 @@ func TestEqualAggWindowTagKeys(t *testing.T) {
 			md: &metricpb.MetricDescriptor{
 				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
 				ValueType:  metricpb.MetricDescriptor_DISTRIBUTION,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
 			},
 			agg:     stats.MeanAggregation{},
 			window:  stats.Cumulative{},
@@ -441,6 +466,7 @@ func TestEqualAggWindowTagKeys(t *testing.T) {
 			md: &metricpb.MetricDescriptor{
 				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
 				ValueType:  metricpb.MetricDescriptor_DISTRIBUTION,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
 			},
 			agg:     stats.CountAggregation{},
 			window:  stats.Cumulative{},
@@ -451,6 +477,7 @@ func TestEqualAggWindowTagKeys(t *testing.T) {
 			md: &metricpb.MetricDescriptor{
 				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
 				ValueType:  metricpb.MetricDescriptor_DOUBLE,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
 			},
 			agg:     stats.MeanAggregation{},
 			window:  stats.Cumulative{},
@@ -461,6 +488,7 @@ func TestEqualAggWindowTagKeys(t *testing.T) {
 			md: &metricpb.MetricDescriptor{
 				MetricKind: metricpb.MetricDescriptor_DELTA,
 				ValueType:  metricpb.MetricDescriptor_DISTRIBUTION,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
 			},
 			agg:     stats.DistributionAggregation{},
 			window:  stats.Interval{},
@@ -471,6 +499,7 @@ func TestEqualAggWindowTagKeys(t *testing.T) {
 			md: &metricpb.MetricDescriptor{
 				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
 				ValueType:  metricpb.MetricDescriptor_DISTRIBUTION,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
 			},
 			agg:     stats.DistributionAggregation{},
 			window:  stats.Interval{},
@@ -484,6 +513,7 @@ func TestEqualAggWindowTagKeys(t *testing.T) {
 				Labels: []*label.LabelDescriptor{
 					{Key: "test_key_one"},
 					{Key: "test_key_two"},
+					{Key: opencensusTaskKey},
 				},
 			},
 			agg:     stats.DistributionAggregation{},
@@ -507,6 +537,7 @@ func TestEqualAggWindowTagKeys(t *testing.T) {
 			md: &metricpb.MetricDescriptor{
 				MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
 				ValueType:  metricpb.MetricDescriptor_INT64,
+				Labels:     []*label.LabelDescriptor{{Key: opencensusTaskKey}},
 			},
 			agg:     &stats.CountAggregation{},
 			window:  &stats.Cumulative{},
@@ -616,6 +647,7 @@ func TestExporter_makeReq_withCustomMonitoredResource(t *testing.T) {
 	end := start.Add(time.Minute)
 	count1 := stats.CountData(10)
 	count2 := stats.CountData(16)
+	taskValue := getTaskValue()
 
 	resource := &monitoredrespb.MonitoredResource{
 		Type:   "gce_instance",
@@ -637,8 +669,11 @@ func TestExporter_makeReq_withCustomMonitoredResource(t *testing.T) {
 				TimeSeries: []*monitoringpb.TimeSeries{
 					{
 						Metric: &metricpb.Metric{
-							Type:   "custom.googleapis.com/opencensus/cumview",
-							Labels: map[string]string{"test_key": "test-value-1"},
+							Type: "custom.googleapis.com/opencensus/cumview",
+							Labels: map[string]string{
+								"test_key":        "test-value-1",
+								opencensusTaskKey: taskValue,
+							},
 						},
 						Resource: resource,
 						Points: []*monitoringpb.Point{
@@ -661,8 +696,11 @@ func TestExporter_makeReq_withCustomMonitoredResource(t *testing.T) {
 					},
 					{
 						Metric: &metricpb.Metric{
-							Type:   "custom.googleapis.com/opencensus/cumview",
-							Labels: map[string]string{"test_key": "test-value-2"},
+							Type: "custom.googleapis.com/opencensus/cumview",
+							Labels: map[string]string{
+								"test_key":        "test-value-2",
+								opencensusTaskKey: taskValue,
+							},
 						},
 						Resource: resource,
 						Points: []*monitoringpb.Point{
@@ -689,7 +727,10 @@ func TestExporter_makeReq_withCustomMonitoredResource(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			e := &Exporter{o: Options{ProjectID: tt.projID, Resource: resource}}
+			e := &Exporter{
+				o:         Options{ProjectID: tt.projID, Resource: resource},
+				taskValue: taskValue,
+			}
 			resps := e.makeReq([]*stats.ViewData{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)
diff --git a/stats/worker_test.go b/stats/worker_test.go
index 24be1cb..fe3da50 100644
--- a/stats/worker_test.go
+++ b/stats/worker_test.go
@@ -442,9 +442,9 @@ func Test_Worker_ViewRegistration(t *testing.T) {
 
 		for _, unreg := range tc.unregs {
 			v := views[unreg.vID]
-			err := v.Unregister()
+			err := UnregisterView(v)
 			if (err != nil) != (unreg.err != nil) {
-				t.Errorf("%v: Unregister() = %v; want %v", tc.label, err, unreg.err)
+				t.Errorf("%v: UnregisterView() = %v; want %v", tc.label, err, unreg.err)
 			}
 		}
 
@@ -605,7 +605,7 @@ func Test_Worker_RecordFloat64(t *testing.T) {
 		}
 
 		for _, v := range tc.registrations {
-			if err := v.Unregister(); err != nil {
+			if err := UnregisterView(v); err != nil {
 				t.Fatalf("%v: Unregistering view %v errrored with %v; want no error", tc.label, v.Name(), err)
 			}
 		}