review round 3 response

Author: lychung83

monitoring/exporter/stackdriver/mock_check_test.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"fmt"
 	"strings"
-	"testing"
 
 	monitoring "cloud.google.com/go/monitoring/apiv3"
 	gax "github.com/googleapis/gax-go"
@@ -50,9 +49,6 @@ var (
 )
 
 func init() {
-	// For testing convenience, we reduce maximum time series that metric client accepts.
-	MaxTimeSeriesPerUpload = 3
-
 	// Mock functions.
 	newMetricClient = mockNewMetricClient
 	createTimeSeries = mockCreateTimeSeries
@@ -126,22 +122,29 @@ func mockAddToBundler(bndler *bundler.Bundler, item interface{}, _ int) error {
 // One of these functions once and only once, and never call NewExporter() directly.
 
 // newTestExp creates an exporter which saves error to errStorage. Caller should not set
-// opts.OnError.
-func newTestExp(t *testing.T, opts Options) *Exporter {
+// opts.OnError and opts.BundleCountThreshold.
+func newTestExp(opts Options) (*Exporter, error) {
 	opts.OnError = testOnError
+	// For testing convenience, we reduce the number of timeseris in one upload monitoring API
+	// call.
+	opts.BundleCountThreshold = 3
 	exp, err := NewExporter(ctx, opts)
 	if err != nil {
-		t.Fatalf("creating exporter failed: %v", err)
+		return nil, fmt.Errorf("creating exporter failed: %v", err)
 	}
 	// Expose projDataMap so that mockAddToBundler() can use it.
 	projDataMap = exp.projDataMap
-	return exp
+	return exp, nil
 }
 
 // newTestProjData creates a projectData object to test behavior of projectData.uploadRowData. Other
 // uses are not recommended. As newTestExp, all errors are saved to errStorage.
-func newTestProjData(t *testing.T, opts Options) *projectData {
-	return newTestExp(t, opts).newProjectData(project1)
+func newTestProjData(opts Options) (*projectData, error) {
+	exp, err := newTestExp(opts)
+	if err != nil {
+		return nil, err
+	}
+	return exp.newProjectData(project1), nil
 }
 
 // We define a storage for all errors happened in export operation.
@@ -157,33 +160,54 @@ func testOnError(err error, rds ...*RowData) {
 	errStorage = append(errStorage, errRowData{err, rds})
 }
 
+// multiError stores a sequence of errors. To convert it to an actual error, call toError().
+type multiError struct {
+	errs []error
+}
+
+func (me *multiError) addf(format string, args ...interface{}) {
+	me.errs = append(me.errs, fmt.Errorf(format, args...))
+}
+
+func (me *multiError) toError() error {
+	switch len(me.errs) {
+	case 0:
+		return nil
+	case 1:
+		return me.errs[0]
+	default:
+		return fmt.Errorf("multiple errors: %q", me.errs)
+	}
+}
+
 // checkMetricClient checks all recorded requests to the metric client. We only compare int64
 // values of the time series. To make this work, we assigned different int64 values for all valid
 // rows in the test.
-func checkMetricClient(t *testing.T, wantReqsValues [][]int64) {
+func checkMetricClient(wantReqsValues [][]int64) error {
 	reqsLen, wantReqsLen := len(timeSeriesReqs), len(wantReqsValues)
 	if reqsLen != wantReqsLen {
-		t.Errorf("number of requests got: %d, want %d", reqsLen, wantReqsLen)
-		return
+		return fmt.Errorf("number of requests got: %d, want %d", reqsLen, wantReqsLen)
 	}
+	var errs multiError
 	for i := 0; i < reqsLen; i++ {
 		prefix := fmt.Sprintf("%d-th request mismatch", i+1)
 		tsArr := timeSeriesReqs[i].TimeSeries
 		wantTsValues := wantReqsValues[i]
 		tsArrLen, wantTsArrLen := len(tsArr), len(wantTsValues)
 		if tsArrLen != wantTsArrLen {
-			t.Errorf("%s: number of time series got: %d, want: %d", prefix, tsArrLen, wantTsArrLen)
+			errs.addf("%s: number of time series got: %d, want: %d", prefix, tsArrLen, wantTsArrLen)
 			continue
 		}
 		for j := 0; j < tsArrLen; j++ {
 			// This is how monitoring API stores the int64 value.
 			tsVal := tsArr[j].Points[0].Value.Value.(*mpb.TypedValue_Int64Value).Int64Value
 			wantTsVal := wantTsValues[j]
 			if tsVal != wantTsVal {
-				t.Errorf("%s: Value got: %d, want: %d", prefix, tsVal, wantTsVal)
+				errs.addf("%s: Value got: %d, want: %d", prefix, tsVal, wantTsVal)
 			}
 		}
 	}
+	return errs.toError()
 }
 
 // errRowDataCheck contains data for checking content of error storage.
@@ -193,96 +217,99 @@ type errRowDataCheck struct {
 }
 
 // checkErrStorage checks content of error storage. For returned errors, we check prefix and suffix.
-func checkErrStorage(t *testing.T, wantErrRdCheck []errRowDataCheck) {
+func checkErrStorage(wantErrRdCheck []errRowDataCheck) error {
 	gotLen, wantLen := len(errStorage), len(wantErrRdCheck)
 	if gotLen != wantLen {
-		t.Errorf("number of reported errors: %d, want: %d", gotLen, wantLen)
-		return
+		return fmt.Errorf("number of reported errors: %d, want: %d", gotLen, wantLen)
 	}
+	var errs multiError
 	for i := 0; i < gotLen; i++ {
 		prefix := fmt.Sprintf("%d-th reported error mismatch", i+1)
 		errRd, wantErrRd := errStorage[i], wantErrRdCheck[i]
 		errStr := errRd.err.Error()
 		if errPrefix := wantErrRd.errPrefix; !strings.HasPrefix(errStr, errPrefix) {
-			t.Errorf("%s: error got: %q, want: prefixed by %q", prefix, errStr, errPrefix)
+			errs.addf("%s: error got: %q, want: prefixed by %q", prefix, errStr, errPrefix)
 		}
 		if errSuffix := wantErrRd.errSuffix; !strings.HasSuffix(errStr, errSuffix) {
-			t.Errorf("%s: error got: %q, want: suffiexd by %q", prefix, errStr, errSuffix)
+			errs.addf("%s: error got: %q, want: suffiexd by %q", prefix, errStr, errSuffix)
 		}
 		if err := checkRowDataArr(errRd.rds, wantErrRd.rds); err != nil {
-			t.Errorf("%s: RowData array mismatch: %v", prefix, err)
+			errs.addf("%s: RowData array mismatch: %v", prefix, err)
 		}
 	}
+	return errs.toError()
 }
 
 func checkRowDataArr(rds, wantRds []*RowData) error {
 	rdLen, wantRdLen := len(rds), len(wantRds)
 	if rdLen != wantRdLen {
 		return fmt.Errorf("number row data got: %d, want: %d", rdLen, wantRdLen)
 	}
+	var errs multiError
 	for i := 0; i < rdLen; i++ {
 		if err := checkRowData(rds[i], wantRds[i]); err != nil {
-			return fmt.Errorf("%d-th row data mismatch: %v", i+1, err)
+			errs.addf("%d-th row data mismatch: %v", i+1, err)
 		}
 	}
-	return nil
+	return errs.toError()
 }
 
 func checkRowData(rd, wantRd *RowData) error {
+	var errs multiError
 	if rd.View != wantRd.View {
-		return fmt.Errorf("View got: %s, want: %s", rd.View.Name, wantRd.View.Name)
+		errs.addf("View got: %s, want: %s", rd.View.Name, wantRd.View.Name)
 	}
 	if rd.Start != wantRd.Start {
-		return fmt.Errorf("Start got: %v, want: %v", rd.Start, wantRd.Start)
+		errs.addf("Start got: %v, want: %v", rd.Start, wantRd.Start)
 	}
 	if rd.End != wantRd.End {
-		return fmt.Errorf("End got: %v, want: %v", rd.End, wantRd.End)
+		errs.addf("End got: %v, want: %v", rd.End, wantRd.End)
 	}
 	if rd.Row != wantRd.Row {
-		return fmt.Errorf("Row got: %v, want: %v", rd.Row, wantRd.Row)
+		errs.addf("Row got: %v, want: %v", rd.Row, wantRd.Row)
 	}
-	return nil
+	return errs.toError()
 }
 
 // checkProjData checks all data passed to the bundler by bundler.Add().
-func checkProjData(t *testing.T, wantProjData map[string][]*RowData) {
-	wantProj := map[string]bool{}
-	for proj := range wantProjData {
-		wantProj[proj] = true
-	}
+func checkProjData(wantProjData map[string][]*RowData) error {
+	var errs multiError
 	for proj := range projRds {
-		if !wantProj[proj] {
-			t.Errorf("project in exporter's project data not wanted: %s", proj)
+		if _, ok := wantProjData[proj]; !ok {
+			errs.addf("project in exporter's project data not wanted: %s", proj)
 		}
 	}
 
 	for proj, wantRds := range wantProjData {
 		rds, ok := projRds[proj]
 		if !ok {
-			t.Errorf("wanted project not found in exporter's project data: %v", proj)
+			errs.addf("wanted project not found in exporter's project data: %v", proj)
 			continue
 		}
 		if err := checkRowDataArr(*rds, wantRds); err != nil {
-			t.Errorf("RowData array mismatch for project %s: %v", proj, err)
+			errs.addf("RowData array mismatch for project %s: %v", proj, err)
 		}
 	}
+	return errs.toError()
 }
 
 // checkLabels checks data in labels.
-func checkLabels(t *testing.T, prefix string, labels, wantLabels map[string]string) {
+func checkLabels(prefix string, labels, wantLabels map[string]string) error {
+	var errs multiError
 	for labelName, value := range labels {
 		wantValue, ok := wantLabels[labelName]
 		if !ok {
-			t.Errorf("%s: label name in time series not wanted: %s", prefix, labelName)
+			errs.addf("%s: label name in time series not wanted: %s", prefix, labelName)
 			continue
 		}
 		if value != wantValue {
-			t.Errorf("%s: value for label name %s got: %s, want: %s", prefix, labelName, value, wantValue)
+			errs.addf("%s: value for label name %s got: %s, want: %s", prefix, labelName, value, wantValue)
 		}
 	}
 	for wantLabelName := range wantLabels {
 		if _, ok := labels[wantLabelName]; !ok {
-			t.Errorf("%s: wanted label name not found in time series: %s", prefix, wantLabelName)
+			errs.addf("%s: wanted label name not found in time series: %s", prefix, wantLabelName)
 		}
 	}
+	return errs.toError()
 }

monitoring/exporter/stackdriver/project_data.go

@@ -21,16 +21,11 @@ import (
 	mpb "google.golang.org/genproto/googleapis/monitoring/v3"
 )
 
-// MaxTimeSeriePerUpload is the maximum number of time series that's uploaded to the stackdriver
-// at once. Consumer may change this value, but note that stackdriver may reject upload request if
-// the number of time series is too large.
-var MaxTimeSeriesPerUpload = 100
-
 // projectData contain per-project data in exporter. It should be created by newProjectData()
 type projectData struct {
 	parent    *Exporter
 	projectID string
-	// We make bundler for each project because call to monitoring RPC can be grouped only in
+	// We make bundler for each project because call to monitoring API can be grouped only in
 	// project level
 	bndler *bundler.Bundler
 }
@@ -55,7 +50,7 @@ func (pd *projectData) uploadRowData(bundle interface{}) {
 		// Time series created. We update both uploadTs and uploadRds.
 		uploadTs = append(uploadTs, ts)
 		uploadRds = append(uploadRds, rd)
-		if len(uploadTs) == MaxTimeSeriesPerUpload {
+		if len(uploadTs) == exp.opts.BundleCountThreshold {
 			pd.uploadTimeSeries(uploadTs, uploadRds)
 			uploadTs = nil
 			uploadRds = nil
@@ -77,7 +72,7 @@ func (pd *projectData) uploadTimeSeries(ts []*mpb.TimeSeries, rds []*RowData) {
 		TimeSeries: ts,
 	}
 	if err := createTimeSeries(exp.client, exp.ctx, req); err != nil {
-		newErr := fmt.Errorf("RPC call to create time series failed for project %s: %v", pd.projectID, err)
+		newErr := fmt.Errorf("monitoring API call to create time series failed for project %s: %v", pd.projectID, err)
 		// We pass all row data not successfully uploaded.
 		exp.opts.OnError(newErr, rds...)
 	}

monitoring/exporter/stackdriver/row_data_to_point.go

@@ -25,7 +25,7 @@ import (
 )
 
 // Functions in this file is used to convert RowData to monitoring point that are used by uploading
-// RPC calls of monitoring client. All functions in this file are copied from
+// monitoring API calls. All functions in this file are copied from
 // contrib.go.opencensus.io/exporter/stackdriver.
 
 func newPoint(v *view.View, row *view.Row, start, end time.Time) *mpb.Point {

monitoring/exporter/stackdriver/stackdriver.go

@@ -44,6 +44,10 @@ import (
 	mpb "google.golang.org/genproto/googleapis/monitoring/v3"
 )
 
+// MaxTimeSeriesPerUpload is the maximum number of timeseries objects that will be uploaded to
+// Stackdriver in one API call.
+const MaxTimeSeriesPerUpload = 100
+
 // Exporter is the exporter that can be registered to opencensus. An Exporter object must be
 // created by NewExporter().
 type Exporter struct {
@@ -62,17 +66,19 @@ type Exporter struct {
 // are valid for use.
 type Options struct {
 	// ClientOptions designates options for creating metric client, especially credentials for
-	// RPC calls.
+	// monitoring API calls.
 	ClientOptions []option.ClientOption
 
 	// Options for bundles amortizing export requests. Note that a bundle is created for each
-	// project. When not provided, default values in bundle package are used.
+	// project.
 
 	// BundleDelayThreshold determines the max amount of time the exporter can wait before
-	// uploading data to the stackdriver.
+	// uploading data to the stackdriver. If this value is not positive, the default value in
+	// the bundle package is used.
 	BundleDelayThreshold time.Duration
 	// BundleCountThreshold determines how many RowData objects can be buffered before batch
-	// uploading them to the backend.
+	// uploading them to the backend. If this value is not between 1 and MaxTimeSeriesPerUpload,
+	// MaxTimeSeriesPerUpload is used.
 	BundleCountThreshold int
 
 	// Callback functions provided by user.
@@ -143,7 +149,7 @@ var (
 
 // NewExporter creates an Exporter object. Once a call to NewExporter is made, any fields in opts
 // must not be modified at all. ctx will also be used throughout entire exporter operation when
-// making RPC call.
+// making monitoring API call.
 func NewExporter(ctx context.Context, opts Options) (*Exporter, error) {
 	client, err := newMetricClient(ctx, opts.ClientOptions...)
 	if err != nil {
@@ -157,6 +163,12 @@ func NewExporter(ctx context.Context, opts Options) (*Exporter, error) {
 		projDataMap: make(map[string]*projectData),
 	}
 
+	if !(0 < e.opts.BundleDelayThreshold) {
+		e.opts.BundleDelayThreshold = bundler.DefaultDelayThreshold
+	}
+	if !(0 < e.opts.BundleCountThreshold && e.opts.BundleCountThreshold <= MaxTimeSeriesPerUpload) {
+		e.opts.BundleCountThreshold = MaxTimeSeriesPerUpload
+	}
 	if e.opts.GetProjectID == nil {
 		e.opts.GetProjectID = defaultGetProjectID
 	}
@@ -238,13 +250,8 @@ func (e *Exporter) newProjectData(projectID string) *projectData {
 	}
 
 	pd.bndler = newBundler((*RowData)(nil), pd.uploadRowData)
-	// Set options for bundler if they are provided by users.
-	if 0 < e.opts.BundleDelayThreshold {
-		pd.bndler.DelayThreshold = e.opts.BundleDelayThreshold
-	}
-	if 0 < e.opts.BundleCountThreshold {
-		pd.bndler.BundleCountThreshold = e.opts.BundleCountThreshold
-	}
+	pd.bndler.DelayThreshold = e.opts.BundleDelayThreshold
+	pd.bndler.BundleCountThreshold = e.opts.BundleCountThreshold
 	return pd
 }