Merge pull request #1 from yeya24/fix-e2e

Fix E2E tests for receive router
thanos-io · Mar 3, 2021 · c35d09c · c35d09c
2 parents 21a645b + 692a14a
commit c35d09c
Show file tree

Hide file tree

Showing 4 changed files with 47 additions and 12 deletions.
diff --git a/cmd/thanos/receive_route.go b/cmd/thanos/receive_route.go
@@ -172,10 +172,11 @@ func runReceiveRoute(
  if !ok {
  return nil
  }
- webHandler.Hashring(h)
  msg := "hashring has changed; server is not ready to receive web requests"
  statusProber.NotReady(errors.New(msg))
  level.Info(logger).Log("msg", msg)
+ webHandler.Hashring(h)
+ statusProber.Ready()
  case <-cancel:
  return nil
  }

diff --git a/test/e2e/e2ethanos/services.go b/test/e2e/e2ethanos/services.go
@@ -192,6 +192,7 @@ func NewReceiver(sharedDir string, networkName string, name string) (*Service, e
 
  dir := filepath.Join(sharedDir, "data", "receive", name)
  dataDir := filepath.Join(dir, "data")
+ container := filepath.Join(e2e.ContainerSharedDir, "data", "receive", name)
  if err := os.MkdirAll(dataDir, 0777); err != nil {
  return nil, errors.Wrap(err, "create receive dir")
  }
@@ -206,6 +207,8 @@ func NewReceiver(sharedDir string, networkName string, name string) (*Service, e
  "--grpc-grace-period": "0s",
  "--http-address": ":8080",
  "--log.level": infoLogLevel,
+ "--tsdb.path": filepath.Join(container, "data"),
+ "--label": fmt.Sprintf(`receive="%s"`, name),
  })...),
  e2e.NewHTTPReadinessProbe(8080, "/-/ready", 200, 200),
  8080,
@@ -218,14 +221,15 @@ func NewReceiver(sharedDir string, networkName string, name string) (*Service, e
  return receiver, nil
 }
 
-func NewReceiverWithConfigWatcher(sharedDir string, networkName string, name string, replicationFactor int, hashring ...route.HashringConfig) (*Service, error) {
+func NewReceiverWithConfigWatcher(sharedDir string, networkName string, name string, hashring ...route.HashringConfig) (*Service, error) {
  localEndpoint := NewService(fmt.Sprintf("receive-%v", name), "", e2e.NewCommand("", ""), nil, 8080, 9091, 8081).GRPCNetworkEndpointFor(networkName)
  if len(hashring) == 0 {
  hashring = []route.HashringConfig{{Endpoints: []string{localEndpoint}}}
  }
 
  dir := filepath.Join(sharedDir, "data", "receive", name)
  dataDir := filepath.Join(dir, "data")
+ container := filepath.Join(e2e.ContainerSharedDir, "data", "receive", name)
  if err := os.MkdirAll(dataDir, 0777); err != nil {
  return nil, errors.Wrap(err, "create receive dir")
  }
@@ -248,6 +252,8 @@ func NewReceiverWithConfigWatcher(sharedDir string, networkName string, name str
  "--grpc-grace-period": "0s",
  "--http-address": ":8080",
  "--log.level": infoLogLevel,
+ "--tsdb.path": filepath.Join(container, "data"),
+ "--label": fmt.Sprintf(`receive="%s"`, name),
  })...),
  e2e.NewHTTPReadinessProbe(8080, "/-/ready", 200, 200),
  8080,

diff --git a/test/e2e/query_test.go b/test/e2e/query_test.go
@@ -26,6 +26,7 @@ import (
  "github.com/thanos-io/thanos/pkg/store/storepb"
 
  "github.com/thanos-io/thanos/pkg/promclient"
+ "github.com/thanos-io/thanos/pkg/route"
  "github.com/thanos-io/thanos/pkg/runutil"
  "github.com/thanos-io/thanos/pkg/testutil"
  "github.com/thanos-io/thanos/test/e2e/e2ethanos"
@@ -95,11 +96,21 @@ func TestQuery(t *testing.T) {
 
  receiver, err := e2ethanos.NewReceiver(s.SharedDir(), s.NetworkName(), "1")
  testutil.Ok(t, err)
- testutil.Ok(t, s.StartAndWaitReady(receiver))
+
+ h := route.HashringConfig{
+ Endpoints: []string{
+ receiver.GRPCNetworkEndpointFor(s.NetworkName()),
+ },
+ }
+
+ // Create a router.
+ route1, err := e2ethanos.NewReceiveRouter(s.SharedDir(), s.NetworkName(), "1", 1, h)
+ testutil.Ok(t, err)
+ testutil.Ok(t, s.StartAndWaitReady(receiver, route1))
 
  prom1, sidecar1, err := e2ethanos.NewPrometheusWithSidecar(s.SharedDir(), "e2e_test_query", "alone", defaultPromConfig("prom-alone", 0, "", ""), e2ethanos.DefaultPrometheusImage())
  testutil.Ok(t, err)
- prom2, sidecar2, err := e2ethanos.NewPrometheusWithSidecar(s.SharedDir(), "e2e_test_query", "remote-and-sidecar", defaultPromConfig("prom-both-remote-write-and-sidecar", 1234, e2ethanos.RemoteWriteEndpoint(receiver.NetworkEndpoint(8081)), ""), e2ethanos.DefaultPrometheusImage())
+ prom2, sidecar2, err := e2ethanos.NewPrometheusWithSidecar(s.SharedDir(), "e2e_test_query", "remote-and-sidecar", defaultPromConfig("prom-both-remote-write-and-sidecar", 1234, e2ethanos.RemoteWriteEndpoint(route1.NetworkEndpoint(8081)), ""), e2ethanos.DefaultPrometheusImage())
  testutil.Ok(t, err)
  prom3, sidecar3, err := e2ethanos.NewPrometheusWithSidecar(s.SharedDir(), "e2e_test_query", "ha1", defaultPromConfig("prom-ha", 0, "", filepath.Join(e2e.ContainerSharedDir, "", "*.yaml")), e2ethanos.DefaultPrometheusImage())
  testutil.Ok(t, err)
@@ -125,6 +136,13 @@ func TestQuery(t *testing.T) {
  "prometheus": "prom-alone",
  "replica": "0",
  },
+ {
+ "job": "myself",
+ "prometheus": "prom-both-remote-write-and-sidecar",
+ "receive": "1",
+ "replica": "1234",
+ "tenant_id": "default-tenant",
+ },
  {
  "job": "myself",
  "prometheus": "prom-both-remote-write-and-sidecar",
@@ -150,6 +168,12 @@ func TestQuery(t *testing.T) {
  "job": "myself",
  "prometheus": "prom-alone",
  },
+ {
+ "job": "myself",
+ "prometheus": "prom-both-remote-write-and-sidecar",
+ "receive": "1",
+ "tenant_id": "default-tenant",
+ },
  {
  "job": "myself",
  "prometheus": "prom-both-remote-write-and-sidecar",

diff --git a/test/e2e/receive_test.go b/test/e2e/receive_test.go
@@ -137,7 +137,7 @@ func TestReceive(t *testing.T) {
  t.Run("hashring with config watcher", func(t *testing.T) {
  t.Parallel()
 
- s, err := e2e.NewScenario("e2e_test_receive_hashring")
+ s, err := e2e.NewScenario("e2e_test_receive_hashring_config_watcher")
  testutil.Ok(t, err)
  t.Cleanup(e2ethanos.CleanScenario(t, s))
 
@@ -158,19 +158,23 @@ func TestReceive(t *testing.T) {
 
  // Recreate again, but with hashring config.
  // TODO(kakkoyun): Update config file and wait config watcher to reconcile hashring.
- r1, err = e2ethanos.NewReceiverWithConfigWatcher(s.SharedDir(), s.NetworkName(), "1", 1, h)
+ r1, err = e2ethanos.NewReceiverWithConfigWatcher(s.SharedDir(), s.NetworkName(), "1", h)
  testutil.Ok(t, err)
- r2, err = e2ethanos.NewReceiverWithConfigWatcher(s.SharedDir(), s.NetworkName(), "2", 1, h)
+ r2, err = e2ethanos.NewReceiverWithConfigWatcher(s.SharedDir(), s.NetworkName(), "2", h)
  testutil.Ok(t, err)
- r3, err = e2ethanos.NewReceiverWithConfigWatcher(s.SharedDir(), s.NetworkName(), "3", 1, h)
+ r3, err = e2ethanos.NewReceiverWithConfigWatcher(s.SharedDir(), s.NetworkName(), "3", h)
  testutil.Ok(t, err)
- testutil.Ok(t, s.StartAndWaitReady(r1, r2, r3))
 
- prom1, _, err := e2ethanos.NewPrometheus(s.SharedDir(), "1", defaultPromConfig("prom1", 0, e2ethanos.RemoteWriteEndpoint(r1.NetworkEndpoint(8081)), ""), e2ethanos.DefaultPrometheusImage())
+ // Create a router.
+ route1, err := e2ethanos.NewReceiveRouter(s.SharedDir(), s.NetworkName(), "1", 1, h)
  testutil.Ok(t, err)
- prom2, _, err := e2ethanos.NewPrometheus(s.SharedDir(), "2", defaultPromConfig("prom2", 0, e2ethanos.RemoteWriteEndpoint(r2.NetworkEndpoint(8081)), ""), e2ethanos.DefaultPrometheusImage())
+ testutil.Ok(t, s.StartAndWaitReady(r1, r2, r3, route1))
+
+ prom1, _, err := e2ethanos.NewPrometheus(s.SharedDir(), "1", defaultPromConfig("prom1", 0, e2ethanos.RemoteWriteEndpoint(route1.NetworkEndpoint(8081)), ""), e2ethanos.DefaultPrometheusImage())
+ testutil.Ok(t, err)
+ prom2, _, err := e2ethanos.NewPrometheus(s.SharedDir(), "2", defaultPromConfig("prom2", 0, e2ethanos.RemoteWriteEndpoint(route1.NetworkEndpoint(8081)), ""), e2ethanos.DefaultPrometheusImage())
  testutil.Ok(t, err)
- prom3, _, err := e2ethanos.NewPrometheus(s.SharedDir(), "3", defaultPromConfig("prom3", 0, e2ethanos.RemoteWriteEndpoint(r3.NetworkEndpoint(8081)), ""), e2ethanos.DefaultPrometheusImage())
+ prom3, _, err := e2ethanos.NewPrometheus(s.SharedDir(), "3", defaultPromConfig("prom3", 0, e2ethanos.RemoteWriteEndpoint(route1.NetworkEndpoint(8081)), ""), e2ethanos.DefaultPrometheusImage())
  testutil.Ok(t, err)
  testutil.Ok(t, s.StartAndWaitReady(prom1, prom2, prom3))