feat: supervisor and async tasks (#1125)

Co-authored-by: Anton Evangelatov <anton.evangelatov@gmail.com>

.circleci/config.yml

@@ -66,6 +66,7 @@ executors:
     machine:
       image: ubuntu-1604:201903-01
       docker_layer_caching: true
+    resource_class: large
     working_directory: << pipeline.parameters.workspace-dir >>/project
     environment:
       GOPATH: << pipeline.parameters.workspace-dir >>/go/<< pipeline.parameters.go-version >>

go.mod

@@ -37,6 +37,7 @@ require (
 	github.com/otiai10/copy v1.0.2
 	github.com/pborman/uuid v1.2.0
 	github.com/pierrec/lz4 v2.5.2+incompatible // indirect
+	github.com/rs/xid v1.2.1
 	github.com/stretchr/testify v1.5.1
 	github.com/syndtr/goleveldb v1.0.0
 	github.com/testground/plan-templates/templates v0.0.0-20200429051153-b24fdc73e401

go.sum

@@ -317,6 +317,8 @@ github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4O
 github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0 h1:MkV+77GLUNo5oJ0jf870itWm3D0Sjh7+Za9gazKc5LQ=
 github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+github.com/rs/xid v1.2.1 h1:mhH9Nq+C1fY2l1XIpgxIiUOfNpRBYH1kKcr+qfKgjRc=
+github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
 github.com/russross/blackfriday/v2 v2.0.1 h1:lPqVAte+HuHNfhJ/0LC98ESWRz8afy9tM/0RK8m9o+Q=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=

integration_tests/01_k8s_kind_placebo_ok.sh

@@ -4,7 +4,7 @@ my_dir="$(dirname "$0")"
 source "$my_dir/header.sh"
 
 testground plan import --from plans/placebo
-testground build single --builder docker:go --plan placebo | tee build.out
+testground build single --builder docker:go --plan placebo --wait | tee build.out
 export ARTIFACT=$(awk -F\" '/generated build artifact/ {print $8}' build.out)
 docker tag $ARTIFACT testplan:placebo
 
@@ -14,12 +14,12 @@ docker tag $ARTIFACT testplan:placebo
 kind load docker-image testplan:placebo
 
 pushd $TEMPDIR
-testground run single --runner cluster:k8s --builder docker:go --use-build testplan:placebo --instances 1 --plan placebo --testcase ok --collect | tee run.out
+testground run single --runner cluster:k8s --builder docker:go --use-build testplan:placebo --instances 1 --plan placebo --testcase ok --collect --wait | tee run.out
 RUNID=$(awk '/finished run with ID/ { print $9 }' run.out)
 echo "checking run $RUNID"
 file $RUNID.tgz
 LENGTH=${#RUNID}
-test $LENGTH -eq 12
+test $LENGTH -eq 20
 tar -xzvvf $RUNID.tgz
 SIZEOUT=$(cat ./"$RUNID"/single/0/run.out | wc -c)
 echo "run.out is $SIZEOUT bytes."

integration_tests/02_k8s_kind_placebo_stall.sh

@@ -4,7 +4,7 @@ my_dir="$(dirname "$0")"
 source "$my_dir/header.sh"
 
 testground plan import --from plans/placebo
-testground build single --builder docker:go --plan placebo | tee build.out
+testground build single --builder docker:go --plan placebo --wait | tee build.out
 export ARTIFACT=$(awk -F\" '/generated build artifact/ {print $8}' build.out)
 docker tag $ARTIFACT testplan:placebo
 
@@ -13,7 +13,7 @@ docker tag $ARTIFACT testplan:placebo
 # The plan is renamed as `testplan:placebo` because kind will check DockerHub if the tag is `latest`.
 kind load docker-image testplan:placebo
 
-testground run single --runner cluster:k8s --builder docker:go --use-build testplan:placebo --instances 2 --plan placebo --testcase stall &
+testground run single --runner cluster:k8s --builder docker:go --use-build testplan:placebo --instances 2 --plan placebo --testcase stall --wait &
 sleep 20
 BEFORE=$(kubectl get pods | grep placebo | grep Running | wc -l)
 testground terminate --runner=cluster:k8s

integration_tests/03_exec_go_placebo_ok.sh

@@ -5,12 +5,12 @@ source "$my_dir/header.sh"
 
 pushd $TEMPDIR
 testground plan import --from plans/placebo
-testground run single --runner local:exec --builder exec:go --instances 2 --plan placebo --testcase ok --collect | tee run.out
+testground run single --runner local:exec --builder exec:go --instances 2 --plan placebo --testcase ok --collect --wait | tee run.out
 RUNID=$(awk '/finished run with ID/ { print $9 }' run.out)
 echo "checking run $RUNID"
 file $RUNID.tgz
 LENGTH=${#RUNID}
-test $LENGTH -eq 12
+test $LENGTH -eq 20
 tar -xzvvf $RUNID.tgz
 SIZEOUT=$(cat ./"$RUNID"/single/0/run.out | wc -c)
 echo "run.out is $SIZEOUT bytes."

integration_tests/04_docker_placebo_ok.sh

@@ -4,18 +4,18 @@ my_dir="$(dirname "$0")"
 source "$my_dir/header.sh"
 
 testground plan import --from plans/placebo
-testground build single --builder docker:go --plan placebo | tee build.out
+testground build single --builder docker:go --plan placebo --wait | tee build.out
 export ARTIFACT=$(awk -F\" '/generated build artifact/ {print $8}' build.out)
 docker tag $ARTIFACT testplan:placebo
 
 pushd $TEMPDIR
 testground healthcheck --runner local:docker --fix
-testground run single --runner local:docker --builder docker:go --use-build testplan:placebo --instances 1 --plan placebo --testcase ok --collect | tee run.out
+testground run single --runner local:docker --builder docker:go --use-build testplan:placebo --instances 1 --plan placebo --testcase ok --collect --wait | tee run.out
 RUNID=$(awk '/finished run with ID/ { print $9 }' run.out)
 echo "checking run $RUNID"
 file $RUNID.tgz
 LENGTH=${#RUNID}
-test $LENGTH -eq 12
+test $LENGTH -eq 20
 tar -xzvvf $RUNID.tgz
 SIZEOUT=$(cat ./"$RUNID"/single/0/run.out | wc -c)
 echo "run.out is $SIZEOUT bytes."

integration_tests/05_docker_placebo_stall.sh

@@ -4,12 +4,12 @@ my_dir="$(dirname "$0")"
 source "$my_dir/header.sh"
 
 testground plan import --from plans/placebo
-testground build single --builder docker:go --plan placebo | tee build.out
+testground build single --builder docker:go --plan placebo --wait | tee build.out
 export ARTIFACT=$(awk -F\" '/generated build artifact/ {print $8}' build.out)
 docker tag $ARTIFACT testplan:placebo
 
 testground healthcheck --runner local:docker --fix
-testground run single --runner local:docker --builder docker:go --use-build testplan:placebo --instances 2 --plan placebo --testcase stall &
+testground run single --runner local:docker --builder docker:go --use-build testplan:placebo --instances 2 --plan placebo --testcase stall --wait &
 sleep 20
 BEFORE=$(docker ps | grep placebo | wc -l)
 testground terminate --runner=local:docker

integration_tests/06_docker_network_ping-pong.sh

@@ -4,13 +4,13 @@ my_dir="$(dirname "$0")"
 source "$my_dir/header.sh"
 
 testground plan import --from plans/network
-testground build single --builder docker:go --plan network | tee build.out
+testground build single --builder docker:go --plan network --wait | tee build.out
 export ARTIFACT=$(awk -F\" '/generated build artifact/ {print $8}' build.out)
 docker tag $ARTIFACT testplan:network
 
 pushd $TEMPDIR
 testground healthcheck --runner local:docker --fix
-testground run single --runner local:docker --builder docker:go --use-build testplan:network --instances 2 --plan network --testcase ping-pong --collect | tee stdout.out
+testground run single --runner local:docker --builder docker:go --use-build testplan:network --instances 2 --plan network --testcase ping-pong --collect --wait | tee stdout.out
 RUNID=$(awk '/finished run with ID/ { print $9 }' stdout.out)
 echo "checking run $RUNID"
 file $RUNID.tgz

integration_tests/07_docker_network_traffic-allowed.sh

@@ -4,18 +4,18 @@ my_dir="$(dirname "$0")"
 source "$my_dir/header.sh"
 
 testground plan import --from plans/network
-testground build single --builder docker:go --plan network | tee build.out
+testground build single --builder docker:go --plan network --wait | tee build.out
 export ARTIFACT=$(awk -F\" '/generated build artifact/ {print $8}' build.out)
 docker tag $ARTIFACT testplan:network
 
 pushd $TEMPDIR
 testground healthcheck --runner local:docker --fix
-testground run single --runner local:docker --builder docker:go --use-build testplan:network --instances 1 --plan network --testcase traffic-allowed --collect | tee stdout.out
+testground run single --runner local:docker --builder docker:go --use-build testplan:network --instances 1 --plan network --testcase traffic-allowed --collect --wait | tee stdout.out
 RUNID=$(awk '/finished run with ID/ { print $9 }' stdout.out)
 echo "checking run $RUNID"
 file $RUNID.tgz
 LENGTH=${#RUNID}
-test $LENGTH -eq 12
+test $LENGTH -eq 20
 tar -xzvvf $RUNID.tgz
 SIZEOUT=$(cat ./"$RUNID"/single/0/run.out | wc -c)
 echo "run.out is $SIZEOUT bytes."

integration_tests/08_docker_network_traffic-blocked.sh

@@ -4,18 +4,18 @@ my_dir="$(dirname "$0")"
 source "$my_dir/header.sh"
 
 testground plan import --from plans/network
-testground build single --builder docker:go --plan network | tee build.out
+testground build single --builder docker:go --plan network --wait | tee build.out
 export ARTIFACT=$(awk -F\" '/generated build artifact/ {print $8}' build.out)
 docker tag $ARTIFACT testplan:network
 
 pushd $TEMPDIR
 testground healthcheck --runner local:docker --fix
-testground run single --runner local:docker --builder docker:go --use-build testplan:network --instances 1 --plan network --testcase traffic-blocked --collect | tee stdout.out
+testground run single --runner local:docker --builder docker:go --use-build testplan:network --instances 1 --plan network --testcase traffic-blocked --collect --wait | tee stdout.out
 RUNID=$(awk '/finished run with ID/ { print $9 }' stdout.out)
 echo "checking run $RUNID"
 file $RUNID.tgz
 LENGTH=${#RUNID}
-test $LENGTH -eq 12
+test $LENGTH -eq 20
 tar -xzvvf $RUNID.tgz
 SIZEOUT=$(cat ./"$RUNID"/single/0/run.out | wc -c)
 echo "run.out is $SIZEOUT bytes."

integration_tests/09_docker_splitbrain_accept.sh

@@ -4,18 +4,18 @@ my_dir="$(dirname "$0")"
 source "$my_dir/header.sh"
 
 testground plan import --from plans/splitbrain
-testground build single --builder docker:go --plan splitbrain | tee build.out
+testground build single --builder docker:go --plan splitbrain --wait | tee build.out
 export ARTIFACT=$(awk -F\" '/generated build artifact/ {print $8}' build.out)
 docker tag $ARTIFACT testplan:splitbrain
 
 pushd $TEMPDIR
 testground healthcheck --runner local:docker --fix
-testground run single --runner local:docker --builder docker:go --use-build testplan:splitbrain --instances 3 --plan splitbrain --testcase accept --collect | tee stdout.out
+testground run single --runner local:docker --builder docker:go --use-build testplan:splitbrain --instances 3 --plan splitbrain --testcase accept --collect --wait | tee stdout.out
 RUNID=$(awk '/finished run with ID/ { print $9 }' stdout.out)
 echo "checking run $RUNID"
 file $RUNID.tgz
 LENGTH=${#RUNID}
-test $LENGTH -eq 12
+test $LENGTH -eq 20
 tar -xzvvf $RUNID.tgz
 SIZEOUT=$(cat ./"$RUNID"/single/0/run.out | wc -c)
 echo "run.out is $SIZEOUT bytes."

integration_tests/10_docker_splitbrain_reject.sh

@@ -4,18 +4,18 @@ my_dir="$(dirname "$0")"
 source "$my_dir/header.sh"
 
 testground plan import --from plans/splitbrain
-testground build single --builder docker:go --plan splitbrain | tee build.out
+testground build single --builder docker:go --plan splitbrain --wait | tee build.out
 export ARTIFACT=$(awk -F\" '/generated build artifact/ {print $8}' build.out)
 docker tag $ARTIFACT testplan:splitbrain
 
 pushd $TEMPDIR
 testground healthcheck --runner local:docker --fix
-testground run single --runner local:docker --builder docker:go --use-build testplan:splitbrain --instances 3 --plan splitbrain --testcase reject --collect | tee stdout.out
+testground run single --runner local:docker --builder docker:go --use-build testplan:splitbrain --instances 3 --plan splitbrain --testcase reject --collect --wait | tee stdout.out
 RUNID=$(awk '/finished run with ID/ { print $9 }' stdout.out)
 echo "checking run $RUNID"
 file $RUNID.tgz
 LENGTH=${#RUNID}
-test $LENGTH -eq 12
+test $LENGTH -eq 20
 tar -xzvvf $RUNID.tgz
 SIZEOUT=$(cat ./"$RUNID"/single/0/run.out | wc -c)
 echo "run.out is $SIZEOUT bytes."

integration_tests/11_docker_splitbrain_drop.sh

@@ -4,18 +4,18 @@ my_dir="$(dirname "$0")"
 source "$my_dir/header.sh"
 
 testground plan import --from plans/splitbrain
-testground build single --builder docker:go --plan splitbrain | tee build.out
+testground build single --builder docker:go --plan splitbrain --wait | tee build.out
 export ARTIFACT=$(awk -F\" '/generated build artifact/ {print $8}' build.out)
 docker tag $ARTIFACT testplan:splitbrain
 
 pushd $TEMPDIR
 testground healthcheck --runner local:docker --fix
-testground run single --runner local:docker --builder docker:go --use-build testplan:splitbrain --instances 3 --plan splitbrain --testcase drop --collect | tee stdout.out
+testground run single --runner local:docker --builder docker:go --use-build testplan:splitbrain --instances 3 --plan splitbrain --testcase drop --collect --wait | tee stdout.out
 RUNID=$(awk '/finished run with ID/ { print $9 }' stdout.out)
 echo "checking run $RUNID"
 file $RUNID.tgz
 LENGTH=${#RUNID}
-test $LENGTH -eq 12
+test $LENGTH -eq 20
 tar -xzvvf $RUNID.tgz
 SIZEOUT=$(cat ./"$RUNID"/single/0/run.out | wc -c)
 echo "run.out is $SIZEOUT bytes."

integration_tests/header.sh

@@ -1,6 +1,7 @@
 #!/bin/bash
 
 set -o errexit
+set -x
 set -e
 
 err_report() {
@@ -19,4 +20,11 @@ mkdir -p /home/circleci/testground
 cp env-kind.toml /home/circleci/testground/.env.toml
 testground daemon > $TEMPDIR/daemon.out 2>&1 &
 DAEMONPID=$!
+
 sleep 2
+
+echo "Waiting for Testground to launch on 8080..."
+while ! nc -z localhost 8080; do
+  sleep 1
+done
+echo "Testground launched"

pkg/api/engine.go

@@ -2,9 +2,9 @@ package api
 
 import (
 	"context"
-
 	"github.com/testground/testground/pkg/config"
 	"github.com/testground/testground/pkg/rpc"
+	"github.com/testground/testground/pkg/task"
 )
 
 type ComponentType string
@@ -19,19 +19,24 @@ const (
 type UnpackedSources struct {
 	// BaseDir is the directory containing the plan under ./plan, and an
 	// optional sdk under ./sdk.
-	BaseDir string
+	BaseDir string `json:"base_dir"`
 
 	// PlanDir is the directory where the test plan's source has been
 	// placed (i.e. BaseSrcPath/plan).
-	PlanDir string
+	PlanDir string `json:"plan_dir"`
 
 	// SDKDir is the directory where the SDK's source has been placed. It
 	// will be a zero-value if no SDK replacement has been requested, or
 	// BaseSrcPath/sdk otherwise.
-	SDKDir string
+	SDKDir string `json:"sdk_dir"`
 
 	// ExtraDir is the directory where any extra sources have been unpacked.
-	ExtraDir string
+	ExtraDir string `json:"extra_dir"`
+}
+
+type TasksFilters struct {
+	Types  []task.Type
+	States []task.State
 }
 
 type Engine interface {
@@ -41,9 +46,14 @@ type Engine interface {
 	ListBuilders() map[string]Builder
 	ListRunners() map[string]Runner
 
-	DoBuild(context.Context, *Composition, *UnpackedSources, *rpc.OutputWriter) ([]*BuildOutput, error)
+	QueueBuild(request *BuildRequest, sources *UnpackedSources) (string, error)
+	QueueRun(request *RunRequest, sources *UnpackedSources) (string, error)
+
+	Status(id string) (*task.Task, error)
+	Logs(ctx context.Context, id string, follow bool, cancel bool, ow *rpc.OutputWriter) (*task.Task, error)
+	Tasks(filters TasksFilters) ([]task.Task, error)
+
 	DoBuildPurge(ctx context.Context, builder, plan string, ow *rpc.OutputWriter) error
-	DoRun(context.Context, *Composition, *rpc.OutputWriter) (*RunOutput, error)
 	DoCollectOutputs(ctx context.Context, runner string, runID string, ow *rpc.OutputWriter) error
 	DoTerminate(ctx context.Context, ctype ComponentType, ref string, ow *rpc.OutputWriter) error
 	DoHealthcheck(ctx context.Context, runner string, fix bool, ow *rpc.OutputWriter) (*HealthcheckReport, error)

pkg/api/rpc.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"bytes"
+	"github.com/testground/testground/pkg/task"
 )
 
 // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -15,12 +16,17 @@ type DescribeRequest struct {
 
 // BuildRequest is the request struct for the `build` function.
 type BuildRequest struct {
-	Composition Composition `json:"composition"`
+	Priority    int              `json:"priority"`
+	Composition Composition      `json:"composition"`
+	Manifest    TestPlanManifest `json:"manifest"`
 }
 
 // RunRequest is the request struct for the `run` function.
 type RunRequest struct {
-	Composition Composition `json:"composition"`
+	Priority    int              `json:"priority"`
+	BuildGroups []int            `json:"build_groups"`
+	Composition Composition      `json:"composition"`
+	Manifest    TestPlanManifest `json:"manifest"`
 }
 
 type OutputsRequest struct {
@@ -43,6 +49,24 @@ type BuildPurgeRequest struct {
 	Testplan string `json:"testplan"`
 }
 
+type TasksRequest = TasksFilters
+
+type StatusRequest struct {
+	TaskID string `json:"task_id"`
+}
+
+type CancelRequest struct {
+	TaskID string `json:"task_id"`
+}
+
+type LogsRequest struct {
+	TaskID string `json:"task_id"`
+	Follow bool   `json:"follow"`
+	// CancelWithContext indicates if the task should be cancelled
+	// on context cancellation.
+	CancelWithContext bool `json:"cancel_with_context"`
+}
+
 // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 // ~~~~~~ Response payloads ~~~~~~
 // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -58,3 +82,7 @@ type CollectResponse struct {
 }
 
 type HealthcheckResponse = HealthcheckReport
+
+type StatusResponse = task.Task
+
+type LogsResponse = task.Task

pkg/api/runner.go

@@ -80,6 +80,9 @@ type RunGroup struct {
 type RunOutput struct {
 	// RunnerID is the ID of the runner used.
 	RunID string
+
+	// Composition that was used for this run.
+	Composition Composition
 }
 
 type CollectionInput struct {