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

Add documentation and try to remove code.

Browse source 
Signed-off-by: Antonio Navarro Perez <antnavper@gmail.com>
This commit is contained in:
Antonio Navarro Perez 2021-03-01 16:25:32 +01:00
parent a906a9036b
commit ea7643f8bf
4 changed files with 53 additions and 137 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 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
}

94
etcd/cache.go
View file

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

38
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():
@ -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)

2
server/server.go
View file

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