Merge pull request #1468 from travisturner/disco-clustertests

adjust clustertests to have etcd config
2026-08-28 10:54:59 +00:00 · 2021-02-26 17:49:20 -06:00 · 2021-02-26 17:49:20 -06:00 · 4add6134b8
commit 4add6134b8
parent d455fb3105 ef40bbb617
4 changed files with 122 additions and 43 deletions
--- a/etcd/embed.go
+++ b/etcd/embed.go
@ -37,6 +37,7 @@ import (
 	"go.etcd.io/etcd/clientv3/concurrency"
 	"go.etcd.io/etcd/embed"
 	"go.etcd.io/etcd/etcdserver/api/v3client"
+	"go.etcd.io/etcd/etcdserver/api/v3rpc/rpctypes"
 	"go.etcd.io/etcd/mvcc"
 	"go.etcd.io/etcd/mvcc/mvccpb"
 	"go.etcd.io/etcd/pkg/types"
@ -217,23 +218,30 @@ func (e *Etcd) Start(ctx context.Context) (disco.InitialClusterState, error) {
 }

 func (e *Etcd) startHeartbeat() error {
-	heartbeatID, ctx, heartbeatCancel, err := e.leaseKeepAlive(context.Background(), e.options.HeartbeatTTL)
+	ctx, heartbeatCancel := context.WithCancel(context.Background())
+	e.heartbeatCancel = heartbeatCancel
+
+	cb := func(heartbeatID clientv3.LeaseID) error {
+		key, value := heartbeatPrefix+e.e.Server.ID().String(), disco.ClusterStateStarting
+		if e.e.Config().ClusterState == embed.ClusterStateFlagExisting {
+			value = disco.ClusterStateResizing
+		}
+
+		if _, err := e.cli.Put(ctx, key, string(value), clientv3.WithLease(heartbeatID)); err != nil {
+			heartbeatCancel()
+			return errors.Wrapf(err, "startHeartbeat: puts a key-value (%s, %s) with lease (%v)", key, value, heartbeatID)
+		}
+
+		e.heartbeatID = heartbeatID
+
+		return nil
+	}
+
+	_, err := e.leaseKeepAlive(ctx, heartbeatCancel, e.options.HeartbeatTTL, cb)
 	if err != nil {
-		return errors.Wrap(err, "startHeartbeat: creates a new hearbeat")
+		return errors.Wrap(err, "startHeartbeat: creates a new heartbeat")
 	}

-	key, value := heartbeatPrefix+e.e.Server.ID().String(), disco.ClusterStateStarting
-	if e.e.Config().ClusterState == embed.ClusterStateFlagExisting {
-		value = disco.ClusterStateResizing
-	}
-
-	if _, err := e.cli.Put(ctx, key, string(value), clientv3.WithLease(heartbeatID)); err != nil {
-		heartbeatCancel()
-		return errors.Wrapf(err, "startHeartbeat: puts a key-value (%s, %s) with lease (%v)", key, value, heartbeatID)
-	}
-
-	e.heartbeatID, e.heartbeatCancel = heartbeatID, heartbeatCancel
-
 	return nil
 }

@ -368,7 +376,11 @@ func (e *Etcd) ClusterState(ctx context.Context) (disco.ClusterState, error) {
 }

 func (e *Etcd) Resize(ctx context.Context) (func([]byte) error, error) {
-	resizeID, ctx, resizeCancel, err := e.leaseKeepAlive(ctx, e.options.HeartbeatTTL)
+	ctx, resizeCancel := context.WithCancel(ctx)
+
+	cb := func(clientv3.LeaseID) error { return nil }
+
+	resizeID, err := e.leaseKeepAlive(ctx, resizeCancel, e.options.HeartbeatTTL, cb)
 	if err != nil {
 		return nil, errors.Wrap(err, "Resize: creates a new hearbeat")
 	}
@ -707,21 +719,24 @@ func (e *Etcd) delKey(ctx context.Context, key string, withPrefix bool) (err err
 // leaseKeepAlive creates a lease with the given ttl (treated as a time.Duration),
 // then refreshes it periodically, and cancels it when done. it yields the lease ID,
 // and also a context and cancelfunc that can be used to abort the heartbeat.
-func (e *Etcd) leaseKeepAlive(ctx context.Context, ttl int64) (clientv3.LeaseID, context.Context, context.CancelFunc, error) {
-	ctx, cancelFunc := context.WithCancel(ctx)
+func (e *Etcd) leaseKeepAlive(ctx context.Context, cancelFunc context.CancelFunc, ttl int64, cb func(clientv3.LeaseID) error) (clientv3.LeaseID, error) {
 	leaseResp, err := e.cli.Grant(ctx, ttl)
 	if err != nil {
 		cancelFunc()
-		return 0, nil, nil, errors.Wrapf(err, "leaseKeepAlive: creates a new lease (TTL: %v)", ttl)
+		return 0, errors.Wrapf(err, "leaseKeepAlive: creates a new lease (TTL: %v)", ttl)
 	}

-	keepaliveFunc := func(tick time.Duration) {
+	keepaliveFunc := func(tick time.Duration) error {
 		ticker := time.NewTicker(tick)
 		defer func() {
 			ticker.Stop()
 			e.wg.Done()
 		}()

+		// leaseResp is a var within the function because we may need to reset
+		// it later if the lease has to be re-granted.
+		var leaseResp *clientv3.LeaseGrantResponse = leaseResp
+
 		for {
 			select {
 			case <-ctx.Done():
@ -733,10 +748,37 @@ func (e *Etcd) leaseKeepAlive(ctx context.Context, ttl int64) (clientv3.LeaseID,

 				if _, err := e.cli.Revoke(revoker, leaseResp.ID); err != nil {
 					log.Printf("leaseKeepAlive: revokes the lease (ID: %x): %#v\n", leaseResp.ID, err)
+					return errors.Wrap(err, "revoking lease")
 				}
-				return
+				return nil
 			case <-ticker.C:
-				if _, err = e.cli.KeepAliveOnce(ctx, leaseResp.ID); err != nil {
+				_, err := e.cli.KeepAliveOnce(ctx, leaseResp.ID)
+				if err == rpctypes.ErrLeaseNotFound {
+					// We create a new client here because in the case where we
+					// have lost track of the lease, it's likely that we've also
+					// lost the client at e.cli.
+					// TODO: should this close/reset e.cli instead?
+					cli := &hookedClient{Client: v3client.New(e.e.Server)}
+					var err error
+					leaseResp, err = cli.Grant(ctx, ttl)
+					cli.Close()
+					if err != nil {
+						cancelFunc()
+						return errors.Wrapf(err, "leaseKeepAlive: creates a new lease (TTL: %v)", ttl)
+					}
+
+					// Call the callback.
+					if err := cb(leaseResp.ID); err != nil {
+						cancelFunc()
+						return errors.Wrap(err, "calling callback")
+					}
+
+					// TODO: this can't be here in this general function because resize doesn't need this.
+					if err := e.Started(ctx); err != nil {
+						cancelFunc()
+						return errors.Wrap(err, "setting to started")
+					}
+				} else if err != nil {
 					log.Printf("leaseKeepAlive: renews the lease (ID: %x): %v\n", leaseResp.ID, err)
 				}
 			}
@ -744,9 +786,17 @@ func (e *Etcd) leaseKeepAlive(ctx context.Context, ttl int64) (clientv3.LeaseID,
 	}

 	e.wg.Add(1)
-	go keepaliveFunc(time.Second)
+	go func() {
+		if err := keepaliveFunc(time.Second); err != nil {
+			log.Printf("leaseKeepAlive: goroutine err: %v\n", err)
+		}
+	}()

-	return leaseResp.ID, ctx, cancelFunc, nil
+	if err := cb(leaseResp.ID); err != nil {
+		return 0, errors.Wrap(err, "calling callback")
+	}
+
+	return leaseResp.ID, nil
 }

 type hookedClient struct {
--- a/internal/clustertests/cluster_test.go
+++ b/internal/clustertests/cluster_test.go
@ -29,17 +29,25 @@ func TestClusterStuff(t *testing.T) {
 	if os.Getenv("ENABLE_PILOSA_CLUSTER_TESTS") != "1" {
 		t.Skip()
 	}
-	cli, err := picli.NewInternalClient("pilosa1:10101", picli.GetHTTPClient(nil))
+	cli1, err := picli.NewInternalClient("pilosa1:10101", picli.GetHTTPClient(nil))
+	if err != nil {
+		t.Fatalf("getting client: %v", err)
+	}
+	cli2, err := picli.NewInternalClient("pilosa2:10101", picli.GetHTTPClient(nil))
+	if err != nil {
+		t.Fatalf("getting client: %v", err)
+	}
+	cli3, err := picli.NewInternalClient("pilosa3:10101", picli.GetHTTPClient(nil))
 	if err != nil {
 		t.Fatalf("getting client: %v", err)
 	}

 	t.Run("long pause", func(t *testing.T) {
-		err := cli.CreateIndex(context.Background(), "testidx", pilosa.IndexOptions{})
+		err := cli1.CreateIndex(context.Background(), "testidx", pilosa.IndexOptions{})
 		if err != nil {
 			t.Fatalf("creating index: %v", err)
 		}
-		err = cli.CreateFieldWithOptions(context.Background(), "testidx", "testf", pilosa.FieldOptions{CacheType: pilosa.CacheTypeRanked, CacheSize: 100})
+		err = cli1.CreateFieldWithOptions(context.Background(), "testidx", "testf", pilosa.FieldOptions{CacheType: pilosa.CacheTypeRanked, CacheSize: 100})
 		if err != nil {
 			t.Fatalf("creating field: %v", err)
 		}
@ -50,19 +58,22 @@ func TestClusterStuff(t *testing.T) {
 			data[i%10].ColumnID = uint64((i/10)*pilosa.ShardWidth + i%10)
 			shard := uint64(i / 10)
 			if i%10 == 9 {
-				err = cli.Import(context.Background(), "testidx", "testf", shard, data)
+				err = cli1.Import(context.Background(), "testidx", "testf", shard, data)
 				if err != nil {
 					t.Fatalf("importing: %v", err)
 				}
 			}
 		}

-		r, err := cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"})
-		if err != nil {
-			t.Fatalf("count querying: %v", err)
-		}
-		if r.Results[0].(uint64) != 1000 {
-			t.Fatalf("count after import is %d", r.Results[0].(uint64))
+		// Check query results from each node.
+		for i, cli := range []*picli.InternalClient{cli1, cli2, cli3} {
+			r, err := cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"})
+			if err != nil {
+				t.Fatalf("count querying pilosa%d: %v", i, err)
+			}
+			if r.Results[0].(uint64) != 1000 {
+				t.Fatalf("count on pilosa%d after import is %d", i, r.Results[0].(uint64))
+			}
 		}

 		pcmd := exec.Command("/pumba", "pause", "clustertests_pilosa3_1", "--duration", "10s")
@ -83,12 +94,15 @@ func TestClusterStuff(t *testing.T) {
 		time.Sleep(time.Second * 20)
 		t.Log("done waiting for stability")

-		r, err = cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"})
-		if err != nil {
-			t.Fatalf("count querying: %v", err)
-		}
-		if r.Results[0].(uint64) != 1000 {
-			t.Fatalf("count after import is %d", r.Results[0].(uint64))
+		// Check query results from each node.
+		for i, cli := range []*picli.InternalClient{cli1, cli2, cli3} {
+			r, err := cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"})
+			if err != nil {
+				t.Fatalf("count querying pilosa%d: %v", i, err)
+			}
+			if r.Results[0].(uint64) != 1000 {
+				t.Fatalf("count on pilosa%d after import is %d", i, r.Results[0].(uint64))
+			}
 		}
 	})

--- a/internal/clustertests/docker-compose.yml
+++ b/internal/clustertests/docker-compose.yml
@ -8,8 +8,12 @@ services:
    ports:
      - "33455:10101"
    environment:
-      - PILOSA_CLUSTER_COORDINATOR=true
-      - PILOSA_GOSSIP_SEEDS=pilosa1:14000
+      - PILOSA_NAME=pilosa1
+      - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
+      - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa1:10201
+      - PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301
+      - PILOSA_ETCD_ADVERTISE_PEER_ADDRESS=http://pilosa1:10301
+      - PILOSA_ETCD_INITIAL_CLUSTER=pilosa1=http://pilosa1:10301,pilosa2=http://pilosa2:10301,pilosa3=http://pilosa3:10301
    networks:
      - pilosanet
    command:
@ -22,7 +26,12 @@ services:
    ports:
      - "33456:10101"
    environment:
-      - PILOSA_GOSSIP_SEEDS=pilosa1:14000
+      - PILOSA_NAME=pilosa2
+      - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
+      - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa2:10201
+      - PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301
+      - PILOSA_ETCD_ADVERTISE_PEER_ADDRESS=http://pilosa2:10301
+      - PILOSA_ETCD_INITIAL_CLUSTER=pilosa1=http://pilosa1:10301,pilosa2=http://pilosa2:10301,pilosa3=http://pilosa3:10301
    networks:
      - pilosanet
    command:
@ -35,7 +44,12 @@ services:
    ports:
      - "33457:10101"
    environment:
-      - PILOSA_GOSSIP_SEEDS=pilosa1:14000,pilosa2:14000
+      - PILOSA_NAME=pilosa3
+      - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
+      - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa3:10201
+      - PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301
+      - PILOSA_ETCD_ADVERTISE_PEER_ADDRESS=http://pilosa3:10301
+      - PILOSA_ETCD_INITIAL_CLUSTER=pilosa1=http://pilosa1:10301,pilosa2=http://pilosa2:10301,pilosa3=http://pilosa3:10301
    networks:
      - pilosanet
    command:
--- a/server/config.go
+++ b/server/config.go
@ -367,6 +367,7 @@ func NewConfig() *Config {
 	c.Etcd.Name = ""
 	c.Etcd.ClusterName = ""
 	c.Etcd.InitCluster = c.Name + "=" + c.Etcd.LPeerURL
+	c.Etcd.HeartbeatTTL = 5

 	return c
 }