Streamlined db, added thanos to e2edb too. (#4)

Signed-off-by: Bartlomiej Plotka <[email protected]>

README.md

@@ -58,14 +58,12 @@ Let's go through an example leveraging `go test` flow:
 
 3. 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,87p' 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, err := e2edb.NewPrometheus(e, "prometheus-1")
-   	testutil.Ok(t, err)
+   	p1 := e2edb.NewPrometheus(e, "prometheus-1")
    	s1 := newThanosSidecar(e, "sidecar-1", p1)
 
-   	p2, err := e2edb.NewPrometheus(e, "prometheus-2")
-   	testutil.Ok(t, err)
+   	p2 := e2edb.NewPrometheus(e, "prometheus-2")
    	s2 := newThanosSidecar(e, "sidecar-2", p2)
 
    	// Create Thanos Query container. We can point the peer network addresses of both Prometheus instance

cli.go

@@ -3,6 +3,8 @@
 
 package e2e
 
+import "strings"
+
 func EmptyFlags() map[string]string {
 	return map[string]string{}
 }
@@ -35,9 +37,27 @@ func BuildArgs(flags map[string]string) []string {
 	for name, value := range flags {
 		if value != "" {
 			args = append(args, name+"="+value)
-		} else {
-			args = append(args, name)
+			continue
+		}
+		args = append(args, name)
+	}
+	return args
+}
+
+// BuildKingpinArgs is like BuildArgs but with special handling of slice args.
+// NOTE(bwplotka): flags with values as comma but not indented to be slice will cause issues.
+func BuildKingpinArgs(flags map[string]string) []string {
+	args := make([]string, 0, len(flags))
+
+	for name, value := range flags {
+		if value != "" {
+			s := strings.Split(value, ",")
+			for _, ss := range s {
+				args = append(args, name+"="+ss)
+			}
+			continue
 		}
+		args = append(args, name)
 	}
 	return args
 }

db/db.go

@@ -7,6 +7,9 @@ package e2edb
 
 import (
 	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
 	"strings"
 
 	"github.com/efficientgo/e2e"
@@ -20,7 +23,8 @@ const (
 type Option func(*options)
 
 type options struct {
-	image string
+	image        string
+	flagOverride map[string]string
 }
 
 func WithImage(image string) Option {
@@ -29,42 +33,49 @@ func WithImage(image string) Option {
 	}
 }
 
+func WithFlagOverride(ov map[string]string) Option {
+	return func(o *options) {
+		o.flagOverride = ov
+	}
+}
+
 const AccessPortName = "http"
 
 // NewMinio returns minio server, used as a local replacement for S3.
 func NewMinio(env e2e.Environment, name, bktName string, opts ...Option) *e2e.InstrumentedRunnable {
-	o := options{image: "minio/minio:RELEASE.2019-12-30T05-45-39Z"}
+	o := options{image: "minio/minio:RELEASE.2021-07-27T02-40-15Z"}
 	for _, opt := range opts {
 		opt(&o)
 	}
 
-	minioKESGithubContent := "https://raw.githubusercontent.com/minio/kes/master"
-	commands := []string{
-		"curl -sSL --tlsv1.2 -O '%s/root.key'	-O '%s/root.cert'",
-		"mkdir -p /data/%s && minio server --address :%v --quiet /data",
+	userID := strconv.Itoa(os.Getuid())
+	ports := map[string]int{AccessPortName: 8090}
+	envVars := []string{
+		"MINIO_ROOT_USER=" + MinioAccessKey,
+		"MINIO_ROOT_PASSWORD=" + MinioSecretKey,
+		"MINIO_BROWSER=" + "off",
+		"ENABLE_HTTPS=" + "0",
+		// https://docs.min.io/docs/minio-kms-quickstart-guide.html
+		"MINIO_KMS_KES_ENDPOINT=" + "https://play.min.io:7373",
+		"MINIO_KMS_KES_KEY_FILE=" + "root.key",
+		"MINIO_KMS_KES_CERT_FILE=" + "root.cert",
+		"MINIO_KMS_KES_KEY_NAME=" + "my-minio-key",
 	}
-
-	return e2e.NewInstrumentedRunnable(
-		env,
-		name,
-		map[string]int{AccessPortName: 8090},
-		AccessPortName,
+	f := e2e.NewFutureInstrumentedRunnable(env, name, ports, AccessPortName)
+	return f.Init(
 		e2e.StartOptions{
 			Image: o.image,
 			// Create the required bucket before starting minio.
-			Command:   e2e.NewCommandWithoutEntrypoint("sh", "-c", fmt.Sprintf(strings.Join(commands, " && "), minioKESGithubContent, minioKESGithubContent, bktName, 8090)),
-			Readiness: e2e.NewHTTPReadinessProbe(AccessPortName, "/minio/health/ready", 200, 200),
-			EnvVars: map[string]string{
-				"MINIO_ACCESS_KEY": MinioAccessKey,
-				"MINIO_SECRET_KEY": MinioSecretKey,
-				"MINIO_BROWSER":    "off",
-				"ENABLE_HTTPS":     "0",
-				// https://docs.min.io/docs/minio-kms-quickstart-guide.html
-				"MINIO_KMS_KES_ENDPOINT":  "https://play.min.io:7373",
-				"MINIO_KMS_KES_KEY_FILE":  "root.key",
-				"MINIO_KMS_KES_CERT_FILE": "root.cert",
-				"MINIO_KMS_KES_KEY_NAME":  "my-minio-key",
-			},
+			Command: e2e.NewCommandWithoutEntrypoint("sh", "-c", fmt.Sprintf(
+				// Hacky: Create user that matches ID with host ID to be able to remove .minio.sys details on the start.
+				// Proper solution would be to contribute/create our own minio image which is non root.
+				"useradd -G root -u %v me && mkdir -p %s && chown -R me %s &&"+
+					"curl -sSL --tlsv1.2 -O 'https://raw.githubusercontent.com/minio/kes/master/root.key' -O 'https://raw.githubusercontent.com/minio/kes/master/root.cert' && "+
+					"cp root.* /home/me/ && "+
+					"su - me -s /bin/sh -c 'mkdir -p %s && %s minio server --address :%v --quiet %v'",
+				userID, f.InternalDir(), f.InternalDir(), filepath.Join(f.InternalDir(), bktName), strings.Join(envVars, " "), ports[AccessPortName], f.InternalDir()),
+			),
+			Readiness: e2e.NewHTTPReadinessProbe(AccessPortName, "/minio/health/live", 200, 200),
 		},
 	)
 }
@@ -75,6 +86,7 @@ func NewConsul(env e2e.Environment, name string, opts ...Option) *e2e.Instrument
 		opt(&o)
 	}
 
+	e2e.MergeFlags()
 	return e2e.NewInstrumentedRunnable(
 		env,
 		name,

db/prometheus.go

@@ -18,7 +18,7 @@ type Prometheus struct {
 	*e2e.InstrumentedRunnable
 }
 
-func NewPrometheus(env e2e.Environment, name string, opts ...Option) (*Prometheus, error) {
+func NewPrometheus(env e2e.Environment, name string, opts ...Option) *Prometheus {
 	o := options{image: "quay.io/prometheus/prometheus:v2.27.0"}
 	for _, opt := range opts {
 		opt(&o)
@@ -27,10 +27,6 @@ func NewPrometheus(env e2e.Environment, name string, opts ...Option) (*Prometheu
 	ports := map[string]int{"http": 9090}
 
 	f := e2e.NewFutureInstrumentedRunnable(env, name, ports, "http")
-	if err := os.MkdirAll(f.Dir(), 0750); err != nil {
-		return nil, errors.Wrap(err, "create prometheus dir")
-	}
-
 	config := fmt.Sprintf(`
 global:
   external_labels:
@@ -48,23 +44,26 @@ scrape_configs:
     action: drop
 `, name, f.InternalEndpoint("http"))
 	if err := ioutil.WriteFile(filepath.Join(f.Dir(), "prometheus.yml"), []byte(config), 0600); err != nil {
-		return nil, errors.Wrap(err, "creating prom config failed")
+		return &Prometheus{InstrumentedRunnable: e2e.NewErrInstrumentedRunnable(name, errors.Wrap(err, "create prometheus config failed"))}
 	}
 
-	args := e2e.BuildArgs(map[string]string{
+	args := map[string]string{
 		"--config.file":                     filepath.Join(f.InternalDir(), "prometheus.yml"),
 		"--storage.tsdb.path":               f.InternalDir(),
-		"--storage.tsdb.max-block-duration": "2h",
+		"--storage.tsdb.max-block-duration": "2h", // No compaction - mostly not needed for quick test.
 		"--log.level":                       "info",
 		"--web.listen-address":              fmt.Sprintf(":%d", ports["http"]),
-	})
+	}
+	if o.flagOverride != nil {
+		args = e2e.MergeFlagsWithoutRemovingEmpty(args, o.flagOverride)
+	}
 
 	return &Prometheus{InstrumentedRunnable: f.Init(e2e.StartOptions{
 		Image:     o.image,
-		Command:   e2e.NewCommandWithoutEntrypoint("prometheus", args...),
+		Command:   e2e.NewCommandWithoutEntrypoint("prometheus", e2e.BuildArgs(args)...),
 		Readiness: e2e.NewHTTPReadinessProbe("http", "/-/ready", 200, 200),
 		User:      strconv.Itoa(os.Getuid()),
-	})}, nil
+	})}
 }
 
 func (p *Prometheus) SetConfig(config string) error {

db/thanos.go

@@ -0,0 +1,113 @@
+// Copyright (c) The EfficientGo Authors.
+// Licensed under the Apache License 2.0.
+
+package e2edb
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+
+	"github.com/efficientgo/e2e"
+)
+
+func NewThanosQuerier(env e2e.Environment, name string, endpointsAddresses []string, opts ...Option) *e2e.InstrumentedRunnable {
+	o := options{image: "quay.io/thanos/thanos:v0.21.1"}
+	for _, opt := range opts {
+		opt(&o)
+	}
+
+	ports := map[string]int{
+		"http": 9090,
+		"grpc": 9091,
+	}
+
+	args := map[string]string{
+		"--debug.name":           name,
+		"--grpc-address":         fmt.Sprintf(":%d", ports["grpc"]),
+		"--http-address":         fmt.Sprintf(":%d", ports["http"]),
+		"--query.replica-label":  "replica",
+		"--log.level":            "info",
+		"--query.max-concurrent": "1",
+	}
+	if len(endpointsAddresses) > 0 {
+		args["--store"] = strings.Join(endpointsAddresses, ",")
+	}
+	if o.flagOverride != nil {
+		args = e2e.MergeFlagsWithoutRemovingEmpty(args, o.flagOverride)
+	}
+
+	return e2e.NewInstrumentedRunnable(env, name, ports, "http", e2e.StartOptions{
+		Image:     o.image,
+		Command:   e2e.NewCommand("query", e2e.BuildKingpinArgs(args)...),
+		Readiness: e2e.NewHTTPReadinessProbe("http", "/-/ready", 200, 200),
+		User:      strconv.Itoa(os.Getuid()),
+	})
+}
+
+func NewThanosSidecar(env e2e.Environment, name string, prom e2e.Linkable, opts ...Option) *e2e.InstrumentedRunnable {
+	o := options{image: "quay.io/thanos/thanos:v0.21.1"}
+	for _, opt := range opts {
+		opt(&o)
+	}
+
+	ports := map[string]int{
+		"http": 9090,
+		"grpc": 9091,
+	}
+
+	args := map[string]string{
+		"--debug.name":     name,
+		"--grpc-address":   fmt.Sprintf(":%d", ports["grpc"]),
+		"--http-address":   fmt.Sprintf(":%d", ports["http"]),
+		"--prometheus.url": "http://" + prom.InternalEndpoint(AccessPortName),
+		"--log.level":      "info",
+	}
+	if o.flagOverride != nil {
+		args = e2e.MergeFlagsWithoutRemovingEmpty(args, o.flagOverride)
+	}
+
+	return e2e.NewInstrumentedRunnable(env, name, ports, "http", e2e.StartOptions{
+		Image:     o.image,
+		Command:   e2e.NewCommand("sidecar", e2e.BuildKingpinArgs(args)...),
+		Readiness: e2e.NewHTTPReadinessProbe("http", "/-/ready", 200, 200),
+		User:      strconv.Itoa(os.Getuid()),
+	})
+}
+
+func NewThanosStore(env e2e.Environment, name string, bktConfigYaml []byte, opts ...Option) *e2e.InstrumentedRunnable {
+	o := options{image: "quay.io/thanos/thanos:v0.21.1"}
+	for _, opt := range opts {
+		opt(&o)
+	}
+
+	ports := map[string]int{
+		"http": 9090,
+		"grpc": 9091,
+	}
+
+	f := e2e.NewFutureInstrumentedRunnable(env, name, ports, "http")
+	args := map[string]string{
+		"--debug.name":      name,
+		"--grpc-address":    fmt.Sprintf(":%d", ports["grpc"]),
+		"--http-address":    fmt.Sprintf(":%d", ports["http"]),
+		"--log.level":       "info",
+		"--data-dir":        f.InternalDir(),
+		"--objstore.config": string(bktConfigYaml),
+		// Accelerated sync time for quicker test (3m by default).
+		//"--sync-block-duration":               "3s",
+		"--block-sync-concurrency":            "1",
+		"--store.grpc.series-max-concurrency": "1",
+		"--consistency-delay":                 "30m",
+	}
+	if o.flagOverride != nil {
+		args = e2e.MergeFlagsWithoutRemovingEmpty(args, o.flagOverride)
+	}
+	return f.Init(e2e.StartOptions{
+		Image:     o.image,
+		Command:   e2e.NewCommand("store", e2e.BuildKingpinArgs(args)...),
+		Readiness: e2e.NewHTTPReadinessProbe("http", "/-/ready", 200, 200),
+		User:      strconv.Itoa(os.Getuid()),
+	})
+}

env_docker.go

@@ -105,6 +105,10 @@ func (e *DockerEnvironment) FutureRunnable(name string, ports map[string]int) Fu
 		logger:    e.logger,
 		hostPorts: map[string]int{},
 	}
+	if err := os.MkdirAll(d.Dir(), 0750); err != nil {
+		return ErrRunnable{name: name, err: err}
+	}
+
 	e.register(name)
 	return d
 }

env_docker_ext_test.go

@@ -25,9 +25,7 @@ func TestDockerEnvironment(t *testing.T) {
 	testutil.Ok(t, err)
 	t.Cleanup(e.Close)
 
-	p1, err := e2edb.NewPrometheus(e, "prometheus-1")
-	testutil.Ok(t, err)
-
+	p1 := e2edb.NewPrometheus(e, "prometheus-1")
 	testutil.Equals(t, "prometheus-1", p1.Name())
 	testutil.Equals(t, filepath.Join(e.SharedDir(), "data", p1.Name()), p1.Dir())
 	testutil.Equals(t, filepath.Join("/shared", "data", p1.Name()), p1.InternalDir())
@@ -45,9 +43,7 @@ func TestDockerEnvironment(t *testing.T) {
 	testutil.Equals(t, "stopped", p1.Endpoint("http"))
 	testutil.Equals(t, "stopped", p1.Endpoint("not-existing"))
 
-	p2, err := e2edb.NewPrometheus(e, "prometheus-2")
-	testutil.Ok(t, err)
-
+	p2 := e2edb.NewPrometheus(e, "prometheus-2")
 	testutil.Ok(t, e2e.StartAndWaitReady(p1, p2))
 	testutil.Ok(t, p1.WaitReady())
 	testutil.Ok(t, p1.WaitReady())
@@ -86,6 +82,6 @@ func TestDockerEnvironment(t *testing.T) {
 	testutil.NotOk(t, p1.Start()) // Starting ok, should fail.
 
 	e.Close()
-	_, err = e2edb.NewPrometheus(e, "prometheus-3") // Should fail.
-	testutil.NotOk(t, err)
+	afterClose := e2edb.NewPrometheus(e, "prometheus-3") // Should fail.
+	testutil.NotOk(t, afterClose.Start())
 }

examples/thanos/standalone.go

@@ -73,16 +73,10 @@ func deploy(ctx context.Context) error {
 	defer e.Close()
 
 	// Create structs for Prometheus containers scraping itself.
-	p1, err := e2edb.NewPrometheus(e, "prometheus-1")
-	if err != nil {
-		return err
-	}
+	p1 := e2edb.NewPrometheus(e, "prometheus-1")
 	s1 := newThanosSidecar(e, "sidecar-1", p1)
 
-	p2, err := e2edb.NewPrometheus(e, "prometheus-2")
-	if err != nil {
-		return err
-	}
+	p2 := e2edb.NewPrometheus(e, "prometheus-2")
 	s2 := newThanosSidecar(e, "sidecar-2", p2)
 
 	// Create Thanos Query container. We can point the peer network addresses of both Prometheus instance