From cc0abbfed09b71b11e61ddb202730ad7c484d2fe Mon Sep 17 00:00:00 2001 From: tdakkota Date: Thu, 24 Sep 2026 13:17:42 +0300 Subject: [PATCH] test(etcdtest): share the embedded etcd and bind it to port 0 etcdtest replaces 4 embedded-etcd copies and 3 freeAddr copies. etcd binds its listeners to port 0, so the probe-then-bind race of #509 is gone. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01KcLq1G8KpWBvM53JaQb1tS --- ARCHITECTURE.md | 2 +- admin_test.go | 3 +- cluster/ARCH.md | 6 ++ cluster/etcd/etcdtest/etcdtest.go | 115 ++++++++++++++++++++ cluster/etcd/etcdtest/etcdtest_test.go | 58 ++++++++++ cluster/etcd/membership_test.go | 56 +--------- cluster/etcd/observe_test.go | 8 +- cluster/etcd/ownership_chaos_test.go | 7 +- cluster/etcd/ownership_restart_test.go | 8 +- cluster/etcd/ownership_test.go | 5 +- cluster/etcd/rejoin_test.go | 118 ++++----------------- cluster/etcd/watch_test.go | 10 +- cluster/router/read_test.go | 11 +- cluster/router/router_test.go | 64 ++--------- cluster/router/trace_test.go | 3 +- cluster_admission_test.go | 3 +- cluster_aggregate_gap_test.go | 5 +- cluster_bootstrap_signals_test.go | 3 +- cluster_bootstrap_test.go | 3 +- cluster_completeness_test.go | 3 +- cluster_ec_chaos_test.go | 3 +- cluster_ec_lazy_test.go | 3 +- cluster_ec_routed_test.go | 4 +- cluster_ec_test.go | 15 +-- cluster_fence_test.go | 3 +- cluster_flushpressure_test.go | 3 +- cluster_labels_test.go | 11 +- cluster_mirror_test.go | 3 +- cluster_readpolicy_test.go | 15 +-- cluster_recovered_seams_test.go | 3 +- cluster_recovery_test.go | 7 +- cluster_repair_test.go | 7 +- cluster_retry_test.go | 5 +- cluster_routed_test.go | 3 +- cluster_rpc_trace_test.go | 3 +- cluster_sampling_test.go | 3 +- cluster_shardwarn_test.go | 7 +- cluster_test.go | 141 +++++-------------------- cluster_trace_test.go | 3 +- clusterprivacy_test.go | 5 +- repair_repro_test.go | 7 +- replica_omitted_repro_test.go | 3 +- replica_reconcile_repro_test.go | 3 +- replica_refresh_repro_test.go | 3 +- traces_keys_test.go | 3 +- 45 files changed, 363 insertions(+), 394 deletions(-) create mode 100644 cluster/etcd/etcdtest/etcdtest.go create mode 100644 cluster/etcd/etcdtest/etcdtest_test.go diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 1a6cfaea..a6f0b199 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -275,7 +275,7 @@ wal/ segmented CRC-framed WAL engine/ metrics vertical recordengine/ shared record engine (logs/traces/profiles/exemplars) query/{fetch,scale,profile,promql} read seam · scale-out decorators · EXPLAIN ANALYZE · Prom adapter -cluster/{,ring,etcd,replica,rebalance,partsync,ec,router} L0 distribution +cluster/{,ring,etcd,etcd/etcdtest,replica,rebalance,partsync,ec,router} L0 distribution · in-process etcd for tests internal/{obs,retry,simd,parallel,partid,diskguard,vfs,memlimit,memsize,watermark,mergestream,reproduce,heaptest,cmd/gensimd} injected observability · reliability · AVX2 kernels · fan-out · part ids · disk-pressure guard · heap accounting · per-series watermark sidecar codec · the merge seam both engines share (k-way key union, seal units, conformance suite) · gated defect reproducers · heap and allocation measurement for tests · filesystem seam + crash model (validated against ext4 on dm-flakey by `internal/vfs/crashmodel`, `-tags crashmodel`) reliability/ public RetryConfig presets ``` diff --git a/admin_test.go b/admin_test.go index a8ade3a3..499c0cf7 100644 --- a/admin_test.go +++ b/admin_test.go @@ -7,6 +7,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/signal" ) @@ -134,7 +135,7 @@ func TestAdminRebalanceSingleNodeNoOp(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestAdminRebalanceClusteredReportsSuccess(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() s := openClusterNodeShared(t, endpoint, "node-a") diff --git a/cluster/ARCH.md b/cluster/ARCH.md index e3196e25..574d3e5b 100644 --- a/cluster/ARCH.md +++ b/cluster/ARCH.md @@ -72,6 +72,12 @@ therefore drops such a key under a value+lease guard and recreates it under the also makes the restart a real new tenure: the recreated claim takes a higher term, so the incarnation's writes outrank the dead one's. +Tests boot etcd through **`cluster/etcd/etcdtest`**, which binds its listeners to port 0 and reads +the client port back from the socket. Probing a free port and releasing it for etcd to bind is +racy: anything, including etcd clients' own ephemeral ports, can take it in between. Only +`Server.Restart` rebinds a known port, since clients must find etcd where it was. `etcdtest` must +not import `cluster/etcd`, whose internal tests use it. + ### Lease fencing — the boundary on acting as primary A node that stops renewing its lease keeps a ring frozen at its last etcd view, so it goes on diff --git a/cluster/etcd/etcdtest/etcdtest.go b/cluster/etcd/etcdtest/etcdtest.go new file mode 100644 index 00000000..52d5b68c --- /dev/null +++ b/cluster/etcd/etcdtest/etcdtest.go @@ -0,0 +1,115 @@ +// Package etcdtest runs an in-process single-node etcd for tests. +package etcdtest + +import ( + "net/url" + "testing" + "time" + + "github.com/stretchr/testify/require" + clientv3 "go.etcd.io/etcd/client/v3" + "go.etcd.io/etcd/server/v3/embed" +) + +const readyTimeout = 30 * time.Second + +// Start boots an etcd that stops with tb and returns its client endpoint URL. +func Start(tb testing.TB) string { + tb.Helper() + + return NewServer(tb).Endpoint() +} + +// Server is an embedded etcd that can be stopped and restarted on the same client URL and data +// directory, which is what an etcd rollout looks like to a node that keeps running across it. +type Server struct { + tb testing.TB + dir string + client url.URL + e *embed.Etcd +} + +// NewServer boots an etcd that stops with tb, on a client port the kernel picks. +func NewServer(tb testing.TB) *Server { + tb.Helper() + + s := &Server{ + tb: tb, + dir: tb.TempDir(), + client: url.URL{Scheme: "http", Host: "127.0.0.1:0"}, + } + s.start() + tb.Cleanup(s.Stop) + + return s +} + +// Endpoint returns the client URL, stable across [Server.Restart]. +func (s *Server) Endpoint() string { + return s.client.String() +} + +// Stop shuts etcd down. It is a no-op on a stopped server. +func (s *Server) Stop() { + if s.e == nil { + return + } + + s.e.Close() + s.e = nil +} + +// Restart starts a stopped server again on its client URL and data directory. Rebinding a fixed +// port can collide with another socket; nothing reserves it while the server is down. +func (s *Server) Restart() { + s.tb.Helper() + + require.Nil(s.tb, s.e, "etcd is still running") + s.start() +} + +// Dial returns a client for this server, closed with the test. +func (s *Server) Dial() *clientv3.Client { + s.tb.Helper() + + c, err := clientv3.New(clientv3.Config{ + Endpoints: []string{s.Endpoint()}, + DialTimeout: 5 * time.Second, + }) + require.NoError(s.tb, err) + s.tb.Cleanup(func() { _ = c.Close() }) + + return c +} + +func (s *Server) start() { + s.tb.Helper() + + // The peer URL is never dialed by a single-node cluster, so it is advertised as the unbound + // port-0 address rather than the one the listener gets. + peer := url.URL{Scheme: "http", Host: "127.0.0.1:0"} + + cfg := embed.NewConfig() + cfg.Dir = s.dir + cfg.LogLevel = "error" + cfg.ListenClientUrls = []url.URL{s.client} + cfg.AdvertiseClientUrls = []url.URL{s.client} + cfg.ListenPeerUrls = []url.URL{peer} + cfg.AdvertisePeerUrls = []url.URL{peer} + cfg.InitialCluster = cfg.Name + "=" + peer.String() + // The gateway dials the configured listen address, which is port 0 on the first start. + cfg.EnableGRPCGateway = false + + e, err := embed.StartEtcd(cfg) + require.NoError(s.tb, err) + + select { + case <-e.Server.ReadyNotify(): + case <-time.After(readyTimeout): + e.Close() + s.tb.Fatal("embedded etcd did not become ready") + } + + s.client.Host = e.Clients[0].Addr().String() + s.e = e +} diff --git a/cluster/etcd/etcdtest/etcdtest_test.go b/cluster/etcd/etcdtest/etcdtest_test.go new file mode 100644 index 00000000..b942f84f --- /dev/null +++ b/cluster/etcd/etcdtest/etcdtest_test.go @@ -0,0 +1,58 @@ +package etcdtest_test + +import ( + "context" + "net/url" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + clientv3 "go.etcd.io/etcd/client/v3" + + "github.com/oteldb/storage/cluster/etcd/etcdtest" +) + +func TestStart(t *testing.T) { + t.Parallel() + + endpoint := etcdtest.Start(t) + u, err := url.Parse(endpoint) + require.NoError(t, err) + assert.NotEqual(t, "0", u.Port()) + + c, err := clientv3.New(clientv3.Config{Endpoints: []string{endpoint}, DialTimeout: 5 * time.Second}) + require.NoError(t, err) + t.Cleanup(func() { _ = c.Close() }) + + _, err = c.Put(t.Context(), "k", "v") + require.NoError(t, err) +} + +func TestServerRestartKeepsEndpointAndData(t *testing.T) { + t.Parallel() + + s := etcdtest.NewServer(t) + endpoint := s.Endpoint() + c := s.Dial() + + _, err := c.Put(t.Context(), "k", "v") + require.NoError(t, err) + + s.Stop() + s.Stop() + s.Restart() + assert.Equal(t, endpoint, s.Endpoint()) + + require.EventuallyWithT(t, func(ct *assert.CollectT) { + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + + resp, err := c.Get(ctx, "k") + if !assert.NoError(ct, err) || !assert.Len(ct, resp.Kvs, 1) { + return + } + + assert.Equal(ct, "v", string(resp.Kvs[0].GetValue())) + }, 30*time.Second, 10*time.Millisecond, "the client from before the restart reconnects to the same data") +} diff --git a/cluster/etcd/membership_test.go b/cluster/etcd/membership_test.go index 86b87b57..662a90f7 100644 --- a/cluster/etcd/membership_test.go +++ b/cluster/etcd/membership_test.go @@ -2,18 +2,14 @@ package etcd import ( "context" - "net" - "net/url" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - clientv3 "go.etcd.io/etcd/client/v3" - "go.etcd.io/etcd/server/v3/embed" -) -const httpScheme = "http" + "github.com/oteldb/storage/cluster/etcd/etcdtest" +) func TestMemberEncodeDecodeRoundTrip(t *testing.T) { t.Parallel() @@ -42,52 +38,6 @@ func TestDecodeMemberRejectsGarbage(t *testing.T) { require.Error(t, err) } -// freeAddr returns a free localhost host:port (the listener is closed before returning, so -// there is a small TOCTOU window — acceptable for a test harness). -func freeAddr(t *testing.T) string { - t.Helper() - var lc net.ListenConfig - l, err := lc.Listen(context.Background(), "tcp", "127.0.0.1:0") - require.NoError(t, err) - addr := l.Addr().String() - require.NoError(t, l.Close()) - - return addr -} - -// startEtcd boots an in-process single-node etcd and returns a client for it. -func startEtcd(t *testing.T) *clientv3.Client { - t.Helper() - - lc := url.URL{Scheme: httpScheme, Host: freeAddr(t)} - lp := url.URL{Scheme: httpScheme, Host: freeAddr(t)} - - cfg := embed.NewConfig() - cfg.Dir = t.TempDir() - cfg.LogLevel = "error" - cfg.ListenClientUrls = []url.URL{lc} - cfg.AdvertiseClientUrls = []url.URL{lc} - cfg.ListenPeerUrls = []url.URL{lp} - cfg.AdvertisePeerUrls = []url.URL{lp} - cfg.InitialCluster = cfg.Name + "=" + lp.String() - - e, err := embed.StartEtcd(cfg) - require.NoError(t, err) - t.Cleanup(e.Close) - - select { - case <-e.Server.ReadyNotify(): - case <-time.After(30 * time.Second): - t.Fatal("embedded etcd did not become ready") - } - - client, err := clientv3.New(clientv3.Config{Endpoints: []string{lc.String()}, DialTimeout: 5 * time.Second}) - require.NoError(t, err) - t.Cleanup(func() { _ = client.Close() }) - - return client -} - func memberIDs(ms []Member) []string { out := make([]string, len(ms)) for i, m := range ms { @@ -99,7 +49,7 @@ func memberIDs(ms []Member) []string { //nolint:paralleltest // owns an embedded etcd; runs serially func TestMembershipJoinWatchLeave(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ctx := context.Background() // Node A joins. diff --git a/cluster/etcd/observe_test.go b/cluster/etcd/observe_test.go index bdfea6c4..1fdefab2 100644 --- a/cluster/etcd/observe_test.go +++ b/cluster/etcd/observe_test.go @@ -6,6 +6,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/oteldb/storage/cluster/etcd/etcdtest" ) // eventually polls cond until it holds, so a watch-driven assertion does not depend on timing. @@ -19,7 +21,7 @@ func TestWatchSeesMembersWithoutJoining(t *testing.T) { t.Parallel() ctx := t.Context() - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() joined, err := Join(ctx, client, "/test", Member{ID: "node-a", Zone: "z1", Addr: "10.0.0.1:9000"}, 0) require.NoError(t, err) @@ -53,7 +55,7 @@ func TestWatchCloseRevokesNothing(t *testing.T) { t.Parallel() ctx := t.Context() - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() joined, err := Join(ctx, client, "/test", Member{ID: "node-a", Addr: "10.0.0.1:9000"}, 0) require.NoError(t, err) @@ -73,7 +75,7 @@ func TestWatchOnEmptyClusterIsUsable(t *testing.T) { t.Parallel() ctx := t.Context() - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() obs, err := Watch(ctx, client, "/test") require.NoError(t, err) diff --git a/cluster/etcd/ownership_chaos_test.go b/cluster/etcd/ownership_chaos_test.go index 08a1c689..b2dc8a28 100644 --- a/cluster/etcd/ownership_chaos_test.go +++ b/cluster/etcd/ownership_chaos_test.go @@ -17,6 +17,7 @@ import ( "github.com/stretchr/testify/require" clientv3 "go.etcd.io/etcd/client/v3" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/cluster/ring" ) @@ -97,7 +98,7 @@ func assertOwnedByPrimary(t *testing.T, client *clientv3.Client, r *ring.Ring, s // //nolint:paralleltest // owns an embedded etcd; runs serially func TestOwnershipConcurrentClaimSingleWinner(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ctx := context.Background() const contenders = 12 @@ -139,7 +140,7 @@ func TestOwnershipConcurrentClaimSingleWinner(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestOwnershipReconcileConvergesUnderChurn(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ids := []string{"a", "b", "c", "d"} owners := make([]*Ownership, len(ids)) @@ -169,7 +170,7 @@ func TestOwnershipReconcileConvergesUnderChurn(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestOwnershipMassFailoverOnLeaseExpiry(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ctx := context.Background() dying, err := client.Grant(ctx, 30) diff --git a/cluster/etcd/ownership_restart_test.go b/cluster/etcd/ownership_restart_test.go index fc66241c..a2872e87 100644 --- a/cluster/etcd/ownership_restart_test.go +++ b/cluster/etcd/ownership_restart_test.go @@ -12,6 +12,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" clientv3 "go.etcd.io/etcd/client/v3" + + "github.com/oteldb/storage/cluster/etcd/etcdtest" ) // claimLease returns the lease a shard's claim key is bound to, and whether the key exists. @@ -35,7 +37,7 @@ func claimLease(t *testing.T, client *clientv3.Client, shard string) (clientv3.L // //nolint:paralleltest // owns an embedded etcd; runs serially func TestOwnershipRestartRebindsClaimToLiveLease(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ctx := context.Background() dead, err := client.Grant(ctx, 30) @@ -100,7 +102,7 @@ func TestOwnershipRestartRebindsClaimToLiveLease(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestOwnershipRestartAdoptionAdvancesTerm(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ctx := context.Background() dead, err := client.Grant(ctx, 30) @@ -134,7 +136,7 @@ func TestOwnershipRestartAdoptionAdvancesTerm(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestOwnershipRestartDoesNotStealPeerClaim(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ctx := context.Background() b := ownerOn(t, client, "b") diff --git a/cluster/etcd/ownership_test.go b/cluster/etcd/ownership_test.go index d29954b2..cd4edf5b 100644 --- a/cluster/etcd/ownership_test.go +++ b/cluster/etcd/ownership_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/cluster/rebalance" "github.com/oteldb/storage/cluster/ring" ) @@ -15,7 +16,7 @@ import ( func twoOwners(t *testing.T) (ctx context.Context, a, b *Ownership) { t.Helper() - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ctx = context.Background() la, err := client.Grant(ctx, 30) @@ -163,7 +164,7 @@ func shardsOf(rs []rebalance.Reassignment) []string { //nolint:paralleltest // owns an embedded etcd func TestOwnershipHandoffOnLeaseExpiry(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ctx := context.Background() la, err := client.Grant(ctx, 1) // short TTL diff --git a/cluster/etcd/rejoin_test.go b/cluster/etcd/rejoin_test.go index 17d63ad7..d00cfa96 100644 --- a/cluster/etcd/rejoin_test.go +++ b/cluster/etcd/rejoin_test.go @@ -13,7 +13,6 @@ package etcd import ( "context" - "net/url" "slices" "strconv" "testing" @@ -22,8 +21,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" clientv3 "go.etcd.io/etcd/client/v3" - "go.etcd.io/etcd/server/v3/embed" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/cluster/ring" ) @@ -36,79 +35,6 @@ const ( tick = 10 * time.Millisecond ) -// etcdServer is an embedded etcd that can be stopped and restarted on the same client URL and -// data directory — what an etcd rollout looks like to a node that keeps running across it. -type etcdServer struct { - t *testing.T - client url.URL - peer url.URL - dir string - e *embed.Etcd -} - -func startEtcdServer(t *testing.T) *etcdServer { - t.Helper() - - s := &etcdServer{ - t: t, - client: url.URL{Scheme: httpScheme, Host: freeAddr(t)}, - peer: url.URL{Scheme: httpScheme, Host: freeAddr(t)}, - dir: t.TempDir(), - } - s.start() - t.Cleanup(s.stop) - - return s -} - -func (s *etcdServer) start() { - s.t.Helper() - - cfg := embed.NewConfig() - cfg.Dir = s.dir - cfg.LogLevel = "error" - cfg.ListenClientUrls = []url.URL{s.client} - cfg.AdvertiseClientUrls = []url.URL{s.client} - cfg.ListenPeerUrls = []url.URL{s.peer} - cfg.AdvertisePeerUrls = []url.URL{s.peer} - cfg.InitialCluster = cfg.Name + "=" + s.peer.String() - - e, err := embed.StartEtcd(cfg) - require.NoError(s.t, err) - - select { - case <-e.Server.ReadyNotify(): - case <-time.After(30 * time.Second): - e.Close() - s.t.Fatal("embedded etcd did not become ready") - } - - s.e = e -} - -func (s *etcdServer) stop() { - if s.e == nil { - return - } - - s.e.Close() - s.e = nil -} - -// dial returns a client for this server, closed with the test. -func (s *etcdServer) dial() *clientv3.Client { - s.t.Helper() - - c, err := clientv3.New(clientv3.Config{ - Endpoints: []string{s.client.String()}, - DialTimeout: 5 * time.Second, - }) - require.NoError(s.t, err) - s.t.Cleanup(func() { _ = c.Close() }) - - return c -} - // memberKeys returns the live member keys with the lease each hangs off, or nil if etcd cannot // be read right now — it is polled from inside Eventually predicates, where a read against a // server that is still coming back is a "not yet", not a failure. @@ -196,8 +122,8 @@ func joinAll(t *testing.T, client *clientv3.Client, ttl time.Duration, ids ...st // //nolint:paralleltest // owns an embedded etcd; runs serially func TestMembershipRejoinsAfterLeaseExpiry(t *testing.T) { - srv := startEtcdServer(t) - client := srv.dial() + srv := etcdtest.NewServer(t) + client := srv.Dial() ctx := context.Background() nodes := joinAll(t, client, 5*time.Second, "node-a", "node-b") @@ -207,7 +133,7 @@ func TestMembershipRejoinsAfterLeaseExpiry(t *testing.T) { require.NotEqual(t, clientv3.NoLease, lost) // Expire node-b's lease from outside, exactly as etcd does when keep-alives stop arriving. - _, err := srv.dial().Revoke(ctx, lost) + _, err := srv.Dial().Revoke(ctx, lost) require.NoError(t, err) require.Eventually(t, func() bool { return b.Rejoins() >= 1 }, settle, tick, @@ -226,13 +152,13 @@ func TestMembershipRejoinsAfterLeaseExpiry(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestMembershipRejoinsAfterWholeRingLeaseLoss(t *testing.T) { - srv := startEtcdServer(t) - client := srv.dial() + srv := etcdtest.NewServer(t) + client := srv.Dial() ctx := context.Background() nodes := joinAll(t, client, 5*time.Second, "node-a", "node-b", "node-c") - killer := srv.dial() + killer := srv.Dial() for _, m := range nodes { _, err := killer.Revoke(ctx, m.LeaseID()) require.NoError(t, err) @@ -253,8 +179,8 @@ func TestMembershipRejoinsAfterWholeRingLeaseLoss(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestMembershipRejoinsAfterExternalKeyDelete(t *testing.T) { - srv := startEtcdServer(t) - client := srv.dial() + srv := etcdtest.NewServer(t) + client := srv.Dial() ctx := context.Background() nodes := joinAll(t, client, 5*time.Second, "node-a", "node-b") @@ -262,7 +188,7 @@ func TestMembershipRejoinsAfterExternalKeyDelete(t *testing.T) { held := b.LeaseID() - _, err := srv.dial().Delete(ctx, testPrefix+"node-b") + _, err := srv.Dial().Delete(ctx, testPrefix+"node-b") require.NoError(t, err) require.Eventually(t, func() bool { return b.Rejoins() >= 1 }, settle, tick, @@ -279,14 +205,14 @@ func TestMembershipRejoinsAfterExternalKeyDelete(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestMembershipRejoinIsIdempotent(t *testing.T) { - srv := startEtcdServer(t) - client := srv.dial() + srv := etcdtest.NewServer(t) + client := srv.Dial() ctx := context.Background() nodes := joinAll(t, client, 5*time.Second, "node-a", "node-b") b := nodes[1] - killer := srv.dial() + killer := srv.Dial() const rounds = 5 @@ -312,19 +238,19 @@ func TestMembershipRejoinIsIdempotent(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestMembershipRejoinsAfterEtcdRestart(t *testing.T) { - srv := startEtcdServer(t) - client := srv.dial() + srv := etcdtest.NewServer(t) + client := srv.Dial() nodes := joinAll(t, client, 2*time.Second, "node-a", "node-b") - srv.stop() + srv.Stop() for _, m := range nodes { require.Eventually(t, m.SelfAbsent, settle, tick, "%s notices it is absent while etcd is down", m.self.ID) } - srv.start() + srv.Restart() requireConverged(t, client, nodes, "node-a", "node-b") } @@ -336,8 +262,8 @@ func TestMembershipRejoinsAfterEtcdRestart(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestOwnershipRebindsToNewLease(t *testing.T) { - srv := startEtcdServer(t) - client := srv.dial() + srv := etcdtest.NewServer(t) + client := srv.Dial() ctx := context.Background() first, err := client.Grant(ctx, 30) @@ -382,8 +308,8 @@ func TestOwnershipRebindsToNewLease(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestWatchObserverNeverRejoins(t *testing.T) { - srv := startEtcdServer(t) - client := srv.dial() + srv := etcdtest.NewServer(t) + client := srv.Dial() ctx := context.Background() nodes := joinAll(t, client, 5*time.Second, "node-a") @@ -395,7 +321,7 @@ func TestWatchObserverNeverRejoins(t *testing.T) { require.Eventually(t, func() bool { return obs.Ring().Len() == 1 }, settle, tick) - _, err = srv.dial().Revoke(ctx, nodes[0].LeaseID()) + _, err = srv.Dial().Revoke(ctx, nodes[0].LeaseID()) require.NoError(t, err) require.Eventually(t, func() bool { return nodes[0].Rejoins() >= 1 }, settle, tick) diff --git a/cluster/etcd/watch_test.go b/cluster/etcd/watch_test.go index 829a7cf5..a09aaf38 100644 --- a/cluster/etcd/watch_test.go +++ b/cluster/etcd/watch_test.go @@ -9,6 +9,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" clientv3 "go.etcd.io/etcd/client/v3" + + "github.com/oteldb/storage/cluster/etcd/etcdtest" ) // compactPast advances the store past rev and compacts, so a watch starting at rev is canceled @@ -40,7 +42,7 @@ func currentRev(ctx context.Context, t *testing.T, client *clientv3.Client) int6 //nolint:paralleltest // owns an embedded etcd; runs serially func TestConsumeReportsCancellation(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ctx := context.Background() m := &Membership{client: client, prefix: "/oteldb/members/", members: map[string]Member{}} @@ -58,7 +60,7 @@ func TestConsumeReportsCancellation(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestWatchResubscribesAfterCancellation(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ctx := context.Background() stale := currentRev(ctx, t, client) @@ -99,7 +101,7 @@ func TestWatchResubscribesAfterCancellation(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestResyncSignalsSelfAbsence(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() ctx := context.Background() m := &Membership{ @@ -142,7 +144,7 @@ func TestResyncSignalsSelfAbsence(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestObserverCheckSelfIsNoop(t *testing.T) { - client := startEtcd(t) + client := etcdtest.NewServer(t).Dial() m := &Membership{client: client, prefix: "/oteldb/members/", members: map[string]Member{}} m.checkSelf() // Must not panic on the nil channel. diff --git a/cluster/router/read_test.go b/cluster/router/read_test.go index 1e1c2830..b50d53c5 100644 --- a/cluster/router/read_test.go +++ b/cluster/router/read_test.go @@ -13,6 +13,7 @@ import ( "github.com/stretchr/testify/require" "github.com/oteldb/storage/cluster" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/cluster/router" "github.com/oteldb/storage/engine" "github.com/oteldb/storage/query/fetch" @@ -46,8 +47,6 @@ type peer struct { func (p *peer) serve(t *testing.T) string { t.Helper() - p.addr = freeAddr(t) - batches := make([]*fetch.Batch, 0, len(p.series)) for i := range p.series { s := p.series[i] @@ -114,9 +113,11 @@ func (p *peer) serve(t *testing.T) string { var lc net.ListenConfig - ln, err := lc.Listen(t.Context(), "tcp", p.addr) + ln, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") require.NoError(t, err) + p.addr = ln.Addr().String() + go func() { _ = srv.Serve(ln) }() t.Cleanup(func() { _ = srv.Close() }) @@ -131,7 +132,7 @@ func openRouter(t *testing.T, peers ...*peer) *router.Router { const root = "/test" - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) for i, p := range peers { joinNode(t, endpoint, root, "node-"+string(rune('a'+i)), p.serve(t)) } @@ -285,7 +286,7 @@ func TestReadsEmptyWhenEveryOwnerDisclaims(t *testing.T) { func TestReadsEmptyOnEmptyRing(t *testing.T) { t.Parallel() - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) r, err := router.Open(t.Context(), router.Config{Etcd: []string{endpoint}, Root: "/test"}) require.NoError(t, err) diff --git a/cluster/router/router_test.go b/cluster/router/router_test.go index 1b7e78aa..2a3a2735 100644 --- a/cluster/router/router_test.go +++ b/cluster/router/router_test.go @@ -4,64 +4,20 @@ import ( "context" "net" "net/http" - "net/url" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" clientv3 "go.etcd.io/etcd/client/v3" - "go.etcd.io/etcd/server/v3/embed" "github.com/oteldb/storage/cluster" "github.com/oteldb/storage/cluster/etcd" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/cluster/router" "github.com/oteldb/storage/signal" ) -const httpScheme = "http" - -func freeAddr(t *testing.T) string { - t.Helper() - - var lc net.ListenConfig - - l, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") - require.NoError(t, err) - defer func() { _ = l.Close() }() - - return l.Addr().String() -} - -// startEtcd runs an embedded etcd for the test and returns its client endpoint. -func startEtcd(t *testing.T) string { - t.Helper() - - lc := url.URL{Scheme: httpScheme, Host: freeAddr(t)} - lp := url.URL{Scheme: httpScheme, Host: freeAddr(t)} - - cfg := embed.NewConfig() - cfg.Dir = t.TempDir() - cfg.LogLevel = "error" - cfg.ListenClientUrls = []url.URL{lc} - cfg.AdvertiseClientUrls = []url.URL{lc} - cfg.ListenPeerUrls = []url.URL{lp} - cfg.AdvertisePeerUrls = []url.URL{lp} - cfg.InitialCluster = cfg.Name + "=" + lp.String() - - e, err := embed.StartEtcd(cfg) - require.NoError(t, err) - t.Cleanup(e.Close) - - select { - case <-e.Server.ReadyNotify(): - case <-time.After(30 * time.Second): - t.Fatal("embedded etcd did not become ready") - } - - return lc.String() -} - // joinNode registers a member the way a storage node does, so the router resolves against a real // membership rather than a hand-built ring. func joinNode(t *testing.T, endpoint, root, id, addr string) { @@ -85,7 +41,7 @@ func joinNode(t *testing.T, endpoint, root, id, addr string) { func TestRouterCloseReportsSuccess(t *testing.T) { const root = "/test" - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) joinNode(t, endpoint, root, "node-a", "10.0.0.1:9000") r, err := router.Open(t.Context(), router.Config{ @@ -101,7 +57,7 @@ func TestRouterResolvesPlacement(t *testing.T) { const root = "/test" - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) for _, n := range []struct{ id, addr string }{ {"node-a", "10.0.0.1:9000"}, {"node-b", "10.0.0.2:9000"}, @@ -145,7 +101,7 @@ func TestRouterShardKeysMatchClusterDerivation(t *testing.T) { shards = 8 ) - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) joinNode(t, endpoint, root, "node-a", "10.0.0.1:9000") r, err := router.Open(t.Context(), router.Config{ @@ -187,19 +143,19 @@ func TestRouterPrimaryWrite(t *testing.T) { return cluster.Reject{OOO: 2}, nil })) - addr := freeAddr(t) - srv := &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second} - var lc net.ListenConfig - ln, err := lc.Listen(t.Context(), "tcp", addr) + ln, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") require.NoError(t, err) + addr := ln.Addr().String() + srv := &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second} + go func() { _ = srv.Serve(ln) }() t.Cleanup(func() { _ = srv.Close() }) - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) joinNode(t, endpoint, root, "node-a", addr) r, err := router.Open(t.Context(), router.Config{Etcd: []string{endpoint}, Root: root, RF: 1}) @@ -219,7 +175,7 @@ func TestRouterPrimaryWrite(t *testing.T) { func TestRouterPrimaryWriteFailsOnEmptyRing(t *testing.T) { t.Parallel() - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) r, err := router.Open(t.Context(), router.Config{Etcd: []string{endpoint}, Root: "/test"}) require.NoError(t, err) diff --git a/cluster/router/trace_test.go b/cluster/router/trace_test.go index 35403683..c849c975 100644 --- a/cluster/router/trace_test.go +++ b/cluster/router/trace_test.go @@ -10,6 +10,7 @@ import ( sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/cluster/router" "github.com/oteldb/storage/query/fetch" "github.com/oteldb/storage/signal" @@ -26,7 +27,7 @@ func openRouterTraced(t *testing.T, rec *tracetest.SpanRecorder, peers ...*peer) tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)) - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) for i, p := range peers { joinNode(t, endpoint, root, "node-"+string(rune('a'+i)), p.serve(t)) } diff --git a/cluster_admission_test.go b/cluster_admission_test.go index 2bab3fd6..4f5ccc40 100644 --- a/cluster_admission_test.go +++ b/cluster_admission_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/require" "github.com/oteldb/storage/backend" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/signal" "github.com/oteldb/storage/tenant" ) @@ -19,7 +20,7 @@ import ( // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterAdmissionStatsTallied(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() // MaxSeries is enforced per engine by the primary, so one metric series and one log stream both diff --git a/cluster_aggregate_gap_test.go b/cluster_aggregate_gap_test.go index eb41d12e..26d388a3 100644 --- a/cluster_aggregate_gap_test.go +++ b/cluster_aggregate_gap_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/engine" "github.com/oteldb/storage/query/fetch" "github.com/oteldb/storage/signal" @@ -52,7 +53,7 @@ func dropProfileEngine(t *testing.T, s *Storage, shard signal.TenantID) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestAggregateDoesNotDivergeAcrossCoordinators(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -106,7 +107,7 @@ func TestAggregateDoesNotDivergeAcrossCoordinators(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestProfileSymbolsDoNotDivergeAcrossCoordinators(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ diff --git a/cluster_bootstrap_signals_test.go b/cluster_bootstrap_signals_test.go index debec89f..042551d7 100644 --- a/cluster_bootstrap_signals_test.go +++ b/cluster_bootstrap_signals_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/query/fetch" "github.com/oteldb/storage/signal" ) @@ -22,7 +23,7 @@ import ( // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterGainedOwnerBootstrapsEverySignal(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ diff --git a/cluster_bootstrap_test.go b/cluster_bootstrap_test.go index 7160627b..d69264b5 100644 --- a/cluster_bootstrap_test.go +++ b/cluster_bootstrap_test.go @@ -9,6 +9,7 @@ import ( "github.com/stretchr/testify/require" "github.com/oteldb/storage/cluster" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/query/fetch" ) @@ -19,7 +20,7 @@ import ( // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterGainedOwnerBootstrap(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ diff --git a/cluster_completeness_test.go b/cluster_completeness_test.go index 70b89275..47b8c2b2 100644 --- a/cluster_completeness_test.go +++ b/cluster_completeness_test.go @@ -9,6 +9,7 @@ import ( "github.com/stretchr/testify/require" "github.com/oteldb/storage/cluster" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/signal" ) @@ -96,7 +97,7 @@ func TestCanAnswerSeparatesAbsenceFromIncompleteness(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestEnumerationFailsOverWhenLocalShardIsIncomplete(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ diff --git a/cluster_ec_chaos_test.go b/cluster_ec_chaos_test.go index db2cfefa..c672373b 100644 --- a/cluster_ec_chaos_test.go +++ b/cluster_ec_chaos_test.go @@ -11,6 +11,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/query/fetch" ) @@ -54,7 +55,7 @@ func TestClusterECChaos(t *testing.T) { func runECChaos(t *testing.T, seed uint64) { t.Helper() - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() rng := rand.New(rand.NewPCG(seed, seed+1)) diff --git a/cluster_ec_lazy_test.go b/cluster_ec_lazy_test.go index 3facd5ff..513410bf 100644 --- a/cluster_ec_lazy_test.go +++ b/cluster_ec_lazy_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/require" "github.com/oteldb/storage/backend/bucketindex" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/signal" ) @@ -49,7 +50,7 @@ func TestMaintainEcPartsLazy(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestMaintainEcPartsSharedSnapshot(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() s := openClusterNodeEC(t, endpoint, "node-a", 2, 1, 0) diff --git a/cluster_ec_routed_test.go b/cluster_ec_routed_test.go index acb1459c..6045b470 100644 --- a/cluster_ec_routed_test.go +++ b/cluster_ec_routed_test.go @@ -7,6 +7,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/oteldb/storage/cluster/etcd/etcdtest" ) // TestClusterECRoutedWritePreservesValues is a regression test for the interaction between a @@ -24,7 +26,7 @@ import ( // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterECRoutedWritePreservesValues(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ diff --git a/cluster_ec_test.go b/cluster_ec_test.go index 395bfcd5..24d41022 100644 --- a/cluster_ec_test.go +++ b/cluster_ec_test.go @@ -15,6 +15,7 @@ import ( "github.com/oteldb/storage/cluster" "github.com/oteldb/storage/cluster/ec" "github.com/oteldb/storage/cluster/etcd" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/query/fetch" "github.com/oteldb/storage/signal" "github.com/oteldb/storage/tenant" @@ -114,7 +115,7 @@ func openClusterNodeECDomains(t *testing.T, endpoint, id string, domains []strin // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterECRackAwarePlacement(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) // Six nodes: 3 racks × 2 servers, one node (disk) each. a := openClusterNodeECDomains(t, endpoint, "n1", []string{"rack1", "s1"}, 4, 2) @@ -158,7 +159,7 @@ func TestClusterECRackAwarePlacement(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterECRackShortfallDetected(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) // Three nodes, all in one rack (different servers) — cannot be rack-safe for ec(2,1). a := openClusterNodeECDomains(t, endpoint, "n1", []string{"rack1", "s1"}, 2, 1) @@ -189,7 +190,7 @@ func TestClusterECRackShortfallDetected(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterECSingleNodeConvertsAndServes(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() // Enough distinct series that the flushed value column clears the 4 KiB shard floor. @@ -227,7 +228,7 @@ func TestClusterECSingleNodeConvertsAndServes(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterECHotPartStaysFullCopy(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() s := openClusterNodeEC(t, endpoint, "node-a", 2, 1, time.Hour) // hot window: 1h @@ -267,7 +268,7 @@ func TestClusterECHotPartStaysFullCopy(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestECSchemeForGating(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) // EC policy but shared backend (PrivateBackend false) ⇒ EC does not apply. shared := openClusterNodeShared(t, endpoint, "node-a", @@ -287,7 +288,7 @@ func TestECSchemeForGating(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterECSlotFiltering(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() // ec(2,1) over three racks: one shard per node, one node per rack. @@ -400,7 +401,7 @@ func distinctSlots(t *testing.T, ctx context.Context, be backend.Backend) int { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterECShardRepair(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ diff --git a/cluster_fence_test.go b/cluster_fence_test.go index 5f755546..b63c0338 100644 --- a/cluster_fence_test.go +++ b/cluster_fence_test.go @@ -16,6 +16,7 @@ import ( "github.com/oteldb/storage/backend" "github.com/oteldb/storage/cluster" "github.com/oteldb/storage/cluster/etcd" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/internal/obs/obstest" "github.com/oteldb/storage/query/fetch" ) @@ -131,7 +132,7 @@ func (p *etcdProxy) restore() { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterFencedPrimaryRejectsWrites(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() proxies := map[string]*etcdProxy{"node-a": newEtcdProxy(t, endpoint), "node-b": newEtcdProxy(t, endpoint)} diff --git a/cluster_flushpressure_test.go b/cluster_flushpressure_test.go index 581b0322..aa3da493 100644 --- a/cluster_flushpressure_test.go +++ b/cluster_flushpressure_test.go @@ -9,6 +9,7 @@ import ( "github.com/stretchr/testify/require" "github.com/oteldb/storage/cluster/etcd" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/signal" ) @@ -23,7 +24,7 @@ func pressureFlushCluster(t *testing.T) (c *readPolicyCluster, owner, replica st t.Helper() ctx := context.Background() - c = newReadPolicyCluster(t, startEtcd(t), WithFlushInterval(int64(time.Hour))) + c = newReadPolicyCluster(t, etcdtest.Start(t), WithFlushInterval(int64(time.Hour))) c.write(t) owner, replica = c.compactionOwnerOf(t) diff --git a/cluster_labels_test.go b/cluster_labels_test.go index 94d7e687..5560478f 100644 --- a/cluster_labels_test.go +++ b/cluster_labels_test.go @@ -15,6 +15,7 @@ import ( "github.com/oteldb/storage/backend" "github.com/oteldb/storage/cluster" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/query/fetch" "github.com/oteldb/storage/signal" "github.com/oteldb/storage/signal/metric" @@ -119,7 +120,7 @@ func eqMatcher(name, value string) fetch.Matcher { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredLabelsUnionAcrossShards(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) const shards = 4 @@ -145,7 +146,7 @@ func TestClusteredLabelsUnionAcrossShards(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredLabelsFailOverFromShardlessOwner(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) const shards = 4 @@ -176,7 +177,7 @@ func TestClusteredLabelsTakeIndexRoute(t *testing.T) { otel.SetTextMapPropagator(propagation.TraceContext{}) t.Cleanup(func() { otel.SetTextMapPropagator(prev) }) - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ids := []string{"node-a", "node-b", "node-c"} recByID := make(map[string]*tracetest.SpanRecorder, len(ids)) @@ -237,7 +238,7 @@ func TestClusteredLabelsTakeIndexRoute(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterLabelsRecordSignalUnsupported(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := t.Context() nodes := map[string]*Storage{ @@ -273,7 +274,7 @@ func TestClusterLabelsRecordSignalUnsupported(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterLabelsRejectsUnpushableMatcher(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) nodes := map[string]*Storage{"node-a": openClusterNode(t, endpoint, "node-a")} awaitMembership(t, nodes) diff --git a/cluster_mirror_test.go b/cluster_mirror_test.go index f688a2a8..47f8d784 100644 --- a/cluster_mirror_test.go +++ b/cluster_mirror_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/require" "github.com/oteldb/storage/backend" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/signal" ) @@ -25,7 +26,7 @@ import ( // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterMirrorSkipsNonOwner(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ diff --git a/cluster_readpolicy_test.go b/cluster_readpolicy_test.go index 296b4df5..0c056711 100644 --- a/cluster_readpolicy_test.go +++ b/cluster_readpolicy_test.go @@ -11,6 +11,7 @@ import ( "github.com/oteldb/storage/backend" "github.com/oteldb/storage/cluster" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/query/fetch" "github.com/oteldb/storage/signal" ) @@ -162,7 +163,7 @@ func fetchLogs(ctx context.Context, s *Storage, start, end int64) ([]*fetch.Batc // //nolint:paralleltest // owns an embedded etcd; runs serially func TestReadOverlappingWantFailsOverToCompleteOwner(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() c := newReadPolicyCluster(t, endpoint) @@ -188,7 +189,7 @@ func TestReadOverlappingWantFailsOverToCompleteOwner(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestReadFailsWhenEveryOwnerIsIncomplete(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() c := newReadPolicyCluster(t, endpoint) @@ -211,7 +212,7 @@ func TestReadFailsWhenEveryOwnerIsIncomplete(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestReadOutsideTheWantIsServedLocally(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() c := newReadPolicyCluster(t, endpoint) @@ -239,7 +240,7 @@ func TestReadOutsideTheWantIsServedLocally(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestShardNoOwnerHoldsStillReadsEmpty(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() c := newReadPolicyCluster(t, endpoint) @@ -265,7 +266,7 @@ func TestShardNoOwnerHoldsStillReadsEmpty(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestSingleNodeReadFailsWithNothingToFailOverTo(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() s := openClusterNodePrivate(t, endpoint, "solo", 1) @@ -327,7 +328,7 @@ func loseOldestLogPart(t *testing.T, s *Storage, shard signal.TenantID) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestEnumerationSoleOwnerIncompleteFails(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() s := openClusterNodePrivate(t, endpoint, "solo", 1) @@ -353,7 +354,7 @@ func TestEnumerationSoleOwnerIncompleteFails(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestEnumerationLocalIncompleteDeniesRemoteAbsence(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() a := openClusterNodePrivate(t, endpoint, "node-a", 2) diff --git a/cluster_recovered_seams_test.go b/cluster_recovered_seams_test.go index 9a0f6774..b61204b0 100644 --- a/cluster_recovered_seams_test.go +++ b/cluster_recovered_seams_test.go @@ -13,6 +13,7 @@ import ( "github.com/oteldb/storage/backend/bucketindex" "github.com/oteldb/storage/backend/file" "github.com/oteldb/storage/cluster" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/engine" "github.com/oteldb/storage/signal" ) @@ -51,7 +52,7 @@ func newRecoveredPair(t *testing.T) *recoveredPair { t.Helper() p := &recoveredPair{ - endpoint: startEtcd(t), + endpoint: etcdtest.Start(t), dirs: map[string]string{"node-a": t.TempDir(), "node-b": t.TempDir()}, nodes: make(map[string]*Storage, 2), } diff --git a/cluster_recovery_test.go b/cluster_recovery_test.go index b7da8ef4..b7cc2b6b 100644 --- a/cluster_recovery_test.go +++ b/cluster_recovery_test.go @@ -11,6 +11,7 @@ import ( "github.com/oteldb/storage/backend/file" "github.com/oteldb/storage/cluster" "github.com/oteldb/storage/cluster/etcd" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/query/fetch" "github.com/oteldb/storage/signal" ) @@ -56,7 +57,7 @@ func TestClusterDurableWithWALAccepted(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterRestartServesUnflushedHead(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() dataDir, walDir := t.TempDir(), t.TempDir() @@ -107,7 +108,7 @@ func TestClusterRestartServesUnflushedHead(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterRestartedReplicaFailsOverInsteadOfServingAHole(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() dataDir := t.TempDir() // one shared durable backend, as a cluster over an object store has walDirs := map[string]string{"node-a": t.TempDir(), "node-b": t.TempDir()} @@ -214,7 +215,7 @@ func TestClusterRestartedReplicaFailsOverInsteadOfServingAHole(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterLaggingMirrorReplicaAnswersCompletely(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ diff --git a/cluster_repair_test.go b/cluster_repair_test.go index 42d00047..0f440d13 100644 --- a/cluster_repair_test.go +++ b/cluster_repair_test.go @@ -11,6 +11,7 @@ import ( "github.com/oteldb/storage/backend/bucketindex" "github.com/oteldb/storage/cluster" "github.com/oteldb/storage/cluster/etcd" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/signal" ) @@ -38,7 +39,7 @@ func openRepairNode(t *testing.T, endpoint, id string, rf int) *Storage { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestRepairCompleteOwnerSetIsAbsence(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() a := openRepairNode(t, endpoint, "a", 2) @@ -65,7 +66,7 @@ func TestRepairCompleteOwnerSetIsAbsence(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestRepairShortOwnerSetIsIncomplete(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() a := openRepairNode(t, endpoint, "a", 3) @@ -92,7 +93,7 @@ func TestRepairShortOwnerSetIsIncomplete(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestRepairSingleOwnerShardIsComplete(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() a := openRepairNode(t, endpoint, "solo", 1) diff --git a/cluster_retry_test.go b/cluster_retry_test.go index 66df28a6..49038308 100644 --- a/cluster_retry_test.go +++ b/cluster_retry_test.go @@ -12,6 +12,7 @@ import ( "github.com/oteldb/storage/backend" "github.com/oteldb/storage/cluster" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/internal/obs" "github.com/oteldb/storage/query/fetch" "github.com/oteldb/storage/reliability" @@ -168,7 +169,7 @@ func partition(t *testing.T, s *Storage) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterReadSurvivesDownOwner(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() ids := []string{"node-a", "node-b", "node-c"} @@ -217,7 +218,7 @@ func TestClusterReadSurvivesDownOwner(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterProfileEnumSurvivesDownOwner(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() ids := []string{"node-a", "node-b", "node-c"} diff --git a/cluster_routed_test.go b/cluster_routed_test.go index 7bd62637..bb49a7e2 100644 --- a/cluster_routed_test.go +++ b/cluster_routed_test.go @@ -9,6 +9,7 @@ import ( "go.opentelemetry.io/otel/sdk/metric/metricdata" "github.com/oteldb/storage/backend" + "github.com/oteldb/storage/cluster/etcd/etcdtest" ) // TestClusterRoutedPointsCountedOnRoutingNode pins the coordinator side of a clustered write: @@ -19,7 +20,7 @@ import ( // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterRoutedPointsCountedOnRoutingNode(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() ids := []string{"node-a", "node-b", "node-c"} diff --git a/cluster_rpc_trace_test.go b/cluster_rpc_trace_test.go index 6ff781c6..f347b35c 100644 --- a/cluster_rpc_trace_test.go +++ b/cluster_rpc_trace_test.go @@ -13,6 +13,7 @@ import ( "go.opentelemetry.io/otel/sdk/trace/tracetest" "github.com/oteldb/storage/backend" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/query/fetch" ) @@ -28,7 +29,7 @@ func TestClusterFetchFanOutSpans(t *testing.T) { otel.SetTextMapPropagator(propagation.TraceContext{}) t.Cleanup(func() { otel.SetTextMapPropagator(prev) }) - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() ids := []string{"node-a", "node-b", "node-c"} diff --git a/cluster_sampling_test.go b/cluster_sampling_test.go index f220a420..e87f0853 100644 --- a/cluster_sampling_test.go +++ b/cluster_sampling_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/require" "github.com/oteldb/storage/backend" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/query/fetch" "github.com/oteldb/storage/signal" "github.com/oteldb/storage/tenant" @@ -20,7 +21,7 @@ import ( func samplingNodes(t *testing.T, budget int64) (nodes map[string]*Storage, nonOwner *Storage) { t.Helper() - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) pol := WithTenancy(tenant.ResolverFunc(func(signal.TenantID) tenant.Policy { return tenant.Policy{Sampling: tenant.Sampling{MaxRowsPerSecond: budget}} })) diff --git a/cluster_shardwarn_test.go b/cluster_shardwarn_test.go index 83767cfa..0e457eb6 100644 --- a/cluster_shardwarn_test.go +++ b/cluster_shardwarn_test.go @@ -11,6 +11,7 @@ import ( "go.uber.org/zap/zaptest/observer" "github.com/oteldb/storage/backend" + "github.com/oteldb/storage/cluster/etcd/etcdtest" ) const singleShardMsg = "cluster is running one shard per tenant: writes cannot spread" @@ -22,7 +23,7 @@ const singleShardMsg = "cluster is running one shard per tenant: writes cannot s // //nolint:paralleltest // owns an embedded etcd; runs serially func TestSingleShardWarning(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() core, logs := observer.New(zap.WarnLevel) @@ -58,7 +59,7 @@ func TestSingleShardWarning(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestSingleShardWarningSilentOnOneNode(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) core, logs := observer.New(zap.WarnLevel) s := openClusterNodeWith(t, endpoint, "solo", backend.Memory(), WithLogger(zap.New(core))) @@ -74,7 +75,7 @@ func TestSingleShardWarningSilentOnOneNode(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestSingleShardWarningSilentWhenSharded(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) core, logs := observer.New(zap.WarnLevel) diff --git a/cluster_test.go b/cluster_test.go index b807b13f..af0c5e04 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -2,22 +2,19 @@ package storage import ( "context" - "net" - "net/url" "strconv" "strings" "testing" "time" - "github.com/go-faster/errors" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "go.etcd.io/etcd/server/v3/embed" "github.com/oteldb/storage/backend" "github.com/oteldb/storage/backend/s3/s3test" "github.com/oteldb/storage/cluster" "github.com/oteldb/storage/cluster/etcd" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/engine" "github.com/oteldb/storage/query/fetch" qprofile "github.com/oteldb/storage/query/profile" @@ -28,86 +25,6 @@ import ( "github.com/oteldb/storage/tenant" ) -// freeAddr returns a free localhost host:port for the embedded etcd's advertise URLs (etcd -// needs concrete URLs up front, unlike the cluster nodes, which bind ":0" and advertise the -// listener's actual port). Probe-then-rebind is racy in principle; etcd is the only remaining -// user, two ports per test. -func freeAddr(t *testing.T) string { - t.Helper() - - var lc net.ListenConfig - - l, err := lc.Listen(context.Background(), "tcp", "127.0.0.1:0") - require.NoError(t, err) - addr := l.Addr().String() - require.NoError(t, l.Close()) - - return addr -} - -// etcdStartAttempts bounds startEtcd's retries. freeAddr can only reserve a port by closing its -// listener, so between that close and etcd's own bind another test — or a socket still in -// TIME_WAIT — can take it, and the whole suite fails on "address already in use". Retrying with -// fresh ports costs nothing when the first attempt works. -const etcdStartAttempts = 5 - -// startEtcd boots an in-process single-node etcd and returns its client endpoint URL, retrying on -// a port collision (see [etcdStartAttempts]). -func startEtcd(t *testing.T) string { - t.Helper() - - var err error - - for range etcdStartAttempts { - var endpoint string - - if endpoint, err = tryStartEtcd(t); err == nil { - return endpoint - } - - t.Logf("embedded etcd failed to start, retrying with fresh ports: %v", err) - } - - require.NoError(t, err, "embedded etcd did not start in %d attempts", etcdStartAttempts) - - return "" -} - -// tryStartEtcd is one startEtcd attempt: it registers the shutdown only once the server is up, so a -// failed attempt leaves nothing behind. -func tryStartEtcd(t *testing.T) (string, error) { - t.Helper() - - lc := url.URL{Scheme: httpScheme, Host: freeAddr(t)} - lp := url.URL{Scheme: httpScheme, Host: freeAddr(t)} - - cfg := embed.NewConfig() - cfg.Dir = t.TempDir() - cfg.LogLevel = "error" - cfg.ListenClientUrls = []url.URL{lc} - cfg.AdvertiseClientUrls = []url.URL{lc} - cfg.ListenPeerUrls = []url.URL{lp} - cfg.AdvertisePeerUrls = []url.URL{lp} - cfg.InitialCluster = cfg.Name + "=" + lp.String() - - e, err := embed.StartEtcd(cfg) - if err != nil { - return "", err - } - - select { - case <-e.Server.ReadyNotify(): - case <-time.After(30 * time.Second): - e.Close() - - return "", errors.New("embedded etcd did not become ready") - } - - t.Cleanup(e.Close) - - return lc.String(), nil -} - func openClusterNode(t *testing.T, endpoint, id string) *Storage { t.Helper() @@ -206,7 +123,7 @@ func awaitMembership(t *testing.T, nodes map[string]*Storage) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterPerSeriesShardingSpreadsAndGathers(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() const shards = 4 @@ -273,7 +190,7 @@ func TestClusterPerSeriesShardingSpreadsAndGathers(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterAggregateWindowGathersAcrossShards(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() const ( @@ -348,7 +265,7 @@ func TestClusterAggregateWindowGathersAcrossShards(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredStorageReplicatesAcrossNodes(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() a := openClusterNode(t, endpoint, "node-a") @@ -383,7 +300,7 @@ func TestClusteredStorageReplicatesAcrossNodes(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterOnlyPrimaryCompacts(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -434,7 +351,7 @@ func TestClusterOnlyPrimaryCompacts(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterReplicaTrimsHeadAfterOwnerFlush(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) newBackend := s3test.Shared(t, "oteldb") ctx := context.Background() @@ -492,7 +409,7 @@ func TestClusterReplicaTrimsHeadAfterOwnerFlush(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredLogsReplicateAndRead(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -520,7 +437,7 @@ func TestClusteredLogsReplicateAndRead(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredTracesReplicateAndRead(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -559,7 +476,7 @@ func TestClusteredTracesReplicateAndRead(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredProfilesReplicateAndRead(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -614,7 +531,7 @@ func TestClusteredProfilesReplicateAndRead(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredLogsAccountForRejected(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() s := openClusterNodeWith(t, endpoint, "node-a", backend.Memory(), WithOOOWindow(50)) @@ -641,7 +558,7 @@ func TestClusteredLogsAccountForRejected(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterPrimaryAccountsForRejectedSamples(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() s := openClusterNodeWith(t, endpoint, "node-a", backend.Memory(), WithOOOWindow(50)) @@ -669,7 +586,7 @@ func TestClusterPrimaryAccountsForRejectedSamples(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterPrimaryAppliesCardinalityLimit(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() s := openClusterNodeWith(t, endpoint, "node-a", backend.Memory(), @@ -699,7 +616,7 @@ func TestClusterPrimaryAppliesCardinalityLimit(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredReadFansOutToOwners(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -766,7 +683,7 @@ func TestClusteredReadFansOutToOwners(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredLogEnumerationFansOut(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -828,7 +745,7 @@ func TestClusteredLogEnumerationFansOut(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestInspectClusterSection(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() s := openClusterNodeShared(t, endpoint, "node-a") @@ -858,7 +775,7 @@ func TestInspectClusterSection(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestAdminCompactOwnershipGate(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -893,7 +810,7 @@ func TestAdminCompactOwnershipGate(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredRecordShardingGathersAllStreams(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() const shards = 4 @@ -929,7 +846,7 @@ func TestClusteredRecordShardingGathersAllStreams(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredRecordShardingEnumeratesAcrossShards(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() const shards = 4 @@ -980,7 +897,7 @@ func TestClusteredRecordShardingEnumeratesAcrossShards(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredShardedTraceByID(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() const shards = 4 @@ -1020,7 +937,7 @@ func TestClusteredShardedTraceByID(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredShardedProfileResolver(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() const shards = 4 @@ -1064,7 +981,7 @@ func TestClusteredShardedProfileResolver(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterReadFailsOverFromShardlessOwner(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() const shards = 4 @@ -1128,7 +1045,7 @@ func TestShardHelpers(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterExplainAnalyzeRemote(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -1189,7 +1106,7 @@ func TestClusterExplainAnalyzeRemote(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterPerTenantRF(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() // Every tenant resolves to RF=3 (the "gold" durability tier); the cluster default stays 2. @@ -1236,7 +1153,7 @@ func TestClusterPerTenantRF(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterRFDefaultsWithoutPolicy(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) s := openClusterNode(t, endpoint, "node-a") // cluster.Config.RF = 2, no Tenancy require.Equal(t, 2, s.rfFor("default"), "zero durability policy falls back to cluster RF") @@ -1267,7 +1184,7 @@ func openClusterNodePrivate(t *testing.T, endpoint, id string, rf int) *Storage // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterSharedNothingReplicatesParts(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -1336,7 +1253,7 @@ func TestClusterSharedNothingReplicatesParts(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterSharedNothingSurvivesNodeLoss(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -1401,7 +1318,7 @@ func TestClusterSharedNothingSurvivesNodeLoss(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterSharedNothingReplicatesLogParts(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -1472,7 +1389,7 @@ func TestSplitEnginePrefix(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterSharedNothingPushNotify(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ @@ -1522,7 +1439,7 @@ func TestClusterSharedNothingPushNotify(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredReadFanOutKeepsScaleFactors(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{ diff --git a/cluster_trace_test.go b/cluster_trace_test.go index a876bca4..7a6f2ec1 100644 --- a/cluster_trace_test.go +++ b/cluster_trace_test.go @@ -14,6 +14,7 @@ import ( "go.uber.org/zap/zaptest/observer" "github.com/oteldb/storage/backend" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/query/fetch" ) @@ -42,7 +43,7 @@ func TestClusterTracePropagationCorrelatesLogs(t *testing.T) { otel.SetTextMapPropagator(propagation.TraceContext{}) t.Cleanup(func() { otel.SetTextMapPropagator(prev) }) - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() ids := []string{"node-a", "node-b", "node-c"} diff --git a/clusterprivacy_test.go b/clusterprivacy_test.go index cf5cfab5..cacbba0d 100644 --- a/clusterprivacy_test.go +++ b/clusterprivacy_test.go @@ -15,6 +15,7 @@ import ( "github.com/oteldb/storage/backend/file" "github.com/oteldb/storage/cluster" "github.com/oteldb/storage/cluster/etcd" + "github.com/oteldb/storage/cluster/etcd/etcdtest" ) // sharedStore is a backend every node can read: it declines the [backend.NodeLocal] capability the @@ -56,7 +57,7 @@ func TestNodeLocalBackendUnshared(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterRefusesNodeLocalBackendUnshared(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) _, err := Open(context.Background(), Options{}, WithBackend(backend.Memory()), WithCluster(&cluster.Config{ @@ -75,7 +76,7 @@ func TestClusterRefusesNodeLocalBackendUnshared(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusterQuietWhenPrivateBackendDeclared(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) core, logs := observer.New(zap.WarnLevel) diff --git a/repair_repro_test.go b/repair_repro_test.go index 45c87d37..5bf7a4a7 100644 --- a/repair_repro_test.go +++ b/repair_repro_test.go @@ -7,6 +7,7 @@ import ( "github.com/stretchr/testify/require" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/query/fetch" "github.com/oteldb/storage/signal" ) @@ -56,7 +57,7 @@ func (c *readPolicyCluster) logStats(id string) (parts int, wanted int, holes in //nolint:paralleltest // owns an embedded etcd; runs serially func TestRepro387ReplicaInstallsDamagedIndexThenOwnerHoles(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() c := newReadPolicyCluster(t, endpoint) @@ -119,7 +120,7 @@ func TestRepro387ReplicaInstallsDamagedIndexThenOwnerHoles(t *testing.T) { //nolint:paralleltest // owns an embedded etcd; runs serially func TestRepro387HoleFreezesReplicaIndexInstall(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() c := newReadPolicyCluster(t, endpoint) @@ -173,7 +174,7 @@ func TestRepro387HoleFreezesReplicaIndexInstall(t *testing.T) { //nolint:paralleltest // owns an embedded etcd; runs serially func TestRepro387ReplicaLossPropagatesToOwner(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() c := newReadPolicyCluster(t, endpoint) diff --git a/replica_omitted_repro_test.go b/replica_omitted_repro_test.go index a93c90c3..450b4b69 100644 --- a/replica_omitted_repro_test.go +++ b/replica_omitted_repro_test.go @@ -9,6 +9,7 @@ import ( "github.com/oteldb/storage/backend" "github.com/oteldb/storage/backend/bucketindex" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/signal" ) @@ -20,7 +21,7 @@ import ( // //nolint:paralleltest // owns an embedded etcd; runs serially func TestRepro556OwnerSilentlyOmitsAPartTheReplicaHolds(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() c := newReadPolicyCluster(t, endpoint) diff --git a/replica_reconcile_repro_test.go b/replica_reconcile_repro_test.go index 891b13b4..7ed46131 100644 --- a/replica_reconcile_repro_test.go +++ b/replica_reconcile_repro_test.go @@ -9,6 +9,7 @@ import ( "github.com/oteldb/storage/backend" "github.com/oteldb/storage/backend/bucketindex" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/signal" ) @@ -30,7 +31,7 @@ func (c *readPolicyCluster) readIndex(t *testing.T, id, indexKey string) []byte // //nolint:paralleltest // owns an embedded etcd; runs serially func TestRepro551ReplicaLossUnderStableOwnerIndex(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() c := newReadPolicyCluster(t, endpoint) diff --git a/replica_refresh_repro_test.go b/replica_refresh_repro_test.go index 855a8397..6556a4d5 100644 --- a/replica_refresh_repro_test.go +++ b/replica_refresh_repro_test.go @@ -9,6 +9,7 @@ import ( "github.com/stretchr/testify/require" "github.com/oteldb/storage/backend/bucketindex" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/signal" ) @@ -19,7 +20,7 @@ import ( // //nolint:paralleltest // owns an embedded etcd; runs serially func TestRepro557ReplicaRefreshRecordsGonePart(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() c := newReadPolicyCluster(t, endpoint) diff --git a/traces_keys_test.go b/traces_keys_test.go index 2cd7fe62..c5cc774d 100644 --- a/traces_keys_test.go +++ b/traces_keys_test.go @@ -7,6 +7,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/oteldb/storage/cluster/etcd/etcdtest" "github.com/oteldb/storage/signal" "github.com/oteldb/storage/signal/trace" ) @@ -157,7 +158,7 @@ func TestFacadeTraceKeysClosed(t *testing.T) { // //nolint:paralleltest // owns an embedded etcd; runs serially func TestClusteredTraceKeysFansOut(t *testing.T) { - endpoint := startEtcd(t) + endpoint := etcdtest.Start(t) ctx := context.Background() nodes := map[string]*Storage{