diff --git a/etcd/embed.go b/etcd/embed.go index 64769b8c9..06c081e74 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -18,12 +18,12 @@ import ( "bytes" "context" "encoding/json" - "fmt" "log" "net" "path" "sort" "strings" + "sync" "time" "github.com/pilosa/pilosa/v2" @@ -36,6 +36,7 @@ import ( "go.etcd.io/etcd/clientv3/clientv3util" "go.etcd.io/etcd/clientv3/concurrency" "go.etcd.io/etcd/embed" + "go.etcd.io/etcd/etcdserver/api/v3client" "go.etcd.io/etcd/mvcc" "go.etcd.io/etcd/mvcc/mvccpb" "go.etcd.io/etcd/pkg/types" @@ -90,13 +91,16 @@ type Etcd struct { lm leaseMetadata - e *embed.Etcd + e *embed.Etcd + cli *hookedClient + wg *sync.WaitGroup } func NewEtcd(opt Options, replicas int) *Etcd { e := &Etcd{ options: opt, replicas: replicas, + wg: &sync.WaitGroup{}, } return e } @@ -104,6 +108,7 @@ func NewEtcd(opt Options, replicas int) *Etcd { // Close implements io.Closer func (e *Etcd) Close() error { _ = testhook.Closed(pilosa.NewAuditor(), e, nil) + if e.e != nil { if e.resizeCancel != nil { e.resizeCancel() @@ -111,10 +116,16 @@ func (e *Etcd) Close() error { if e.heartbeatCancel != nil { e.heartbeatCancel() } + + e.wg.Wait() e.e.Close() <-e.e.Server.StopNotify() } + if e.cli != nil { + e.cli.Close() + } + return nil } @@ -190,6 +201,7 @@ func (e *Etcd) Start(ctx context.Context) (disco.InitialClusterState, error) { } _ = testhook.Opened(pilosa.NewAuditor(), e, nil) e.e = etcd + e.cli = &hookedClient{Client: v3client.New(e.e.Server)} select { case <-ctx.Done(): @@ -205,12 +217,6 @@ func (e *Etcd) Start(ctx context.Context) (disco.InitialClusterState, error) { } func (e *Etcd) startHeartbeat() error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "startHeartbeat: creates a new client") - } - defer cli.Close() - heartbeatID, ctx, heartbeatCancel, err := e.leaseKeepAlive(context.Background(), e.options.HeartbeatTTL) if err != nil { return errors.Wrap(err, "startHeartbeat: creates a new hearbeat") @@ -221,7 +227,7 @@ func (e *Etcd) startHeartbeat() error { value = disco.ClusterStateResizing } - if _, err := cli.Put(ctx, key, string(value), clientv3.WithLease(heartbeatID)); err != nil { + 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) } @@ -232,17 +238,12 @@ func (e *Etcd) startHeartbeat() error { } func (e *Etcd) NodeState(ctx context.Context, peerID string) (disco.NodeState, error) { - cli, err := e.client() - if err != nil { - return disco.NodeStateUnknown, errors.Wrap(err, "NodeState: creates a new client") - } - defer cli.Close() - return e.nodeState(ctx, peerID) } func (e *Etcd) nodeState(ctx context.Context, peerID string) (disco.NodeState, error) { - resp, err := e.e.Server.KV().Range([]byte(path.Join(resizePrefix, peerID)), nil, mvcc.RangeOptions{Count: true}) + kv := e.e.Server.KV() + resp, err := kv.Range([]byte(path.Join(resizePrefix, peerID)), nil, mvcc.RangeOptions{Count: true}) if err != nil { return disco.NodeStateUnknown, err } @@ -250,31 +251,25 @@ func (e *Etcd) nodeState(ctx context.Context, peerID string) (disco.NodeState, e return disco.NodeStateResizing, nil } - resp, err = e.e.Server.KV().Range([]byte(path.Join(heartbeatPrefix, peerID)), nil, mvcc.RangeOptions{}) + resp, err = kv.Range([]byte(path.Join(heartbeatPrefix, peerID)), nil, mvcc.RangeOptions{}) if err != nil { return disco.NodeStateUnknown, err } + kvs := resp.KVs - if len(resp.KVs) > 1 { + if len(kvs) > 1 { return disco.NodeStateUnknown, disco.ErrTooManyResults } - if len(resp.KVs) == 0 { + if len(kvs) == 0 { return disco.NodeStateUnknown, disco.ErrNoResults } - return disco.NodeState(resp.KVs[0].Value), nil + return disco.NodeState(kvs[0].Value), nil } func (e *Etcd) NodeStates(ctx context.Context) (map[string]disco.NodeState, error) { out := make(map[string]disco.NodeState) - - cli, err := e.client() - if err != nil { - return nil, errors.Wrap(err, "NodeStates") - } - defer cli.Close() - members := e.e.Server.Cluster().Members() for _, member := range members { s, err := e.nodeState(ctx, member.ID.String()) @@ -288,15 +283,9 @@ func (e *Etcd) NodeStates(ctx context.Context) (map[string]disco.NodeState, erro return out, nil } -func (e *Etcd) Started(ctx context.Context) error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "Started") - } - defer cli.Close() - +func (e *Etcd) Started(ctx context.Context) (err error) { key, value := heartbeatPrefix+e.e.Server.ID().String(), disco.NodeStateStarted - if _, err = cli.Put(ctx, key, string(value), clientv3.WithLease(e.heartbeatID)); err == nil { + if _, err = e.cli.Put(ctx, key, string(value), clientv3.WithLease(e.heartbeatID)); err == nil { e.lm.started = true } return err @@ -335,12 +324,6 @@ func (e *Etcd) ClusterState(ctx context.Context) (disco.ClusterState, error) { return disco.ClusterStateUnknown, nil } - cli, err := e.client() - if err != nil { - return disco.ClusterStateUnknown, errors.WithMessage(err, "ClusterState: creates a new client") - } - defer cli.Close() - var ( heartbeats int = 0 resize bool @@ -385,12 +368,6 @@ func (e *Etcd) ClusterState(ctx context.Context) (disco.ClusterState, error) { } func (e *Etcd) Resize(ctx context.Context) (func([]byte) error, error) { - cli, err := e.client() - if err != nil { - return nil, errors.Wrap(err, "Resize: creates a new client") - } - defer cli.Close() - resizeID, ctx, resizeCancel, err := e.leaseKeepAlive(ctx, e.options.HeartbeatTTL) if err != nil { return nil, errors.Wrap(err, "Resize: creates a new hearbeat") @@ -398,7 +375,7 @@ func (e *Etcd) Resize(ctx context.Context) (func([]byte) error, error) { // Check if key exists - maybe we are still resizing key := path.Join(resizePrefix, e.e.Server.ID().String()) - txnResp, err := cli.Txn(ctx). + txnResp, err := e.cli.Txn(ctx). If(clientv3util.KeyMissing(key)). Then(clientv3.OpPut(key, "", clientv3.WithLease(resizeID))). Commit() @@ -428,14 +405,8 @@ func (e *Etcd) DoneResize() error { } func (e *Etcd) Watch(ctx context.Context, peerID string, onUpdate func([]byte) error) error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "Watch: creates a new client") - } - defer cli.Close() - key := path.Join(resizePrefix, peerID) - for resp := range cli.Watch(ctx, key) { + for resp := range e.cli.Watch(ctx, key) { if err := resp.Err(); err != nil { return errors.Wrapf(err, "Watch: key (%s) response", key) } @@ -465,13 +436,7 @@ func (e *Etcd) DeleteNode(ctx context.Context, nodeID string) error { return err } - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "DeleteNode: creates a new client") - } - defer cli.Close() - - _, err = cli.MemberRemove(ctx, uint64(id)) + _, err = e.cli.MemberRemove(ctx, uint64(id)) if err != nil { return errors.Wrap(err, "DeleteNode: removes an existing member from the cluster") } @@ -480,7 +445,7 @@ func (e *Etcd) DeleteNode(ctx context.Context, nodeID string) error { } func (e *Etcd) Schema(ctx context.Context) (disco.Schema, error) { - keys, vals, err := e.getKey(ctx, schemaPrefix) + keys, vals, err := e.getKeyWithPrefix(ctx, schemaPrefix) if err != nil { return nil, err } @@ -535,26 +500,22 @@ func (e *Etcd) Schema(ctx context.Context) (disco.Schema, error) { } func (e *Etcd) Metadata(ctx context.Context, peerID string) ([]byte, error) { - cli, err := e.client() - if err != nil { - return nil, errors.Wrap(err, "Metadata") - } - defer cli.Close() - - resp, err := cli.Get(ctx, path.Join(metadataPrefix, peerID)) + kv := e.e.Server.KV() + resp, err := kv.Range([]byte(path.Join(metadataPrefix, peerID)), nil, mvcc.RangeOptions{}) if err != nil { return nil, err } + kvs := resp.KVs - if len(resp.Kvs) > 1 { + if len(kvs) > 1 { return nil, disco.ErrTooManyResults } - if len(resp.Kvs) == 0 { + if len(kvs) == 0 { return nil, disco.ErrNoResults } - return resp.Kvs[0].Value, nil + return kvs[0].Value, nil } func (e *Etcd) SetMetadata(ctx context.Context, metadata []byte) error { @@ -570,12 +531,6 @@ func (e *Etcd) SetMetadata(ctx context.Context, metadata []byte) error { } func (e *Etcd) CreateIndex(ctx context.Context, name string, val []byte) error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "CreateIndex: creating client") - } - defer cli.Close() - key := schemaPrefix + name // Set up Op to write index value as bytes. @@ -583,7 +538,7 @@ func (e *Etcd) CreateIndex(ctx context.Context, name string, val []byte) error { op.WithValueBytes(val) // Check for key existence, and execute Op within a transaction. - resp, err := cli.KV.Txn(ctx). + resp, err := e.cli.Txn(ctx). If(clientv3util.KeyMissing(key)). Then(op). Commit() @@ -602,16 +557,10 @@ func (e *Etcd) Index(ctx context.Context, name string) ([]byte, error) { return e.getKeyBytes(ctx, schemaPrefix+name) } -func (e *Etcd) DeleteIndex(ctx context.Context, name string) error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "DeleteIndex: creating client") - } - defer cli.Close() - +func (e *Etcd) DeleteIndex(ctx context.Context, name string) (err error) { key := schemaPrefix + name // Deleting index and fields in one transaction. - _, err = cli.KV.Txn(ctx). + _, err = e.cli.Txn(ctx). If(clientv3.Compare(clientv3.Version(key), ">", -1)). Then( clientv3.OpDelete(key+"/", clientv3.WithPrefix()), // deleting index fields @@ -627,12 +576,6 @@ func (e *Etcd) Field(ctx context.Context, indexName string, name string) ([]byte } func (e *Etcd) CreateField(ctx context.Context, indexName string, name string, val []byte) error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "CreateField: creating client") - } - defer cli.Close() - key := schemaPrefix + indexName + "/" + name // Set up Op to write field value as bytes. @@ -640,7 +583,7 @@ func (e *Etcd) CreateField(ctx context.Context, indexName string, name string, v op.WithValueBytes(val) // Check for key existence, and execute Op within a transaction. - resp, err := cli.KV.Txn(ctx). + resp, err := e.cli.Txn(ctx). If(clientv3util.KeyMissing(key)). Then(op). Commit() @@ -655,16 +598,10 @@ func (e *Etcd) CreateField(ctx context.Context, indexName string, name string, v return nil } -func (e *Etcd) DeleteField(ctx context.Context, indexname string, name string) error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "DeleteField: creating client") - } - defer cli.Close() - +func (e *Etcd) DeleteField(ctx context.Context, indexname string, name string) (err error) { key := schemaPrefix + indexname + "/" + name // Deleting field and views in one transaction. - _, err = cli.KV.Txn(ctx). + _, err = e.cli.Txn(ctx). If(clientv3.Compare(clientv3.Version(key), ">", -1)). Then( clientv3.OpDelete(key+"/", clientv3.WithPrefix()), // deleting field views @@ -682,17 +619,11 @@ func (e *Etcd) View(ctx context.Context, indexName, fieldName, name string) (boo // CreateView differs from CreateIndex and CreateField in that it does not // return an error if the view already exists. If this logic needs to be // changed, we likely need to return disco.ErrViewExists. -func (e *Etcd) CreateView(ctx context.Context, indexName, fieldName, name string) error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "CreateView: creating client") - } - defer cli.Close() - +func (e *Etcd) CreateView(ctx context.Context, indexName, fieldName, name string) (err error) { key := schemaPrefix + indexName + "/" + fieldName + "/" + name // Check for key existence, and execute Op within a transaction. - _, err = cli.KV.Txn(ctx). + _, err = e.cli.Txn(ctx). If(clientv3util.KeyMissing(key)). Then(clientv3.OpPut(key, "")). Commit() @@ -708,13 +639,7 @@ func (e *Etcd) DeleteView(ctx context.Context, indexName, fieldName, name string } func (e *Etcd) putKey(ctx context.Context, key, val string, opts ...clientv3.OpOption) error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "putKey: creates a new client") - } - defer cli.Close() - - if _, err := cli.KV.Put(ctx, key, val, opts...); err != nil { + if _, err := e.cli.Put(ctx, key, val, opts...); err != nil { return errors.Wrapf(err, "putKey: Put(%s, %s)", key, val) } @@ -722,68 +647,45 @@ func (e *Etcd) putKey(ctx context.Context, key, val string, opts ...clientv3.OpO } func (e *Etcd) getKeyBytes(ctx context.Context, key string) ([]byte, error) { - cli, err := e.client() - if err != nil { - return nil, errors.Wrap(err, "getKeyBytes: creates a new client") - } - defer cli.Close() - // Get the current value for the key. - resp, err := cli.Get(ctx, key) + kv := e.e.Server.KV() + resp, err := kv.Range([]byte(key), nil, mvcc.RangeOptions{}) if err != nil { return nil, err } + kvs := resp.KVs // TODO: consider returning a "key does not exist" error instead of (nil, nil) - if len(resp.Kvs) == 0 { + if len(kvs) == 0 { return nil, nil } - return resp.Kvs[0].Value, nil + return kvs[0].Value, nil } -func (e *Etcd) getKey(ctx context.Context, key string) ([]string, [][]byte, error) { - cli, err := e.client() - if err != nil { - return nil, nil, errors.Wrap(err, "getKey: creates a new client") - } - defer cli.Close() - - resp, err := cli.KV.Txn(ctx). - If(clientv3.Compare(clientv3.Version(key), ">", -1)). - Then(clientv3.OpGet(key, clientv3.WithPrefix())). - Commit() +func (e *Etcd) getKeyWithPrefix(ctx context.Context, key string) ([]string, [][]byte, error) { + resp, err := e.cli.Get(ctx, key, clientv3.WithPrefix()) if err != nil { return nil, nil, err } - - if !resp.Succeeded { - return nil, nil, fmt.Errorf("key %s does not exist", key) - } + kvs := resp.Kvs var ( keys []string values [][]byte ) - for _, r := range resp.Responses { - for _, kv := range r.GetResponseRange().Kvs { - keys = append(keys, string(kv.Key)) - values = append(values, kv.Value) - } + for _, kv := range kvs { + keys = append(keys, string(kv.Key)) + values = append(values, kv.Value) } return keys, values, nil } func (e *Etcd) keyExists(ctx context.Context, key string) (bool, error) { - cli, err := e.client() - if err != nil { - return false, errors.Wrap(err, "keyExists: creates a new client") - } - defer cli.Close() - - resp, err := cli.Get(ctx, key, clientv3.WithCountOnly()) + kv := e.e.Server.KV() + resp, err := kv.Range([]byte(key), nil, mvcc.RangeOptions{Count: true}) if err != nil { return false, err } @@ -793,23 +695,12 @@ func (e *Etcd) keyExists(ctx context.Context, key string) (bool, error) { return false, nil } -func (e *Etcd) delKey(ctx context.Context, key string, withPrefix bool) error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "delKey: creates a new client") - } - defer cli.Close() - - var opts []clientv3.OpOption +func (e *Etcd) delKey(ctx context.Context, key string, withPrefix bool) (err error) { if withPrefix { - opts = append(opts, clientv3.WithPrefix()) + _, err = e.cli.Delete(ctx, key, clientv3.WithPrefix()) + } else { + _, err = e.cli.Delete(ctx, key) } - - _, err = cli.KV.Txn(ctx). - If(clientv3.Compare(clientv3.Version(key), ">", -1)). - Then(clientv3.OpDelete(key, opts...)). - Commit() - return err } @@ -817,15 +708,8 @@ func (e *Etcd) delKey(ctx context.Context, key string, withPrefix bool) error { // 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) { - cli, err := e.client() - if err != nil { - return 0, nil, nil, errors.Wrap(err, "leaseKeepAlive: creates a new client") - } - defer cli.Close() - ctx, cancelFunc := context.WithCancel(ctx) - - leaseResp, err := cli.Grant(ctx, ttl) + 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) @@ -833,7 +717,10 @@ func (e *Etcd) leaseKeepAlive(ctx context.Context, ttl int64) (clientv3.LeaseID, keepaliveFunc := func(tick time.Duration) { ticker := time.NewTicker(tick) - defer ticker.Stop() + defer func() { + ticker.Stop() + e.wg.Done() + }() for { select { @@ -843,28 +730,21 @@ func (e *Etcd) leaseKeepAlive(ctx context.Context, ttl int64) (clientv3.LeaseID, // here, resulting in massive piles of excess goroutines. revoker, cancel := context.WithTimeout(context.Background(), time.Duration(ttl)) defer cancel() - if cli, err := e.client(); err != nil { - log.Printf("leaseKeepAlive: creates a new client: %v\n", err) - } else { - if _, err := cli.Revoke(revoker, leaseResp.ID); err != nil { - log.Printf("leaseKeepAlive: revokes the lease (ID: %x): %#v\n", leaseResp.ID, err) - } - cli.Close() + + if _, err := e.cli.Revoke(revoker, leaseResp.ID); err != nil { + log.Printf("leaseKeepAlive: revokes the lease (ID: %x): %#v\n", leaseResp.ID, err) } return case <-ticker.C: - if cli, err := e.client(); err != nil { - log.Printf("leaseKeepAlive: creates a new client: %v\n", err) - } else { - if _, err = cli.KeepAliveOnce(ctx, leaseResp.ID); err != nil { - log.Printf("leaseKeepAlive: renews the lease (ID: %x): %v\n", leaseResp.ID, err) - } - cli.Close() + if _, err = e.cli.KeepAliveOnce(ctx, leaseResp.ID); err != nil { + log.Printf("leaseKeepAlive: renews the lease (ID: %x): %v\n", leaseResp.ID, err) } } } } - go keepaliveFunc(1 * time.Second) + + e.wg.Add(1) + go keepaliveFunc(time.Second) return leaseResp.ID, ctx, cancelFunc, nil } @@ -883,19 +763,6 @@ func (h *hookedClient) Close() { h.Client.Close() } -func (e *Etcd) client() (*hookedClient, error) { - urls := e.e.Server.Cluster().ClientURLs() - - cli, err := clientv3.NewFromURLs(urls) - if err != nil { - return nil, errors.Wrapf(err, "creates a new etcd client from URLs (%v)", urls) - } - - // Temporarily disabled, see comment in Close above. - // _ = testhook.Opened(pilosa.NewAuditor(), cli, nil) - return &hookedClient{Client: cli}, nil -} - func memberList(cli *hookedClient) (ids []uint64, names []string, urls []string) { ml, err := cli.MemberList(context.TODO()) if err != nil { @@ -923,20 +790,14 @@ func memberAdd(cli *hookedClient, peerURL string) (id uint64, name string) { // Shards implements the Sharder interface. func (e *Etcd) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) { - cli, err := e.client() - if err != nil { - return nil, errors.Wrap(err, "Shards: creating client") - } - defer cli.Close() - - return e.shards(ctx, cli, index, field) + return e.shards(ctx, index, field) } -func (e *Etcd) shards(ctx context.Context, cli *hookedClient, index, field string) (*roaring.Bitmap, error) { +func (e *Etcd) shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) { key := path.Join(shardPrefix, index, field) // Get the current shards for the field. - resp, err := cli.Get(ctx, key) + resp, err := e.cli.Get(ctx, key) if err != nil { return nil, err } @@ -957,12 +818,6 @@ func (e *Etcd) shards(ctx context.Context, cli *hookedClient, index, field strin // AddShards implements the Sharder interface. func (e *Etcd) AddShards(ctx context.Context, index, field string, shards *roaring.Bitmap) (*roaring.Bitmap, error) { - cli, err := e.client() - if err != nil { - return nil, errors.Wrap(err, "AddShards: creating client") - } - defer cli.Close() - key := path.Join(shardPrefix, index, field) // This tended to add more overhead than it saved. @@ -975,7 +830,7 @@ func (e *Etcd) AddShards(ctx context.Context, index, field string, shards *roari // } // Create a session to acquire a lock. - sess, _ := concurrency.NewSession(cli.Client) + sess, _ := concurrency.NewSession(e.cli.Client) defer sess.Close() muKey := path.Join(lockPrefix, index, field) @@ -987,7 +842,7 @@ func (e *Etcd) AddShards(ctx context.Context, index, field string, shards *roari } // Read shards within lock. - globalShards, err := e.shards(ctx, cli, index, field) + globalShards, err := e.shards(ctx, index, field) if err != nil { return nil, errors.Wrap(err, "reading shards") } @@ -1004,7 +859,7 @@ func (e *Etcd) AddShards(ctx context.Context, index, field string, shards *roari op := clientv3.OpPut(key, "") op.WithValueBytes(buf.Bytes()) - if _, err := cli.Do(ctx, op); err != nil { + if _, err := e.cli.Do(ctx, op); err != nil { return nil, errors.Wrap(err, "doing op") } @@ -1018,17 +873,11 @@ func (e *Etcd) AddShards(ctx context.Context, index, field string, shards *roari // AddShard implements the Sharder interface. func (e *Etcd) AddShard(ctx context.Context, index, field string, shard uint64) error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "AddShard: creating client") - } - defer cli.Close() - key := path.Join(shardPrefix, index, field) // Read shards outside of a lock just to check if shard is already included. // If shard is already included, no-op. - if shards, err := e.shards(ctx, cli, index, field); err != nil { + if shards, err := e.shards(ctx, index, field); err != nil { return errors.Wrap(err, "reading shards") } else if shards.Contains(shard) { return nil @@ -1040,7 +889,7 @@ func (e *Etcd) AddShard(ctx context.Context, index, field string, shard uint64) // write shards to etcd. // Create a session to acquire a lock. - sess, _ := concurrency.NewSession(cli.Client) + sess, _ := concurrency.NewSession(e.cli.Client) defer sess.Close() muKey := path.Join(lockPrefix, index, field) @@ -1052,7 +901,7 @@ func (e *Etcd) AddShard(ctx context.Context, index, field string, shard uint64) } // Read shards again (within lock). - shards, err := e.shards(ctx, cli, index, field) + shards, err := e.shards(ctx, index, field) if err != nil { return errors.Wrap(err, "reading shards") } @@ -1073,7 +922,7 @@ func (e *Etcd) AddShard(ctx context.Context, index, field string, shard uint64) op := clientv3.OpPut(key, "") op.WithValueBytes(buf.Bytes()) - if _, err := cli.Do(ctx, op); err != nil { + if _, err := e.cli.Do(ctx, op); err != nil { return errors.Wrap(err, "doing op") } @@ -1087,17 +936,11 @@ func (e *Etcd) AddShard(ctx context.Context, index, field string, shard uint64) // RemoveShard implements the Sharder interface. func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint64) error { - cli, err := e.client() - if err != nil { - return errors.Wrap(err, "RemoveShard: creating client") - } - defer cli.Close() - key := path.Join(shardPrefix, index, field) // Read shards outside of a lock just to check if shard is already excluded. // If shard is already excluded, no-op. - if shards, err := e.shards(ctx, cli, index, field); err != nil { + if shards, err := e.shards(ctx, index, field); err != nil { return errors.Wrap(err, "reading shards") } else if !shards.Contains(shard) { return nil @@ -1109,7 +952,7 @@ func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint6 // write shards to etcd. // Create a session to acquire a lock. - sess, _ := concurrency.NewSession(cli.Client) + sess, _ := concurrency.NewSession(e.cli.Client) defer sess.Close() muKey := path.Join(lockPrefix, index, field) @@ -1121,7 +964,7 @@ func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint6 } // Read shards again (within lock). - shards, err := e.shards(ctx, cli, index, field) + shards, err := e.shards(ctx, index, field) if err != nil { return errors.Wrap(err, "reading shards") } @@ -1138,7 +981,7 @@ func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint6 // If this is removing the last bit from the shards bitmap, then instead of // writing an empty bitmap, just delete the key. if shards.Count() == 0 { - _, err := cli.Delete(ctx, key) + _, err := e.cli.Delete(ctx, key) return err } @@ -1151,7 +994,7 @@ func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint6 op := clientv3.OpPut(key, "") op.WithValueBytes(buf.Bytes()) - if _, err := cli.Do(ctx, op); err != nil { + if _, err := e.cli.Do(ctx, op); err != nil { return errors.Wrap(err, "doing op") }