cloud: switch to cloud output v2 by default but backwards-compatible

pkg/cloud/aggregation.go

@@ -8,39 +8,64 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
-var aggregationVarNames = []string{
-	"K6_CLOUD_AGGREGATION_MIN_SAMPLES",
-	"K6_CLOUD_AGGREGATION_PERIOD",
-	"K6_CLOUD_AGGREGATION_WAIT_PERIOD",
-	"K6_CLOUD_METRIC_PUSH_INTERVAL",
-	"K6_CLOUD_MAX_METRIC_SAMPLES_PER_PACKAGE",
-	"K6_CLOUD_MAX_METRIC_PUSH_CONCURRENCY",
+var aggregationVarNames = map[int][]string{
+	1: []string{
+		// cloud output v1: to be removed in the future
+		"K6_CLOUD_AGGREGATION_MIN_SAMPLES",
+		"K6_CLOUD_AGGREGATION_PERIOD",
+		"K6_CLOUD_AGGREGATION_WAIT_PERIOD",
+		"K6_CLOUD_METRIC_PUSH_INTERVAL",
+		"K6_CLOUD_MAX_METRIC_SAMPLES_PER_PACKAGE",
+		"K6_CLOUD_MAX_METRIC_PUSH_CONCURRENCY",
+	},
+	2: []string{
+		// cloud output v2
+		"K6_CLOUD_API_VERSION",
+		"K6_CLOUD_AGGREGATION_PERIOD",
+		"K6_CLOUD_AGGREGATION_WAIT_PERIOD",
+		"K6_CLOUD_METRIC_PUSH_INTERVAL",
+		"K6_CLOUD_METRIC_PUSH_CONCURRENCY",
+	},
 }
 
 func EncodeAggregationConfig(testRun *cloudapi.CreateTestRunResponse) string {
-	return fmt.Sprintf("%d|%s|%s|%s|%d|%d",
-		testRun.ConfigOverride.AggregationMinSamples.Int64,
+	return fmt.Sprintf("%d|%s|%s|%s|%d",
+		2, // use v2 for all new test runs
 		testRun.ConfigOverride.AggregationPeriod.String(),
 		testRun.ConfigOverride.AggregationWaitPeriod.String(),
 		testRun.ConfigOverride.MetricPushInterval.String(),
-		testRun.ConfigOverride.MaxMetricSamplesPerPackage.Int64,
 		testRun.ConfigOverride.MetricPushConcurrency.Int64)
 }
 
 func DecodeAggregationConfig(encoded string) ([]corev1.EnvVar, error) {
 	values := strings.Split(encoded, "|")
-	if len(values) != len(aggregationVarNames) {
+
+	// in order not to break existing deployments,
+	// let's support decoding of cloud output v1 for some time
+	var (
+		apiV1VarNames = len(aggregationVarNames[1])
+		apiV2VarNames = len(aggregationVarNames[2])
+	)
+
+	if len(values) != apiV1VarNames && len(values) != apiV2VarNames {
 		return nil, fmt.Errorf(
-			"Aggregation vars got corrupted: there are %d values instead of %d. Encoded value: `%s`.",
+			"Aggregation vars got corrupted: there are %d values instead of %d or %d. Encoded value: `%s`.",
 			len(values),
-			len(aggregationVarNames),
+			apiV1VarNames, apiV2VarNames,
 			encoded)
 	}
 
+	var varNames []string
+	if len(values) == apiV1VarNames {
+		varNames = aggregationVarNames[1]
+	} else {
+		varNames = aggregationVarNames[2]
+	}
+
 	vars := make([]corev1.EnvVar, len(values))
 	for i := range values {
 		vars[i] = corev1.EnvVar{
-			Name:  aggregationVarNames[i],
+			Name:  varNames[i],
 			Value: values[i],
 		}
 	}

pkg/cloud/aggregation_test.go

@@ -0,0 +1,100 @@
+package cloud
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"go.k6.io/k6/cloudapi"
+	"go.k6.io/k6/lib/types"
+	"gopkg.in/guregu/null.v3"
+	corev1 "k8s.io/api/core/v1"
+)
+
+func Test_EncodeAggregationConfig(t *testing.T) {
+	expected := "2|5s|3s|10s|10"
+
+	testRunResponse := &cloudapi.CreateTestRunResponse{
+		ReferenceID: "test-run-id",
+		ConfigOverride: &cloudapi.Config{
+			AggregationPeriod:     types.NullDurationFrom(time.Second * 5),
+			AggregationWaitPeriod: types.NullDurationFrom(time.Second * 3),
+			MetricPushInterval:    types.NullDurationFrom(time.Second * 10),
+			MetricPushConcurrency: null.IntFrom(10),
+		},
+	}
+
+	encodedAggregation := EncodeAggregationConfig(testRunResponse)
+	assert.Equal(t, expected, encodedAggregation)
+}
+
+func Test_DecodeAggregationConfig(t *testing.T) {
+	var (
+		// For now, we support both versions in decoding.
+		v1Encoded = "50|3s|8s|6s|10000|10"
+		v2Encoded = "2|5s|3s|10s|10"
+
+		v1EnvVars = []corev1.EnvVar{
+			{
+				Name:  "K6_CLOUD_AGGREGATION_MIN_SAMPLES",
+				Value: "50",
+			},
+			{
+				Name:  "K6_CLOUD_AGGREGATION_PERIOD",
+				Value: "3s",
+			},
+			{
+				Name:  "K6_CLOUD_AGGREGATION_WAIT_PERIOD",
+				Value: "8s",
+			},
+			{
+				Name:  "K6_CLOUD_METRIC_PUSH_INTERVAL",
+				Value: "6s",
+			},
+			{
+				Name:  "K6_CLOUD_MAX_METRIC_SAMPLES_PER_PACKAGE",
+				Value: "10000",
+			},
+			{
+				Name:  "K6_CLOUD_MAX_METRIC_PUSH_CONCURRENCY",
+				Value: "10",
+			},
+		}
+
+		v2EnvVars = []corev1.EnvVar{
+			{
+				Name:  "K6_CLOUD_API_VERSION",
+				Value: "2",
+			},
+			{
+				Name:  "K6_CLOUD_AGGREGATION_PERIOD",
+				Value: "5s",
+			},
+			{
+				Name:  "K6_CLOUD_AGGREGATION_WAIT_PERIOD",
+				Value: "3s",
+			},
+			{
+				Name:  "K6_CLOUD_METRIC_PUSH_INTERVAL",
+				Value: "10s",
+			},
+			{
+				Name:  "K6_CLOUD_METRIC_PUSH_CONCURRENCY",
+				Value: "10",
+			},
+		}
+	)
+
+	envVars, err := DecodeAggregationConfig(v1Encoded)
+	assert.Equal(t, nil, err)
+
+	for i, expectedEnvVar := range v1EnvVars {
+		assert.Equal(t, expectedEnvVar, envVars[i])
+	}
+
+	envVars, err = DecodeAggregationConfig(v2Encoded)
+	assert.Equal(t, nil, err)
+	for i, expectedEnvVar := range v2EnvVars {
+		assert.Equal(t, expectedEnvVar, envVars[i])
+	}
+}