Revert "Removed RunOnce, extended Exec."

This reverts commit f047a4f.

README.md

@@ -27,11 +27,11 @@ Let's go through an example leveraging `go test` flow:
 2. Implement test. Start by creating environment. Currently `e2e` supports Docker environment only. Use unique name for all your tests. It's recommended to keep it stable so resources are consistently cleaned.
 
    ```go mdox-exec="sed -n '22,26p' examples/thanos/unittest_test.go"
-
    	// Start isolated environment with given ref.
    	e, err := e2e.NewDockerEnvironment("e2e_example")
    	testutil.Ok(t, err)
    	// Make sure resources (e.g docker containers, network, dir) are cleaned.
+   	t.Cleanup(e.Close)
    ```
 
 3. Implement the workload by embedding`e2e.Runnable` or `*e2e.InstrumentedRunnable`. Or you can use existing ones in [e2edb](db/) package. For example implementing function that schedules Jaeger with our desired configuration could look like this:
@@ -49,8 +49,7 @@ Let's go through an example leveraging `go test` flow:
 
 4. Program your scenario as you want. You can start, wait for their readiness, stop, check their metrics and use their network endpoints from both unit test (`Endpoint`) as well as within each workload (`InternalEndpoint`). You can also access workload directory. There is a shared directory across all workloads. Check `Dir` and `InternalDir` runnable methods.
 
-   ```go mdox-exec="sed -n '28,93p' examples/thanos/unittest_test.go"
-
+   ```go mdox-exec="sed -n '28,86p' examples/thanos/unittest_test.go"
    	// Create structs for Prometheus containers scraping itself.
    	p1 := e2edb.NewPrometheus(e, "prometheus-1")
    	s1 := e2edb.NewThanosSidecar(e, "sidecar-1", p1)
@@ -78,16 +77,16 @@ Let's go through an example leveraging `go test` flow:
    	testutil.Ok(t, err)
 
    	{
-   		now := model.Now()
-   		v, w, err := v1.NewAPI(a).Query(context.Background(), "up{}", now.Time())
-   		testutil.Ok(t, err)
-   		testutil.Equals(t, 0, len(w))
-   		testutil.Equals(
-   			t,
-   			fmt.Sprintf(`up{instance="%v", job="myself", prometheus="prometheus-1"} => 1 @[%v]
+        now := model.Now()
+        v, w, err := v1.NewAPI(a).Query(context.Background(), "up{}", now.Time())
+        testutil.Ok(t, err)
+        testutil.Equals(t, 0, len(w))
+        testutil.Equals(
+            t,
+            fmt.Sprintf(`up{instance="%v", job="myself", prometheus="prometheus-1"} => 1 @[%v]
    up{instance="%v", job="myself", prometheus="prometheus-2"} => 1 @[%v]`, p1.InternalEndpoint(e2edb.AccessPortName), now, p2.InternalEndpoint(e2edb.AccessPortName), now),
-   			v.String(),
-   		)
+            v.String(),
+        )
    	}
 
    	// Stop first Prometheus and sidecar.
@@ -98,24 +97,17 @@ Let's go through an example leveraging `go test` flow:
    	testutil.Ok(t, t1.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"thanos_store_nodes_grpc_connections"}, e2e.WaitMissingMetrics()))
 
    	{
-   		now := model.Now()
-   		v, w, err := v1.NewAPI(a).Query(context.Background(), "up{}", now.Time())
-   		testutil.Ok(t, err)
-   		testutil.Equals(t, 0, len(w))
-   		testutil.Equals(
-   			t,
-   			fmt.Sprintf(`up{instance="%v", job="myself", prometheus="prometheus-2"} => 1 @[%v]`, p2.InternalEndpoint(e2edb.AccessPortName), now),
-   			v.String(),
-   		)
+        now := model.Now()
+        v, w, err := v1.NewAPI(a).Query(context.Background(), "up{}", now.Time())
+        testutil.Ok(t, err)
+        testutil.Equals(t, 0, len(w))
+        testutil.Equals(
+            t,
+            fmt.Sprintf(`up{instance="%v", job="myself", prometheus="prometheus-2"} => 1 @[%v]`, p2.InternalEndpoint(e2edb.AccessPortName), now),
+            v.String(),
+        )
    	}
-
-   	// Batch job example.
-   	batch := e.Runnable("batch").Init(e2e.StartOptions{Image: "ubuntu:20.04", Command: e2e.NewCommandRunUntilStop()})
-   	testutil.Ok(t, batch.Start())
-
-   	var out bytes.Buffer
-   	testutil.Ok(t, batch.Exec(e2e.NewCommand("echo", "it works"), e2e.WithExecOptionStdout(&out)))
-   	testutil.Equals(t, "it works\n", out.String())
+   }
    ```
 
 ### Interactive

db/prometheus.go

@@ -72,7 +72,8 @@ func (p *Prometheus) SetConfig(config string) error {
 	}
 
 	if p.IsRunning() {
-		return p.Exec(e2e.NewCommand("kill", "-SIGHUP", "1"))
+		_, _, err := p.Exec(e2e.NewCommand("kill", "-SIGHUP", "1"))
+		return err
 	}
 	return nil
 }

env.go

@@ -7,7 +7,6 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -138,6 +137,9 @@ type runnable interface {
 	// Start tells Runnable to start.
 	Start() error
 
+	// RunOnce tells Runnable to start as batch job and wait until completion with output capture.
+	RunOnce(ctx context.Context) (output string, err error)
+
 	// WaitReady waits until the Runnable is ready. It should return error if runnable is stopped in mean time or
 	// it was stopped before.
 	WaitReady() error
@@ -150,10 +152,9 @@ type runnable interface {
 	// It should be ok to Stop and Kill more than once, with next invokes being noop.
 	Stop() error
 
-	// Exec runs the provided command inside the same process context (e.g. in the running docker container).
-	// It returns error response from attempting to run the command.
-	// See ExecOptions for more options like returning output or attaching to e2e logging.
-	Exec(Command, ...ExecOption) error
+	// Exec runs the provided command inside the same process context (e.g in the docker container).
+	// It returns the stdout, stderr, and error response from attempting to run the command.
+	Exec(command Command) (stdout string, stderr string, err error)
 
 	// Endpoint returns external runnable endpoint (host:port) for given port name.
 	// External means that it will be accessible only from host, but not from docker containers.
@@ -162,29 +163,6 @@ type runnable interface {
 	Endpoint(portName string) string
 }
 
-type ExecOption func(o *ExecOptions)
-
-type ExecOptions struct {
-	Stdout io.Writer
-	Stderr io.Writer
-}
-
-// WithExecOptionStdout sets stdout writer to be used when exec is performed.
-// By default, it is streaming to the env logger.
-func WithExecOptionStdout(stdout io.Writer) ExecOption {
-	return func(o *ExecOptions) {
-		o.Stdout = stdout
-	}
-}
-
-// WithExecOptionStderr sets stderr writer to be used when exec is performed.
-// By default, it is streaming to the env logger.
-func WithExecOptionStderr(stderr io.Writer) ExecOption {
-	return func(o *ExecOptions) {
-		o.Stderr = stderr
-	}
-}
-
 // Runnable is the entity that environment returns to manage single instance.
 type Runnable interface {
 	runnable
@@ -239,11 +217,6 @@ func NewCommandWithoutEntrypoint(cmd string, args ...string) Command {
 	}
 }
 
-// NewCommandRunUntilStop is a command that allows to keep container running.
-func NewCommandRunUntilStop() Command {
-	return NewCommandWithoutEntrypoint("tail", "-f", "/dev/null")
-}
-
 type ReadinessProbe interface {
 	Ready(runnable Runnable) (err error)
 }
@@ -353,5 +326,6 @@ func NewCmdReadinessProbe(cmd Command) *CmdReadinessProbe {
 }
 
 func (p *CmdReadinessProbe) Ready(runnable Runnable) error {
-	return runnable.Exec(p.cmd)
+	_, _, err := runnable.Exec(p.cmd)
+	return err
 }

env_docker.go

@@ -4,10 +4,12 @@
 package e2e
 
 import (
+	"bytes"
 	"context"
 	"crypto/rand"
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
 	"os/exec"
@@ -176,10 +178,11 @@ func (e Errorer) Name() string                              { return e.name }
 func (Errorer) Dir() string                                 { return "" }
 func (Errorer) InternalDir() string                         { return "" }
 func (e Errorer) Start() error                              { return e.err }
+func (e Errorer) RunOnce(context.Context) (string, error)   { return "", nil }
 func (e Errorer) WaitReady() error                          { return e.err }
 func (e Errorer) Kill() error                               { return e.err }
 func (e Errorer) Stop() error                               { return e.err }
-func (e Errorer) Exec(Command, ...ExecOption) error         { return e.err }
+func (e Errorer) Exec(Command) (string, string, error)      { return "", "", e.err }
 func (Errorer) Endpoint(string) string                      { return "" }
 func (Errorer) InternalEndpoint(string) string              { return "" }
 func (Errorer) IsRunning() bool                             { return false }
@@ -345,6 +348,37 @@ func (d *dockerRunnable) IsRunning() bool {
 	return d.usedNetworkName != ""
 }
 
+func (d *dockerRunnable) RunOnce(ctx context.Context) (output string, err error) {
+	if d.IsRunning() {
+		return "", errors.Errorf("%v is running. Stop or kill it first to restart.", d.Name())
+	}
+
+	i, ok := d.concreteType.(identificable)
+	if !ok {
+		return "", errors.Errorf("concrete type has at least embed runnable or future runnable instance provided by Runnable builder, got %T; not implementing identificable", d.concreteType)
+	}
+	if i.id() != d.id() {
+		return "", errors.Errorf("concrete type has at least embed runnable or future runnable instance provided by Runnable builder, got %T; id %v, expected %v", d.concreteType, i.id(), d.id())
+	}
+
+	d.logger.Log("Starting", d.Name())
+
+	// Make sure the image is available locally; if not wait for it to download.
+	if err = d.prePullImage(ctx); err != nil {
+		return "", err
+	}
+
+	cmd := d.env.execContext(ctx, "docker", append([]string{"run", "-t"}, d.env.buildDockerRunArgs(d.name, d.ports, d.opts)...)...)
+	out := bytes.Buffer{}
+	l := &LinePrefixLogger{prefix: d.Name() + ": ", logger: d.logger}
+	ml := io.MultiWriter(&out, l)
+
+	cmd.Stdout = ml
+	cmd.Stderr = ml
+	err = cmd.Run()
+	return strings.TrimSuffix(strings.TrimSuffix(out.String(), "\n"), "\r"), err
+}
+
 // Start starts runnable.
 func (d *dockerRunnable) Start() (err error) {
 	if d.IsRunning() {
@@ -472,7 +506,7 @@ func (d *dockerRunnable) Endpoint(portName string) string {
 		return ""
 	}
 
-	// Do not use "localhost", because it doesn't work with the AWS DynamoDB client.
+	// Do not use "localhost" cause it doesn't work with the AWS DynamoDB client.
 	return fmt.Sprintf("127.0.0.1:%d", localPort)
 }
 
@@ -599,26 +633,27 @@ func (d *dockerRunnable) WaitReady() (err error) {
 	return errors.Wrapf(err, "the service %s is not ready", d.Name())
 }
 
-// Exec runs the provided command against the docker container specified by this
-// service.
-func (d *dockerRunnable) Exec(command Command, opts ...ExecOption) error {
+// Exec runs the provided command against a the docker container specified by this
+// service. It returns the stdout, stderr, and error response from attempting
+// to run the command.
+func (d *dockerRunnable) Exec(command Command) (string, string, error) {
 	if !d.IsRunning() {
-		return errors.Errorf("service %s is stopped", d.Name())
-	}
-
-	l := &LinePrefixLogger{prefix: d.Name() + "-exec: ", logger: d.logger}
-	o := ExecOptions{Stdout: l, Stderr: l}
-	for _, opt := range opts {
-		opt(&o)
+		return "", "", errors.Errorf("service %s is stopped", d.Name())
 	}
 
 	args := []string{"exec", d.containerName()}
 	args = append(args, command.Cmd)
 	args = append(args, command.Args...)
 	cmd := d.env.exec("docker", args...)
-	cmd.Stdout = o.Stdout
-	cmd.Stderr = o.Stderr
-	return cmd.Run()
+
+	var stdout bytes.Buffer
+	cmd.Stdout = &stdout
+
+	var stderr bytes.Buffer
+	cmd.Stderr = &stderr
+
+	err := cmd.Run()
+	return stdout.String(), stderr.String(), err
 }
 
 func (e *DockerEnvironment) existDockerNetwork() (bool, error) {

env_docker_ext_test.go

@@ -4,7 +4,7 @@
 package e2e_test
 
 import (
-	"bytes"
+	"context"
 	"io/ioutil"
 	"net/http"
 	"path/filepath"
@@ -39,7 +39,8 @@ func TestDockerEnvironment(t *testing.T) {
 	testutil.Ok(t, p1.Stop())
 	testutil.Ok(t, p1.Kill())
 
-	testutil.NotOk(t, p1.Exec(wgetFlagsCmd("localhost:9090")))
+	_, _, err = p1.Exec(wgetFlagsCmd("localhost:9090"))
+	testutil.NotOk(t, err)
 	testutil.Equals(t, "stopped", p1.Endpoint("http"))
 	testutil.Equals(t, "stopped", p1.Endpoint("not-existing"))
 
@@ -61,9 +62,9 @@ func TestDockerEnvironment(t *testing.T) {
 		expectedFlagsOutputProm1 = "{\"status\":\"success\",\"data\":{\"alertmanager.notification-queue-capacity\":\"10000\",\"alertmanager.timeout\":\"\",\"config.file\":\"/shared/data/prometheus-1/prometheus.yml\",\"enable-feature\":\"\",\"log.format\":\"logfmt\",\"log.level\":\"info\",\"query.lookback-delta\":\"5m\",\"query.max-concurrency\":\"20\",\"query.max-samples\":\"50000000\",\"query.timeout\":\"2m\",\"rules.alert.for-grace-period\":\"10m\",\"rules.alert.for-outage-tolerance\":\"1h\",\"rules.alert.resend-delay\":\"1m\",\"scrape.adjust-timestamps\":\"true\",\"storage.exemplars.exemplars-limit\":\"0\",\"storage.remote.flush-deadline\":\"1m\",\"storage.remote.read-concurrent-limit\":\"10\",\"storage.remote.read-max-bytes-in-frame\":\"1048576\",\"storage.remote.read-sample-limit\":\"50000000\",\"storage.tsdb.allow-overlapping-blocks\":\"false\",\"storage.tsdb.max-block-chunk-segment-size\":\"0B\",\"storage.tsdb.max-block-duration\":\"2h\",\"storage.tsdb.min-block-duration\":\"2h\",\"storage.tsdb.no-lockfile\":\"false\",\"storage.tsdb.path\":\"/shared/data/prometheus-1\",\"storage.tsdb.retention\":\"0s\",\"storage.tsdb.retention.size\":\"0B\",\"storage.tsdb.retention.time\":\"0s\",\"storage.tsdb.wal-compression\":\"true\",\"storage.tsdb.wal-segment-size\":\"0B\",\"web.config.file\":\"\",\"web.console.libraries\":\"console_libraries\",\"web.console.templates\":\"consoles\",\"web.cors.origin\":\".*\",\"web.enable-admin-api\":\"false\",\"web.enable-lifecycle\":\"false\",\"web.external-url\":\"\",\"web.listen-address\":\":9090\",\"web.max-connections\":\"512\",\"web.page-title\":\"Prometheus Time Series Collection and Processing Server\",\"web.read-timeout\":\"5m\",\"web.route-prefix\":\"/\",\"web.user-assets\":\"\"}}"
 		expectedFlagsOutputProm2 = "{\"status\":\"success\",\"data\":{\"alertmanager.notification-queue-capacity\":\"10000\",\"alertmanager.timeout\":\"\",\"config.file\":\"/shared/data/prometheus-2/prometheus.yml\",\"enable-feature\":\"\",\"log.format\":\"logfmt\",\"log.level\":\"info\",\"query.lookback-delta\":\"5m\",\"query.max-concurrency\":\"20\",\"query.max-samples\":\"50000000\",\"query.timeout\":\"2m\",\"rules.alert.for-grace-period\":\"10m\",\"rules.alert.for-outage-tolerance\":\"1h\",\"rules.alert.resend-delay\":\"1m\",\"scrape.adjust-timestamps\":\"true\",\"storage.exemplars.exemplars-limit\":\"0\",\"storage.remote.flush-deadline\":\"1m\",\"storage.remote.read-concurrent-limit\":\"10\",\"storage.remote.read-max-bytes-in-frame\":\"1048576\",\"storage.remote.read-sample-limit\":\"50000000\",\"storage.tsdb.allow-overlapping-blocks\":\"false\",\"storage.tsdb.max-block-chunk-segment-size\":\"0B\",\"storage.tsdb.max-block-duration\":\"2h\",\"storage.tsdb.min-block-duration\":\"2h\",\"storage.tsdb.no-lockfile\":\"false\",\"storage.tsdb.path\":\"/shared/data/prometheus-2\",\"storage.tsdb.retention\":\"0s\",\"storage.tsdb.retention.size\":\"0B\",\"storage.tsdb.retention.time\":\"0s\",\"storage.tsdb.wal-compression\":\"true\",\"storage.tsdb.wal-segment-size\":\"0B\",\"web.config.file\":\"\",\"web.console.libraries\":\"console_libraries\",\"web.console.templates\":\"consoles\",\"web.cors.origin\":\".*\",\"web.enable-admin-api\":\"false\",\"web.enable-lifecycle\":\"false\",\"web.external-url\":\"\",\"web.listen-address\":\":9090\",\"web.max-connections\":\"512\",\"web.page-title\":\"Prometheus Time Series Collection and Processing Server\",\"web.read-timeout\":\"5m\",\"web.route-prefix\":\"/\",\"web.user-assets\":\"\"}}"
 	)
-	var out bytes.Buffer
-	testutil.Ok(t, p1.Exec(wgetFlagsCmd("localhost:9090"), e2e.WithExecOptionStdout(&out)))
-	testutil.Equals(t, expectedFlagsOutputProm1, out.String())
+	out, errout, err := p1.Exec(wgetFlagsCmd("localhost:9090"))
+	testutil.Ok(t, err, errout)
+	testutil.Equals(t, expectedFlagsOutputProm1, out)
 
 	resp, err := http.Get("http://" + p1.Endpoint("http") + "/api/v1/status/flags")
 	testutil.Ok(t, err)
@@ -75,19 +76,18 @@ func TestDockerEnvironment(t *testing.T) {
 	testutil.Equals(t, "", p1.Endpoint("not-existing"))
 
 	// Now try the same but cross containers.
-	out.Reset()
-	testutil.Ok(t, p1.Exec(wgetFlagsCmd(p2.InternalEndpoint("http")), e2e.WithExecOptionStdout(&out)))
-	testutil.Equals(t, expectedFlagsOutputProm2, out.String())
+	out, errout, err = p1.Exec(wgetFlagsCmd(p2.InternalEndpoint("http")))
+	testutil.Ok(t, err, errout)
+	testutil.Equals(t, expectedFlagsOutputProm2, out)
 
 	testutil.NotOk(t, p1.Start()) // Starting ok, should fail.
 
-	// Batch job example and test.
-	batch := e.Runnable("batch").Init(e2e.StartOptions{Image: "ubuntu:20.04", Command: e2e.NewCommandRunUntilStop()})
-	testutil.Ok(t, batch.Start())
+	// Batch job.
+	batch := e.Runnable("batch").Init(e2e.StartOptions{Image: "ubuntu:20.04", Command: e2e.NewCommandWithoutEntrypoint("echo", "yolo")})
 	for i := 0; i < 3; i++ {
-		out.Reset()
-		testutil.Ok(t, batch.Exec(e2e.NewCommand("echo", "yolo"), e2e.WithExecOptionStdout(&out)))
-		testutil.Equals(t, "yolo\n", out.String())
+		out, err := batch.RunOnce(context.Background())
+		testutil.Ok(t, err)
+		testutil.Equals(t, "yolo", out)
 	}
 
 	e.Close()

examples/thanos/unittest_test.go

@@ -4,7 +4,6 @@
 package main
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"testing"
@@ -83,12 +82,4 @@ up{instance="%v", job="myself", prometheus="prometheus-2"} => 1 @[%v]`, p1.Inter
 			v.String(),
 		)
 	}
-
-	// Batch job example.
-	batch := e.Runnable("batch").Init(e2e.StartOptions{Image: "ubuntu:20.04", Command: e2e.NewCommandRunUntilStop()})
-	testutil.Ok(t, batch.Start())
-
-	var out bytes.Buffer
-	testutil.Ok(t, batch.Exec(e2e.NewCommand("echo", "it works"), e2e.WithExecOptionStdout(&out)))
-	testutil.Equals(t, "it works\n", out.String())
 }