Repository: census-instrumentation/opencensus-go
Pull request #1078 (2019-03-22 20:34:47)

stats worker as metric producer.
This change provides stats data (view data model) as metric data. Stats worker registers as a metric producer with global producer manager. Read() function is then invoked by Readers.

rghetia: @songy23 PTAL.
rghetia: @songy23 PTAL

Patch:
diff --git a/stats/view/aggregation_data.go b/stats/view/aggregation_data.go
index 8774a07..d500e67 100644
--- a/stats/view/aggregation_data.go
+++ b/stats/view/aggregation_data.go
@@ -30,6 +30,7 @@ type AggregationData interface {
 	addSample(v float64, attachments map[string]interface{}, t time.Time)
 	clone() AggregationData
 	equal(other AggregationData) bool
+	toPoint(t metricdata.Type, time time.Time) metricdata.Point
 }
 
 const epsilon = 1e-9
@@ -61,6 +62,15 @@ func (a *CountData) equal(other AggregationData) bool {
 	return a.Value == a2.Value
 }
 
+func (a *CountData) toPoint(metricType metricdata.Type, t time.Time) metricdata.Point {
+	switch metricType {
+	case metricdata.TypeCumulativeInt64:
+		return metricdata.NewInt64Point(t, a.Value)
+	default:
+		panic("unsupported metricdata.Type")
+	}
+}
+
 // SumData is the aggregated data for the Sum aggregation.
 // A sum aggregation processes data and sums up the recordings.
 //
@@ -87,6 +97,17 @@ func (a *SumData) equal(other AggregationData) bool {
 	return math.Pow(a.Value-a2.Value, 2) < epsilon
 }
 
+func (a *SumData) toPoint(metricType metricdata.Type, t time.Time) metricdata.Point {
+	switch metricType {
+	case metricdata.TypeCumulativeInt64:
+		return metricdata.NewInt64Point(t, int64(a.Value))
+	case metricdata.TypeCumulativeFloat64:
+		return metricdata.NewFloat64Point(t, a.Value)
+	default:
+		panic("unsupported metricdata.Type")
+	}
+}
+
 // DistributionData is the aggregated data for the
 // Distribution aggregation.
 //
@@ -208,6 +229,33 @@ func (a *DistributionData) equal(other AggregationData) bool {
 	return a.Count == a2.Count && a.Min == a2.Min && a.Max == a2.Max && math.Pow(a.Mean-a2.Mean, 2) < epsilon && math.Pow(a.variance()-a2.variance(), 2) < epsilon
 }
 
+func (a *DistributionData) toPoint(metricType metricdata.Type, t time.Time) metricdata.Point {
+	switch metricType {
+	case metricdata.TypeCumulativeDistribution:
+		buckets := []metricdata.Bucket{}
+		for i := 0; i < len(a.CountPerBucket); i++ {
+			buckets = append(buckets, metricdata.Bucket{
+				Count:    a.CountPerBucket[i],
+				Exemplar: a.ExemplarsPerBucket[i],
+			})
+		}
+		bucketOptions := &metricdata.BucketOptions{Bounds: a.bounds}
+
+		val := &metricdata.Distribution{
+			Count:                 a.Count,
+			Sum:                   a.Sum(),
+			SumOfSquaredDeviation: a.SumOfSquaredDev,
+			BucketOptions:         bucketOptions,
+			Buckets:               buckets,
+		}
+		return metricdata.NewDistributionPoint(t, val)
+
+	default:
+		// TODO: [rghetia] when we have a use case for TypeGaugeDistribution.
+		panic("unsupported metricdata.Type")
+	}
+}
+
 // LastValueData returns the last value recorded for LastValue aggregation.
 type LastValueData struct {
 	Value float64
@@ -232,3 +280,14 @@ func (l *LastValueData) equal(other AggregationData) bool {
 	}
 	return l.Value == a2.Value
 }
+
+func (l *LastValueData) toPoint(metricType metricdata.Type, t time.Time) metricdata.Point {
+	switch metricType {
+	case metricdata.TypeGaugeInt64:
+		return metricdata.NewInt64Point(t, int64(l.Value))
+	case metricdata.TypeGaugeFloat64:
+		return metricdata.NewFloat64Point(t, l.Value)
+	default:
+		panic("unsupported metricdata.Type")
+	}
+}
diff --git a/stats/view/view.go b/stats/view/view.go
index 95f01ad..37f88e1 100644
--- a/stats/view/view.go
+++ b/stats/view/view.go
@@ -24,6 +24,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"go.opencensus.io/metric/metricdata"
 	"go.opencensus.io/stats"
 	"go.opencensus.io/tag"
 )
@@ -116,15 +117,17 @@ func dropZeroBounds(bounds ...float64) []float64 {
 
 // viewInternal is the internal representation of a View.
 type viewInternal struct {
-	view       *View  // view is the canonicalized View definition associated with this view.
-	subscribed uint32 // 1 if someone is subscribed and data need to be exported, use atomic to access
-	collector  *collector
+	view             *View  // view is the canonicalized View definition associated with this view.
+	subscribed       uint32 // 1 if someone is subscribed and data need to be exported, use atomic to access
+	collector        *collector
+	metricDescriptor *metricdata.Descriptor
 }
 
 func newViewInternal(v *View) (*viewInternal, error) {
 	return &viewInternal{
-		view:      v,
-		collector: &collector{make(map[string]AggregationData), v.Aggregation},
+		view:             v,
+		collector:        &collector{make(map[string]AggregationData), v.Aggregation},
+		metricDescriptor: viewToMetricDescriptor(v),
 	}, nil
 }
 
diff --git a/stats/view/view_to_metric.go b/stats/view/view_to_metric.go
new file mode 100644
index 0000000..284299f
--- /dev/null
+++ b/stats/view/view_to_metric.go
@@ -0,0 +1,131 @@
+// 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 view
+
+import (
+	"time"
+
+	"go.opencensus.io/metric/metricdata"
+	"go.opencensus.io/stats"
+)
+
+func getUnit(unit string) metricdata.Unit {
+	switch unit {
+	case "1":
+		return metricdata.UnitDimensionless
+	case "ms":
+		return metricdata.UnitMilliseconds
+	case "By":
+		return metricdata.UnitBytes
+	}
+	return metricdata.UnitDimensionless
+}
+
+func getType(v *View) metricdata.Type {
+	m := v.Measure
+	agg := v.Aggregation
+
+	switch agg.Type {
+	case AggTypeSum:
+		switch m.(type) {
+		case *stats.Int64Measure:
+			return metricdata.TypeCumulativeInt64
+		case *stats.Float64Measure:
+			return metricdata.TypeCumulativeFloat64
+		default:
+			panic("unexpected measure type")
+		}
+	case AggTypeDistribution:
+		return metricdata.TypeCumulativeDistribution
+	case AggTypeLastValue:
+		switch m.(type) {
+		case *stats.Int64Measure:
+			return metricdata.TypeGaugeInt64
+		case *stats.Float64Measure:
+			return metricdata.TypeGaugeFloat64
+		default:
+			panic("unexpected measure type")
+		}
+	case AggTypeCount:
+		switch m.(type) {
+		case *stats.Int64Measure:
+			return metricdata.TypeCumulativeInt64
+		case *stats.Float64Measure:
+			return metricdata.TypeCumulativeInt64
+		default:
+			panic("unexpected measure type")
+		}
+	default:
+		panic("unexpected aggregation type")
+	}
+}
+
+func getLableKeys(v *View) []string {
+	labelKeys := []string{}
+	for _, k := range v.TagKeys {
+		labelKeys = append(labelKeys, k.Name())
+	}
+	return labelKeys
+}
+
+func viewToMetricDescriptor(v *View) *metricdata.Descriptor {
+	return &metricdata.Descriptor{
+		Name:        v.Name,
+		Description: v.Description,
+		Unit:        getUnit(v.Measure.Unit()),
+		Type:        getType(v),
+		LabelKeys:   getLableKeys(v),
+	}
+}
+
+func toLabelValues(row *Row) []metricdata.LabelValue {
+	labelValues := []metricdata.LabelValue{}
+	for _, tag := range row.Tags {
+		labelValues = append(labelValues, metricdata.NewLabelValue(tag.Value))
+	}
+	return labelValues
+}
+
+func rowToTimeseries(v *viewInternal, row *Row, now time.Time, startTime time.Time) *metricdata.TimeSeries {
+	return &metricdata.TimeSeries{
+		Points:      []metricdata.Point{row.Data.toPoint(v.metricDescriptor.Type, now)},
+		LabelValues: toLabelValues(row),
+		StartTime:   startTime,
+	}
+}
+
+func viewToMetric(v *viewInternal, now time.Time, startTime time.Time) *metricdata.Metric {
+	if v.metricDescriptor.Type == metricdata.TypeGaugeInt64 ||
+		v.metricDescriptor.Type == metricdata.TypeGaugeFloat64 {
+		startTime = time.Time{}
+	}
+
+	rows := v.collectedRows()
+	if len(rows) == 0 {
+		return nil
+	}
+
+	ts := []*metricdata.TimeSeries{}
+	for _, row := range rows {
+		ts = append(ts, rowToTimeseries(v, row, now, startTime))
+	}
+
+	m := &metricdata.Metric{
+		Descriptor: *v.metricDescriptor,
+		TimeSeries: ts,
+	}
+	return m
+}
diff --git a/stats/view/worker.go b/stats/view/worker.go
index d29dbae..37279b3 100644
--- a/stats/view/worker.go
+++ b/stats/view/worker.go
@@ -17,8 +17,11 @@ package view
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
+	"go.opencensus.io/metric/metricdata"
+	"go.opencensus.io/metric/metricproducer"
 	"go.opencensus.io/stats"
 	"go.opencensus.io/stats/internal"
 	"go.opencensus.io/tag"
@@ -43,6 +46,7 @@ type worker struct {
 	timer      *time.Ticker
 	c          chan command
 	quit, done chan bool
+	mu         sync.RWMutex
 }
 
 var defaultWorker *worker
@@ -143,6 +147,9 @@ func newWorker() *worker {
 }
 
 func (w *worker) start() {
+	prodMgr := metricproducer.GlobalManager()
+	prodMgr.AddProducer(w)
+
 	for {
 		select {
 		case cmd := <-w.c:
@@ -159,6 +166,9 @@ func (w *worker) start() {
 }
 
 func (w *worker) stop() {
+	prodMgr := metricproducer.GlobalManager()
+	prodMgr.DeleteProducer(w)
+
 	w.quit <- true
 	<-w.done
 }
@@ -176,6 +186,8 @@ func (w *worker) getMeasureRef(name string) *measureRef {
 }
 
 func (w *worker) tryRegisterView(v *View) (*viewInternal, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
 	vi, err := newViewInternal(v)
 	if err != nil {
 		return nil, err
@@ -195,6 +207,12 @@ func (w *worker) tryRegisterView(v *View) (*viewInternal, error) {
 	return vi, nil
 }
 
+func (w *worker) unregisterView(viewName string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	delete(w.views, viewName)
+}
+
 func (w *worker) reportView(v *viewInternal, now time.Time) {
 	if !v.isSubscribed() {
 		return
@@ -222,3 +240,40 @@ func (w *worker) reportUsage(now time.Time) {
 		w.reportView(v, now)
 	}
 }
+
+func (w *worker) toMetric(v *viewInternal, now time.Time) *metricdata.Metric {
+	if !v.isSubscribed() {
+		return nil
+	}
+
+	_, ok := w.startTimes[v]
+	if !ok {
+		w.startTimes[v] = now
+	}
+
+	var startTime time.Time
+	if v.metricDescriptor.Type == metricdata.TypeGaugeInt64 ||
+		v.metricDescriptor.Type == metricdata.TypeGaugeFloat64 {
+		startTime = time.Time{}
+	} else {
+		startTime = w.startTimes[v]
+	}
+
+	return viewToMetric(v, now, startTime)
+}
+
+// Read reads all view data and returns them as metrics.
+// It is typically invoked by metric reader to export stats in metric format.
+func (w *worker) Read() []*metricdata.Metric {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	now := time.Now()
+	metrics := make([]*metricdata.Metric, 0, len(w.views))
+	for _, v := range w.views {
+		metric := w.toMetric(v, now)
+		if metric != nil {
+			metrics = append(metrics, metric)
+		}
+	}
+	return metrics
+}
diff --git a/stats/view/worker_commands.go b/stats/view/worker_commands.go
index e27f294..ba6203a 100644
--- a/stats/view/worker_commands.go
+++ b/stats/view/worker_commands.go
@@ -103,7 +103,7 @@ func (cmd *unregisterFromViewReq) handleCommand(w *worker) {
 			// The collected data can be cleared.
 			vi.clearRows()
 		}
-		delete(w.views, name)
+		w.unregisterView(name)
 	}
 	cmd.done <- struct{}{}
 }


Test patch:
diff --git a/stats/view/view_to_metric_test.go b/stats/view/view_to_metric_test.go
new file mode 100644
index 0000000..d9a5a3d
--- /dev/null
+++ b/stats/view/view_to_metric_test.go
@@ -0,0 +1,424 @@
+// 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 view
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"encoding/json"
+	"github.com/google/go-cmp/cmp"
+	"go.opencensus.io/metric/metricdata"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/tag"
+)
+
+type recordValWithTag struct {
+	tags  []tag.Tag
+	value interface{}
+}
+type testToMetrics struct {
+	vi          *viewInternal
+	view        *View
+	recordValue []recordValWithTag
+	wantMetric  *metricdata.Metric
+}
+
+var (
+	// tag objects.
+	tk1         tag.Key
+	tk2         tag.Key
+	tk3         tag.Key
+	tk1v1       tag.Tag
+	tk1v2       tag.Tag
+	tk2v2       tag.Tag
+	tk3v3       tag.Tag
+	tags        []tag.Tag
+	labelValues []metricdata.LabelValue
+	labelKeys   []string
+
+	recordsInt64   []recordValWithTag
+	recordsFloat64 []recordValWithTag
+
+	// distribution objects.
+	aggDist *Aggregation
+	aggCnt  *Aggregation
+	aggS    *Aggregation
+	aggL    *Aggregation
+	buckOpt *metricdata.BucketOptions
+
+	// exemplar objects.
+	attachments metricdata.Attachments
+
+	// views and descriptors
+	viewTypeFloat64Distribution         *View
+	viewTypeInt64Distribution           *View
+	viewTypeInt64Count                  *View
+	viewTypeFloat64Count                *View
+	viewTypeFloat64Sum                  *View
+	viewTypeInt64Sum                    *View
+	viewTypeFloat64LastValue            *View
+	viewTypeInt64LastValue              *View
+	mdTypeFloat64CumulativeDistribution metricdata.Descriptor
+	mdTypeInt64CumulativeDistribution   metricdata.Descriptor
+	mdTypeInt64CumulativeCount          metricdata.Descriptor
+	mdTypeFloat64CumulativeCount        metricdata.Descriptor
+	mdTypeInt64CumulativeSum            metricdata.Descriptor
+	mdTypeFloat64CumulativeSum          metricdata.Descriptor
+	mdTypeInt64CumulativeLastValue      metricdata.Descriptor
+	mdTypeFloat64CumulativeLastValue    metricdata.Descriptor
+)
+
+const (
+	nameInt64DistM1        = "viewToMetricTest_Int64_Distribution/m1"
+	nameFloat64DistM1      = "viewToMetricTest_Float64_Distribution/m1"
+	nameInt64CountM1       = "viewToMetricTest_Int64_Count/m1"
+	nameFloat64CountM1     = "viewToMetricTest_Float64_Count/m1"
+	nameInt64SumM1         = "viewToMetricTest_Int64_Sum/m1"
+	nameFloat64SumM1       = "viewToMetricTest_Float64_Sum/m1"
+	nameInt64LastValueM1   = "viewToMetricTest_Int64_LastValue/m1"
+	nameFloat64LastValueM1 = "viewToMetricTest_Float64_LastValue/m1"
+	v1                     = "v1"
+	v2                     = "v2"
+	v3                     = "v3"
+)
+
+func init() {
+	initTags()
+	initAgg()
+	initViews()
+	initMetricDescriptors()
+
+}
+
+func initTags() {
+	tk1, _ = tag.NewKey("k1")
+	tk2, _ = tag.NewKey("k2")
+	tk3, _ = tag.NewKey("k3")
+	tk1v1 = tag.Tag{Key: tk1, Value: v1}
+	tk1v2 = tag.Tag{Key: tk1, Value: v2}
+	tk2v2 = tag.Tag{Key: tk2, Value: v2}
+	tk3v3 = tag.Tag{Key: tk3, Value: v3}
+
+	tags = []tag.Tag{tk1v1, tk2v2}
+	labelValues = []metricdata.LabelValue{
+		{Value: v1, Present: true},
+		{Value: v2, Present: true},
+	}
+	labelKeys = []string{tk1.Name(), tk2.Name()}
+
+	recordsInt64 = []recordValWithTag{
+		{tags: tags, value: int64(2)},
+		{tags: tags, value: int64(4)},
+	}
+	recordsFloat64 = []recordValWithTag{
+		{tags: tags, value: float64(1.5)},
+		{tags: tags, value: float64(5.4)},
+	}
+}
+
+func initAgg() {
+	aggDist = Distribution(2.0)
+	aggCnt = Count()
+	aggS = Sum()
+	aggL = LastValue()
+	buckOpt = &metricdata.BucketOptions{Bounds: []float64{2.0}}
+}
+
+func initViews() {
+	// View objects
+	viewTypeInt64Distribution = &View{
+		Name:        nameInt64DistM1,
+		TagKeys:     []tag.Key{tk1, tk2},
+		Measure:     stats.Int64(nameInt64DistM1, "", stats.UnitDimensionless),
+		Aggregation: aggDist,
+	}
+	viewTypeFloat64Distribution = &View{
+		Name:        nameFloat64DistM1,
+		TagKeys:     []tag.Key{tk1, tk2},
+		Measure:     stats.Float64(nameFloat64DistM1, "", stats.UnitDimensionless),
+		Aggregation: aggDist,
+	}
+	viewTypeInt64Count = &View{
+		Name:        nameInt64CountM1,
+		TagKeys:     []tag.Key{tk1, tk2},
+		Measure:     stats.Int64(nameInt64CountM1, "", stats.UnitDimensionless),
+		Aggregation: aggCnt,
+	}
+	viewTypeFloat64Count = &View{
+		Name:        nameFloat64CountM1,
+		TagKeys:     []tag.Key{tk1, tk2},
+		Measure:     stats.Float64(nameFloat64CountM1, "", stats.UnitDimensionless),
+		Aggregation: aggCnt,
+	}
+	viewTypeInt64Sum = &View{
+		Name:        nameInt64SumM1,
+		TagKeys:     []tag.Key{tk1, tk2},
+		Measure:     stats.Int64(nameInt64SumM1, "", stats.UnitBytes),
+		Aggregation: aggS,
+	}
+	viewTypeFloat64Sum = &View{
+		Name:        nameFloat64SumM1,
+		TagKeys:     []tag.Key{tk1, tk2},
+		Measure:     stats.Float64(nameFloat64SumM1, "", stats.UnitMilliseconds),
+		Aggregation: aggS,
+	}
+	viewTypeInt64LastValue = &View{
+		Name:        nameInt64LastValueM1,
+		TagKeys:     []tag.Key{tk1, tk2},
+		Measure:     stats.Int64(nameInt64LastValueM1, "", stats.UnitDimensionless),
+		Aggregation: aggL,
+	}
+	viewTypeFloat64LastValue = &View{
+		Name:        nameFloat64LastValueM1,
+		TagKeys:     []tag.Key{tk1, tk2},
+		Measure:     stats.Float64(nameFloat64LastValueM1, "", stats.UnitDimensionless),
+		Aggregation: aggL,
+	}
+}
+
+func initMetricDescriptors() {
+	// Metric objects
+	mdTypeFloat64CumulativeDistribution = metricdata.Descriptor{
+		Name: nameFloat64DistM1, Description: "", Unit: metricdata.UnitDimensionless,
+		Type: metricdata.TypeCumulativeDistribution, LabelKeys: labelKeys,
+	}
+	mdTypeInt64CumulativeDistribution = metricdata.Descriptor{
+		Name: nameInt64DistM1, Description: "", Unit: metricdata.UnitDimensionless,
+		Type: metricdata.TypeCumulativeDistribution, LabelKeys: labelKeys,
+	}
+	mdTypeInt64CumulativeCount = metricdata.Descriptor{
+		Name: nameInt64CountM1, Description: "", Unit: metricdata.UnitDimensionless,
+		Type: metricdata.TypeCumulativeInt64, LabelKeys: labelKeys,
+	}
+	mdTypeFloat64CumulativeCount = metricdata.Descriptor{
+		Name: nameFloat64CountM1, Description: "", Unit: metricdata.UnitDimensionless,
+		Type: metricdata.TypeCumulativeInt64, LabelKeys: labelKeys,
+	}
+	mdTypeInt64CumulativeSum = metricdata.Descriptor{
+		Name: nameInt64SumM1, Description: "", Unit: metricdata.UnitBytes,
+		Type: metricdata.TypeCumulativeInt64, LabelKeys: labelKeys,
+	}
+	mdTypeFloat64CumulativeSum = metricdata.Descriptor{
+		Name: nameFloat64SumM1, Description: "", Unit: metricdata.UnitMilliseconds,
+		Type: metricdata.TypeCumulativeFloat64, LabelKeys: labelKeys,
+	}
+	mdTypeInt64CumulativeLastValue = metricdata.Descriptor{
+		Name: nameInt64LastValueM1, Description: "", Unit: metricdata.UnitDimensionless,
+		Type: metricdata.TypeGaugeInt64, LabelKeys: labelKeys,
+	}
+	mdTypeFloat64CumulativeLastValue = metricdata.Descriptor{
+		Name: nameFloat64LastValueM1, Description: "", Unit: metricdata.UnitDimensionless,
+		Type: metricdata.TypeGaugeFloat64, LabelKeys: labelKeys,
+	}
+}
+
+func Test_ViewToMetric(t *testing.T) {
+	startTime := time.Now().Add(-time.Duration(60 * time.Second))
+	now := time.Now()
+	tests := []*testToMetrics{
+		{
+			view:        viewTypeInt64Distribution,
+			recordValue: recordsInt64,
+			wantMetric: &metricdata.Metric{
+				Descriptor: mdTypeInt64CumulativeDistribution,
+				TimeSeries: []*metricdata.TimeSeries{
+					{Points: []metricdata.Point{
+						{Value: &metricdata.Distribution{
+							Count:                 2,
+							Sum:                   6.0,
+							SumOfSquaredDeviation: 2,
+							BucketOptions:         buckOpt,
+							Buckets: []metricdata.Bucket{
+								{Count: 0, Exemplar: nil},
+								{Count: 2, Exemplar: nil},
+							},
+						},
+							Time: now,
+						},
+					},
+						LabelValues: labelValues,
+						StartTime:   startTime,
+					},
+				},
+			},
+		},
+		{
+			view:        viewTypeFloat64Distribution,
+			recordValue: recordsFloat64,
+			wantMetric: &metricdata.Metric{
+				Descriptor: mdTypeFloat64CumulativeDistribution,
+				TimeSeries: []*metricdata.TimeSeries{
+					{
+						Points: []metricdata.Point{
+							{
+								Value: &metricdata.Distribution{
+									Count:                 2,
+									Sum:                   6.9,
+									SumOfSquaredDeviation: 7.605000000000001,
+									BucketOptions:         buckOpt,
+									Buckets: []metricdata.Bucket{
+										{Count: 1, Exemplar: nil}, // TODO: [rghetia] add exemplar test.
+										{Count: 1, Exemplar: nil},
+									},
+								},
+								Time: now,
+							},
+						},
+						LabelValues: labelValues,
+						StartTime:   startTime,
+					},
+				},
+			},
+		},
+		{
+			view:        viewTypeInt64Count,
+			recordValue: recordsInt64,
+			wantMetric: &metricdata.Metric{
+				Descriptor: mdTypeInt64CumulativeCount,
+				TimeSeries: []*metricdata.TimeSeries{
+					{Points: []metricdata.Point{
+						metricdata.NewInt64Point(now, 2),
+					},
+						LabelValues: labelValues,
+						StartTime:   startTime,
+					},
+				},
+			},
+		},
+		{
+			view:        viewTypeFloat64Count,
+			recordValue: recordsFloat64,
+			wantMetric: &metricdata.Metric{
+				Descriptor: mdTypeFloat64CumulativeCount,
+				TimeSeries: []*metricdata.TimeSeries{
+					{Points: []metricdata.Point{
+						metricdata.NewInt64Point(now, 2),
+					},
+						LabelValues: labelValues,
+						StartTime:   startTime,
+					},
+				},
+			},
+		},
+		{
+			view:        viewTypeInt64Sum,
+			recordValue: recordsInt64,
+			wantMetric: &metricdata.Metric{
+				Descriptor: mdTypeInt64CumulativeSum,
+				TimeSeries: []*metricdata.TimeSeries{
+					{Points: []metricdata.Point{
+						metricdata.NewInt64Point(now, 6),
+					},
+						LabelValues: labelValues,
+						StartTime:   startTime,
+					},
+				},
+			},
+		},
+		{
+			view:        viewTypeFloat64Sum,
+			recordValue: recordsFloat64,
+			wantMetric: &metricdata.Metric{
+				Descriptor: mdTypeFloat64CumulativeSum,
+				TimeSeries: []*metricdata.TimeSeries{
+					{Points: []metricdata.Point{
+						metricdata.NewFloat64Point(now, 6.9),
+					},
+						LabelValues: labelValues,
+						StartTime:   startTime,
+					},
+				},
+			},
+		},
+		{
+			view:        viewTypeInt64LastValue,
+			recordValue: recordsInt64,
+			wantMetric: &metricdata.Metric{
+				Descriptor: mdTypeInt64CumulativeLastValue,
+				TimeSeries: []*metricdata.TimeSeries{
+					{Points: []metricdata.Point{
+						metricdata.NewInt64Point(now, 4),
+					},
+						LabelValues: labelValues,
+						StartTime:   time.Time{},
+					},
+				},
+			},
+		},
+		{
+			view:        viewTypeFloat64LastValue,
+			recordValue: recordsFloat64,
+			wantMetric: &metricdata.Metric{
+				Descriptor: mdTypeFloat64CumulativeLastValue,
+				TimeSeries: []*metricdata.TimeSeries{
+					{Points: []metricdata.Point{
+						metricdata.NewFloat64Point(now, 5.4),
+					},
+						LabelValues: labelValues,
+						StartTime:   time.Time{},
+					},
+				},
+			},
+		},
+	}
+
+	wantMetrics := []*metricdata.Metric{}
+	for _, tc := range tests {
+		tc.vi, _ = defaultWorker.tryRegisterView(tc.view)
+		tc.vi.clearRows()
+		tc.vi.subscribe()
+		wantMetrics = append(wantMetrics, tc.wantMetric)
+	}
+
+	for i, tc := range tests {
+		for _, r := range tc.recordValue {
+			mods := []tag.Mutator{}
+			for _, tg := range r.tags {
+				mods = append(mods, tag.Insert(tg.Key, tg.Value))
+			}
+			ctx, err := tag.New(context.Background(), mods...)
+			if err != nil {
+				t.Errorf("%v: New = %v", tc.view.Name, err)
+			}
+			var v float64
+			switch i := r.value.(type) {
+			case float64:
+				v = float64(i)
+			case int64:
+				v = float64(i)
+			default:
+				t.Errorf("unexpected value type %v", r.tags)
+			}
+			tc.vi.addSample(tag.FromContext(ctx), v, nil, now)
+		}
+
+		gotMetric := viewToMetric(tc.vi, now, startTime)
+		if !cmp.Equal(gotMetric, tc.wantMetric) {
+			// JSON format is strictly for checking the content when test fails. Do not use JSON
+			// format to determine if the two values are same as it doesn't differentiate between
+			// int64(2) and float64(2.0)
+			t.Errorf("#%d: Unmatched \nGot:\n\t%v\nWant:\n\t%v\nGot Serialized:%s\nWant Serialized:%s\n",
+				i, gotMetric, tc.wantMetric, serializeAsJSON(gotMetric), serializeAsJSON(tc.wantMetric))
+		}
+	}
+}
+
+func serializeAsJSON(v interface{}) string {
+	blob, _ := json.MarshalIndent(v, "", "  ")
+	return string(blob)
+}