diff --git a/disco/disco.go b/disco/disco.go index 117e1f6e3..39f15e762 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 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 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) + + // 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 in 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 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) 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 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 } diff --git a/etcd/cache.go b/etcd/cache.go index 8005c87ef..4cd5bc684 100644 --- a/etcd/cache.go +++ b/etcd/cache.go @@ -15,7 +15,6 @@ package etcd import ( - "context" "sync" "time" @@ -28,9 +27,6 @@ import ( type EtcdWithCache struct { *Etcd - peerMetadataMu sync.RWMutex - peerMetadata map[string][]byte - peersMu sync.Mutex // peer-list cache updates nodes []*topology.Node // unmarshalled Node data @@ -44,28 +40,9 @@ func NewEtcdWithCache(opt Options, replicas int) *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() @@ -78,17 +55,3 @@ func (c *EtcdWithCache) Nodes() []*topology.Node { } 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..977f2a6c8 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(): @@ -512,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 @@ -762,7 +755,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 +796,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 +812,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 +863,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 +922,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 +985,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)