perf(bigtable): parallelize the exportTimeSeries function (#13004)

bhshkh · web-flow · commit bafd691de181 · 2025-10-09T17:56:41.000-07:00
Fix Bigtable test timeout in TestBatchingExport The `TestBatchingExport/1199_metrics` test was timing out after 45 minutes. This was caused by the `exportTimeSeries` function sending metric batches sequentially. With 1199 metrics, this resulted in 6 separate, synchronous gRPC calls, which caused a hang. This change parallelizes the `exportTimeSeries` function to send metric batches concurrently using goroutines. This resolves the timeout and improves performance. Fixes: #12920
diff --git a/bigtable/metrics_monitoring_exporter.go b/bigtable/metrics_monitoring_exporter.go
@@ -142,21 +142,40 @@ func (me *monitoringExporter) exportTimeSeries(ctx context.Context, rm *otelmetr
 
 	name := fmt.Sprintf("projects/%s", me.projectID)
 
-	errs := []error{err}
+	var wg sync.WaitGroup
+	errs := make(chan error, len(tss)/sendBatchSize+1)
+	if err != nil {
+		errs <- err
+	}
+
 	for i := 0; i < len(tss); i += sendBatchSize {
 		j := i + sendBatchSize
-		if j >= len(tss) {
+		if j > len(tss) {
 			j = len(tss)
 		}
 
-		req := &monitoringpb.CreateTimeSeriesRequest{
-			Name:       name,
-			TimeSeries: tss[i:j],
-		}
-		errs = append(errs, me.client.CreateServiceTimeSeries(ctx, req))
+		wg.Add(1)
+		go func(i, j int) {
+			defer wg.Done()
+			req := &monitoringpb.CreateTimeSeriesRequest{
+				Name:       name,
+				TimeSeries: tss[i:j],
+			}
+			if err := me.client.CreateServiceTimeSeries(ctx, req); err != nil {
+				errs <- err
+			}
+		}(i, j)
+	}
+
+	wg.Wait()
+	close(errs)
+
+	var allErrors []error
+	for err := range errs {
+		allErrors = append(allErrors, err)
 	}
 
-	return errors.Join(errs...)
+	return errors.Join(allErrors...)
 }
 
 // recordToMetricAndMonitoredResourcePbs converts data from records to Metric and Monitored resource proto type for Cloud Monitoring.
