Introducing the MaxTimeSeriesInBatch config option

Moving cloud output v2 from MaxMetricSamplesPerPackage to MaxTimeSeriesInBatch, to have a flexibility and better reflect the actual usage of the option.
grafana · Jun 30, 2023 · ecbe215 · ecbe215
1 parent 340b0da
commit ecbe215
Show file tree

Hide file tree

Showing 5 changed files with 15 additions and 10 deletions.
diff --git a/cloudapi/config.go b/cloudapi/config.go
@@ -31,10 +31,12 @@ type Config struct {
 	StopOnError    null.Bool   `json:"stopOnError" envconfig:"K6_CLOUD_STOP_ON_ERROR"`
 	APIVersion     null.Int    `json:"apiVersion" envconfig:"K6_CLOUD_API_VERSION"`
 
-	// TODO: rename the config field to align to the new logic by time series
-	// when the migration from the version 1 is completed.
+	// TODO: remove this when migration from the cloud output v1 will be completed
 	MaxMetricSamplesPerPackage null.Int `json:"maxMetricSamplesPerPackage" envconfig:"K6_CLOUD_MAX_METRIC_SAMPLES_PER_PACKAGE"`
 
+	// Defines the max allowed number of time series in a single batch.
+	MaxTimeSeriesInBatch null.Int `json:"maxTimeSeriesInBatch" envconfig:"K6_CLOUD_MAX_TIME_SERIES_IN_BATCH"`
+
 	// The time interval between periodic API calls for sending samples to the cloud ingest service.
 	MetricPushInterval types.NullDuration `json:"metricPushInterval" envconfig:"K6_CLOUD_METRIC_PUSH_INTERVAL"`
 
@@ -166,6 +168,7 @@ func NewConfig() Config {
 		TracesPushInterval: types.NewNullDuration(1*time.Second, false),
 
 		MaxMetricSamplesPerPackage: null.NewInt(100000, false),
+		MaxTimeSeriesInBatch:       null.NewInt(100000, false),
 		Timeout:                    types.NewNullDuration(1*time.Minute, false),
 		APIVersion:                 null.NewInt(1, false),
 		// Aggregation is disabled by default, since AggregationPeriod has no default value
@@ -225,6 +228,9 @@ func (c Config) Apply(cfg Config) Config {
 	if cfg.MaxMetricSamplesPerPackage.Valid {
 		c.MaxMetricSamplesPerPackage = cfg.MaxMetricSamplesPerPackage
 	}
+	if cfg.MaxTimeSeriesInBatch.Valid {
+		c.MaxTimeSeriesInBatch = cfg.MaxTimeSeriesInBatch
+	}
 	if cfg.MetricPushInterval.Valid {
 		c.MetricPushInterval = cfg.MetricPushInterval
 	}

diff --git a/cloudapi/config_test.go b/cloudapi/config_test.go
@@ -36,6 +36,7 @@ func TestConfigApply(t *testing.T) {
 		StopOnError:                     null.NewBool(true, true),
 		APIVersion:                      null.NewInt(2, true),
 		MaxMetricSamplesPerPackage:      null.NewInt(2, true),
+		MaxTimeSeriesInBatch:            null.NewInt(3, true),
 		MetricPushInterval:              types.NewNullDuration(1*time.Second, true),
 		MetricPushConcurrency:           null.NewInt(3, true),
 		TracesEnabled:                   null.NewBool(true, true),

diff --git a/output/cloud/expv2/flush.go b/output/cloud/expv2/flush.go
@@ -17,7 +17,7 @@ type metricsFlusher struct {
 	bq                         *bucketQ
 	client                     pusher
 	aggregationPeriodInSeconds uint32
-	maxSeriesInSingleBatch     int
+	maxSeriesInBatch           int
 }
 
 // flush flushes the queued buckets sending them to the remote Cloud service.
@@ -42,7 +42,7 @@ func (f *metricsFlusher) flush(_ context.Context) error {
 	msb := newMetricSetBuilder(f.referenceID, f.aggregationPeriodInSeconds)
 	for i := 0; i < len(buckets); i++ {
 		msb.addTimeBucket(buckets[i])
-		if len(msb.seriesIndex) < f.maxSeriesInSingleBatch {
+		if len(msb.seriesIndex) < f.maxSeriesInBatch {
 			continue
 		}
 

diff --git a/output/cloud/expv2/flush_test.go b/output/cloud/expv2/flush_test.go
@@ -132,9 +132,9 @@ func TestMetricsFlusherFlushChunk(t *testing.T) {
 		bq := &bucketQ{}
 		pm := &pusherMock{}
 		mf := metricsFlusher{
-			bq:                     bq,
-			client:                 pm,
-			maxSeriesInSingleBatch: 3,
+			bq:               bq,
+			client:           pm,
+			maxSeriesInBatch: 3,
 		}
 
 		bq.buckets = make([]timeBucket, 0, tc.series)

diff --git a/output/cloud/expv2/output.go b/output/cloud/expv2/output.go
@@ -107,9 +107,7 @@ func (o *Output) Start() error {
 		bq:                         &o.collector.bq,
 		client:                     mc,
 		aggregationPeriodInSeconds: uint32(o.config.AggregationPeriod.TimeDuration().Seconds()),
-		// TODO: rename the config field to align to the new logic by time series
-		// when the migration from the version 1 is completed.
-		maxSeriesInSingleBatch: int(o.config.MaxMetricSamplesPerPackage.Int64),
+		maxSeriesInBatch:           int(o.config.MaxTimeSeriesInBatch.Int64),
 	}
 
 	o.runFlushWorkers()