Removed RunOnce, extended Exec. (#32)

Fixes #31

Signed-off-by: Bartlomiej Plotka <bwplotka@gmail.com>

Author: bwplotka

README.md

@@ -27,14 +27,14 @@ 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:
+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:
 
    ```go mdox-exec="sed -n '35,42p' examples/thanos/standalone.go"
    	// Setup Jaeger for example purposes, on how easy is to setup tracing pipeline in e2e framework.
@@ -49,7 +49,8 @@ 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,86p' examples/thanos/unittest_test.go"
+   ```go mdox-exec="sed -n '28,93p' 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)
@@ -77,16 +78,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.
@@ -97,17 +98,24 @@ 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,8 +72,7 @@ func (p *Prometheus) SetConfig(config string) error {
 	}
 
 	if p.IsRunning() {
-		_, _, err := p.Exec(e2e.NewCommand("kill", "-SIGHUP", "1"))
-		return err
+		return p.Exec(e2e.NewCommand("kill", "-SIGHUP", "1"))
 	}
 	return nil
 }

env.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -137,9 +138,6 @@ 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
@@ -152,9 +150,10 @@ 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 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)
+	// 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
 
 	// 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.
@@ -163,6 +162,29 @@ 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
@@ -217,6 +239,11 @@ 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)
 }
@@ -326,6 +353,5 @@ func NewCmdReadinessProbe(cmd Command) *CmdReadinessProbe {
 }
 
 func (p *CmdReadinessProbe) Ready(runnable Runnable) error {
-	_, _, err := runnable.Exec(p.cmd)
-	return err
+	return runnable.Exec(p.cmd)
 }

env_docker.go

@@ -4,12 +4,10 @@
 package e2e
 
 import (
-	"bytes"
 	"context"
 	"crypto/rand"
 	"encoding/json"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"os"
 	"os/exec"
@@ -178,11 +176,10 @@ 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) (string, string, error)      { return "", "", e.err }
+func (e Errorer) Exec(Command, ...ExecOption) error         { return e.err }
 func (Errorer) Endpoint(string) string                      { return "" }
 func (Errorer) InternalEndpoint(string) string              { return "" }
 func (Errorer) IsRunning() bool                             { return false }
@@ -348,37 +345,6 @@ 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() {
@@ -506,7 +472,7 @@ func (d *dockerRunnable) Endpoint(portName string) string {
 		return ""
 	}
 
-	// Do not use "localhost" cause it doesn't work with the AWS DynamoDB client.
+	// Do not use "localhost", because it doesn't work with the AWS DynamoDB client.
 	return fmt.Sprintf("127.0.0.1:%d", localPort)
 }
 
@@ -633,27 +599,26 @@ func (d *dockerRunnable) WaitReady() (err error) {
 	return errors.Wrapf(err, "the service %s is not ready", d.Name())
 }
 
-// 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) {
+// Exec runs the provided command against the docker container specified by this
+// service.
+func (d *dockerRunnable) Exec(command Command, opts ...ExecOption) error {
 	if !d.IsRunning() {
-		return "", "", errors.Errorf("service %s is stopped", d.Name())
+		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)
 	}
 
 	args := []string{"exec", d.containerName()}
 	args = append(args, command.Cmd)
 	args = append(args, command.Args...)
 	cmd := d.env.exec("docker", args...)
-
-	var stdout bytes.Buffer
-	cmd.Stdout = &stdout
-
-	var stderr bytes.Buffer
-	cmd.Stderr = &stderr
-
-	err := cmd.Run()
-	return stdout.String(), stderr.String(), err
+	cmd.Stdout = o.Stdout
+	cmd.Stderr = o.Stderr
+	return cmd.Run()
 }
 
 func (e *DockerEnvironment) existDockerNetwork() (bool, error) {

env_docker_ext_test.go

@@ -4,7 +4,7 @@
 package e2e_test
 
 import (
-	"context"
+	"bytes"
 	"io/ioutil"
 	"net/http"
 	"path/filepath"
@@ -39,8 +39,7 @@ func TestDockerEnvironment(t *testing.T) {
 	testutil.Ok(t, p1.Stop())
 	testutil.Ok(t, p1.Kill())
 
-	_, _, err = p1.Exec(wgetFlagsCmd("localhost:9090"))
-	testutil.NotOk(t, err)
+	testutil.NotOk(t, p1.Exec(wgetFlagsCmd("localhost:9090")))
 	testutil.Equals(t, "stopped", p1.Endpoint("http"))
 	testutil.Equals(t, "stopped", p1.Endpoint("not-existing"))
 
@@ -62,9 +61,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\":\"\"}}"
 	)
-	out, errout, err := p1.Exec(wgetFlagsCmd("localhost:9090"))
-	testutil.Ok(t, err, errout)
-	testutil.Equals(t, expectedFlagsOutputProm1, out)
+	var out bytes.Buffer
+	testutil.Ok(t, p1.Exec(wgetFlagsCmd("localhost:9090"), e2e.WithExecOptionStdout(&out)))
+	testutil.Equals(t, expectedFlagsOutputProm1, out.String())
 
 	resp, err := http.Get("http://" + p1.Endpoint("http") + "/api/v1/status/flags")
 	testutil.Ok(t, err)
@@ -76,18 +75,19 @@ func TestDockerEnvironment(t *testing.T) {
 	testutil.Equals(t, "", p1.Endpoint("not-existing"))
 
 	// Now try the same but cross containers.
-	out, errout, err = p1.Exec(wgetFlagsCmd(p2.InternalEndpoint("http")))
-	testutil.Ok(t, err, errout)
-	testutil.Equals(t, expectedFlagsOutputProm2, out)
+	out.Reset()
+	testutil.Ok(t, p1.Exec(wgetFlagsCmd(p2.InternalEndpoint("http")), e2e.WithExecOptionStdout(&out)))
+	testutil.Equals(t, expectedFlagsOutputProm2, out.String())
 
 	testutil.NotOk(t, p1.Start()) // Starting ok, should fail.
 
-	// Batch job.
-	batch := e.Runnable("batch").Init(e2e.StartOptions{Image: "ubuntu:20.04", Command: e2e.NewCommandWithoutEntrypoint("echo", "yolo")})
+	// Batch job example and test.
+	batch := e.Runnable("batch").Init(e2e.StartOptions{Image: "ubuntu:20.04", Command: e2e.NewCommandRunUntilStop()})
+	testutil.Ok(t, batch.Start())
 	for i := 0; i < 3; i++ {
-		out, err := batch.RunOnce(context.Background())
-		testutil.Ok(t, err)
-		testutil.Equals(t, "yolo", out)
+		out.Reset()
+		testutil.Ok(t, batch.Exec(e2e.NewCommand("echo", "yolo"), e2e.WithExecOptionStdout(&out)))
+		testutil.Equals(t, "yolo\n", out.String())
 	}
 
 	e.Close()

examples/thanos/unittest_test.go

@@ -4,6 +4,7 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"testing"
@@ -82,4 +83,12 @@ 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())
 }