perf: limit goroutine usage while unmarshaling prometheus and influx points/values (#909)

Signed-off-by: Chris Randles <randles.chris@gmail.com>

Author: crandles

pkg/backends/influxdb/influxql/unmarshal.go

@@ -21,14 +21,15 @@ import (
 	"encoding/json"
 	"errors"
 	"io"
+	"runtime"
 	"sort"
-	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/trickstercache/trickster/v2/pkg/timeseries"
 	"github.com/trickstercache/trickster/v2/pkg/timeseries/dataset"
 	"github.com/trickstercache/trickster/v2/pkg/timeseries/epoch"
+	"golang.org/x/sync/errgroup"
 )
 
 // Unmarshal performs a standard unmarshal of the bytes into the InfluxDB Wire Format Document,
@@ -97,17 +98,18 @@ func UnmarshalTimeseriesReader(reader io.Reader, trq *timeseries.TimeRangeQuery)
 			sh.CalculateSize()
 			pts := make(dataset.Points, len(wfd.Results[i].SeriesList[j].Values))
 			var sz int64
-			var wg sync.WaitGroup
+			var eg errgroup.Group
+			eg.SetLimit(runtime.GOMAXPROCS(0))
 			errs := make([]error, len(wfd.Results[i].SeriesList[j].Values))
 			for vi, v := range wfd.Results[i].SeriesList[j].Values {
-				wg.Go(func() {
+				eg.Go(func() error {
 					pt, cols, err := pointFromValues(v, sh.TimestampField.OutputPosition)
 					if err != nil {
 						errs[vi] = err
-						return
+						return err
 					}
 					if pt.Epoch == 0 {
-						return
+						return nil
 					}
 					if vi == 0 {
 						for x := range cols {
@@ -117,9 +119,10 @@ func UnmarshalTimeseriesReader(reader io.Reader, trq *timeseries.TimeRangeQuery)
 					pts[vi] = pt
 					atomic.AddInt64(&sz, int64(pt.Size))
 					wfd.Results[i].SeriesList[j].Values[vi] = nil
+					return nil
 				})
 			}
-			wg.Wait()
+			eg.Wait()
 			if err := errors.Join(errs...); err != nil {
 				return nil, err
 			}

pkg/backends/prometheus/model/timeseries.go

@@ -21,16 +21,17 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"runtime"
 	"sort"
 	"strconv"
-	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/trickstercache/trickster/v2/pkg/errors"
 	"github.com/trickstercache/trickster/v2/pkg/timeseries"
 	"github.com/trickstercache/trickster/v2/pkg/timeseries/dataset"
 	"github.com/trickstercache/trickster/v2/pkg/timeseries/epoch"
+	"golang.org/x/sync/errgroup"
 )
 
 // WFMatrixDocument is the Wire Format Document for prometheus range / timeseries
@@ -239,17 +240,19 @@ func populateSeries(ds *dataset.DataSet, result []*WFResult,
 		var ps int64 = 16
 		if !isVector && l > 0 {
 			pts = make(dataset.Points, l)
-			var wg sync.WaitGroup
+			var eg errgroup.Group
+			eg.SetLimit(runtime.GOMAXPROCS(0))
 			for i, v := range pr.Values {
-				wg.Go(func() {
+				eg.Go(func() error {
 					pt, _ := pointFromValues(v)
 					if pt.Epoch > 0 {
 						atomic.AddInt64(&ps, int64(pt.Size))
 						pts[i] = pt
 					}
+					return nil
 				})
 			}
-			wg.Wait()
+			eg.Wait()
 		} else if isVector && len(pr.Value) == 2 {
 			pts = make(dataset.Points, 1)
 			pt, _ := pointFromValues(pr.Value)