Repository: census-ecosystem/opencensus-go-exporter-stackdriver
Pull request #213 (2019-09-13 03:42:27)

Allow sending CreateTimeSeriesReq in multiple goroutines

songy23: I'd prefer to leave the tests in this PR in `metrics_batcher_test.go` since they only tested the batcher. We can add separate tests to metrics_proto_api_test.go once #215 is submitted.
rghetia: > I'd prefer to leave the tests in this PR in `metrics_batcher_test.go` since they only tested the batcher. We can add separate tests to metrics_proto_api_test.go once #215 is submitted.

SG

Patch:
diff --git a/metrics_batcher.go b/metrics_batcher.go
index 921d967..b6b34ba 100644
--- a/metrics_batcher.go
+++ b/metrics_batcher.go
@@ -18,62 +18,85 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync"
 
 	monitoring "cloud.google.com/go/monitoring/apiv3"
 	monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
 )
 
+const minNumWorkers = 1
+
 type metricsBatcher struct {
 	projectName string
-	allReqs     []*monitoringpb.CreateTimeSeriesRequest
 	allTss      []*monitoringpb.TimeSeries
 	allErrs     []error
-	// Counts all dropped TimeSeries by this exporter.
+
+	// Counts all dropped TimeSeries by this metricsBatcher.
 	droppedTimeSeries int
+
+	workers []*worker
+	// reqsChan, respsChan and wg are shared between metricsBatcher and worker goroutines.
+	reqsChan  chan *monitoringpb.CreateTimeSeriesRequest
+	respsChan chan *response
+	wg        *sync.WaitGroup
 }
 
-func newMetricsBatcher(projectID string) *metricsBatcher {
+func newMetricsBatcher(ctx context.Context, projectID string, numWorkers int, mc *monitoring.MetricClient) *metricsBatcher {
+	if numWorkers < minNumWorkers {
+		numWorkers = minNumWorkers
+	}
+	workers := make([]*worker, 0, numWorkers)
+	reqsChan := make(chan *monitoringpb.CreateTimeSeriesRequest, 3*numWorkers)
+	respsChan := make(chan *response, numWorkers)
+	var wg sync.WaitGroup
+	wg.Add(numWorkers)
+	for i := 0; i < numWorkers; i++ {
+		w := newWorker(ctx, mc, reqsChan, respsChan, &wg)
+		workers = append(workers, w)
+		go w.start()
+	}
 	return &metricsBatcher{
 		projectName:       fmt.Sprintf("projects/%s", projectID),
 		allTss:            make([]*monitoringpb.TimeSeries, 0, maxTimeSeriesPerUpload),
 		droppedTimeSeries: 0,
+		workers:           workers,
+		wg:                &wg,
+		reqsChan:          reqsChan,
+		respsChan:         respsChan,
 	}
 }
 
-func (mb *metricsBatcher) recordDroppedTimeseries(numTimeSeries int, err error) {
+func (mb *metricsBatcher) recordDroppedTimeseries(numTimeSeries int, errs ...error) {
 	mb.droppedTimeSeries += numTimeSeries
-	mb.allErrs = append(mb.allErrs, err)
+	for _, err := range errs {
+		if err != nil {
+			mb.allErrs = append(mb.allErrs, err)
+		}
+	}
 }
 
 func (mb *metricsBatcher) addTimeSeries(ts *monitoringpb.TimeSeries) {
 	mb.allTss = append(mb.allTss, ts)
 	if len(mb.allTss) == maxTimeSeriesPerUpload {
-		mb.allReqs = append(mb.allReqs, &monitoringpb.CreateTimeSeriesRequest{
-			Name:       mb.projectName,
-			TimeSeries: mb.allTss,
-		})
+		mb.sendReqToChan()
 		mb.allTss = make([]*monitoringpb.TimeSeries, 0, maxTimeSeriesPerUpload)
 	}
 }
 
-func (mb *metricsBatcher) export(ctx context.Context, mc *monitoring.MetricClient) {
-	// Last batch, if any.
+func (mb *metricsBatcher) close(ctx context.Context) error {
+	// Send any remaining time series, must be <200
 	if len(mb.allTss) > 0 {
-		mb.allReqs = append(mb.allReqs, &monitoringpb.CreateTimeSeriesRequest{
-			Name:       mb.projectName,
-			TimeSeries: mb.allTss,
-		})
+		mb.sendReqToChan()
 	}
 
-	// Send create time series requests to Stackdriver.
-	for _, req := range mb.allReqs {
-		if err := createTimeSeries(ctx, mc, req); err != nil {
-			mb.recordDroppedTimeseries(len(req.TimeSeries), err)
-		}
+	close(mb.reqsChan)
+	mb.wg.Wait()
+	for i := 0; i < len(mb.workers); i++ {
+		resp := <-mb.respsChan
+		mb.recordDroppedTimeseries(resp.droppedTimeSeries, resp.errs...)
 	}
-}
+	close(mb.respsChan)
 
-func (mb *metricsBatcher) finalError() error {
 	numErrors := len(mb.allErrs)
 	if numErrors == 0 {
 		return nil
@@ -89,3 +112,73 @@ func (mb *metricsBatcher) finalError() error {
 	}
 	return fmt.Errorf("[%s]", strings.Join(errMsgs, "; "))
 }
+
+// sendReqToChan grabs all the timeseies in this metricsBatcher, puts them
+// to a CreateTimeSeriesRequest and sends the request to reqsChan.
+func (mb *metricsBatcher) sendReqToChan() {
+	req := &monitoringpb.CreateTimeSeriesRequest{
+		Name:       mb.projectName,
+		TimeSeries: mb.allTss,
+	}
+	mb.reqsChan <- req
+}
+
+// sendReq sends create time series requests to Stackdriver,
+// and returns the count of dropped time series and error.
+func sendReq(ctx context.Context, c *monitoring.MetricClient, req *monitoringpb.CreateTimeSeriesRequest) (int, error) {
+	if c != nil { // c==nil only happens in unit tests where we don't make real calls to Stackdriver server
+		err := createTimeSeries(ctx, c, req)
+		if err != nil {
+			return len(req.TimeSeries), err
+		}
+	}
+	return 0, nil
+}
+
+type worker struct {
+	ctx context.Context
+	mc  *monitoring.MetricClient
+
+	resp *response
+
+	respsChan chan *response
+	reqsChan  chan *monitoringpb.CreateTimeSeriesRequest
+
+	wg *sync.WaitGroup
+}
+
+func newWorker(
+	ctx context.Context,
+	mc *monitoring.MetricClient,
+	reqsChan chan *monitoringpb.CreateTimeSeriesRequest,
+	respsChan chan *response,
+	wg *sync.WaitGroup) *worker {
+	return &worker{
+		ctx:       ctx,
+		mc:        mc,
+		resp:      &response{},
+		reqsChan:  reqsChan,
+		respsChan: respsChan,
+		wg:        wg,
+	}
+}
+
+func (w *worker) start() {
+	for req := range w.reqsChan {
+		w.recordDroppedTimeseries(sendReq(w.ctx, w.mc, req))
+	}
+	w.respsChan <- w.resp
+	w.wg.Done()
+}
+
+func (w *worker) recordDroppedTimeseries(numTimeSeries int, err error) {
+	w.resp.droppedTimeSeries += numTimeSeries
+	if err != nil {
+		w.resp.errs = append(w.resp.errs, err)
+	}
+}
+
+type response struct {
+	droppedTimeSeries int
+	errs              []error
+}
diff --git a/metrics_proto.go b/metrics_proto.go
index 41f7bb8..af2ed37 100644
--- a/metrics_proto.go
+++ b/metrics_proto.go
@@ -60,7 +60,7 @@ func (se *statsExporter) PushMetricsProto(ctx context.Context, node *commonpb.No
 	// Caches the resources seen so far
 	seenResources := make(map[*resourcepb.Resource]*monitoredrespb.MonitoredResource)
 
-	mb := newMetricsBatcher(se.o.ProjectID)
+	mb := newMetricsBatcher(ctx, se.o.ProjectID, se.o.NumberOfWorkers, se.c)
 	for _, metric := range metrics {
 		if len(metric.GetTimeseries()) == 0 {
 			// No TimeSeries to export, skip this metric.
@@ -85,8 +85,7 @@ func (se *statsExporter) PushMetricsProto(ctx context.Context, node *commonpb.No
 		}
 	}
 
-	mb.export(ctx, se.c)
-	return mb.droppedTimeSeries, mb.finalError()
+	return mb.droppedTimeSeries, mb.close(ctx)
 }
 
 func (se *statsExporter) convertSummaryMetrics(summary *metricspb.Metric) []*metricspb.Metric {
diff --git a/stackdriver.go b/stackdriver.go
index c219e1b..5e36c5f 100644
--- a/stackdriver.go
+++ b/stackdriver.go
@@ -250,6 +250,11 @@ type Options struct {
 	// ReportingInterval sets the interval between reporting metrics.
 	// If it is set to zero then default value is used.
 	ReportingInterval time.Duration
+
+	// NumberOfWorkers sets the number of go rountines that send requests
+	// to Stackdriver Monitoring. This is only used for Proto metrics export
+	// for now. The minimum number of workers is 1.
+	NumberOfWorkers int
 }
 
 const defaultTimeout = 5 * time.Second


Test patch:
diff --git a/metrics_batcher_test.go b/metrics_batcher_test.go
new file mode 100644
index 0000000..8c72524
--- /dev/null
+++ b/metrics_batcher_test.go
@@ -0,0 +1,111 @@
+// 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 stackdriver
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	monitoring "cloud.google.com/go/monitoring/apiv3"
+	"google.golang.org/api/option"
+	googlemetricpb "google.golang.org/genproto/googleapis/api/metric"
+	monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
+	"google.golang.org/grpc"
+)
+
+func TestWorkers(t *testing.T) {
+	server, addr, doneFn := createFakeServer(t)
+	defer doneFn()
+	ctx := context.Background()
+
+	c1, err := makeClient(addr)
+	if err != nil {
+		t.Fatalf("Failed to create metric client %v", err)
+	}
+	m1 := newMetricsBatcher(ctx, "test", 1, c1) // batcher with 1 worker
+
+	c2, err := makeClient(addr)
+	if err != nil {
+		t.Fatalf("Failed to create metric client %v", err)
+	}
+	m2 := newMetricsBatcher(ctx, "test", 2, c2) // batcher with 2 workers
+
+	tss := make([]*monitoringpb.TimeSeries, 0, 500) // make 500 time series, should be split to 3 reqs
+	for i := 0; i < 500; i++ {
+		tss = append(tss, makeTs(i))
+	}
+
+	for _, ts := range tss {
+		m1.addTimeSeries(ts)
+	}
+	if err := m1.close(ctx); err != nil {
+		t.Fatalf("Want no error, got %v", err)
+	}
+	reqs1 := server.stackdriverTimeSeries
+	server.resetStackdriverTimeSeries()
+	server.resetStackdriverMetricDescriptors()
+
+	for _, ts := range tss {
+		m2.addTimeSeries(ts)
+	}
+	if err := m2.close(ctx); err != nil {
+		t.Fatalf("Want no error, got %v", err)
+	}
+	reqs2 := server.stackdriverTimeSeries
+
+	if len(reqs1) != 3 {
+		t.Fatalf("Want 3 CreateTimeSeriesReqs, got %v", len(reqs1))
+	}
+	if len(reqs2) != 3 {
+		t.Fatalf("Want 3 CreateTimeSeriesReqs, got %v", len(reqs2))
+	}
+	if m1.droppedTimeSeries != m2.droppedTimeSeries {
+		t.Fatalf("Dropped time series counts don't match, FromOneWorker: %v, FromTwoWorkers: %v", m1.droppedTimeSeries, m2.droppedTimeSeries)
+	}
+	if diff := cmpTSReqs(reqs1, reqs2); diff != "" {
+		t.Fatalf("CreateTimeSeriesRequests don't match -FromOneWorker +FromTwoWorkers: %s", diff)
+	}
+}
+
+func makeClient(addr string) (*monitoring.MetricClient, error) {
+	conn, err := grpc.Dial(addr, grpc.WithInsecure())
+	if err != nil {
+		return nil, err
+	}
+	return monitoring.NewMetricClient(context.Background(), option.WithGRPCConn(conn))
+}
+
+func makeTs(i int) *monitoringpb.TimeSeries {
+	return &monitoringpb.TimeSeries{
+		Metric: &googlemetricpb.Metric{
+			Type: fmt.Sprintf("custom.googleapis.com/opencensus/test/metric/%v", i),
+			Labels: map[string]string{
+				"key": fmt.Sprintf("test_%v", i),
+			},
+		},
+		MetricKind: googlemetricpb.MetricDescriptor_CUMULATIVE,
+		ValueType:  googlemetricpb.MetricDescriptor_INT64,
+		Points: []*monitoringpb.Point{
+			{
+				Value: &monitoringpb.TypedValue{
+					Value: &monitoringpb.TypedValue_Int64Value{
+						Int64Value: int64(i),
+					},
+				},
+			},
+		},
+	}
+}
diff --git a/metrics_proto_test.go b/metrics_proto_test.go
index 3da3ce2..fd9e031 100644
--- a/metrics_proto_test.go
+++ b/metrics_proto_test.go
@@ -1083,7 +1083,7 @@ func makePercentileValue(val, percentile float64) *metricspb.SummaryValue_Snapsh
 }
 
 func protoMetricToTimeSeries(ctx context.Context, se *statsExporter, mappedRsc *monitoredrespb.MonitoredResource, metric *metricspb.Metric) ([]*monitoringpb.TimeSeries, error) {
-	mb := newMetricsBatcher(se.o.ProjectID)
+	mb := newMetricsBatcher(ctx, se.o.ProjectID, se.o.NumberOfWorkers, se.c)
 	se.protoMetricToTimeSeries(ctx, mappedRsc, metric, mb)
-	return mb.allTss, mb.finalError()
+	return mb.allTss, mb.close(ctx)
 }