From ea7643f8bfbc0102acb9e26e3193aa93e742b926 Mon Sep 17 00:00:00 2001 From: Antonio Navarro Perez Date: Mon, 1 Mar 2021 16:25:32 +0100 Subject: [PATCH 1/5] Add documentation and try to remove code. Signed-off-by: Antonio Navarro Perez --- disco/disco.go | 56 ++++++++++++++++++++++------- etcd/cache.go | 94 ------------------------------------------------ etcd/embed.go | 38 +++++--------------- server/server.go | 2 +- 4 files changed, 53 insertions(+), 137 deletions(-) delete mode 100644 etcd/cache.go diff --git a/disco/disco.go b/disco/disco.go index 117e1f6e3..7d842f9b9 100644 --- a/disco/disco.go +++ b/disco/disco.go @@ -57,20 +57,21 @@ type DisCo interface { type ( InitialClusterState string - ClusterState string + + // ClusterState represents the state returned in the /status endpoint. + ClusterState string ) const ( InitialClusterStateNew InitialClusterState = "new" InitialClusterStateExisting InitialClusterState = "existing" - // ClusterState represents the state returned in the /status endpoint. - ClusterStateUnknown ClusterState = "UNKNOWN" - ClusterStateStarting ClusterState = "STARTING" - ClusterStateDegraded ClusterState = "DEGRADED" // cluster is running but we've lost some # of hosts >0 but < replicaN - ClusterStateNormal ClusterState = "NORMAL" - ClusterStateResizing ClusterState = "RESIZING" // cluster is replicating data to other nodes - ClusterStateDown ClusterState = "DOWN" // cluster is unable to serve queries + ClusterStateUnknown ClusterState = "UNKNOWN" // default cluster state. It is returned when we are not able to get the real actual state. + ClusterStateStarting ClusterState = "STARTING" // cluster is starting and some internal services are not ready yet. + ClusterStateDegraded ClusterState = "DEGRADED" // cluster is running but we've lost some # of hosts >0 but < replicaN. Only read queries are allowed. + ClusterStateNormal ClusterState = "NORMAL" // cluster is up and running. + ClusterStateResizing ClusterState = "RESIZING" // cluster is replicating data to other nodes. + ClusterStateDown ClusterState = "DOWN" // cluster is unable to serve queries. ) type NodeState string @@ -83,9 +84,26 @@ const ( ) type Stator interface { + + // Started will mark the actual node as already started. + // It must be called after all initialization processes + // are up and running. Started(ctx context.Context) error + + // ClusterState summarize the state of all nodes and gives + // a general cluster state. The output calculation is as follows: + // - If any of the nodes are still starting: "STARTING" + // - If all nodes are up and running: "NORMAL" + // - If number of nodes down is lower than number of replicas: "DEGRADED" + // - If number of nodes down is bigger than number of replicas: "DOWN" + // - If any of the nodes started a resize operation, or a new + // node was specifically added or removed from the cluster: "RESIZING" ClusterState(context.Context) (ClusterState, error) + + // NodeState returns the specific state of a node giving its ID. NodeState(context.Context, string) (NodeState, error) + + // NodeStates will return all the states by node ID of the actual nodes on the cluster. NodeStates(context.Context) (map[string]NodeState, error) } @@ -107,9 +125,17 @@ type Field struct { Views map[string]struct{} } +// Schemator is the source of truth for different schema elements. +// All nodes will store and retrieve information from the same source, +// having the same information at the same time. type Schemator interface { + + // Schema return the actual pilosa schema. If the schema is not present, an error is returned. Schema(ctx context.Context) (Schema, error) + + // Index gets a specific index data by name. Index(ctx context.Context, name string) ([]byte, error) + CreateIndex(ctx context.Context, name string, val []byte) error DeleteIndex(ctx context.Context, name string) error Field(ctx context.Context, index, field string) ([]byte, error) @@ -120,11 +146,8 @@ type Schemator interface { DeleteView(ctx context.Context, index, field, view string) error } -type Metadata interface { - Marshal() ([]byte, error) - Unmarshal([]byte) error -} - +// Metadator is in charge of store specific metadata per node. +// This metadata can be retrieved by any node using the specific peerID. type Metadator interface { Metadata(ctx context.Context, peerID string) ([]byte, error) SetMetadata(ctx context.Context, metadata []byte) error @@ -133,8 +156,15 @@ type Metadator interface { // Resizer triggers resizing the node and changes cluster state into RESIZING. // We can also return some kind of handler from Resize function (e.g. key-value) type Resizer interface { + // Resize will trigger a resize event. Node state will change to RESIZE state. + // The returned function can be used to send info about the resize process to other nodes. Resize(ctx context.Context) (func([]byte) error, error) + + // DoneResize will mark the resize event as done. This will be called when all the resize actions are done. DoneResize() error + + // Watch will give information about a resize event in other node, using its peerID. + // onUpdate function will be called per each event sent by the node in RESIZE state. Watch(ctx context.Context, peerID string, onUpdate func([]byte) error) error } diff --git a/etcd/cache.go b/etcd/cache.go deleted file mode 100644 index 8005c87ef..000000000 --- a/etcd/cache.go +++ /dev/null @@ -1,94 +0,0 @@ -// Copyright 2017 Pilosa Corp. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package etcd - -import ( - "context" - "sync" - "time" - - "github.com/pilosa/pilosa/v2/topology" -) - -// EtcdWithCache is a wrapper around the Etcd type which will return a -// cached value when the number of requests come in below a configured -// frequency. It also breaks the cache after a configured TTL. -type EtcdWithCache struct { - *Etcd - - peerMetadataMu sync.RWMutex - peerMetadata map[string][]byte - - peersMu sync.Mutex // peer-list cache updates - - nodes []*topology.Node // unmarshalled Node data - nodesTTL int // seconds - nodesLastRequest time.Time // last time requested -} - -// NewEtcdWithCache returns a new instance of Cache. -func NewEtcdWithCache(opt Options, replicas int) *EtcdWithCache { - return &EtcdWithCache{ - Etcd: NewEtcd(opt, replicas), - - nodesTTL: 6, - - peerMetadata: make(map[string][]byte), - } -} - -// Metadata is a cache wrapper around the Metadator.Metadata method. -func (c *EtcdWithCache) Metadata(ctx context.Context, peerID string) ([]byte, error) { - c.peerMetadataMu.RLock() - v, ok := c.peerMetadata[peerID] - c.peerMetadataMu.RUnlock() - if ok { - return v, nil - } - v, err := c.Etcd.Metadata(ctx, peerID) - if err == nil { - c.peerMetadataMu.Lock() - c.peerMetadata[peerID] = v - c.peerMetadataMu.Unlock() - } - return v, err -} - -// Nodes caches the result of the underlying implementation's node list. -func (c *EtcdWithCache) Nodes() []*topology.Node { - c.peersMu.Lock() - defer c.peersMu.Unlock() - - now := time.Now() - if now.Sub(c.nodesLastRequest) > (time.Duration(c.nodesTTL) * time.Second) { - c.nodes = c.Etcd.Nodes() - c.nodesLastRequest = now - } - return c.nodes -} - -// SetNodes implements the Noder interface as NOP -// (because we can't force to set nodes for etcd). -func (c *EtcdWithCache) SetNodes(nodes []*topology.Node) {} - -// AppendNode implements the Noder interface as NOP -// (because resizer is responsible for adding new nodes). -func (c *EtcdWithCache) AppendNode(node *topology.Node) {} - -// RemoveNode implements the Noder interface as NOP -// (because resizer is responsible for removing existing nodes) -func (c *EtcdWithCache) RemoveNode(nodeID string) bool { - return false -} diff --git a/etcd/embed.go b/etcd/embed.go index 08f784f84..1077463fa 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -26,10 +26,8 @@ import ( "sync" "time" - "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/disco" "github.com/pilosa/pilosa/v2/roaring" - "github.com/pilosa/pilosa/v2/testhook" "github.com/pilosa/pilosa/v2/topology" "github.com/pkg/errors" "go.etcd.io/etcd/clientv3" @@ -93,7 +91,7 @@ type Etcd struct { lm leaseMetadata e *embed.Etcd - cli *hookedClient + cli *clientv3.Client wg *sync.WaitGroup } @@ -108,8 +106,6 @@ 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() @@ -169,11 +165,10 @@ func parseOptions(opt Options) *embed.Config { if opt.ClusterURL != "" { cfg.ClusterState = embed.ClusterStateFlagExisting - t, err := clientv3.NewFromURL(opt.ClusterURL) + cli, err := clientv3.NewFromURL(opt.ClusterURL) if err != nil { panic(err) } - cli := &hookedClient{Client: t} defer cli.Close() log.Println("Cluster Members:") @@ -200,9 +195,8 @@ func (e *Etcd) Start(ctx context.Context) (disco.InitialClusterState, error) { if err != nil { return state, errors.Wrap(err, "starting etcd") } - _ = testhook.Opened(pilosa.NewAuditor(), e, nil) e.e = etcd - e.cli = &hookedClient{Client: v3client.New(e.e.Server)} + e.cli = v3client.New(e.e.Server) select { case <-ctx.Done(): @@ -762,7 +756,7 @@ func (e *Etcd) leaseKeepAlive(ctx context.Context, cancelFunc context.CancelFunc // 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)} + cli := v3client.New(e.e.Server) var err error leaseResp, err = cli.Grant(ctx, ttl) cli.Close() @@ -803,21 +797,7 @@ func (e *Etcd) leaseKeepAlive(ctx context.Context, cancelFunc context.CancelFunc return leaseResp.ID, nil } -type hookedClient struct { - *clientv3.Client -} - -func (h *hookedClient) Close() { - // The hook open/closed test here is disabled because there's a - // slight delay before the client actually gets closed in - // some cases, which is long enough to frequently be caught - // if there was a client in the last test run, even though it'd - // be fine a few seconds later. - // _ = testhook.Closed(pilosa.NewAuditor(), h.Client, nil) - h.Client.Close() -} - -func memberList(cli *hookedClient) (ids []uint64, names []string, urls []string) { +func memberList(cli *clientv3.Client) (ids []uint64, names []string, urls []string) { ml, err := cli.MemberList(context.TODO()) if err != nil { panic(err) @@ -833,7 +813,7 @@ func memberList(cli *hookedClient) (ids []uint64, names []string, urls []string) return } -func memberAdd(cli *hookedClient, peerURL string) (id uint64, name string) { +func memberAdd(cli *clientv3.Client, peerURL string) (id uint64, name string) { ma, err := cli.MemberAdd(context.TODO(), []string{peerURL}) if err != nil { return 0, "" @@ -884,7 +864,7 @@ func (e *Etcd) AddShards(ctx context.Context, index, field string, shards *roari // } // Create a session to acquire a lock. - sess, _ := concurrency.NewSession(e.cli.Client) + sess, _ := concurrency.NewSession(e.cli) defer sess.Close() muKey := path.Join(lockPrefix, index, field) @@ -943,7 +923,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(e.cli.Client) + sess, _ := concurrency.NewSession(e.cli) defer sess.Close() muKey := path.Join(lockPrefix, index, field) @@ -1006,7 +986,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(e.cli.Client) + sess, _ := concurrency.NewSession(e.cli) defer sess.Close() muKey := path.Join(lockPrefix, index, field) diff --git a/server/server.go b/server/server.go index ab94b40f5..d607c6341 100644 --- a/server/server.go +++ b/server/server.go @@ -391,7 +391,7 @@ func (m *Command) SetupServer() error { m.Config.Etcd.Dir = filepath.Join(path, pilosa.DefaultDiscoDir) } - e := petcd.NewEtcdWithCache(m.Config.Etcd, m.Config.Cluster.ReplicaN) + e := petcd.NewEtcd(m.Config.Etcd, m.Config.Cluster.ReplicaN) discoOpt := pilosa.OptServerDisCo(e, e, e, e, e, e, e) serverOptions := []pilosa.ServerOption{ From 6f5770bf656bb0bfb01fa178bb50c291b39bcde7 Mon Sep 17 00:00:00 2001 From: Antonio Navarro Perez Date: Mon, 1 Mar 2021 18:39:04 +0100 Subject: [PATCH 2/5] Fix nil pointer exception Signed-off-by: Antonio Navarro Perez --- etcd/embed.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/etcd/embed.go b/etcd/embed.go index 1077463fa..977f2a6c8 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -506,12 +506,11 @@ func (e *Etcd) Schema(ctx context.Context) (disco.Schema, error) { } func (e *Etcd) Metadata(ctx context.Context, peerID string) ([]byte, error) { - kv := e.e.Server.KV() - resp, err := kv.Range([]byte(path.Join(metadataPrefix, peerID)), nil, mvcc.RangeOptions{}) + resp, err := e.cli.KV.Get(ctx, path.Join(metadataPrefix, peerID)) if err != nil { return nil, err } - kvs := resp.KVs + kvs := resp.Kvs if len(kvs) > 1 { return nil, disco.ErrTooManyResults From 50fd72f98033d59fa1a6e2dcb2082e36a8b38e0d Mon Sep 17 00:00:00 2001 From: Antonio Navarro Perez Date: Mon, 1 Mar 2021 18:41:58 +0100 Subject: [PATCH 3/5] Apply suggestions from code review Co-authored-by: Travis Turner --- disco/disco.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/disco/disco.go b/disco/disco.go index 7d842f9b9..49c0970e8 100644 --- a/disco/disco.go +++ b/disco/disco.go @@ -90,20 +90,20 @@ type Stator interface { // are up and running. Started(ctx context.Context) error - // ClusterState summarize the state of all nodes and gives + // ClusterState considers the state of all nodes and gives // a general cluster state. The output calculation is as follows: // - If any of the nodes are still starting: "STARTING" // - If all nodes are up and running: "NORMAL" // - If number of nodes down is lower than number of replicas: "DEGRADED" - // - If number of nodes down is bigger than number of replicas: "DOWN" + // - If number of nodes down is bigger than (or equal to) the number of replicas: "DOWN" // - If any of the nodes started a resize operation, or a new // node was specifically added or removed from the cluster: "RESIZING" ClusterState(context.Context) (ClusterState, error) - // NodeState returns the specific state of a node giving its ID. + // NodeState returns the specific state of a node given its ID. NodeState(context.Context, string) (NodeState, error) - // NodeStates will return all the states by node ID of the actual nodes on the cluster. + // NodeStates will return all the states by node ID of the actual nodes in the cluster. NodeStates(context.Context) (map[string]NodeState, error) } @@ -146,7 +146,7 @@ type Schemator interface { DeleteView(ctx context.Context, index, field, view string) error } -// Metadator is in charge of store specific metadata per node. +// Metadator is in charge of storing specific metadata per node. // This metadata can be retrieved by any node using the specific peerID. type Metadator interface { Metadata(ctx context.Context, peerID string) ([]byte, error) @@ -163,7 +163,7 @@ type Resizer interface { // DoneResize will mark the resize event as done. This will be called when all the resize actions are done. DoneResize() error - // Watch will give information about a resize event in other node, using its peerID. + // Watch will give information about a resize event in another node, using its peerID. // onUpdate function will be called per each event sent by the node in RESIZE state. Watch(ctx context.Context, peerID string, onUpdate func([]byte) error) error } From f67db5035e4d05999c7dbad03480bc9b167ccc0b Mon Sep 17 00:00:00 2001 From: Antonio Navarro Perez Date: Mon, 1 Mar 2021 19:09:31 +0100 Subject: [PATCH 4/5] Add part of cache back. Signed-off-by: Antonio Navarro Perez --- etcd/cache.go | 57 ++++++++++++++++++++++++++++++++++++++++++++++++ server/server.go | 2 +- 2 files changed, 58 insertions(+), 1 deletion(-) create mode 100644 etcd/cache.go diff --git a/etcd/cache.go b/etcd/cache.go new file mode 100644 index 000000000..4cd5bc684 --- /dev/null +++ b/etcd/cache.go @@ -0,0 +1,57 @@ +// Copyright 2017 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package etcd + +import ( + "sync" + "time" + + "github.com/pilosa/pilosa/v2/topology" +) + +// EtcdWithCache is a wrapper around the Etcd type which will return a +// cached value when the number of requests come in below a configured +// frequency. It also breaks the cache after a configured TTL. +type EtcdWithCache struct { + *Etcd + + peersMu sync.Mutex // peer-list cache updates + + nodes []*topology.Node // unmarshalled Node data + nodesTTL int // seconds + nodesLastRequest time.Time // last time requested +} + +// NewEtcdWithCache returns a new instance of Cache. +func NewEtcdWithCache(opt Options, replicas int) *EtcdWithCache { + return &EtcdWithCache{ + Etcd: NewEtcd(opt, replicas), + + nodesTTL: 6, + } +} + +// Nodes caches the result of the underlying implementation's node list. +func (c *EtcdWithCache) Nodes() []*topology.Node { + c.peersMu.Lock() + defer c.peersMu.Unlock() + + now := time.Now() + if now.Sub(c.nodesLastRequest) > (time.Duration(c.nodesTTL) * time.Second) { + c.nodes = c.Etcd.Nodes() + c.nodesLastRequest = now + } + return c.nodes +} diff --git a/server/server.go b/server/server.go index d607c6341..ab94b40f5 100644 --- a/server/server.go +++ b/server/server.go @@ -391,7 +391,7 @@ func (m *Command) SetupServer() error { m.Config.Etcd.Dir = filepath.Join(path, pilosa.DefaultDiscoDir) } - e := petcd.NewEtcd(m.Config.Etcd, m.Config.Cluster.ReplicaN) + e := petcd.NewEtcdWithCache(m.Config.Etcd, m.Config.Cluster.ReplicaN) discoOpt := pilosa.OptServerDisCo(e, e, e, e, e, e, e) serverOptions := []pilosa.ServerOption{ From 274540ea0b8557caa7ab9cd9b7158fead9ecf7f3 Mon Sep 17 00:00:00 2001 From: Antonio Navarro Perez Date: Tue, 2 Mar 2021 13:46:37 +0100 Subject: [PATCH 5/5] Requested changes Signed-off-by: Antonio Navarro Perez --- disco/disco.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/disco/disco.go b/disco/disco.go index 49c0970e8..39f15e762 100644 --- a/disco/disco.go +++ b/disco/disco.go @@ -94,8 +94,8 @@ type Stator interface { // a general cluster state. The output calculation is as follows: // - If any of the nodes are still starting: "STARTING" // - If all nodes are up and running: "NORMAL" - // - If number of nodes down is lower than number of replicas: "DEGRADED" - // - If number of nodes down is bigger than (or equal to) the number of replicas: "DOWN" + // - If number of DOWN nodes is lower than number of replicas: "DEGRADED" + // - If number of unresponsive nodes is greater than (or equal to) the number of replicas: "DOWN" // - If any of the nodes started a resize operation, or a new // node was specifically added or removed from the cluster: "RESIZING" ClusterState(context.Context) (ClusterState, error)