Support for the Cloud output

README.md

@@ -154,6 +154,52 @@ Defines options for the starter pod. This includes:
 * passing in custom image
 * passing in labels and annotations
 
+### k6 outputs
+
+#### k6 Cloud output
+
+k6 supports [output to its Cloud](https://k6.io/docs/results-visualization/cloud) with `k6 run --out cloud script.js` command. This feature is available in k6-operator as well for subscribed users. Note that it supports only `parallelism: 20` or less.
+
+To use this option in k6-operator, set the argument in yaml:
+
+```yaml
+...
+  script:
+    configMap:
+      name: "<configmap>"
+  arguments: --out cloud
+...
+```
+
+Then uncomment cloud output section in `config/default/kustomization.yaml` and copy your token from the Cloud there:
+
+```yaml
+# Uncomment this section if you need cloud output and copy-paste your token
+secretGenerator:
+- name: cloud-token
+  literals:
+  - token=<copy-paste-token-string-here>
+  options:
+    annotations:
+      kubernetes.io/service-account.name: k6-operator-controller
+    labels:
+      k6cloud: token
+```
+
+This is sufficient to run k6 with the Cloud output and default values of `projectID` and `name` (`"k6-operator-test"`). For non-default values, extended script options can be used like this:
+
+```js
+export let options = {
+  ...
+  ext: {
+    loadimpact: {
+      name: 'Configured k6-operator test',
+      projectID: 1234567,
+    }
+  }
+};
+```
+
 ### Cleaning up between test runs
 After completing a test run, you need to clean up the test jobs created. This is done by running the following command:
 ```bash

api/v1alpha1/k6_types.go

@@ -90,7 +90,7 @@ type K6Configmap struct {
 type Cleanup string
 
 // Stage describes which stage of the test execution lifecycle our runners are in
-// +kubebuilder:validation:Enum=created;started;finished
+// +kubebuilder:validation:Enum=initialization;initialized;created;started;finished
 type Stage string
 
 // K6Status defines the observed state of K6

config/crd/bases/k6.io_k6s.yaml

@@ -1956,6 +1956,8 @@ spec:
                 description: Stage describes which stage of the test execution lifecycle
                   our runners are in
                 enum:
+                - initialization
+                - initialized
                 - created
                 - started
                 - finished

config/default/kustomization.yaml

@@ -62,3 +62,14 @@ vars:
 #    kind: Service
 #    version: v1
 #    name: webhook-service
+
+# Uncomment this section if you need cloud output and copy-paste your token
+# secretGenerator:
+# - name: cloud-token
+#   literals:
+#   - token=
+#   options:
+#     annotations:
+#       kubernetes.io/service-account.name: k6-operator-controller
+#     labels:
+#       k6cloud: token

config/rbac/role.yaml

@@ -45,6 +45,7 @@ rules:
       - ""
     resources:
       - pods
+      - pods/log
     verbs:
       - get
       - list
@@ -70,3 +71,11 @@ rules:
       - get
       - patch
       - update
+  - apiGroups:
+      - ""
+    resources:
+      - secrets
+    verbs:
+      - list
+      - get
+      - watch

controllers/k6_controller.go

@@ -56,12 +56,22 @@ func (r *K6Reconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 		return ctrl.Result{Requeue: true}, err
 	}
 
+	log.Info(fmt.Sprintf("Reconcile(); stage = %s", k6.Status.Stage))
+
 	switch k6.Status.Stage {
 	case "":
+		return InitializeJobs(ctx, log, k6, r)
+	case "initialization":
+		// here we're just waiting until initialize is done
+		// Note: it is present as a separate stage to ensure there's only one
+		// initialization job at a time
+		return ctrl.Result{}, nil
+	case "initialized":
 		return CreateJobs(ctx, log, k6, r)
 	case "created":
 		return StartJobs(ctx, log, k6, r)
 	case "started":
+		// wait for test to finish and then mark as finished
 		return FinishJobs(ctx, log, k6, r)
 	case "finished":
 		// delete if configured

controllers/k6_create.go

@@ -14,7 +14,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 )
 
-// CreateJobs that will spawn k6 pods, running distributed tests
+// CreateJobs creates jobs that will spawn k6 pods for distributed test
 func CreateJobs(ctx context.Context, log logr.Logger, k6 *v1alpha1.K6, r *K6Reconciler) (ctrl.Result, error) {
 	var err error
 	var res ctrl.Result
@@ -25,6 +25,7 @@ func CreateJobs(ctx context.Context, log logr.Logger, k6 *v1alpha1.K6, r *K6Reco
 		return res, err
 	}
 
+	log.Info("Changing stage of K6 status to created")
 	k6.Status.Stage = "created"
 	if err = r.Client.Status().Update(ctx, k6); err != nil {
 		log.Error(err, "Could not update status of custom resource")
@@ -62,11 +63,14 @@ func launchTest(ctx context.Context, k6 *v1alpha1.K6, index int, log logr.Logger
 	msg := fmt.Sprintf("Launching k6 test #%d", index)
 	log.Info(msg)
 
-	if job, err = jobs.NewRunnerJob(k6, index); err != nil {
+	if job, err = jobs.NewRunnerJob(k6, index, testRunId, token); err != nil {
 		log.Error(err, "Failed to generate k6 test job")
 		return err
 	}
 
+	log.Info(fmt.Sprintf("Runner job is ready to start with image `%s` and command `%s`",
+		job.Spec.Template.Spec.Containers[0].Image, job.Spec.Template.Spec.Containers[0].Command))
+
 	if err = ctrl.SetControllerReference(k6, job, r.Scheme); err != nil {
 		log.Error(err, "Failed to set controller reference for job")
 		return err

controllers/k6_finish.go

@@ -3,47 +3,85 @@ package controllers
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/go-logr/logr"
 	"github.com/grafana/k6-operator/api/v1alpha1"
+	"github.com/grafana/k6-operator/pkg/cloud"
+	"github.com/grafana/k6-operator/pkg/types"
 	batchv1 "k8s.io/api/batch/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/wait"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-// Mark k6 as finished as jobs finish
+// FinishJobs waits for the pods to finish, performs finishing call for cloud output and moves state to "finished".
 func FinishJobs(ctx context.Context, log logr.Logger, k6 *v1alpha1.K6, r *K6Reconciler) (ctrl.Result, error) {
-	selector := labels.SelectorFromSet(map[string]string{
-		"app":   "k6",
-		"k6_cr": k6.Name,
-	})
+	log.Info("Waiting for pods to finish")
 
-	opts := &client.ListOptions{LabelSelector: selector, Namespace: k6.Namespace}
-	jl := &batchv1.JobList{}
+	// Here we assume that the test runs for some time and there is no need to
+	// check it more often than twice in a minute.
+	//
+	// The total timeout for the test is set to duration of the test + 2 min.
+	// These 2 min are meant to cover the time needed to start the pods: sometimes
+	// pods are ready a bit later than operator reaches this stage so from the
+	// viewpoint of operator it takes longer. This behaviour depends on the setup of
+	// cluster. 2 min are meant to be a sufficient safeguard for such cases.
 
-	if err := r.List(ctx, jl, opts); err != nil {
-		log.Error(err, "Could not list jobs")
-		return ctrl.Result{}, err
-	}
+	testDuration := inspectOutput.TotalDuration.TimeDuration()
+
+	err := wait.PollImmediate(time.Second*30, testDuration+time.Minute*2, func() (done bool, err error) {
+		selector := labels.SelectorFromSet(map[string]string{
+			"app":    "k6",
+			"k6_cr":  k6.Name,
+			"runner": "true",
+		})
+
+		opts := &client.ListOptions{LabelSelector: selector, Namespace: k6.Namespace}
+		jl := &batchv1.JobList{}
 
-	//TODO: We should distinguish between suceeded/failed
-	var finished int32
-	for _, job := range jl.Items {
-		if job.Status.Active != 0 {
-			continue
+		if err := r.List(ctx, jl, opts); err != nil {
+			log.Error(err, "Could not list jobs")
+			return false, nil
+		}
+
+		// TODO: We should distinguish between Suceeded/Failed/Unknown
+		var finished int32
+		for _, job := range jl.Items {
+			if job.Status.Active != 0 {
+				continue
+			}
+			finished++
 		}
-		finished++
-	}
 
-	log.Info(fmt.Sprintf("%d/%d jobs complete", finished, k6.Spec.Parallelism+1))
+		log.Info(fmt.Sprintf("%d/%d jobs complete", finished, k6.Spec.Parallelism))
 
-	// parallelism (pods) + starter pod = total expected
-	if finished == k6.Spec.Parallelism+1 {
-		k6.Status.Stage = "finished"
-		if err := r.Client.Status().Update(ctx, k6); err != nil {
-			log.Error(err, "Could not update status of custom resource")
+		if finished >= k6.Spec.Parallelism {
+			return true, nil
 		}
+
+		return false, nil
+	})
+
+	if err != nil {
+		log.Error(err, "Waiting for pods to finish ended with error")
+	}
+
+	// If this is a test run with cloud output, try to finalize it regardless.
+	if cli := types.ParseCLI(&k6.Spec); cli.HasCloudOut {
+		if err = cloud.FinishTestRun(testRunId); err != nil {
+			log.Error(err, "Could not finalize the test run with cloud output")
+		} else {
+			log.Info(fmt.Sprintf("Cloud test run %s was finalized succesfully", testRunId))
+		}
+	}
+
+	log.Info("Changing stage of K6 status to finished")
+	k6.Status.Stage = "finished"
+	if err = r.Client.Status().Update(ctx, k6); err != nil {
+		log.Error(err, "Could not update status of custom resource")
+		return ctrl.Result{}, err
 	}
 
 	return ctrl.Result{}, nil