Introducing the MaxTimeSeriesInBatch config option

cloudapi/config.go

@@ -31,9 +31,8 @@ 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.
-	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"`
@@ -150,6 +149,9 @@ type Config struct {
 
 	// Connection or request times with how many IQRs above Q3 to consier as non-aggregatable outliers.
 	AggregationOutlierIqrCoefUpper null.Float `json:"aggregationOutlierIqrCoefUpper" envconfig:"K6_CLOUD_AGGREGATION_OUTLIER_IQR_COEF_UPPER"`
+
+	// Deprecated: 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"`
 }
 
 // NewConfig creates a new Config instance with default values for some fields.
@@ -166,6 +168,7 @@ func NewConfig() Config {
 		TracesPushInterval: types.NewNullDuration(1*time.Second, false),
 
 		MaxMetricSamplesPerPackage: null.NewInt(100000, false),
+		MaxTimeSeriesInBatch:       null.NewInt(10000, 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
 	}

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),

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
 		}
 

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)

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()

output/cloud/output.go

@@ -118,6 +118,11 @@ func newOutput(params output.Params) (*Output, error) {
 			conf.MaxMetricSamplesPerPackage.Int64)
 	}
 
+	if conf.MaxTimeSeriesInBatch.Int64 < 1 {
+		return nil, fmt.Errorf("max allowed number of time series in a single batch must be a positive number but is %d",
+			conf.MaxTimeSeriesInBatch.Int64)
+	}
+
 	apiClient := cloudapi.NewClient(
 		logger, conf.Token.String, conf.Host.String, consts.Version, conf.Timeout.TimeDuration())