Repository: census-ecosystem/opencensus-go-exporter-stackdriver
Pull request #222 (2019-09-17 00:52:28)

Apply timeout to each worker in metrics_proto export

codecov-io: # [Codecov](https://codecov.io/gh/census-ecosystem/opencensus-go-exporter-stackdriver/pull/222?src=pr&el=h1) Report
> Merging [#222](https://codecov.io/gh/census-ecosystem/opencensus-go-exporter-stackdriver/pull/222?src=pr&el=desc) into [master](https://codecov.io/gh/census-ecosystem/opencensus-go-exporter-stackdriver/commit/633d9eaf679a7cf6fc85b331ff6dcd8d80979c2a?src=pr&el=desc) will **increase** coverage by `0.03%`.
> The diff coverage is `100%`.

[![Impacted file tree graph](https://codecov.io/gh/census-ecosystem/opencensus-go-exporter-stackdriver/pull/222/graphs/tree.svg?width=650&token=os6w8WyKNy&height=150&src=pr)](https://codecov.io/gh/census-ecosystem/opencensus-go-exporter-stackdriver/pull/222?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #222      +/-   ##
==========================================
+ Coverage   72.75%   72.78%   +0.03%     
==========================================
  Files          15       15              
  Lines        1593     1595       +2     
==========================================
+ Hits         1159     1161       +2     
  Misses        355      355              
  Partials       79       79
```


| [Impacted Files](https://codecov.io/gh/census-ecosystem/opencensus-go-exporter-stackdriver/pull/222?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [metrics\_proto.go](https://codecov.io/gh/census-ecosystem/opencensus-go-exporter-stackdriver/pull/222/diff?src=pr&el=tree#diff-bWV0cmljc19wcm90by5nbw==) | `87.57% <100%> (-0.08%)` | :arrow_down: |
| [metrics\_batcher.go](https://codecov.io/gh/census-ecosystem/opencensus-go-exporter-stackdriver/pull/222/diff?src=pr&el=tree#diff-bWV0cmljc19iYXRjaGVyLmdv) | `84.33% <100%> (+0.79%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/census-ecosystem/opencensus-go-exporter-stackdriver/pull/222?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/census-ecosystem/opencensus-go-exporter-stackdriver/pull/222?src=pr&el=footer). Last update [633d9ea...7425a93](https://codecov.io/gh/census-ecosystem/opencensus-go-exporter-stackdriver/pull/222?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/metrics_batcher.go b/metrics_batcher.go
index 3468183..ccd6ee4 100644
--- a/metrics_batcher.go
+++ b/metrics_batcher.go
@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"time"
 
 	monitoring "cloud.google.com/go/monitoring/apiv3"
 	monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
@@ -44,7 +45,7 @@ type metricsBatcher struct {
 	wg        *sync.WaitGroup
 }
 
-func newMetricsBatcher(ctx context.Context, projectID string, numWorkers int, mc *monitoring.MetricClient) *metricsBatcher {
+func newMetricsBatcher(ctx context.Context, projectID string, numWorkers int, mc *monitoring.MetricClient, timeout time.Duration) *metricsBatcher {
 	if numWorkers < minNumWorkers {
 		numWorkers = minNumWorkers
 	}
@@ -58,7 +59,7 @@ func newMetricsBatcher(ctx context.Context, projectID string, numWorkers int, mc
 	var wg sync.WaitGroup
 	wg.Add(numWorkers)
 	for i := 0; i < numWorkers; i++ {
-		w := newWorker(ctx, mc, reqsChan, respsChan, &wg)
+		w := newWorker(ctx, mc, reqsChan, respsChan, &wg, timeout)
 		workers = append(workers, w)
 		go w.start()
 	}
@@ -143,8 +144,9 @@ func sendReq(ctx context.Context, c *monitoring.MetricClient, req *monitoringpb.
 }
 
 type worker struct {
-	ctx context.Context
-	mc  *monitoring.MetricClient
+	ctx     context.Context
+	timeout time.Duration
+	mc      *monitoring.MetricClient
 
 	resp *response
 
@@ -159,7 +161,8 @@ func newWorker(
 	mc *monitoring.MetricClient,
 	reqsChan chan *monitoringpb.CreateTimeSeriesRequest,
 	respsChan chan *response,
-	wg *sync.WaitGroup) *worker {
+	wg *sync.WaitGroup,
+	timeout time.Duration) *worker {
 	return &worker{
 		ctx:       ctx,
 		mc:        mc,
@@ -172,12 +175,19 @@ func newWorker(
 
 func (w *worker) start() {
 	for req := range w.reqsChan {
-		w.recordDroppedTimeseries(sendReq(w.ctx, w.mc, req))
+		w.sendReqWithTimeout(req)
 	}
 	w.respsChan <- w.resp
 	w.wg.Done()
 }
 
+func (w *worker) sendReqWithTimeout(req *monitoringpb.CreateTimeSeriesRequest) {
+	ctx, cancel := newContextWithTimeout(w.ctx, w.timeout)
+	defer cancel()
+
+	w.recordDroppedTimeseries(sendReq(ctx, w.mc, req))
+}
+
 func (w *worker) recordDroppedTimeseries(numTimeSeries int, err error) {
 	w.resp.droppedTimeSeries += numTimeSeries
 	if err != nil {
diff --git a/metrics_proto.go b/metrics_proto.go
index af2ed37..12da732 100644
--- a/metrics_proto.go
+++ b/metrics_proto.go
@@ -54,13 +54,10 @@ func (se *statsExporter) PushMetricsProto(ctx context.Context, node *commonpb.No
 		return 0, errNilMetricOrMetricDescriptor
 	}
 
-	ctx, cancel := newContextWithTimeout(ctx, se.o.Timeout)
-	defer cancel()
-
 	// Caches the resources seen so far
 	seenResources := make(map[*resourcepb.Resource]*monitoredrespb.MonitoredResource)
 
-	mb := newMetricsBatcher(ctx, se.o.ProjectID, se.o.NumberOfWorkers, se.c)
+	mb := newMetricsBatcher(ctx, se.o.ProjectID, se.o.NumberOfWorkers, se.c, se.o.Timeout)
 	for _, metric := range metrics {
 		if len(metric.GetTimeseries()) == 0 {
 			// No TimeSeries to export, skip this metric.
@@ -70,14 +67,14 @@ func (se *statsExporter) PushMetricsProto(ctx context.Context, node *commonpb.No
 		if metric.GetMetricDescriptor().GetType() == metricspb.MetricDescriptor_SUMMARY {
 			summaryMtcs := se.convertSummaryMetrics(metric)
 			for _, summaryMtc := range summaryMtcs {
-				if err := se.protoCreateMetricDescriptor(ctx, summaryMtc); err != nil {
+				if err := se.createMetricDescriptorWithTimeout(ctx, summaryMtc); err != nil {
 					mb.recordDroppedTimeseries(len(summaryMtc.GetTimeseries()), err)
 					continue
 				}
 				se.protoMetricToTimeSeries(ctx, mappedRsc, summaryMtc, mb)
 			}
 		} else {
-			if err := se.protoCreateMetricDescriptor(ctx, metric); err != nil {
+			if err := se.createMetricDescriptorWithTimeout(ctx, metric); err != nil {
 				mb.recordDroppedTimeseries(len(metric.GetTimeseries()), err)
 				continue
 			}
@@ -298,6 +295,13 @@ func labelsPerTimeSeries(defaults map[string]labelValue, labelKeys []string, lab
 	return labels, nil
 }
 
+func (se *statsExporter) createMetricDescriptorWithTimeout(ctx context.Context, metric *metricspb.Metric) error {
+	ctx, cancel := newContextWithTimeout(ctx, se.o.Timeout)
+	defer cancel()
+
+	return se.protoCreateMetricDescriptor(ctx, metric)
+}
+
 // createMetricDescriptor creates a metric descriptor from the OpenCensus proto metric
 // and then creates it remotely using Stackdriver's API.
 func (se *statsExporter) protoCreateMetricDescriptor(ctx context.Context, metric *metricspb.Metric) error {


Test patch:
diff --git a/metrics_batcher_test.go b/metrics_batcher_test.go
index 8c72524..b7cc4cf 100644
--- a/metrics_batcher_test.go
+++ b/metrics_batcher_test.go
@@ -35,13 +35,13 @@ func TestWorkers(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Failed to create metric client %v", err)
 	}
-	m1 := newMetricsBatcher(ctx, "test", 1, c1) // batcher with 1 worker
+	m1 := newMetricsBatcher(ctx, "test", 1, c1, defaultTimeout) // 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
+	m2 := newMetricsBatcher(ctx, "test", 2, c2, defaultTimeout) // 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++ {
diff --git a/metrics_proto_test.go b/metrics_proto_test.go
index fd9e031..dc0ef06 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(ctx, se.o.ProjectID, se.o.NumberOfWorkers, se.c)
+	mb := newMetricsBatcher(ctx, se.o.ProjectID, se.o.NumberOfWorkers, se.c, defaultTimeout)
 	se.protoMetricToTimeSeries(ctx, mappedRsc, metric, mb)
 	return mb.allTss, mb.close(ctx)
 }