Batch requests

Per https://cloud.google.com/monitoring/quotas we can include up to 200
TimeSeries per RPC. Batch them to limit the number of RPCs.

Author: tam7t

stackdriver.go

@@ -234,10 +234,6 @@ func (s *Sink) deep() (time.Time, map[string]*gauge, map[string]*counter, map[st
 func (s *Sink) report(ctx context.Context) {
 	end, rGauges, rCounters, rHistograms := s.deep()
 
-	if s.client == nil {
-		return
-	}
-
 	// https://cloud.google.com/monitoring/api/resources
 	resource := &monitoredrespb.MonitoredResource{
 		Type: "generic_task",
@@ -250,72 +246,56 @@ func (s *Sink) report(ctx context.Context) {
 		},
 	}
 
+	ts := []*monitoringpb.TimeSeries{}
+
 	for _, v := range rCounters {
-		err := s.client.CreateTimeSeries(ctx, &monitoringpb.CreateTimeSeriesRequest{
-			Name: fmt.Sprintf("projects/%s", s.taskInfo.ProjectID),
-			TimeSeries: []*monitoringpb.TimeSeries{
-				{
-					Metric: &metricpb.Metric{
-						Type:   path.Join("custom.googleapis.com", "go-metrics", v.name.name),
-						Labels: v.name.labelMap(),
+		ts = append(ts, &monitoringpb.TimeSeries{
+			Metric: &metricpb.Metric{
+				Type:   path.Join("custom.googleapis.com", "go-metrics", v.name.name),
+				Labels: v.name.labelMap(),
+			},
+			MetricKind: metric.MetricDescriptor_GAUGE,
+			Resource:   resource,
+			Points: []*monitoringpb.Point{
+				&monitoringpb.Point{
+					Interval: &monitoringpb.TimeInterval{
+						EndTime: &googlepb.Timestamp{
+							Seconds: end.Unix(),
+						},
 					},
-					MetricKind: metric.MetricDescriptor_GAUGE,
-					Resource:   resource,
-					Points: []*monitoringpb.Point{
-						&monitoringpb.Point{
-							Interval: &monitoringpb.TimeInterval{
-								EndTime: &googlepb.Timestamp{
-									Seconds: end.Unix(),
-								},
-							},
-							Value: &monitoringpb.TypedValue{
-								Value: &monitoringpb.TypedValue_DoubleValue{
-									DoubleValue: v.value,
-								},
-							},
+					Value: &monitoringpb.TypedValue{
+						Value: &monitoringpb.TypedValue_DoubleValue{
+							DoubleValue: v.value,
 						},
 					},
 				},
 			},
 		})
-
-		if err != nil {
-			log.Printf("Failed to write time series data: %v", err)
-		}
 	}
 
 	for _, v := range rGauges {
-		err := s.client.CreateTimeSeries(ctx, &monitoringpb.CreateTimeSeriesRequest{
-			Name: fmt.Sprintf("projects/%s", s.taskInfo.ProjectID),
-			TimeSeries: []*monitoringpb.TimeSeries{
-				{
-					Metric: &metricpb.Metric{
-						Type:   path.Join("custom.googleapis.com", "go-metrics", v.name.name),
-						Labels: v.name.labelMap(),
+		ts = append(ts, &monitoringpb.TimeSeries{
+			Metric: &metricpb.Metric{
+				Type:   path.Join("custom.googleapis.com", "go-metrics", v.name.name),
+				Labels: v.name.labelMap(),
+			},
+			MetricKind: metric.MetricDescriptor_GAUGE,
+			Resource:   resource,
+			Points: []*monitoringpb.Point{
+				&monitoringpb.Point{
+					Interval: &monitoringpb.TimeInterval{
+						EndTime: &googlepb.Timestamp{
+							Seconds: end.Unix(),
+						},
 					},
-					MetricKind: metric.MetricDescriptor_GAUGE,
-					Resource:   resource,
-					Points: []*monitoringpb.Point{
-						&monitoringpb.Point{
-							Interval: &monitoringpb.TimeInterval{
-								EndTime: &googlepb.Timestamp{
-									Seconds: end.Unix(),
-								},
-							},
-							Value: &monitoringpb.TypedValue{
-								Value: &monitoringpb.TypedValue_DoubleValue{
-									DoubleValue: float64(v.value),
-								},
-							},
+					Value: &monitoringpb.TypedValue{
+						Value: &monitoringpb.TypedValue_DoubleValue{
+							DoubleValue: float64(v.value),
 						},
 					},
 				},
 			},
 		})
-
-		if err != nil {
-			log.Printf("Failed to write time series data: %v", err)
-		}
 	}
 
 	for _, v := range rHistograms {
@@ -325,46 +305,58 @@ func (s *Sink) report(ctx context.Context) {
 			count += int64(i)
 		}
 
-		err := s.client.CreateTimeSeries(ctx, &monitoringpb.CreateTimeSeriesRequest{
-			Name: fmt.Sprintf("projects/%s", s.taskInfo.ProjectID),
-			TimeSeries: []*monitoringpb.TimeSeries{
-				{
-					Metric: &metricpb.Metric{
-						Type:   path.Join("custom.googleapis.com", "go-metrics", v.name.name),
-						Labels: v.name.labelMap(),
+		ts = append(ts, &monitoringpb.TimeSeries{
+			Metric: &metricpb.Metric{
+				Type:   path.Join("custom.googleapis.com", "go-metrics", v.name.name),
+				Labels: v.name.labelMap(),
+			},
+			MetricKind: metric.MetricDescriptor_CUMULATIVE,
+			Resource:   resource,
+			Points: []*monitoringpb.Point{
+				&monitoringpb.Point{
+					Interval: &monitoringpb.TimeInterval{
+						StartTime: &googlepb.Timestamp{
+							Seconds: s.firstTime.Unix(),
+						},
+						EndTime: &googlepb.Timestamp{
+							Seconds: end.Unix(),
+						},
 					},
-					MetricKind: metric.MetricDescriptor_CUMULATIVE,
-					Resource:   resource,
-					Points: []*monitoringpb.Point{
-						&monitoringpb.Point{
-							Interval: &monitoringpb.TimeInterval{
-								StartTime: &googlepb.Timestamp{
-									Seconds: s.firstTime.Unix(),
-								},
-								EndTime: &googlepb.Timestamp{
-									Seconds: end.Unix(),
-								},
-							},
-							Value: &monitoringpb.TypedValue{
-								Value: &monitoringpb.TypedValue_DistributionValue{
-									DistributionValue: &distributionpb.Distribution{
-										BucketOptions: &distributionpb.Distribution_BucketOptions{
-											Options: &distributionpb.Distribution_BucketOptions_ExplicitBuckets{
-												ExplicitBuckets: &distributionpb.Distribution_BucketOptions_Explicit{
-													Bounds: v.buckets,
-												},
-											},
+					Value: &monitoringpb.TypedValue{
+						Value: &monitoringpb.TypedValue_DistributionValue{
+							DistributionValue: &distributionpb.Distribution{
+								BucketOptions: &distributionpb.Distribution_BucketOptions{
+									Options: &distributionpb.Distribution_BucketOptions_ExplicitBuckets{
+										ExplicitBuckets: &distributionpb.Distribution_BucketOptions_Explicit{
+											Bounds: v.buckets,
 										},
-										BucketCounts: v.counts,
-										Count:        count,
 									},
 								},
+								BucketCounts: v.counts,
+								Count:        count,
 							},
 						},
 					},
 				},
 			},
 		})
+	}
+
+	if s.client == nil {
+		return
+	}
+
+	for i := 0; i < len(ts); i += 200 {
+		end := i + 200
+
+		if end > len(ts) {
+			end = len(ts)
+		}
+
+		err := s.client.CreateTimeSeries(ctx, &monitoringpb.CreateTimeSeriesRequest{
+			Name:       fmt.Sprintf("projects/%s", s.taskInfo.ProjectID),
+			TimeSeries: ts[i:end],
+		})
 
 		if err != nil {
 			log.Printf("Failed to write time series data: %v", err)

stackdriver_test.go

@@ -224,6 +224,24 @@ func TestSample(t *testing.T) {
 				}
 			},
 		},
+		{
+			name: "batching",
+			collect: func() {
+				for i := 0; i < 300; i++ {
+					ss.SetGauge([]string{"foo", fmt.Sprintf("%d", i)}, 50.0)
+				}
+			},
+			createFn: func(t *testing.T) func(context.Context, *monitoringpb.CreateTimeSeriesRequest) (*emptypb.Empty, error) {
+				return func(_ context.Context, req *monitoringpb.CreateTimeSeriesRequest) (*emptypb.Empty, error) {
+					// 300 TimeSeries were created, we expect 2 RPCs (first with 200 and second with 100 TS)
+					if len(req.TimeSeries) == 200 || len(req.TimeSeries) == 100 {
+						return &emptypb.Empty{}, nil
+					}
+					t.Errorf("unexpected CreateTimeSeriesRequest\ngot(# of TimeSeries): %v", len(req.TimeSeries))
+					return nil, errors.New("unexpected CreateTimeSeriesRequest")
+				}
+			},
+		},
 	}
 
 	for _, tc := range tests {