README.md

@@ -9,7 +9,7 @@
 
 The `TestRun` CRD is a representation of a single k6 test executed once. `TestRun` supports various configuration options that allow you to adapt to different Kubernetes setups. You can find a description of the more common options [here](https://grafana.com/docs/k6/latest/set-up/set-up-distributed-k6/usage/common-options/), and the full list of options can be found in the [definition itself](https://github.com/grafana/k6-operator/blob/main/config/crd/bases/k6.io_testruns.yaml).
 
-The `PrivateLoadZone` CRD is a representation of a [load zone](https://grafana.com/docs/grafana-cloud/testing/k6/author-run/use-load-zones/), which is a k6 term for a set of nodes within a cluster designated to execute k6 test runs. `PrivateLoadZone` is integrated with [Grafana Cloud k6](https://grafana.com/products/cloud/k6/) and requires a [Grafana Cloud account](https://grafana.com/auth/sign-up/create-user). You can find a guide describing how to set up a `PrivateLoadZone` [here](https://grafana.com/docs/grafana-cloud/testing/k6/author-run/set-up-private-load-zones/), while billing details can be found [here](https://grafana.com/docs/grafana-cloud/cost-management-and-billing/understand-your-invoice/k6-invoice/).
+The `PrivateLoadZone` CRD is a representation of a [load zone](https://grafana.com/docs/grafana-cloud/testing/k6/author-run/use-load-zones/), which is a k6 term for a set of nodes within a cluster designated to execute k6 test runs. `PrivateLoadZone` is integrated with [Grafana Cloud k6](https://grafana.com/products/cloud/k6/) and requires a [Grafana Cloud account](https://grafana.com/auth/sign-up/create-user). You can find a guide describing how to set up a `PrivateLoadZone` [here](https://grafana.com/docs/grafana-cloud/testing/k6/author-run/private-load-zone-v2/), while billing details can be found [here](https://grafana.com/docs/grafana-cloud/cost-management-and-billing/understand-your-invoice/k6-invoice/).
 
 ## Documentation
 

charts/k6-operator/templates/deployment.yaml

@@ -53,7 +53,9 @@ spec:
           command:
             - /manager
           args:
+            {{- if gt (int .Values.manager.replicas) 1 }}
             - --enable-leader-election
+            {{- end }}
             {{- if .Values.authProxy.enabled }}
             - --metrics-addr=127.0.0.1:8080
             {{- end }}

controllers/common.go

@@ -100,7 +100,7 @@ func inspectTestRun(ctx context.Context, log logr.Logger, k6 v1alpha1.TestRunI,
 	defer podLogs.Close()
 
 	buf := new(bytes.Buffer)
-	_, err = io.Copy(buf, podLogs)
+	_, returnErr = io.Copy(buf, podLogs)
 	if err != nil {
 		log.Error(err, "unable to copy logs from the pod")
 		return

controllers/plz_controller.go

@@ -55,6 +55,7 @@ type PrivateLoadZoneReconciler struct {
 	// poller should be made PLZ specific;
 	// e.g. with a map: PLZ name -> poller.
 	poller *cloud.TestRunPoller
+	token  string // needed for cloud logs
 }
 
 //+kubebuilder:rbac:groups=k6.io,resources=privateloadzones,verbs=get;list;watch;create;update;patch;delete
@@ -88,6 +89,7 @@ func (r *PrivateLoadZoneReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		}
 
 		r.poller = cloud.NewTestRunPoller(cloud.ApiURL(k6CloudHost()), token, plz.Name, logger)
+		r.token = token
 	}
 
 	if plz.DeletionTimestamp.IsZero() && (plz.IsUnknown(v1alpha1.PLZRegistered) || plz.IsFalse(v1alpha1.PLZRegistered)) {

controllers/plz_factory.go

@@ -41,7 +41,7 @@ func (r *PrivateLoadZoneReconciler) startFactory(plz *v1alpha1.PrivateLoadZone,
 				continue
 			}
 
-			k6 = testrun.NewPLZTestRun(plz, trData, k6CloudHost())
+			k6 = testrun.NewPLZTestRun(plz, r.token, trData, k6CloudHost())
 
 			logger.Info(fmt.Sprintf("PLZ test run has been prepared with image `%s` and `%d` instances",
 				k6.Spec.Runner.Image, k6.Spec.Parallelism), "testRunId", testRunId)

e2e/ipv6/kind-ipv6.yaml

@@ -0,0 +1,5 @@
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+networking:
+  ipFamily: ipv6
+  apiServerAddress: 127.0.0.1

pkg/cloud/aggregation.go

@@ -8,29 +8,18 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
-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",
-	},
+var aggregationVarNames = []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.Config) string {
 	return fmt.Sprintf("%d|%s|%s|%s|%d",
-		2, // use v2 for all new test runs
+		2, // version of protocol
 		testRun.AggregationPeriod.String(),
 		testRun.AggregationWaitPeriod.String(),
 		testRun.MetricPushInterval.String(),
@@ -40,32 +29,18 @@ func EncodeAggregationConfig(testRun *cloudapi.Config) string {
 func DecodeAggregationConfig(encoded string) ([]corev1.EnvVar, error) {
 	values := strings.Split(encoded, "|")
 
-	// 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 {
+	if len(values) != len(aggregationVarNames) {
 		return nil, fmt.Errorf(
-			"Aggregation vars got corrupted: there are %d values instead of %d or %d. Encoded value: `%s`.",
+			"Aggregation vars got corrupted: there are %d values instead of %d. Encoded value: `%s`.",
 			len(values),
-			apiV1VarNames, apiV2VarNames,
+			len(aggregationVarNames),
 			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:  varNames[i],
+			Name:  aggregationVarNames[i],
 			Value: values[i],
 		}
 	}

pkg/cloud/aggregation_test.go

@@ -30,38 +30,9 @@ func Test_EncodeAggregationConfig(t *testing.T) {
 
 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"
+		encoded = "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{
+		expected = []corev1.EnvVar{
 			{
 				Name:  "K6_CLOUD_API_VERSION",
 				Value: "2",
@@ -85,16 +56,10 @@ func Test_DecodeAggregationConfig(t *testing.T) {
 		}
 	)
 
-	envVars, err := DecodeAggregationConfig(v1Encoded)
+	envVars, err := DecodeAggregationConfig(encoded)
 	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 {
+	for i, expectedEnvVar := range expected {
 		assert.Equal(t, expectedEnvVar, envVars[i])
 	}
 }

pkg/resources/containers/curl_start.go

@@ -3,6 +3,7 @@ package containers
 import (
 	"encoding/json"
 	"fmt"
+	"net"
 	"strings"
 
 	"github.com/grafana/k6-operator/pkg/types"
@@ -26,7 +27,7 @@ func NewStartContainer(hostnames []string, image string, imagePullPolicy corev1.
 
 	var parts []string
 	for _, hostname := range hostnames {
-		parts = append(parts, fmt.Sprintf("curl --retry 3 -X PATCH -H 'Content-Type: application/json' http://%s:6565/v1/status -d '%s'", hostname, req))
+		parts = append(parts, fmt.Sprintf("curl --retry 3 -X PATCH -H 'Content-Type: application/json' http://%s/v1/status -d '%s'", net.JoinHostPort(hostname, "6565"), req))
 	}
 
 	return corev1.Container{

pkg/resources/containers/curl_stop.go

@@ -3,6 +3,7 @@ package containers
 import (
 	"encoding/json"
 	"fmt"
+	"net"
 	"strings"
 
 	"github.com/grafana/k6-operator/pkg/types"
@@ -26,7 +27,7 @@ func NewStopContainer(hostnames []string, image string, imagePullPolicy corev1.P
 
 	var parts []string
 	for _, hostname := range hostnames {
-		parts = append(parts, fmt.Sprintf("curl --retry 3 -X PATCH -H 'Content-Type: application/json' http://%s:6565/v1/status -d '%s'", hostname, req))
+		parts = append(parts, fmt.Sprintf("curl --retry 3 -X PATCH -H 'Content-Type: application/json' http://%s/v1/status -d '%s'", net.JoinHostPort(hostname, "6565"), req))
 	}
 
 	return corev1.Container{

pkg/resources/jobs/runner_test.go

@@ -17,23 +17,20 @@ import (
 
 // these are default values hard-coded in k6
 var aggregationEnvVars = []corev1.EnvVar{
-	corev1.EnvVar{
-		Name:  "K6_CLOUD_AGGREGATION_MIN_SAMPLES",
-		Value: "50",
-	}, corev1.EnvVar{
+	{
+		Name:  "K6_CLOUD_API_VERSION",
+		Value: "2",
+	}, {
 		Name:  "K6_CLOUD_AGGREGATION_PERIOD",
-		Value: "3s",
-	}, corev1.EnvVar{
+		Value: "5s",
+	}, {
 		Name:  "K6_CLOUD_AGGREGATION_WAIT_PERIOD",
-		Value: "8s",
-	}, corev1.EnvVar{
+		Value: "3s",
+	}, {
 		Name:  "K6_CLOUD_METRIC_PUSH_INTERVAL",
-		Value: "6s",
-	}, corev1.EnvVar{
-		Name:  "K6_CLOUD_MAX_METRIC_SAMPLES_PER_PACKAGE",
-		Value: "10000",
-	}, corev1.EnvVar{
-		Name:  "K6_CLOUD_MAX_METRIC_PUSH_CONCURRENCY",
+		Value: "10s",
+	}, {
+		Name:  "K6_CLOUD_METRIC_PUSH_CONCURRENCY",
 		Value: "10",
 	},
 }
@@ -1114,7 +1111,7 @@ func TestNewRunnerJobCloud(t *testing.T) {
 		// testrunid has to be set hard-coded here.
 		Status: v1alpha1.TestRunStatus{
 			TestRunID:       "testrunid",
-			AggregationVars: "50|3s|8s|6s|10000|10",
+			AggregationVars: "2|5s|3s|10s|10",
 		},
 	}
 

pkg/testrun/plz.go

@@ -16,7 +16,7 @@ func TestName(testRunId string) string {
 }
 
 // ingestURL is a temp hack
-func NewPLZTestRun(plz *v1alpha1.PrivateLoadZone, trData *cloud.TestRunData, ingestUrl string) *v1alpha1.TestRun {
+func NewPLZTestRun(plz *v1alpha1.PrivateLoadZone, token string, trData *cloud.TestRunData, ingestUrl string) *v1alpha1.TestRun {
 	volume := corev1.Volume{
 		Name: "archive-volume",
 		VolumeSource: corev1.VolumeSource{
@@ -72,8 +72,11 @@ func NewPLZTestRun(plz *v1alpha1.PrivateLoadZone, trData *cloud.TestRunData, ing
 			},
 			Parallelism: int32(trData.InstanceCount),
 			Separate:    false,
-			Arguments:   "--out cloud --no-thresholds",
-			Cleanup:     v1alpha1.Cleanup("post"),
+			Arguments: fmt.Sprintf(`--out cloud --no-thresholds --log-output=loki=https://cloudlogs.k6.io/api/v1/push,label.lz=%s,label.test_run_id=%s,header.Authorization="Token %s"`,
+				plz.Name,
+				trData.TestRunID(),
+				token),
+			Cleanup: v1alpha1.Cleanup("post"),
 
 			TestRunID: trData.TestRunID(),
 			Token:     plz.Spec.Token,

pkg/testrun/plz_test.go

@@ -2,6 +2,7 @@ package testrun
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/go-test/deep"
@@ -53,7 +54,7 @@ func Test_NewPLZTestRun(t *testing.T) {
 				},
 				Parallelism: int32(0),
 				Separate:    false,
-				Arguments:   "--out cloud --no-thresholds",
+				Arguments:   "--out cloud --no-thresholds --log-output=loki=https://cloudlogs.k6.io/api/v1/push,label.lz=,label.test_run_id=0,header.Authorization=\"Token token\"",
 				Cleanup:     v1alpha1.Cleanup("post"),
 
 				TestRunID: "0",
@@ -101,6 +102,10 @@ func Test_NewPLZTestRun(t *testing.T) {
 	cloudFieldsTestRun = requiredFieldsTestRun // build up on top of required field case
 	cloudFieldsTestRun.ObjectMeta.Name = TestName(fmt.Sprintf("%d", someTestRunID))
 	cloudFieldsTestRun.Spec.TestRunID = fmt.Sprintf("%d", someTestRunID)
+	cloudFieldsTestRun.Spec.Arguments = strings.Replace(requiredFieldsTestRun.Spec.Arguments,
+		"test_run_id=0",
+		fmt.Sprintf("test_run_id=%d", someTestRunID),
+		1)
 	cloudFieldsTestRun.Spec.Runner.InitContainers = []v1alpha1.InitContainer{
 		containers.NewS3InitContainer(
 			someArchiveURL,
@@ -223,7 +228,7 @@ func Test_NewPLZTestRun(t *testing.T) {
 		testCase := testCase
 		t.Run(testCase.name, func(t *testing.T) {
 			t.Parallel()
-			got := NewPLZTestRun(testCase.plz, testCase.cloudData, testCase.ingestUrl)
+			got := NewPLZTestRun(testCase.plz, "token", testCase.cloudData, testCase.ingestUrl)
 			if diff := deep.Equal(got, testCase.expected); diff != nil {
 				t.Errorf("NewPLZTestRun returned unexpected data, diff: %s", diff)
 			}