scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 09:05:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1486 from ajnavarro/disco/remove-unused-code-and-godoc

Browse source 
[DISCO] Add documentation and try to remove code.
This commit is contained in:
Matthew Jaffee 2021-03-02 14:07:06 -06:00 • committed by GitHub
commit 0568daaa14
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 54 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
disco/disco.go
View file

@ -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
}

37
etcd/cache.go
View file

@ -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
}

43
etcd/embed.go
View file

@ -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)