e2e tests: Move to efficientgo/e2e

.gitignore

@@ -12,7 +12,7 @@ kube/.minikube
 
 # Ignore e2e working dirs.
 data/
-test/e2e/e2e_integration_test*
+test/e2e/e2e_*
 
 # Ignore promu artifacts.
 /.build

go.mod

@@ -20,6 +20,7 @@ require (
  github.com/cortexproject/cortex v1.10.1-0.20210820081236-70dddb6b70b8
  github.com/davecgh/go-spew v1.1.1
  github.com/efficientgo/e2e v0.11.1-0.20210829161758-f4cc6dbdc6ea
+ github.com/efficientgo/tools/core v0.0.0-20210129205121-421d0828c9a6
  github.com/efficientgo/tools/extkingpin v0.0.0-20210609125236-d73259166f20
  github.com/facette/natsort v0.0.0-20181210072756-2cd4dd1e2dcb
  github.com/fatih/structtag v1.1.0
@@ -33,7 +34,7 @@ require (
  github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da
  github.com/golang/snappy v0.0.4
  github.com/googleapis/gax-go v2.0.2+incompatible
- github.com/grafana/dskit v0.0.0-20210819132858-471020752967
+ github.com/grafana/dskit v0.0.0-20210819132858-471020752967 // indirect
  github.com/grpc-ecosystem/go-grpc-middleware/providers/kit/v2 v2.0.0-20201002093600-73cf2ae9d891
  github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.0-rc.2.0.20201207153454-9f6bf00c00a7
  github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0

pkg/objstore/s3/s3_e2e_test.go

@@ -9,8 +9,8 @@ import (
  "strings"
  "testing"
 
- "github.com/cortexproject/cortex/integration/e2e"
- e2edb "github.com/cortexproject/cortex/integration/e2e/db"
+ "github.com/efficientgo/e2e"
+ e2edb "github.com/efficientgo/e2e/db"
  "github.com/go-kit/kit/log"
  "github.com/thanos-io/thanos/pkg/objstore/s3"
  "github.com/thanos-io/thanos/test/e2e/e2ethanos"
@@ -23,19 +23,19 @@ func BenchmarkUpload(b *testing.B) {
  b.ReportAllocs()
  ctx := context.Background()
 
- s, err := e2e.NewScenario("e2e_bench_mino_client")
+ e, err := e2e.NewDockerEnvironment("e2e_bench_mino_client")
  testutil.Ok(b, err)
- b.Cleanup(e2ethanos.CleanScenario(b, s))
+ b.Cleanup(e2ethanos.CleanScenario(b, e))
 
- const bucket = "test"
- m := e2edb.NewMinio(8080, bucket)
- testutil.Ok(b, s.StartAndWaitReady(m))
+ const bucket = "benchmark"
+ m := e2ethanos.NewMinio(e, "benchmark", bucket)
+ testutil.Ok(b, e2e.StartAndWaitReady(m))
 
  bkt, err := s3.NewBucketWithConfig(log.NewNopLogger(), s3.Config{
  Bucket: bucket,
  AccessKey: e2edb.MinioAccessKey,
  SecretKey: e2edb.MinioSecretKey,
- Endpoint: m.HTTPEndpoint(),
+ Endpoint: m.Endpoint("http"),
  Insecure: true,
  }, "test-feed")
  testutil.Ok(b, err)

test/e2e/compact_test.go

@@ -16,8 +16,9 @@ import (
  "testing"
  "time"
 
- "github.com/cortexproject/cortex/integration/e2e"
- e2edb "github.com/cortexproject/cortex/integration/e2e/db"
+ "github.com/efficientgo/e2e"
+ e2edb "github.com/efficientgo/e2e/db"
+ "github.com/efficientgo/e2e/matchers"
  "github.com/go-kit/kit/log"
  "github.com/oklog/ulid"
  "github.com/prometheus/client_golang/prometheus"
@@ -33,6 +34,7 @@ import (
  "github.com/thanos-io/thanos/pkg/objstore/client"
  "github.com/thanos-io/thanos/pkg/objstore/s3"
  "github.com/thanos-io/thanos/pkg/promclient"
+ "github.com/thanos-io/thanos/pkg/runutil"
  "github.com/thanos-io/thanos/pkg/testutil"
  "github.com/thanos-io/thanos/pkg/testutil/e2eutil"
  "github.com/thanos-io/thanos/test/e2e/e2ethanos"
@@ -336,22 +338,22 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
  if penaltyDedup {
  name = "e2e_test_compact_penalty_dedup"
  }
- s, err := e2e.NewScenario(name)
+ e, err := e2e.NewDockerEnvironment(name)
  testutil.Ok(t, err)
- t.Cleanup(e2ethanos.CleanScenario(t, s))
+ t.Cleanup(e2ethanos.CleanScenario(t, e))
 
- dir := filepath.Join(s.SharedDir(), "tmp")
+ dir := filepath.Join(e.SharedDir(), "tmp")
  testutil.Ok(t, os.MkdirAll(dir, os.ModePerm))
 
  const bucket = "compact_test"
- m := e2edb.NewMinio(8080, bucket)
- testutil.Ok(t, s.StartAndWaitReady(m))
+ m := e2ethanos.NewMinio(e, "minio", bucket)
+ testutil.Ok(t, e2e.StartAndWaitReady(m))
 
  bkt, err := s3.NewBucketWithConfig(logger, s3.Config{
  Bucket: bucket,
  AccessKey: e2edb.MinioAccessKey,
  SecretKey: e2edb.MinioSecretKey,
- Endpoint: m.HTTPEndpoint(), // We need separate client config, when connecting to minio from outside.
+ Endpoint: m.Endpoint("http"), // We need separate client config, when connecting to minio from outside.
  Insecure: true,
  }, "test-feed")
  testutil.Ok(t, err)
@@ -363,7 +365,10 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
  for _, b := range blocks {
  id, err := b.Create(ctx, dir, justAfterConsistencyDelay, b.hashFunc)
  testutil.Ok(t, err)
- testutil.Ok(t, objstore.UploadDir(ctx, logger, bkt, path.Join(dir, id.String()), id.String()))
+ testutil.Ok(t, runutil.Retry(time.Second, ctx.Done(), func() error {
+ return objstore.UploadDir(ctx, logger, bkt, path.Join(dir, id.String()), id.String())
+ }))
+
  rawBlockIDs[id] = struct{}{}
  if b.markedForNoCompact {
  testutil.Ok(t, block.MarkForNoCompact(ctx, logger, bkt, id, metadata.ManualNoCompactReason, "why not", promauto.With(nil).NewCounter(prometheus.CounterOpts{})))
@@ -442,26 +447,26 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
  Bucket: bucket,
  AccessKey: e2edb.MinioAccessKey,
  SecretKey: e2edb.MinioSecretKey,
- Endpoint: m.NetworkHTTPEndpoint(),
+ Endpoint: m.InternalEndpoint("http"),
  Insecure: true,
  },
  }
- str, err := e2ethanos.NewStoreGW(s.SharedDir(), "1", svcConfig)
+ str, err := e2ethanos.NewStoreGW(e, "1", svcConfig)
  testutil.Ok(t, err)
- testutil.Ok(t, s.StartAndWaitReady(str))
+ testutil.Ok(t, e2e.StartAndWaitReady(str))
  testutil.Ok(t, str.WaitSumMetrics(e2e.Equals(float64(len(rawBlockIDs)+7)), "thanos_blocks_meta_synced"))
  testutil.Ok(t, str.WaitSumMetrics(e2e.Equals(0), "thanos_blocks_meta_sync_failures_total"))
  testutil.Ok(t, str.WaitSumMetrics(e2e.Equals(0), "thanos_blocks_meta_modified"))
 
- q, err := e2ethanos.NewQuerierBuilder(s.SharedDir(), "1", str.GRPCNetworkEndpoint()).Build()
+ q, err := e2ethanos.NewQuerierBuilder(e, "1", str.InternalEndpoint("grpc")).Build()
  testutil.Ok(t, err)
- testutil.Ok(t, s.StartAndWaitReady(q))
+ testutil.Ok(t, e2e.StartAndWaitReady(q))
 
  ctx, cancel = context.WithTimeout(context.Background(), 3*time.Minute)
  t.Cleanup(cancel)
 
  // Check if query detects current series, even if overlapped.
- queryAndAssert(t, ctx, q.HTTPEndpoint(),
+ queryAndAssert(t, ctx, q.Endpoint("http"),
  fmt.Sprintf(`count_over_time({a="1"}[13h] offset %ds)`, int64(time.Since(now.Add(12*time.Hour)).Seconds())),
  promclient.QueryOptions{
  Deduplicate: false, // This should be false, so that we can be sure deduplication was offline.
@@ -599,7 +604,7 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
  // Precreate a directory. It should be deleted.
  // In a hypothetical scenario, the directory could be a left-over from
  // a compaction that had crashed.
- p := filepath.Join(s.SharedDir(), "data", "compact", "expect-to-halt", "compact")
+ p := filepath.Join(e.SharedDir(), "data", "compact", "expect-to-halt", "compact")
 
  testutil.Assert(t, len(blocksWithHashes) > 0)
 
@@ -613,9 +618,9 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
  testutil.Ok(t, err)
  testutil.Ok(t, f.Close())
 
- c, err := e2ethanos.NewCompactor(s.SharedDir(), "expect-to-halt", svcConfig, nil)
+ c, err := e2ethanos.NewCompactor(e, "expect-to-halt", svcConfig, nil)
  testutil.Ok(t, err)
- testutil.Ok(t, s.StartAndWaitReady(c))
+ testutil.Ok(t, e2e.StartAndWaitReady(c))
 
  // Expect compactor halted and for one cleanup iteration to happen.
  testutil.Ok(t, c.WaitSumMetrics(e2e.Equals(1), "thanos_compact_halted"))
@@ -626,10 +631,10 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
  testutil.Ok(t, c.WaitSumMetrics(e2e.Equals(0), "thanos_blocks_meta_modified"))
 
  // The compact directory is still there.
- dataDir := filepath.Join(s.SharedDir(), "data", "compact", "expect-to-halt")
+ dataDir := filepath.Join(e.SharedDir(), "data", "compact", "expect-to-halt")
  empty, err := isEmptyDir(dataDir)
  testutil.Ok(t, err)
- testutil.Equals(t, false, empty, "directory %s should not be empty", dataDir)
+ testutil.Equals(t, false, empty, "directory %e should not be empty", dataDir)
 
  // We expect no ops.
  testutil.Ok(t, c.WaitSumMetrics(e2e.Equals(0), "thanos_compact_iterations_total"))
@@ -646,10 +651,10 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
  testutil.Ok(t, c.WaitSumMetrics(e2e.Equals(2), "thanos_compact_blocks_cleaned_total"))
 
  // Ensure bucket UI.
- ensureGETStatusCode(t, http.StatusOK, "http://"+path.Join(c.HTTPEndpoint(), "global"))
- ensureGETStatusCode(t, http.StatusOK, "http://"+path.Join(c.HTTPEndpoint(), "loaded"))
+ ensureGETStatusCode(t, http.StatusOK, "http://"+path.Join(c.Endpoint("http"), "global"))
+ ensureGETStatusCode(t, http.StatusOK, "http://"+path.Join(c.Endpoint("http"), "loaded"))
 
- testutil.Ok(t, s.Stop(c))
+ testutil.Ok(t, c.Stop())
 
  _, err = os.Stat(randBlockDir)
  testutil.NotOk(t, err)
@@ -661,7 +666,7 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
  // Dedup enabled; compactor should work as expected.
  {
  // Predownload block dirs with hashes. We should not try downloading them again.
- p := filepath.Join(s.SharedDir(), "data", "compact", "working")
+ p := filepath.Join(e.SharedDir(), "data", "compact", "working")
 
  for _, id := range blocksWithHashes {
  m, err := block.DownloadMeta(ctx, logger, bkt, id)
@@ -677,9 +682,9 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
  }
 
  // We expect 2x 4-block compaction, 2-block vertical compaction, 2x 3-block compaction.
- c, err := e2ethanos.NewCompactor(s.SharedDir(), "working", svcConfig, nil, extArgs...)
+ c, err := e2ethanos.NewCompactor(e, "working", svcConfig, nil, extArgs...)
  testutil.Ok(t, err)
- testutil.Ok(t, s.StartAndWaitReady(c))
+ testutil.Ok(t, e2e.StartAndWaitReady(c))
 
  // NOTE: We cannot assert on intermediate `thanos_blocks_meta_` metrics as those are gauge and change dynamically due to many
  // compaction groups. Wait for at least first compaction iteration (next is in 5m).
@@ -706,9 +711,9 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
 
  testutil.Ok(t, c.WaitSumMetrics(e2e.Equals(0), "thanos_compact_halted"))
 
- bucketMatcher, err := labels.NewMatcher(labels.MatchEqual, "bucket", bucket)
+ bucketMatcher, err := matchers.NewMatcher(matchers.MatchEqual, "bucket", bucket)
  testutil.Ok(t, err)
- operationMatcher, err := labels.NewMatcher(labels.MatchEqual, "operation", "get")
+ operationMatcher, err := matchers.NewMatcher(matchers.MatchEqual, "operation", "get")
  testutil.Ok(t, err)
  testutil.Ok(t, c.WaitSumMetricsWithOptions(e2e.Equals(478),
  []string{"thanos_objstore_bucket_operations_total"}, e2e.WithLabelMatchers(
@@ -718,13 +723,13 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
  )
 
  // Make sure compactor does not modify anything else over time.
- testutil.Ok(t, s.Stop(c))
+ testutil.Ok(t, c.Stop())
 
  ctx, cancel = context.WithTimeout(context.Background(), 3*time.Minute)
  t.Cleanup(cancel)
 
  // Check if query detects new blocks.
- queryAndAssert(t, ctx, q.HTTPEndpoint(),
+ queryAndAssert(t, ctx, q.Endpoint("http"),
  fmt.Sprintf(`count_over_time({a="1"}[13h] offset %ds)`, int64(time.Since(now.Add(12*time.Hour)).Seconds())),
  promclient.QueryOptions{
  Deduplicate: false, // This should be false, so that we can be sure deduplication was offline.
@@ -742,9 +747,9 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
  if penaltyDedup {
  extArgs = append(extArgs, "--deduplication.func=penalty")
  }
- c, err := e2ethanos.NewCompactor(s.SharedDir(), "working", svcConfig, nil, extArgs...)
+ c, err := e2ethanos.NewCompactor(e, "working-dedup", svcConfig, nil, extArgs...)
  testutil.Ok(t, err)
- testutil.Ok(t, s.StartAndWaitReady(c))
+ testutil.Ok(t, e2e.StartAndWaitReady(c))
 
  // NOTE: We cannot assert on intermediate `thanos_blocks_meta_` metrics as those are gauge and change dynamically due to many
  // compaction groups. Wait for at least first compaction iteration (next is in 5m).
@@ -767,13 +772,13 @@ func testCompactWithStoreGateway(t *testing.T, penaltyDedup bool) {
 
  testutil.Ok(t, c.WaitSumMetrics(e2e.Equals(0), "thanos_compact_halted"))
  // Make sure compactor does not modify anything else over time.
- testutil.Ok(t, s.Stop(c))
+ testutil.Ok(t, c.Stop())
 
  ctx, cancel = context.WithTimeout(context.Background(), 3*time.Minute)
  t.Cleanup(cancel)
 
  // Check if query detects new blocks.
- queryAndAssert(t, ctx, q.HTTPEndpoint(),
+ queryAndAssert(t, ctx, q.Endpoint("http"),
  fmt.Sprintf(`count_over_time({a="1"}[13h] offset %ds)`, int64(time.Since(now.Add(12*time.Hour)).Seconds())),
  promclient.QueryOptions{
  Deduplicate: false, // This should be false, so that we can be sure deduplication was offline.

test/e2e/e2ethanos/helpers.go

@@ -11,16 +11,16 @@ import (
  "strings"
  "testing"
 
- "github.com/cortexproject/cortex/integration/e2e"
+ "github.com/efficientgo/e2e"
 
  "github.com/thanos-io/thanos/pkg/testutil"
 )
 
-func CleanScenario(t testing.TB, s *e2e.Scenario) func() {
+func CleanScenario(t testing.TB, e *e2e.DockerEnvironment) func() {
  return func() {
  // Make sure Clean can properly delete everything.
- testutil.Ok(t, exec.Command("chmod", "-R", "777", s.SharedDir()).Run())
- s.Close()
+ testutil.Ok(t, exec.Command("chmod", "-R", "777", e.SharedDir()).Run())
+ e.Close()
  }
 }
 

test/e2e/e2ethanos/service.go

@@ -4,35 +4,74 @@
 package e2ethanos
 
 import (
- "github.com/cortexproject/cortex/integration/e2e"
-)
+ "os"
+ "strconv"
 
-type Service struct {
- *e2e.HTTPService
+ "github.com/efficientgo/e2e"
+)
 
- grpc int
+type Port struct {
+ Name string
+ PortNum int
+ IsMetrics bool
 }
 
 func NewService(
+ e e2e.Environment,
  name string,
  image string,
- command *e2e.Command,
+ command e2e.Command,
  readiness *e2e.HTTPReadinessProbe,
  http, grpc int,
- otherPorts ...int,
-) *Service {
- return &Service{
- HTTPService: e2e.NewHTTPService(name, image, command, readiness, http, append(otherPorts, grpc)...),
- grpc: grpc,
- }
+ otherPorts ...Port,
+) *e2e.InstrumentedRunnable {
+ return newUninitiatedService(e, name, http, grpc, otherPorts...).Init(
+ e2e.StartOptions{
+ Image: image,
+ Command: command,
+ Readiness: readiness,
+ User: strconv.Itoa(os.Getuid()),
+ WaitReadyBackoff: &defaultBackoffConfig,
+ },
+ )
 }
 
-func (s *Service) GRPCEndpoint() string { return s.Endpoint(s.grpc) }
+func newUninitiatedService(
+ e e2e.Environment,
+ name string,
+ http, grpc int,
+ otherPorts ...Port,
+) *e2e.FutureInstrumentedRunnable {
+ metricsPorts := "http"
+ ports := map[string]int{
+ "http": http,
+ "grpc": grpc,
+ }
 
-func (s *Service) GRPCNetworkEndpoint() string {
- return s.NetworkEndpoint(s.grpc)
+ for _, op := range otherPorts {
+ ports[op.Name] = op.PortNum
+
+ if op.IsMetrics {
+ metricsPorts = op.Name
+ }
+ }
+
+ return e2e.NewInstrumentedRunnable(e, name, ports, metricsPorts)
 }
 
-func (s *Service) GRPCNetworkEndpointFor(networkName string) string {
- return s.NetworkEndpointFor(networkName, s.grpc)
+func initiateService(
+ service *e2e.FutureInstrumentedRunnable,
+ image string,
+ command e2e.Command,
+ readiness *e2e.HTTPReadinessProbe,
+) *e2e.InstrumentedRunnable {
+ return service.Init(
+ e2e.StartOptions{
+ Image: image,
+ Command: command,
+ Readiness: readiness,
+ User: strconv.Itoa(os.Getuid()),
+ WaitReadyBackoff: &defaultBackoffConfig,
+ },
+ )
 }