review round 2 fix

Author: lychung83

monitoring/exporter/stackdriver/mock_check_test.go

@@ -127,7 +127,7 @@ func mockAddToBundler(bndler *bundler.Bundler, item interface{}, _ int) error {
 
 // newTestExp creates an exporter which saves error to errStorage. Caller should not set
 // opts.OnError.
-func newTestExp(t *testing.T, opts *Options) *Exporter {
+func newTestExp(t *testing.T, opts Options) *Exporter {
 	opts.OnError = testOnError
 	exp, err := NewExporter(ctx, opts)
 	if err != nil {
@@ -140,7 +140,7 @@ func newTestExp(t *testing.T, opts *Options) *Exporter {
 
 // 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 {
+func newTestProjData(t *testing.T, opts Options) *projectData {
 	return newTestExp(t, opts).newProjectData(project1)
 }
 

monitoring/exporter/stackdriver/project_data.go

@@ -17,15 +17,14 @@ package stackdriver
 import (
 	"fmt"
 
-	"go.opencensus.io/tag"
 	"google.golang.org/api/support/bundler"
-	metricpb "google.golang.org/genproto/googleapis/api/metric"
 	mpb "google.golang.org/genproto/googleapis/monitoring/v3"
 )
 
-// MaxTimeSeriePerUpload is the maximum number of time series that stackdriver accepts. Only test
-// may change this value.
-var MaxTimeSeriesPerUpload = 200
+// 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 {
@@ -36,121 +35,50 @@ type projectData struct {
 	bndler *bundler.Bundler
 }
 
-func (e *Exporter) newProjectData(projectID string) *projectData {
-	pd := &projectData{
-		parent:    e,
-		projectID: projectID,
-	}
-
-	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
-	}
-	return pd
-}
-
 // uploadRowData is called by bundler to upload row data, and report any error happened meanwhile.
 func (pd *projectData) uploadRowData(bundle interface{}) {
 	exp := pd.parent
 	rds := bundle.([]*RowData)
 
-	// reqRds contains RowData objects those are uploaded to stackdriver at given iteration.
-	// It's main usage is for error reporting. For actual uploading operation, we use req.
-	// remainingRds are RowData that has not been processed at all.
-	var reqRds, remainingRds []*RowData
-	for ; len(rds) != 0; rds = remainingRds {
-		var req *mpb.CreateTimeSeriesRequest
-		req, reqRds, remainingRds = pd.makeReq(rds)
-		if req == nil {
-			// No need to perform RPC call for empty set of requests.
-			continue
-		}
-		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)
-			// We pass all row data not successfully uploaded.
-			exp.onError(newErr, reqRds...)
-		}
-	}
-}
+	// uploadTs contains TimeSeries objects that needs to be uploaded.
+	var uploadTs []*mpb.TimeSeries = nil
+	// uploadRds contains RowData objects corresponds to uploadTs. It's used for error reporting
+	// when upload operation fails.
+	var uploadRds []*RowData = nil
 
-// makeReq creates a request that's suitable to be passed to create time series RPC call.
-//
-// reqRds contains rows those are contained in req. Main use of reqRds is to be returned to users if
-// creating time series failed. (We don't want users to investigate structure of timeseries.)
-// remainingRds contains rows those are not used at all in makeReq because of the length limitation
-// or request. Another call of makeReq() with remainigRds will handle (some) rows in them. When req
-// is nil, then there's nothing to request and reqRds will also contain nothing.
-//
-// Some rows in rds may fail while converting them to time series, and in that case makeReq() calls
-// exporter's onError() directly, not propagating errors to the caller.
-func (pd *projectData) makeReq(rds []*RowData) (req *mpb.CreateTimeSeriesRequest, reqRds, remainingRds []*RowData) {
-	exp := pd.parent
-	timeSeries := []*mpb.TimeSeries{}
-
-	var i int
-	var rd *RowData
-	for i, rd = range rds {
-		pt := newPoint(rd.View, rd.Row, rd.Start, rd.End)
-		if pt.Value == nil {
-			err := fmt.Errorf("inconsistent data found in view %s", rd.View.Name)
-			pd.parent.onError(err, rd)
-			continue
-		}
-		resource, err := exp.makeResource(rd)
+	for _, rd := range rds {
+		ts, err := exp.makeTS(rd)
 		if err != nil {
-			newErr := fmt.Errorf("failed to construct resource of view %s: %v", rd.View.Name, err)
-			pd.parent.onError(newErr, rd)
+			exp.opts.OnError(err, rd)
 			continue
 		}
-
-		ts := &mpb.TimeSeries{
-			Metric: &metricpb.Metric{
-				Type:   rd.View.Name,
-				Labels: exp.makeLabels(rd.Row.Tags),
-			},
-			Resource: resource,
-			Points:   []*mpb.Point{pt},
-		}
-		// Growing timeseries and reqRds are done at same time.
-		timeSeries = append(timeSeries, ts)
-		reqRds = append(reqRds, rd)
-		// Don't grow timeseries over the limit.
-		if len(timeSeries) == MaxTimeSeriesPerUpload {
-			break
+		// Time series created. We update both uploadTs and uploadRds.
+		uploadTs = append(uploadTs, ts)
+		uploadRds = append(uploadRds, rd)
+		if len(uploadTs) == MaxTimeSeriesPerUpload {
+			pd.uploadTimeSeries(uploadTs, uploadRds)
+			uploadTs = nil
+			uploadRds = nil
 		}
 	}
-
-	// Since i is the last index processed, remainingRds should start from i+1.
-	remainingRds = rds[i+1:]
-	if len(timeSeries) == 0 {
-		req = nil
-	} else {
-		req = &mpb.CreateTimeSeriesRequest{
-			Name:       fmt.Sprintf("projects/%s", pd.projectID),
-			TimeSeries: timeSeries,
-		}
+	// Upload any remaining time series.
+	if len(uploadTs) != 0 {
+		pd.uploadTimeSeries(uploadTs, uploadRds)
 	}
-	return req, reqRds, remainingRds
 }
 
-// makeLables constructs label that's ready for being uploaded to stackdriver.
-func (e *Exporter) makeLabels(tags []tag.Tag) map[string]string {
-	opts := e.opts
-	labels := make(map[string]string, len(opts.DefaultLabels)+len(tags))
-	for key, val := range opts.DefaultLabels {
-		labels[key] = val
-	}
-	// If there's overlap When combining exporter's default label and tags, values in tags win.
-	for _, tag := range tags {
-		labels[tag.Key.Name()] = tag.Value
+// uploadTimeSeries uploads timeSeries. ts and rds must contain matching data, and ts must not be
+// empty. When uploading fails, this function calls exporter's OnError() directly, not propagating
+// errors to the caller.
+func (pd *projectData) uploadTimeSeries(ts []*mpb.TimeSeries, rds []*RowData) {
+	exp := pd.parent
+	req := &mpb.CreateTimeSeriesRequest{
+		Name:       fmt.Sprintf("projects/%s", pd.projectID),
+		TimeSeries: ts,
 	}
-	// Some labels are not for exporting.
-	for _, key := range opts.UnexportedLabels {
-		delete(labels, key)
+	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)
+		// We pass all row data not successfully uploaded.
+		exp.opts.OnError(newErr, rds...)
 	}
-	return labels
 }

monitoring/exporter/stackdriver/stackdriver.go

@@ -36,9 +36,12 @@ import (
 
 	monitoring "cloud.google.com/go/monitoring/apiv3"
 	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
 	"google.golang.org/api/option"
 	"google.golang.org/api/support/bundler"
+	metricpb "google.golang.org/genproto/googleapis/api/metric"
 	mrpb "google.golang.org/genproto/googleapis/api/monitoredres"
+	mpb "google.golang.org/genproto/googleapis/monitoring/v3"
 )
 
 // Exporter is the exporter that can be registered to opencensus. An Exporter object must be
@@ -47,12 +50,7 @@ type Exporter struct {
 	// TODO(lawrencechung): If possible, find a way to not storing ctx in the struct.
 	ctx    context.Context
 	client *monitoring.MetricClient
-	opts   *Options
-
-	// copy of some option values which may be modified by exporter.
-	getProjectID func(*RowData) (string, error)
-	onError      func(error, ...*RowData)
-	makeResource func(*RowData) (*mrpb.MonitoredResource, error)
+	opts   Options
 
 	// mu protects access to projDataMap
 	mu sync.Mutex
@@ -67,9 +65,14 @@ type Options struct {
 	// RPC calls.
 	ClientOptions []option.ClientOption
 
-	// options for bundles amortizing export requests. Note that a bundle is created for each
+	// 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.
+
+	// BundleDelayThreshold determines the max amount of time the exporter can wait before
+	// uploading data to the stackdriver.
 	BundleDelayThreshold time.Duration
+	// BundleCountThreshold determines how many RowData objects can be buffered before batch
+	// uploading them to the backend.
 	BundleCountThreshold int
 
 	// Callback functions provided by user.
@@ -85,8 +88,8 @@ type Options struct {
 	GetProjectID func(*RowData) (projectID string, err error)
 	// OnError is used to report any error happened while exporting view data fails. Whenever
 	// this function is called, it's guaranteed that at least one row data is also passed to
-	// OnError. Row data passed to OnError must not be modified. When OnError is not set, all
-	// errors happened on exporting are ignored.
+	// OnError. Row data passed to OnError must not be modified and OnError must be
+	// non-blocking. When OnError is not set, all errors happened on exporting are ignored.
 	OnError func(error, ...*RowData)
 	// MakeResource creates monitored resource from RowData. It is guaranteed that only RowData
 	// that passes GetProjectID will be given to this function. Though not recommended, error
@@ -129,8 +132,8 @@ func defaultMakeResource(rd *RowData) (*mrpb.MonitoredResource, error) {
 	return &mrpb.MonitoredResource{Type: "global"}, nil
 }
 
-// Following functions are wrapper of functions that may show non-deterministic behavior. Only tests
-// can modify these functions.
+// Following functions are wrapper of functions those will be mocked by tests. Only tests can modify
+// these functions.
 var (
 	newMetricClient  = monitoring.NewMetricClient
 	createTimeSeries = (*monitoring.MetricClient).CreateTimeSeries
@@ -141,7 +144,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.
-func NewExporter(ctx context.Context, opts *Options) (*Exporter, error) {
+func NewExporter(ctx context.Context, opts Options) (*Exporter, error) {
 	client, err := newMetricClient(ctx, opts.ClientOptions...)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create a metric client: %v", err)
@@ -154,19 +157,14 @@ func NewExporter(ctx context.Context, opts *Options) (*Exporter, error) {
 		projDataMap: make(map[string]*projectData),
 	}
 
-	// We don't want to modify user-supplied options, so save default options directly in
-	// exporter.
-	e.getProjectID = defaultGetProjectID
-	if opts.GetProjectID != nil {
-		e.getProjectID = opts.GetProjectID
+	if e.opts.GetProjectID == nil {
+		e.opts.GetProjectID = defaultGetProjectID
 	}
-	e.onError = defaultOnError
-	if opts.OnError != nil {
-		e.onError = opts.OnError
+	if e.opts.OnError == nil {
+		e.opts.OnError = defaultOnError
 	}
-	e.makeResource = defaultMakeResource
-	if opts.MakeResource != nil {
-		e.makeResource = opts.MakeResource
+	if e.opts.MakeResource == nil {
+		e.opts.MakeResource = defaultMakeResource
 	}
 
 	return e, nil
@@ -201,12 +199,12 @@ var RowDataNotApplicableError = errors.New("row data is not applicable to the ex
 
 // exportRowData exports a single row data.
 func (e *Exporter) exportRowData(rd *RowData) {
-	projID, err := e.getProjectID(rd)
+	projID, err := e.opts.GetProjectID(rd)
 	if err != nil {
 		// We ignore non-applicable RowData.
 		if err != RowDataNotApplicableError {
 			newErr := fmt.Errorf("failed to get project ID on row data with view %s: %v", rd.View.Name, err)
-			e.onError(newErr, rd)
+			e.opts.OnError(newErr, rd)
 		}
 		return
 	}
@@ -217,7 +215,7 @@ func (e *Exporter) exportRowData(rd *RowData) {
 		go pd.uploadRowData(rd)
 	default:
 		newErr := fmt.Errorf("failed to add row data with view %s to bundle for project %s: %v", rd.View.Name, projID, err)
-		e.onError(newErr, rd)
+		e.opts.OnError(newErr, rd)
 	}
 }
 
@@ -233,6 +231,23 @@ func (e *Exporter) getProjectData(projectID string) *projectData {
 	return pd
 }
 
+func (e *Exporter) newProjectData(projectID string) *projectData {
+	pd := &projectData{
+		parent:    e,
+		projectID: projectID,
+	}
+
+	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
+	}
+	return pd
+}
+
 // Close flushes and closes the exporter. Close must be called after the exporter is unregistered
 // and no further calls to ExportView() are made. Once Close() is returned no further access to the
 // exporter is allowed in any way.
@@ -248,3 +263,42 @@ func (e *Exporter) Close() error {
 	}
 	return nil
 }
+
+// makeTS constructs a time series from a row data.
+func (e *Exporter) makeTS(rd *RowData) (*mpb.TimeSeries, error) {
+	pt := newPoint(rd.View, rd.Row, rd.Start, rd.End)
+	if pt.Value == nil {
+		return nil, fmt.Errorf("inconsistent data found in view %s", rd.View.Name)
+	}
+	resource, err := e.opts.MakeResource(rd)
+	if err != nil {
+		return nil, fmt.Errorf("failed to construct resource of view %s: %v", rd.View.Name, err)
+	}
+	ts := &mpb.TimeSeries{
+		Metric: &metricpb.Metric{
+			Type:   rd.View.Name,
+			Labels: e.makeLabels(rd.Row.Tags),
+		},
+		Resource: resource,
+		Points:   []*mpb.Point{pt},
+	}
+	return ts, nil
+}
+
+// makeLables constructs label that's ready for being uploaded to stackdriver.
+func (e *Exporter) makeLabels(tags []tag.Tag) map[string]string {
+	opts := e.opts
+	labels := make(map[string]string, len(opts.DefaultLabels)+len(tags))
+	for key, val := range opts.DefaultLabels {
+		labels[key] = val
+	}
+	// If there's overlap When combining exporter's default label and tags, values in tags win.
+	for _, tag := range tags {
+		labels[tag.Key.Name()] = tag.Value
+	}
+	// Some labels are not for exporting.
+	for _, key := range opts.UnexportedLabels {
+		delete(labels, key)
+	}
+	return labels
+}