diff --git a/etcd/embed.go b/etcd/embed.go index 06c081e74..a40b343f8 100644 --- 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 { diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go index 95e804ec0..c156e9125 100644 --- 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)) + } } }) diff --git a/internal/clustertests/docker-compose.yml b/internal/clustertests/docker-compose.yml index 36b418921..37b7519e3 100644 --- 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: diff --git a/server/config.go b/server/config.go index bf13b15c1..29c3434a6 100644 --- 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 }