review round 2 answer 1

Author: lychung83

monitoring/exporter/stackdriver/stackdriver.go

@@ -67,9 +67,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 +90,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
@@ -221,6 +226,15 @@ func (e *Exporter) exportRowData(rd *RowData) {
 	}
 }
 
+// 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
+	// project level
+	bndler *bundler.Bundler
+}
+
 func (e *Exporter) getProjectData(projectID string) *projectData {
 	e.mu.Lock()
 	defer e.mu.Unlock()
@@ -233,6 +247,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.

monitoring/exporter/stackdriver/upload.go

@@ -18,123 +18,69 @@ 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
+// This file contains features used for uploading row data to stackdriver when bundler makes an
+// upload request by calling projectData.uploadRowData().
 
-// 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
-	// project level
-	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
-}
+// 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
 
 // 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...)
-		}
-	}
-}
-
-// 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{}
+	// 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
 
-	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.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
 		}
 	}
+	// Upload any remaining time series.
+	if len(uploadTs) != 0 {
+		pd.uploadTimeSeries(uploadTs, uploadRds)
+	}
+}
 
-	// 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,
-		}
+// 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)
 	}
-	return req, reqRds, remainingRds
+	resource, err := e.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.
@@ -154,3 +100,19 @@ func (e *Exporter) makeLabels(tags []tag.Tag) map[string]string {
 	}
 	return labels
 }
+
+// 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,
+	}
+	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, rds...)
+	}
+}