scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

simplify, streamline, and possibly debug embedded etcd

Browse source 
The root problem this is attempting to address is sporadic
weird cases in which etcd mistakenly thinks it's down even when
it's up. I am not confident that this is addressed, but there's
a reasonable chance that it is, and I can't trigger it at the
moment, but it was always sporadic, so that doesn't prove much.

There's a lot going on here, and it comes into roughly three
categories.

First: Dropping unused/unneeded code. There's a lot of leftover
bits from the initial development and refactoring of this.

Second: Unifying and shuffling some of the design. We had
multiple interfaces which are functionally impossible to
usefully implement separately, so they're combined together,
and in some cases, moved.

Third: Streamlining logic and simplifying design choices.

This is combined into one commit because the changes are
thoroughly entertwined with each other and you can't usefully
break most of them out.

Also, a bunch of test coverage for most of these changes.

Big changes:

We merge the topology and disco packages.  The topology and disco
packages being separate creates a complicated tangle of problems
and dependencies.  The fundamental problem, approximately, is that
topology.Node has to track disco.NodeState.

There's three core interfaces interacting here:
	topology.Noder (maintains list of nodes)
	disco.Stator (maintains the state of a node)
	disco.Metadator (stores, possibly retrieves, node metadata)
But the node state mantained by the Noder *is* the set of node
metadata, plus state updates produced by Stators. The only actual
non-trivial and usable implementation of these interfaces is a single
thing which implements all three, and in which the implementations
share a single backend data source which they are all modifying.

But you can't move Noder into disco, because Noder has to refer
to topology.Node, but topology.Node refers to disco.

Solution: First, merge these two packages. Second, merge these
three interfaces, to provide a single interface which is more
clear about the fact that (metadator.)SetMetadata() and
(stator.)Started() are both changing the output we'll get from
(noder.)Nodes().

We rework the node state tracking.

We have this nodeStates map which is almost unused. Really, we
don't need it at all. Every node's state is either its last heartbeat
state or "Unknown", so we simplify this a bit. Also, we ensure that
the populateNodeStates function itself is yielding the sorted nodes
list, so we don't have to be as worried about possible later lookups
of sortedNodes happening outside a lock. We also add diagnostics
for deleting nodes from the metadata list (this should never happen),
and try to track heartbeat state more closely.

This is *probably* what fixes the underlying reported problem,
if anything did.

Still an open issue: Make heartbeat state changes aware of when
they're talking about *this* node and possibly not try to
mark it down? Except this may have a flaw: That would result in
each node disagreeing with other nodes in etcd about the state
of that node in the failure cases, and undermine the point of
using etcd to keep these states consistent.

We reduce the number of contexts and cancelfuncs in the etcd wrapper.

We create a shared context for the non-etcd.embed children of our
etcd wrapper, the heartbeat/keepalive and the node watcher, so we
can cancel that one context and cancel all of those at once, so
we don't need to separately track a function to call to cancel
the watch, AND be closing another channel. Also, our shutdown
now propagates automatically to the various etcd API calls we've
made for things like the node watcher and keepalive calls.

We still need to watch that channel in watchNodesOnce, though,
because apparently the watch doesn't yield an error even if the
context calling it is canceled. Whee.

This should reduce the risk of ending up in an inconsistent state,
and also the Close() function is probably idempotent now.

Smaller changes:

* Remove config-generators that existed to generate etcd
  configs but were used only for tests that no longer exist
  or make sense.
* Move the logic to generate etcd configs into the etcd
  package, instead of the "testing" subpackage. This allows
  us to write a self-contained config generator for
  clusters where the nodes know about each other, but do
  this just with etcd, not with full featurebase servers.
* Move the thing generating `fake:%d` socket names into
  the etcd package, which is the only place we use it.
  Also simplify it slightly.
* Don't panic on invalid URLs, report errors from them.
* At least try to use etcd's config.Validate functionality.
  It's underdocumented, so we're not sure what it will report,
  but at least if it does we'll get reports from it and
  know what they are?
* Try to handle CompactRevision errors from watches more
  correctly -- after a CompactRevision, any future attempt
  to watch from a lower revision will necessarily fail, so
  we adjust our target revision up. We don't have good
  testing for this.
* Drop the Metadata() method (that used to be in Metadator)
  because nothing ever used it and it didn't make much sense
  to try.
* Convert SetMetadata from taking an arbitrary json blob
  to taking the only data that would ever be valid since
  we always use it to extract node information anyway.
* Drop several unused functions, unexport things only used
  internally.
* Replace Started() with SetState("STARTED"), allowing us
  to write tests that mess with states. We weren't really thinking
  carefully about state transitions sometimes and now it's much
  easier to do that thinking.
* Stop leaving stray localhost:2380 and localhost:2379 in
  our embed config. We still sometimes see peer requests from
  those and I honestly don't know why, but at least it should
  be rarer.
This commit is contained in:
Seebs 2022-06-23 10:58:41 -05:00 • committed by seebs
parent 5451312515
commit b3a4e52a13
34 changed files with 648 additions and 574 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
api.go
View file

@ -30,7 +30,6 @@ import (
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/stats"
"github.com/molecula/featurebase/v3/topology"
"github.com/molecula/featurebase/v3/tracing"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
@ -763,7 +762,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
}
// ShardNodes returns the node and all replicas which should contain a shard's data.
func (api *API) ShardNodes(ctx context.Context, indexName string, shard uint64) ([]*topology.Node, error) {
func (api *API) ShardNodes(ctx context.Context, indexName string, shard uint64) ([]*disco.Node, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.ShardNodes")
defer span.Finish()
@ -778,7 +777,7 @@ func (api *API) ShardNodes(ctx context.Context, indexName string, shard uint64)
}
// PartitionNodes returns the node and all replicas which should contain a partition key data.
func (api *API) PartitionNodes(ctx context.Context, partitionID int) ([]*topology.Node, error) {
func (api *API) PartitionNodes(ctx context.Context, partitionID int) ([]*disco.Node, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.PartitionNodes")
defer span.Finish()
@ -896,7 +895,7 @@ func (api *API) TranslateData(ctx context.Context, indexName string, partition i
// Find the node that can service the request.
snap := api.cluster.NewSnapshot()
nodes := snap.PartitionNodes(partition)
var upNode *topology.Node
var upNode *disco.Node
for _, node := range nodes {
// we all UNKNOWN state here because we often mistakenly think
// a node is not up under heavy load, but prefer STARTED if we
@ -963,14 +962,14 @@ func (api *API) FieldTranslateData(ctx context.Context, indexName, fieldName str
// Hosts returns a list of the hosts in the cluster including their ID,
// URL, and which is the primary.
func (api *API) Hosts(ctx context.Context) []*topology.Node {
func (api *API) Hosts(ctx context.Context) []*disco.Node {
span, _ := tracing.StartSpanFromContext(ctx, "API.Hosts")
defer span.Finish()
return api.cluster.Nodes()
}
// Node gets the ID, URI and primary status for this particular node.
func (api *API) Node() *topology.Node {
func (api *API) Node() *disco.Node {
return api.server.node()
}
@ -981,7 +980,7 @@ func (api *API) NodeID() string {
}
// PrimaryNode returns the primary node for the cluster.
func (api *API) PrimaryNode() *topology.Node {
func (api *API) PrimaryNode() *disco.Node {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := api.cluster.NewSnapshot()
return snap.PrimaryFieldTranslationNode()

4
boltdb/translate_test.go
View file

@ -12,9 +12,9 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/testhook"
"github.com/molecula/featurebase/v3/topology"
)
//var vv = pilosa.VV
@ -382,7 +382,7 @@ func MustNewTranslateStore(tb testing.TB) *boltdb.TranslateStore {
panic(err)
}
s := boltdb.NewTranslateStore("I", "F", 0, topology.DefaultPartitionN, false)
s := boltdb.NewTranslateStore("I", "F", 0, disco.DefaultPartitionN, false)
s.Path = f.Name()
return s
}

6
broadcast.go
View file

@ -4,7 +4,7 @@ package pilosa
import (
"fmt"
"github.com/molecula/featurebase/v3/topology"
"github.com/molecula/featurebase/v3/disco"
"github.com/pkg/errors"
)
@ -29,7 +29,7 @@ func (*nopSerializer) Unmarshal([]byte, Message) error { return nil }
type broadcaster interface {
SendSync(Message) error
SendAsync(Message) error
SendTo(*topology.Node, Message) error
SendTo(*disco.Node, Message) error
}
// Message is the interface implemented by all core pilosa types which can be serialized to messages.
@ -48,7 +48,7 @@ func (nopBroadcaster) SendSync(Message) error { return nil }
func (nopBroadcaster) SendAsync(Message) error { return nil }
// SendTo is a no-op implementation of Broadcaster SendTo method.
func (nopBroadcaster) SendTo(*topology.Node, Message) error { return nil }
func (nopBroadcaster) SendTo(*disco.Node, Message) error { return nil }
// Broadcast message types.
const (

55
cluster.go
View file

@ -11,7 +11,6 @@ import (
"github.com/molecula/featurebase/v3/ingest"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
@ -23,13 +22,13 @@ const (
// cluster represents a collection of nodes.
type cluster struct { // nolint: maligned
noder topology.Noder
noder disco.Noder
id string
Node *topology.Node
Node *disco.Node
// Hashing algorithm used to assign partitions to nodes.
Hasher topology.Hasher
Hasher disco.Hasher
// The number of partitions in the cluster.
partitionN int
@ -48,7 +47,6 @@ type cluster struct { // nolint: maligned
// Distributed Consensus
disCo disco.DisCo
stator disco.Stator
sharder disco.Sharder
holder *Holder
@ -78,8 +76,8 @@ type cluster struct { // nolint: maligned
// newCluster returns a new instance of Cluster with defaults.
func newCluster() *cluster {
return &cluster{
Hasher: &topology.Jmphasher{},
partitionN: topology.DefaultPartitionN,
Hasher: &disco.Jmphasher{},
partitionN: disco.DefaultPartitionN,
ReplicaN: 1,
closing: make(chan struct{}),
@ -93,9 +91,8 @@ func newCluster() *cluster {
confirmDownRetries: defaultConfirmDownRetries,
confirmDownSleep: defaultConfirmDownSleep,
disCo: disco.NopDisCo,
noder: topology.NewEmptyLocalNoder(),
stator: disco.NopStator,
disCo: disco.NopDisCo,
noder: disco.NewEmptyLocalNoder(),
}
}
@ -126,12 +123,12 @@ func (c *cluster) abortAntiEntropy() {
}
}
func (c *cluster) primaryNode() *topology.Node {
func (c *cluster) primaryNode() *disco.Node {
return c.unprotectedPrimaryNode()
}
// unprotectedPrimaryNode returns the primary node.
func (c *cluster) unprotectedPrimaryNode() *topology.Node {
func (c *cluster) unprotectedPrimaryNode() *disco.Node {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := c.NewSnapshot()
return snap.PrimaryFieldTranslationNode()
@ -161,7 +158,7 @@ func (c *cluster) applySchemaWithNewShards(schema *Schema) error {
// unprotectedStatus returns the the cluster's status including what nodes it contains, its ID, and current state.
func (c *cluster) unprotectedStatus() (*ClusterStatus, error) {
state, err := c.stator.ClusterState(context.Background())
state, err := c.noder.ClusterState(context.Background())
if err != nil {
return nil, err
}
@ -196,21 +193,21 @@ func (c *cluster) remoteSchema() (*Schema, error) {
// nodeIDs returns the list of IDs in the cluster.
func (c *cluster) nodeIDs() []string {
return topology.Nodes(c.Nodes()).IDs()
return disco.Nodes(c.Nodes()).IDs()
}
func (c *cluster) State() (disco.ClusterState, error) {
return c.stator.ClusterState(context.Background())
return c.noder.ClusterState(context.Background())
}
func (c *cluster) nodeByID(id string) *topology.Node {
func (c *cluster) nodeByID(id string) *disco.Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.unprotectedNodeByID(id)
}
// unprotectedNodeByID returns a node reference by ID.
func (c *cluster) unprotectedNodeByID(id string) *topology.Node {
func (c *cluster) unprotectedNodeByID(id string) *disco.Node {
for _, n := range c.noder.Nodes() {
if n.ID == id {
return n
@ -231,13 +228,13 @@ func (c *cluster) nodePositionByID(nodeID string) int {
// Nodes returns a copy of the slice of nodes in the cluster. Safe for
// concurrent use, result may be modified.
func (c *cluster) Nodes() []*topology.Node {
func (c *cluster) Nodes() []*disco.Node {
nodes := c.noder.Nodes()
// duplicate the nodes since we're going to be altering them
copiedNodes := make([]topology.Node, len(nodes))
result := make([]*topology.Node, len(nodes))
copiedNodes := make([]disco.Node, len(nodes))
result := make([]*disco.Node, len(nodes))
primary := topology.PrimaryNode(nodes, c.Hasher)
primary := disco.PrimaryNode(nodes, c.Hasher)
// Set node states and IsPrimary.
for i, node := range nodes {
@ -293,13 +290,13 @@ func (c *cluster) close() error {
// PrimaryReplicaNode returns the node listed before the current node in c.Nodes.
// This is different than "previous node" as the first node always returns nil.
func (c *cluster) PrimaryReplicaNode() *topology.Node {
func (c *cluster) PrimaryReplicaNode() *disco.Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.unprotectedPrimaryReplicaNode()
}
func (c *cluster) unprotectedPrimaryReplicaNode() *topology.Node {
func (c *cluster) unprotectedPrimaryReplicaNode() *disco.Node {
pos := c.nodePositionByID(c.Node.ID)
if pos <= 0 {
return nil
@ -642,7 +639,7 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s
// TODO: use local replicas to short-circuit network traffic
// Group keys by node.
keysByNode := make(map[*topology.Node][]string)
keysByNode := make(map[*disco.Node][]string)
for partitionID, keys := range keysByPartition {
// Find the primary node for this partition.
primary := snap.PrimaryPartitionNode(partitionID)
@ -752,7 +749,7 @@ func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys ..
// Group keys by node.
// Delete remote keys from the by-partition map so that it can be used for local translation.
keysByNode := make(map[*topology.Node][]string)
keysByNode := make(map[*disco.Node][]string)
for partitionID, keys := range keysByPartition {
// Find the primary node for this partition.
primary := snap.PrimaryPartitionNode(partitionID)
@ -909,8 +906,8 @@ func (c *cluster) translateIndexIDSet(ctx context.Context, indexName string, idS
return idMap, nil
}
func (c *cluster) NewSnapshot() *topology.ClusterSnapshot {
return topology.NewClusterSnapshot(c.noder, c.Hasher, c.partitionAssigner, c.ReplicaN)
func (c *cluster) NewSnapshot() *disco.ClusterSnapshot {
return disco.NewClusterSnapshot(c.noder, c.Hasher, c.partitionAssigner, c.ReplicaN)
}
// ClusterStatus describes the status of the cluster including its
@ -918,7 +915,7 @@ func (c *cluster) NewSnapshot() *topology.ClusterSnapshot {
type ClusterStatus struct {
ClusterID string
State string
Nodes []*topology.Node
Nodes []*disco.Node
Schema *Schema
}
@ -997,7 +994,7 @@ type NodeStateMessage struct {
// NodeStatus is an internal message representing the contents of a node.
type NodeStatus struct {
Node *topology.Node
Node *disco.Node
Indexes []*IndexStatus
Schema *Schema
}

36
cluster_internal_test.go
View file

@ -10,10 +10,10 @@ import (
"time"
"github.com/davecgh/go-spew/spew"
"github.com/molecula/featurebase/v3/disco"
pnet "github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/testhook"
"github.com/molecula/featurebase/v3/topology"
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
)
@ -57,7 +57,7 @@ func newIndexWithTempPath(tb testing.TB, name string) *Index {
// Ensure the cluster can fairly distribute partitions across the nodes.
func TestCluster_Owners(t *testing.T) {
c := cluster{
noder: topology.NewLocalNoder([]*topology.Node{
noder: disco.NewLocalNoder([]*disco.Node{
{URI: NewTestURIFromHostPort("serverA", 1000)},
{URI: NewTestURIFromHostPort("serverB", 1000)},
{URI: NewTestURIFromHostPort("serverC", 1000)},
@ -72,12 +72,12 @@ func TestCluster_Owners(t *testing.T) {
snap := c.NewSnapshot()
// Verify nodes are distributed.
if a := snap.PartitionNodes(0); !reflect.DeepEqual(a, []*topology.Node{cNodes[0], cNodes[1]}) {
if a := snap.PartitionNodes(0); !reflect.DeepEqual(a, []*disco.Node{cNodes[0], cNodes[1]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
// Verify nodes go around the ring.
if a := snap.PartitionNodes(2); !reflect.DeepEqual(a, []*topology.Node{cNodes[2], cNodes[0]}) {
if a := snap.PartitionNodes(2); !reflect.DeepEqual(a, []*disco.Node{cNodes[2], cNodes[0]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
}
@ -88,7 +88,7 @@ func TestCluster_Partition(t *testing.T) {
c := newCluster()
c.partitionN = partitionN
partitionID := topology.ShardToShardPartition(index, shard, partitionN)
partitionID := disco.ShardToShardPartition(index, shard, partitionN)
if partitionID < 0 || partitionID >= partitionN {
t.Errorf("partition out of range: shard=%d, p=%d, n=%d", shard, partitionID, partitionN)
}
@ -118,7 +118,7 @@ func TestHasher(t *testing.T) {
{0x0ddc0ffeebadf00d, []int{0, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 15, 15, 15, 15}},
} {
for i, v := range tt.bucket {
hasher := &topology.Jmphasher{}
hasher := &disco.Jmphasher{}
if got := hasher.Hash(tt.key, i+1); got != v {
t.Errorf("hash(%v,%v)=%v, want %v", tt.key, i+1, got, v)
}
@ -150,15 +150,15 @@ func TestCluster_Nodes(t *testing.T) {
uris = append(uris, NewTestURIFromHostPort(fmt.Sprintf("node%d", i), uint16(arbitraryPorts[i])))
}
node0 := &topology.Node{ID: "node0", URI: uris[0]}
node1 := &topology.Node{ID: "node1", URI: uris[1]}
node2 := &topology.Node{ID: "node2", URI: uris[2]}
node3 := &topology.Node{ID: "node3", URI: uris[3]}
node0 := &disco.Node{ID: "node0", URI: uris[0]}
node1 := &disco.Node{ID: "node1", URI: uris[1]}
node2 := &disco.Node{ID: "node2", URI: uris[2]}
node3 := &disco.Node{ID: "node3", URI: uris[3]}
nodes := []*topology.Node{node0, node1, node2}
nodes := []*disco.Node{node0, node1, node2}
t.Run("NodeIDs", func(t *testing.T) {
actual := topology.Nodes(nodes).IDs()
actual := disco.Nodes(nodes).IDs()
expected := []string{node0.ID, node1.ID, node2.ID}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
@ -166,7 +166,7 @@ func TestCluster_Nodes(t *testing.T) {
})
t.Run("Filter", func(t *testing.T) {
actual := topology.Nodes(topology.Nodes(nodes).Filter(nodes[1])).URIs()
actual := disco.Nodes(disco.Nodes(nodes).Filter(nodes[1])).URIs()
expected := []pnet.URI{uris[0], uris[2]}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
@ -174,7 +174,7 @@ func TestCluster_Nodes(t *testing.T) {
})
t.Run("FilterURI", func(t *testing.T) {
actual := topology.Nodes(topology.Nodes(nodes).FilterURI(uris[1])).URIs()
actual := disco.Nodes(disco.Nodes(nodes).FilterURI(uris[1])).URIs()
expected := []pnet.URI{uris[0], uris[2]}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
@ -182,8 +182,8 @@ func TestCluster_Nodes(t *testing.T) {
})
t.Run("Contains", func(t *testing.T) {
actualTrue := topology.Nodes(nodes).Contains(node1)
actualFalse := topology.Nodes(nodes).Contains(node3)
actualTrue := disco.Nodes(nodes).Contains(node1)
actualFalse := disco.Nodes(nodes).Contains(node3)
if !reflect.DeepEqual(actualTrue, true) {
t.Errorf("expected: %v, but got: %v", true, actualTrue)
}
@ -193,8 +193,8 @@ func TestCluster_Nodes(t *testing.T) {
})
t.Run("Clone", func(t *testing.T) {
clone := topology.Nodes(nodes).Clone()
actual := topology.Nodes(clone).URIs()
clone := disco.Nodes(nodes).Clone()
actual := disco.Nodes(clone).URIs()
expected := []pnet.URI{uris[0], uris[1], uris[2]}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)

6
ctl/backup.go
View file

@ -14,9 +14,9 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/authn"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
@ -300,7 +300,7 @@ func (cmd *BackupCommand) backupShard(ctx context.Context, indexName string, sha
}
// backupShardNode backs up a single shard from a single index on a specific node.
func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string, shard uint64, node *topology.Node) error {
func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string, shard uint64, node *disco.Node) error {
logger := cmd.Logger()
logger.Printf("backing up shard: index=%q id=%d", indexName, shard)
@ -334,7 +334,7 @@ func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string,
}
func (cmd *BackupCommand) backupIndexTranslateData(ctx context.Context, name string) error {
partitionN := topology.DefaultPartitionN
partitionN := disco.DefaultPartitionN
ch := make(chan int, partitionN)
for partitionID := 0; partitionID < partitionN; partitionID++ {

12
ctl/restore.go
View file

@ -19,9 +19,9 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/authn"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
@ -103,7 +103,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
return err
}
var primary *topology.Node
var primary *disco.Node
for _, node := range nodes {
if node.IsPrimary {
primary = node
@ -138,7 +138,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
return nil
}
func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.Node) error {
func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *disco.Node) error {
f, err := os.Open(filepath.Join(cmd.Path, "schema"))
if err != nil {
return err
@ -235,7 +235,7 @@ func (cmd *RestoreCommand) newClient() *retryablehttp.Client {
return client
}
func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology.Node) error {
func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *disco.Node) error {
logger := cmd.Logger()
f, err := os.Open(filepath.Join(cmd.Path, "idalloc"))
@ -411,7 +411,7 @@ func (cmd *RestoreCommand) restoreIndexTranslationFile(ctx context.Context, file
return nil
}
func (cmd *RestoreCommand) restoreFieldTranslation(ctx context.Context, nodes []*topology.Node) error {
func (cmd *RestoreCommand) restoreFieldTranslation(ctx context.Context, nodes []*disco.Node) error {
filenames, err := filepath.Glob(filepath.Join(cmd.Path, "indexes", "*", "fields", "*", "translate"))
if err != nil {
return err
@ -443,7 +443,7 @@ func (cmd *RestoreCommand) restoreFieldTranslation(ctx context.Context, nodes []
return g.Wait()
}
func (cmd *RestoreCommand) restoreFieldTranslationFile(ctx context.Context, nodes []*topology.Node, filename string) error {
func (cmd *RestoreCommand) restoreFieldTranslationFile(ctx context.Context, nodes []*disco.Node, filename string) error {
logger := cmd.Logger()
rel, err := filepath.Rel(cmd.Path, filename)

48
disco/disco.go
View file

@ -68,21 +68,6 @@ const (
NodeStateStarted NodeState = "STARTED"
)
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"
ClusterState(context.Context) (ClusterState, error)
}
// Schema is a map of all indexes, each of those being a map of fields, then
// views.
type Schema map[string]*Index
@ -123,13 +108,6 @@ type Schemator interface {
DeleteView(ctx context.Context, index, field, view string) 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
}
// Sharder is an interface used to maintain the set of availableShards bitmaps
// per field.
type Sharder interface {
@ -177,32 +155,6 @@ func (n *nopDisCo) DeleteNode(context.Context, string) error {
return nil
}
// NopStator represents a Stator that doesn't do anything.
var NopStator Stator = &nopStator{}
type nopStator struct{}
// ClusterState is a no-op implementation of the Stator ClusterState method.
func (n *nopStator) ClusterState(context.Context) (ClusterState, error) {
return ClusterStateUnknown, nil
}
func (n *nopStator) Started(ctx context.Context) error {
return nil
}
// NopMetadator represents a Metadator that doesn't do anything.
var NopMetadator Metadator = &nopMetadator{}
type nopMetadator struct{}
func (*nopMetadator) Metadata(context.Context, string) ([]byte, error) {
return nil, nil
}
func (*nopMetadator) SetMetadata(context.Context, []byte) error {
return nil
}
// NopSharder represents a Sharder that doesn't do anything.
var NopSharder Sharder = &nopSharder{}

2
topology/hasher.go → disco/hasher.go
View file

@ -1,5 +1,5 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package topology
package disco
// Hasher represents an interface to hash integers into buckets.
type Hasher interface {

13
topology/node.go → disco/node.go
View file

@ -1,20 +1,19 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package topology
package disco
import (
"fmt"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/net"
)
// Node represents a node in the cluster.
type Node struct {
ID string `json:"id"`
URI net.URI `json:"uri"`
GRPCURI net.URI `json:"grpc-uri"`
IsPrimary bool `json:"isPrimary"`
State disco.NodeState `json:"state"`
ID string `json:"id"`
URI net.URI `json:"uri"`
GRPCURI net.URI `json:"grpc-uri"`
IsPrimary bool `json:"isPrimary"`
State NodeState `json:"state"`
}
func (n *Node) Clone() *Node {

31
topology/noder.go → disco/noder.go
View file

@ -1,7 +1,8 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package topology
package disco
import (
"context"
"sort"
)
@ -10,6 +11,20 @@ import (
type Noder interface {
Nodes() []*Node // Remember: this has to be sorted correctly!!
PrimaryNodeID(hasher Hasher) string
// SetMetadata records the local node's metadata.
SetMetadata(ctx context.Context, node *Node) error
// SetState changes a node to a given state.
SetState(ctx context.Context, state NodeState) 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"
ClusterState(context.Context) (ClusterState, error)
}
// localNoder is a simple implementation of the Noder interface
@ -64,3 +79,17 @@ func (n *localNoder) PrimaryNodeID(hasher Hasher) string {
}
return primaryNode.ID
}
// ClusterState is a no-op implementation of the Stator ClusterState method.
func (n *localNoder) ClusterState(context.Context) (ClusterState, error) {
return ClusterStateUnknown, nil
}
func (n *localNoder) SetState(ctx context.Context, state NodeState) error {
return nil
}
// localNoder doesn't really implement metadata support.
func (*localNoder) SetMetadata(context.Context, *Node) error {
return nil
}

2
topology/snapshot.go → disco/snapshot.go
View file

@ -1,5 +1,5 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package topology
package disco
import (
"encoding/binary"

21
encoding/proto/proto.go
View file

@ -14,7 +14,6 @@ import (
"github.com/molecula/featurebase/v3/pb"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
)
@ -159,7 +158,7 @@ func (s Serializer) Unmarshal(buf []byte, m pilosa.Message) error {
}
s.decodeNodeStatus(msg, mt)
return nil
case *topology.Node:
case *disco.Node:
msg := &pb.Node{}
err := proto.Unmarshal(buf, msg)
if err != nil {
@ -345,7 +344,7 @@ func (s Serializer) encodeToProto(m pilosa.Message) proto.Message {
return s.encodeNodeEventMessage(mt)
case *pilosa.NodeStatus:
return s.encodeNodeStatus(mt)
case *topology.Node:
case *disco.Node:
return s.encodeNode(mt)
case *pilosa.QueryRequest:
return s.encodeQueryRequest(mt)
@ -633,7 +632,7 @@ func (s Serializer) encodeFieldOptions(o *pilosa.FieldOptions) *pb.FieldOptions
}
// s.encodeNodes converts a slice of Nodes into its pb.representation.
func (s Serializer) encodeNodes(a []*topology.Node) []*pb.Node {
func (s Serializer) encodeNodes(a []*disco.Node) []*pb.Node {
other := make([]*pb.Node, len(a))
for i := range a {
other[i] = s.encodeNode(a[i])
@ -642,7 +641,7 @@ func (s Serializer) encodeNodes(a []*topology.Node) []*pb.Node {
}
// s.encodeNode converts a Node into its pb.representation.
func (s Serializer) encodeNode(m *topology.Node) *pb.Node {
func (s Serializer) encodeNode(m *disco.Node) *pb.Node {
n := m.Clone()
return &pb.Node{
ID: n.ID,
@ -1019,9 +1018,9 @@ func (s Serializer) decodeDecimal(d *pb.Decimal, m *pql.Decimal) {
m.Scale = d.Scale
}
func (s Serializer) decodeNodes(a []*pb.Node, m []*topology.Node) {
func (s Serializer) decodeNodes(a []*pb.Node, m []*disco.Node) {
for i := range a {
m[i] = &topology.Node{}
m[i] = &disco.Node{}
s.decodeNode(a[i], m[i])
}
}
@ -1029,13 +1028,13 @@ func (s Serializer) decodeNodes(a []*pb.Node, m []*topology.Node) {
func (s Serializer) decodeClusterStatus(cs *pb.ClusterStatus, m *pilosa.ClusterStatus) {
m.State = cs.State
m.ClusterID = cs.ClusterID
m.Nodes = make([]*topology.Node, len(cs.Nodes))
m.Nodes = make([]*disco.Node, len(cs.Nodes))
s.decodeNodes(cs.Nodes, m.Nodes)
m.Schema = &pilosa.Schema{}
s.decodeSchema(cs.Schema, m.Schema)
}
func (s Serializer) decodeNode(node *pb.Node, m *topology.Node) {
func (s Serializer) decodeNode(node *pb.Node, m *disco.Node) {
m.ID = node.ID
s.decodeURI(node.URI, &m.URI)
s.decodeURI(node.GRPCURI, &m.GRPCURI)
@ -1120,12 +1119,12 @@ func (s Serializer) decodeNodeStateMessage(pb *pb.NodeStateMessage, m *pilosa.No
func (s Serializer) decodeNodeEventMessage(pb *pb.NodeEventMessage, m *pilosa.NodeEvent) {
m.Event = pilosa.NodeEventType(pb.Event)
m.Node = &topology.Node{}
m.Node = &disco.Node{}
s.decodeNode(pb.Node, m.Node)
}
func (s Serializer) decodeNodeStatus(pb *pb.NodeStatus, m *pilosa.NodeStatus) {
m.Node = &topology.Node{}
m.Node = &disco.Node{}
m.Indexes = s.decodeIndexStatuses(pb.Indexes)
m.Schema = &pilosa.Schema{}
s.decodeSchema(pb.Schema, m.Schema)

71
etcd/config_gen.go Normal file
View file

@ -0,0 +1,71 @@
package etcd
import (
"fmt"
"os"
"strings"
"sync/atomic"
)
// Helper utilities to allow us to create etcd cluster configs
// that can clean up after themselves.
// DirCleaner represents the subset of the testing.TB interface
// we care about, allowing us to take objects which behave like
// that without importing all of testing to get them.
type DirCleaner interface {
Cleanup(func())
Fatalf(string, ...interface{})
Logf(string, ...interface{})
Name() string
TempDir() string
}
var unixSocketCounter uint64
// unixSocket returns a url for use in test etcd clusters, referring
// to a file which will be cleaned up after a test completes.
func unixSocket(dc DirCleaner) string {
count := atomic.AddUint64(&unixSocketCounter, 1)
addr := fmt.Sprintf("fake:%d", count)
dc.Cleanup(func() {
err := os.Remove(addr)
if err != nil && !os.IsNotExist(err) { // not an error if the socket is not present
dc.Logf("could not remove '%s', %v", addr, err)
}
})
return fmt.Sprintf("unix://%s", addr)
}
// GenEtcdConfigs generates a set of etcd configs, and an initial cluster
// URL. It also makes temporary etcd dirs. Everything it creates
// is cleaned up by `dc.Cleanup` after the test/benchmark completes. On
// error, it calls `dc.Fatalf`, presumably terminating the test.
func GenEtcdConfigs(dc DirCleaner, n int) (clusterName string, cfgs []Options) {
allPeerURLs := make([]string, n)
cfgs = make([]Options, n)
clusterName = fmt.Sprintf("cluster-%s", dc.Name())
for i := 0; i < n; i++ {
name := fmt.Sprintf("server%d", i)
discoDir := dc.TempDir()
clientURL := unixSocket(dc)
peerURL := unixSocket(dc)
cfgs[i] = Options{
Name: name,
Dir: discoDir,
LClientURL: clientURL,
AClientURL: clientURL,
LPeerURL: peerURL,
APeerURL: peerURL,
HeartbeatTTL: 60,
UnsafeNoFsync: true,
}
allPeerURLs[i] = fmt.Sprintf("%s=%s", name, peerURL)
}
peerURLs := strings.Join(allPeerURLs, ",")
for i := range cfgs {
cfgs[i].InitCluster = peerURLs
}
return clusterName, cfgs
}

356
etcd/embed.go
View file

@ -16,7 +16,6 @@ import (
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/monitor"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
"go.etcd.io/etcd/api/v3/mvccpb"
"go.etcd.io/etcd/client/pkg/v3/transport"
@ -55,8 +54,7 @@ type Options struct {
var (
_ disco.DisCo = &Etcd{}
_ disco.Schemator = &Etcd{}
_ disco.Stator = &Etcd{}
_ disco.Metadator = &Etcd{}
_ disco.Noder = &Etcd{}
_ disco.Sharder = &Etcd{}
)
@ -71,7 +69,6 @@ const (
)
var (
etcdLeaderChanged = etcdserver.ErrLeaderChanged.Error()
errEtcdShuttingDown = errors.New("etcd shutting down")
)
@ -79,7 +76,7 @@ var (
// nodes in etcd, which we update from data collected either directly
// from the KV, or via heartbeats.
//
// Any change to the topology.Node should create a new Node rather
// Any change to the disco.Node should create a new Node rather
// than reusing the old one, so we can return the structure and not worry
// about data races.
//
@ -88,16 +85,9 @@ var (
// but still handle cases where we get updates to several fields that
// reach us out of order.
type nodeData struct {
heartbeatState string
metadata []byte
topologyNode *topology.Node
}
func (n *nodeData) computedState() disco.NodeState {
if n.heartbeatState != "" {
return disco.NodeState(n.heartbeatState)
}
return disco.NodeStateUnknown
heartbeat disco.NodeState
metadata []byte
node *disco.Node
}
type Etcd struct {
@ -110,20 +100,20 @@ type Etcd struct {
heartbeatLeasedKV *leasedKV
// We have a watcher running. watchCancel() cancels its context.
watchCancel func()
closeWatch chan struct{}
// A context for our children (node watcher, lease keepalive)
childContext context.Context
// function to cancel the child contexts when we're done
childCancel func()
// knownNodes and sortedNodes get updated by data coming in from
// watchers. Any change to the contents of a *topology.Node here
// watchers. Any change to the contents of a *disco.Node here
// should be implemented by making a new one and replacing the pointer,
// so the old pointer stays valid and can be used.
nodeMu sync.Mutex
nodeRev int64
knownNodes map[string]*nodeData
sortedNodes []*topology.Node
nodeStates map[string]disco.NodeState
nodeStatesDirty bool // do we need to remake the nodeStates map to use it?
nodeMu sync.Mutex
nodeRev int64
knownNodes map[string]*nodeData
sortedNodes []*disco.Node // immutable nodes kept in sorted order
nodesDirty bool // do we need to recompute sortedNodes?
version string
@ -137,7 +127,6 @@ func NewEtcd(opt Options, logger logger.Logger, replicas int, version string) *E
logger: logger,
replicas: replicas,
knownNodes: make(map[string]*nodeData),
nodeStates: make(map[string]disco.NodeState),
version: version,
}
@ -149,21 +138,25 @@ func NewEtcd(opt Options, logger logger.Logger, replicas int, version string) *E
// Close implements io.Closer
func (e *Etcd) Close() error {
if e.closeWatch != nil {
close(e.closeWatch)
// tell the heartbeat to stop. we do this before canceling
// the context because we want the heartbeat to get a chance
// to notify other nodes that it's down.
if e.heartbeatLeasedKV != nil {
e.heartbeatLeasedKV.Stop()
}
if e.watchCancel != nil {
e.watchCancel()
// cancel the contexts that heartbeat and watcher are using.
if e.childCancel != nil {
e.childCancel()
}
// shut down the server, if we have one.
if e.e != nil {
if e.heartbeatLeasedKV != nil {
e.heartbeatLeasedKV.Stop()
}
e.e.Close()
<-e.e.Server.StopNotify()
}
// shut down the client, if we have one. if something's still
// using it, we anticipate the client failing its current call,
// and any retry will be checking for the child context being
// cancelled, first, we hope.
if e.cli != nil {
e.cli.Close()
}
@ -171,17 +164,6 @@ func (e *Etcd) Close() error {
return nil
}
// retryClient attempts to do a thing, but also tries to handle the
// specific case where the client fails because of a leader election,
// in which case we need to restart the client and retry the thing.
//
// We have to let go of the lock while calling `fn` because some fn are
// long-lasting ones, like watchNodesOnce. So we grab a local copy of
// the client object, then call things on that object. This should error
// out sanely instead of panicing if we close the client while something
// is running on it.
//
// New feature: retryClient can also retry on errTimeout.
const etcdRetryTimes = 3
// newClient requests a new client which is different from the one
@ -200,6 +182,17 @@ func (e *Etcd) newClient(cli *clientv3.Client) *clientv3.Client {
return e.cli
}
// retryClient attempts to do a thing, but also tries to handle the
// specific case where the client fails because of a leader election,
// in which case we need to restart the client and retry the thing.
//
// We have to let go of the lock while calling `fn` because some fn are
// long-lasting ones, like watchNodesOnce. So we grab a local copy of
// the client object, then call things on that object. This should error
// out sanely instead of panicing if we close the client while something
// is running on it.
//
// New feature: retryClient can also retry on errTimeout.
func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) {
e.cliMu.Lock()
cli := e.cli
@ -210,7 +203,6 @@ func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) {
switch err {
case etcdserver.ErrLeaderChanged:
cli = e.newClient(cli)
break
case nil:
return nil
default:
@ -242,7 +234,6 @@ func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) {
// timeout to give us a reasonable backoff period and keep us
// from spamming these.
time.Sleep(100 * time.Millisecond)
break
}
}
// if we got here, we got a total of three of some combination of
@ -257,16 +248,29 @@ func (e *Etcd) parseOptions() (*embed.Config, error) {
cfg.Name = e.options.Name
cfg.Dir = e.options.Dir
cfg.InitialClusterToken = e.options.ClusterName
cfg.LCUrls = types.MustNewURLs([]string{e.options.LClientURL})
var err error
cfg.LCUrls, err = types.NewURLs([]string{e.options.LClientURL})
if err != nil {
return nil, fmt.Errorf("parsing listen client URL %q: %v", e.options.LClientURL, err)
}
cfg.UnsafeNoFsync = e.options.UnsafeNoFsync
if e.options.AClientURL != "" {
cfg.ACUrls = types.MustNewURLs([]string{e.options.AClientURL})
cfg.ACUrls, err = types.NewURLs([]string{e.options.AClientURL})
if err != nil {
return nil, fmt.Errorf("parsing advertise client URL %q: %v", e.options.AClientURL, err)
}
} else {
cfg.ACUrls = cfg.LCUrls
}
cfg.LPUrls = types.MustNewURLs([]string{e.options.LPeerURL})
cfg.LPUrls, err = types.NewURLs([]string{e.options.LPeerURL})
if err != nil {
return nil, fmt.Errorf("parsing listen peer URL %q: %v", e.options.LPeerURL, err)
}
if e.options.APeerURL != "" {
cfg.APUrls = types.MustNewURLs([]string{e.options.APeerURL})
cfg.APUrls, err = types.NewURLs([]string{e.options.APeerURL})
if err != nil {
return nil, fmt.Errorf("parsing advertise peer URL %q: %v", e.options.APeerURL, err)
}
} else {
cfg.APUrls = cfg.LPUrls
}
@ -306,7 +310,9 @@ func (e *Etcd) parseOptions() (*embed.Config, error) {
KeyFile: e.options.PeerKeyFile,
}
return cfg, nil
// We might get an error from Validate. etcd docs don't tell us what
// that error might be, though!
return cfg, cfg.Validate()
}
// Start starts etcd and hearbeat
@ -316,6 +322,8 @@ func (e *Etcd) Start(ctx context.Context) (_ disco.InitialClusterState, err erro
return disco.InitialClusterStateNew, err
}
state := disco.InitialClusterState(opts.ClusterState)
// create a context that can be used for our watch processes, etcetera.
e.childContext, e.childCancel = context.WithCancel(context.Background())
e.e, err = embed.StartEtcd(opts)
if err != nil {
@ -325,7 +333,8 @@ func (e *Etcd) Start(ctx context.Context) (_ disco.InitialClusterState, err erro
// later, so we have to stop the server ourselves.
defer func() {
if err != nil {
e.e.Server.Stop()
// shut down everything on our way out.
e.Close()
}
}()
e.cli = v3client.New(e.e.Server)
@ -345,15 +354,9 @@ func (e *Etcd) Start(ctx context.Context) (_ disco.InitialClusterState, err erro
// heard from yet.
for _, member := range members {
peerID := member.ID.String()
e.knownNodes[peerID] = &nodeData{
topologyNode: &topology.Node{
ID: peerID,
State: disco.NodeStateUnknown,
},
}
e.nodeStates[peerID] = disco.NodeStateUnknown
_ = e.seeNode(peerID)
}
e.nodeStatesDirty = true
e.nodesDirty = true
return state, e.startHeartbeatAndWatcher(ctx)
}
}
@ -362,22 +365,21 @@ func (e *Etcd) Start(ctx context.Context) (_ disco.InitialClusterState, err erro
// watcher that watches for changes to events we care about.
func (e *Etcd) startHeartbeatAndWatcher(ctx context.Context) error {
key := heartbeatPrefix + e.e.Server.ID().String()
e.heartbeatLeasedKV = newLeasedKV(e, key, e.options.HeartbeatTTL)
e.closeWatch = make(chan struct{})
e.heartbeatLeasedKV = newLeasedKV(e, e.childContext, key, e.options.HeartbeatTTL)
if err := e.heartbeatLeasedKV.Start(string(disco.NodeStateStarting)); err != nil {
return errors.Wrap(err, "startHeartbeat: starting a new heartbeat")
}
// WatchNodes does not check for an error, and will need to be shut
// watchNodes does not check for an error, and will need to be shut
// down later. We only get this far at a point where we're returning
// a nil error, and thus, the caller is expected to cleanly shut down
// the server later.
go e.WatchNodes()
go e.watchNodes()
return nil
}
func (e *Etcd) Started(ctx context.Context) (err error) {
return e.heartbeatLeasedKV.Set(ctx, string(disco.NodeStateStarted))
func (e *Etcd) SetState(ctx context.Context, state disco.NodeState) (err error) {
return e.heartbeatLeasedKV.Set(ctx, string(state))
}
func (e *Etcd) ID() string {
@ -423,15 +425,14 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er
starting bool
)
e.nodeMu.Lock()
err = e.populateNodeStates(ctx)
states := e.nodeStates
nodes := e.populateNodeStates(ctx)
e.nodeMu.Unlock()
if err != nil {
e.logger.Printf("ClusterState %q: getting node states: %v", e.options.Name, states)
e.logger.Errorf("requesting cluster state %q: getting node states: %v", e.options.Name, err)
return disco.ClusterStateUnknown, err
}
for _, state := range states {
switch state {
for _, node := range nodes {
switch node.State {
case disco.NodeStateStarting:
starting = true
case disco.NodeStateUnknown:
@ -445,8 +446,8 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er
return disco.ClusterStateStarting, nil
}
if heartbeats < len(states) {
if len(states)-heartbeats >= e.replicas {
if heartbeats < len(e.knownNodes) {
if len(e.knownNodes)-heartbeats >= e.replicas {
return disco.ClusterStateDown, nil
}
@ -469,6 +470,24 @@ func parseNodeKey(key []byte) (prefix string, peerID string, err error) {
return string(key[:peerIndex+1]), string(key[peerIndex+1:]), nil
}
// seeNode encapsulates the practice of creating a new node, when needed,
// and checking for duplicates.
func (e *Etcd) seeNode(peerID string) *nodeData {
var node *nodeData
if node = e.knownNodes[peerID]; node == nil {
e.logger.Debugf("previously unseen node, peer ID %s", peerID)
node = &nodeData{
node: &disco.Node{
ID: peerID,
State: disco.NodeStateUnknown,
},
heartbeat: disco.NodeStateUnknown,
}
e.knownNodes[peerID] = node
}
return node
}
// deleteNodeData is like putNodeData, but handles deletes rather than cases
// where a value exists. you should call it with the node mutex locked.
func (e *Etcd) deleteNodeData(key []byte, revision int64) error {
@ -481,18 +500,14 @@ func (e *Etcd) deleteNodeData(key []byte, revision int64) error {
}
switch prefix {
case heartbeatPrefix:
if e.knownNodes[peerID] == nil {
e.knownNodes[peerID] = &nodeData{}
}
e.knownNodes[peerID].heartbeatState = ""
e.nodeStatesDirty = true
node := e.seeNode(peerID)
// mark state as unknown because we deleted the heartbeat.
node.heartbeat = disco.NodeStateUnknown
e.nodesDirty = true
case metadataPrefix:
if e.knownNodes[peerID] == nil {
e.knownNodes[peerID] = &nodeData{}
}
e.knownNodes[peerID].metadata = nil
e.knownNodes[peerID].topologyNode = &topology.Node{}
e.nodeStatesDirty = true
e.logger.Infof("deleting a previously-seen node, peer ID %q", peerID)
delete(e.knownNodes, peerID)
e.nodesDirty = true
default:
return fmt.Errorf("node watch: invalid prefix %q", prefix)
}
@ -512,25 +527,21 @@ func (e *Etcd) putNodeData(key []byte, value []byte, revision int64) (err error)
}
switch prefix {
case heartbeatPrefix:
if e.knownNodes[peerID] == nil {
e.knownNodes[peerID] = &nodeData{}
}
e.knownNodes[peerID].heartbeatState = string(value)
e.nodeStatesDirty = true
node := e.seeNode(peerID)
node.heartbeat = disco.NodeState(value)
e.nodesDirty = true
case metadataPrefix:
if e.knownNodes[peerID] == nil {
e.knownNodes[peerID] = &nodeData{}
}
e.knownNodes[peerID].metadata = value
var newNode topology.Node
node := e.seeNode(peerID)
node.metadata = value
var newNode disco.Node
err := json.Unmarshal(value, &newNode)
if err != nil {
return fmt.Errorf("json unmarshal of node metadata: %v", err)
}
e.knownNodes[peerID].topologyNode = &newNode
// This saves us one remake of the node later, probably.
e.knownNodes[peerID].topologyNode.State = e.knownNodes[peerID].computedState()
e.nodeStatesDirty = true
// start with the heartbeat state
newNode.State = node.heartbeat
node.node = &newNode
e.nodesDirty = true
default:
return fmt.Errorf("node watch: invalid prefix %q", prefix)
}
@ -540,51 +551,70 @@ func (e *Etcd) putNodeData(key []byte, value []byte, revision int64) (err error)
// compute the states of all the nodes. we compute all of them because
// we might have returned the old map in response to a query, so we want to
// make a new one. You should have the node state lock held when you call this.
func (e *Etcd) populateNodeStates(ctx context.Context) error {
if !e.nodeStatesDirty {
return nil
// Returns an immutable sorted list of nodes; future updates will not
// modify the slice or the nodes in it.
func (e *Etcd) populateNodeStates(ctx context.Context) []*disco.Node {
if !e.nodesDirty {
return e.sortedNodes
}
e.nodeStates = make(map[string]disco.NodeState, len(e.knownNodes))
e.sortedNodes = make([]*topology.Node, 0, len(e.knownNodes))
for peerID, data := range e.knownNodes {
newState := data.computedState()
e.nodeStates[peerID] = newState
e.sortedNodes = make([]*disco.Node, 0, len(e.knownNodes))
for _, data := range e.knownNodes {
newState := data.heartbeat
// update the state with the current state, so we can
// reuse these nodes later. sortedNodes may end up shorter
// than the whole node list if we don't have all the nodes
// yet!
if data.topologyNode != nil {
if data.topologyNode.State != newState {
newNode := *data.topologyNode
if data.node != nil {
// The only part that should ever change is the state, which
// will be either "unknown" or the state from a heartbeat.
// If that computed state is different, we make a new node
// at this point. The reason is that, if we previously returned
// the sorted list of nodes, someone else could have a
// pointer to the existing node. We don't want to clone these
// every time anyone reads them, so instead we make them
// immutable and copy-on-write.
if data.node.State != newState {
newNode := *data.node
newNode.State = newState
data.topologyNode = &newNode
data.node = &newNode
}
e.sortedNodes = append(e.sortedNodes, data.topologyNode)
e.sortedNodes = append(e.sortedNodes, data.node)
}
}
// sort list by ID. list now contains sorted nodes which have their
// current states.
sort.Sort(topology.ByID(e.sortedNodes))
e.nodeStatesDirty = false
return nil
sort.Sort(disco.ByID(e.sortedNodes))
e.nodesDirty = false
return e.sortedNodes
}
// watchNodesOnce is a helper function to use with the retry logic
// to let us restart the client if we need to.
func (e *Etcd) watchNodesOnce(ctx context.Context, cli *clientv3.Client) (err error) {
func (e *Etcd) watchNodesOnce(cli *clientv3.Client) (err error) {
e.nodeMu.Lock()
// we are looking for revisions HIGHER than the highest revision we've
// currently seen, we don't want one equal to it.
minRev := e.nodeRev + 1
done := e.childContext.Done()
e.nodeMu.Unlock()
watcher := cli.Watch(ctx, nodePrefix, clientv3.WithPrefix(), clientv3.WithRev(minRev))
watcher := cli.Watch(clientv3.WithRequireLeader(e.childContext), nodePrefix, clientv3.WithPrefix(), clientv3.WithRev(minRev))
for {
select {
case <-e.closeWatch:
return errEtcdShuttingDown
case <-done:
// we're done, this is not an error
return nil
case resp := <-watcher:
if err := resp.Err(); err != nil {
if resp.CompactRevision > minRev {
e.logger.Infof("watching node status, wanted rev %d, minimum now %d",
minRev, resp.CompactRevision)
// We've been told that any request with a revision under
// CompactRevision will always fail. Set nodeRev to one less than that,
// so we'll specify it as the minimum when we retry.
//
// We currently have no obvious way to verify that this will work.
e.nodeRev = resp.CompactRevision - 1
}
return err
}
// lock the node mutex for this whole process of updating so
@ -596,41 +626,37 @@ func (e *Etcd) watchNodesOnce(ctx context.Context, cli *clientv3.Client) (err er
case mvccpb.PUT:
err := e.putNodeData(ev.Kv.Key, ev.Kv.Value, ev.Kv.ModRevision)
if err != nil {
e.logger.Printf("put event: %v", err)
e.logger.Warnf("put event: %v", err)
}
case mvccpb.DELETE:
err := e.deleteNodeData(ev.Kv.Key, ev.Kv.ModRevision)
if err != nil {
e.logger.Printf("delete event: %v", err)
e.logger.Warnf("delete event: %v", err)
}
default:
e.logger.Printf("watchp %q: unknown event %#v", e.options.Name, ev)
e.logger.Warnf("watchp %q: unknown event %#v", e.options.Name, ev)
}
}
e.nodeMu.Unlock()
}
}
}
// WatchNodes monitors changes to /heartbeat/ and /metadata/;
// watchNodes monitors changes to /heartbeat/ and /metadata/;
// basically, it catches changes to cluster state, but ignores the schema.
func (e *Etcd) WatchNodes() {
ctx, cancel := context.WithCancel(context.Background())
e.watchCancel = cancel
watchInContext := func(cli *clientv3.Client) error {
return e.watchNodesOnce(ctx, cli)
}
func (e *Etcd) watchNodes() {
// retryClient will retry on leader failure, but not for other failures
// such as ErrCompacted which can terminate a watch. But we want to resume
// watching again as long as our context isn't cancelled. The context
// should get cancelled when this Etcd gets shut down.
for ctx.Err() == nil {
err := e.retryClient(watchInContext)
for e.childContext.Err() == nil {
err := e.retryClient(func(cli *clientv3.Client) error {
return e.watchNodesOnce(cli)
})
if err != nil {
e.logger.Printf("WatchNodes: error from watch client: %v", err)
e.logger.Warnf("watchNodes: error from watch client: %v", err)
}
// delay slightly on error so we don't go completely crazy
// delay slightly on watch termination so we don't go completely crazy
time.Sleep(1 * time.Second)
}
}
@ -704,24 +730,15 @@ func (e *Etcd) Schema(ctx context.Context) (disco.Schema, error) {
return m, nil
}
func (e *Etcd) Metadata(ctx context.Context, peerID string) ([]byte, error) {
e.nodeMu.Lock()
defer e.nodeMu.Unlock()
err := e.populateNodeStates(ctx)
func (e *Etcd) SetMetadata(ctx context.Context, node *disco.Node) error {
// Set metadata for this node.
data, err := json.Marshal(node)
if err != nil {
return nil, err
return errors.Wrap(err, "marshaling json metadata")
}
data, ok := e.knownNodes[peerID]
if !ok {
return nil, errors.New("node not found")
}
return data.metadata, nil
}
func (e *Etcd) SetMetadata(ctx context.Context, metadata []byte) error {
err := e.putKey(ctx, path.Join(metadataPrefix,
err = e.putKey(ctx, path.Join(metadataPrefix,
e.e.Server.ID().String()),
string(metadata),
string(data),
)
if err != nil {
return errors.Wrap(err, "SetMetadata")
@ -979,31 +996,6 @@ func (e *Etcd) delKey(ctx context.Context, key string, withPrefix bool) (err err
return err
}
func memberList(cli *clientv3.Client) (ids []uint64, names []string, urls []string) {
ml, err := cli.MemberList(context.TODO())
if err != nil {
panic(err)
}
n := len(ml.Members)
ids = make([]uint64, n)
names = make([]string, n)
urls = make([]string, n)
for i, m := range ml.Members {
ids[i], names[i], urls[i] = m.ID, m.Name, m.PeerURLs[0]
}
return
}
func memberAdd(cli *clientv3.Client, peerURL string) (id uint64, name string) {
ma, err := cli.MemberAdd(context.TODO(), []string{peerURL})
if err != nil {
return 0, ""
}
return ma.Member.ID, ma.Member.Name
}
// Shards implements the Sharder interface.
func (e *Etcd) Shards(ctx context.Context, index, field string) ([][]byte, error) {
key := path.Join(shardPrefix, index, field)
@ -1031,19 +1023,15 @@ func (e *Etcd) SetShards(ctx context.Context, index, field string, shards []byte
// Nodes implements the Noder interface. It returns the sorted list of nodes
// based on the etcd peers.
func (e *Etcd) Nodes() []*topology.Node {
func (e *Etcd) Nodes() []*disco.Node {
e.nodeMu.Lock()
defer e.nodeMu.Unlock()
err := e.populateNodeStates(context.TODO())
if err != nil {
return nil
}
return e.sortedNodes
return e.populateNodeStates(context.TODO())
}
// PrimaryNodeID implements the Noder interface.
func (e *Etcd) PrimaryNodeID(hasher topology.Hasher) string {
return topology.PrimaryNodeID(e.NodeIDs(), hasher)
func (e *Etcd) PrimaryNodeID(hasher disco.Hasher) string {
return disco.PrimaryNodeID(e.NodeIDs(), hasher)
}
// NodeIDs returns the list of node IDs in the etcd cluster.

25
etcd/embed_test.go
View file

@ -6,7 +6,6 @@ import (
"testing"
"time"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/logger"
"go.etcd.io/etcd/server/v3/embed"
@ -15,10 +14,14 @@ import (
func TestRestartEtcd(t *testing.T) {
cfg := embed.NewConfig()
cfg.Dir = "default.etcd"
curl, _ := url.Parse(pilosa.EtcdUnixSocket(t))
cfg.LPUrls = append(cfg.LPUrls, *curl)
curl, _ = url.Parse(pilosa.EtcdUnixSocket(t))
cfg.LCUrls = append(cfg.LCUrls, *curl)
curl, _ := url.Parse(unixSocket(t))
// append-in-place to replace any existing URLs with our new URL
cfg.LPUrls = append(cfg.LPUrls[:0], *curl)
cfg.APUrls = cfg.LPUrls
cfg.InitialCluster = "default=" + cfg.LPUrls[0].String()
curl, _ = url.Parse(unixSocket(t))
cfg.LCUrls = append(cfg.LCUrls[:0], *curl)
cfg.ACUrls = cfg.LCUrls
e, err := embed.StartEtcd(cfg)
if err != nil {
t.Fatal(err)
@ -49,13 +52,19 @@ func TestRestartEtcd(t *testing.T) {
func TestParseOptions(t *testing.T) {
var e = &Etcd{options: Options{ClusterURL: "http://foo"}, logger: logger.NewLogfLogger(t)}
curl, _ := url.Parse(pilosa.EtcdUnixSocket(t))
curl, _ := url.Parse(unixSocket(t))
e.options.LClientURL = curl.String()
curl, _ = url.Parse(pilosa.EtcdUnixSocket(t))
e.options.LPeerURL = "i'm a teapot"
_, err := e.parseOptions()
if err == nil {
t.Fatalf("invalid peer URL should be rejected")
}
curl, _ = url.Parse(unixSocket(t))
e.options.LPeerURL = curl.String()
e.options.ClusterURL = "http://foo"
_, err := e.parseOptions()
_, err = e.parseOptions()
if err == nil {
t.Fatalf("cluster URL should be rejected")
}

126
etcd/fake_test.go Normal file
View file

@ -0,0 +1,126 @@
package etcd
import (
"context"
"fmt"
"testing"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/logger"
"golang.org/x/sync/errgroup"
)
// fakeCluster is a cluster of just our local etcd wrappers, without the
// rest of featurebase present. This code is expected to migrate into a
// _test.go file later, but for now it's here so we can see code coverage.
type fakeCluster struct {
tb testing.TB
nodes []*Etcd
}
func (f *fakeCluster) Start() error {
eg, ctx := errgroup.WithContext(context.Background())
for _, node := range f.nodes {
node := node
eg.Go(func() error {
_, err := node.Start(ctx)
return err
})
}
return eg.Wait()
}
func (f *fakeCluster) BringUp() error {
eg, ctx := errgroup.WithContext(context.Background())
for _, node := range f.nodes {
node := node
eg.Go(func() error {
return node.SetState(ctx, disco.NodeStateStarted)
})
}
return eg.Wait()
}
// AwaitClusterState verifies that node 0 thinks the cluster is in the
// requested state, or tells you why it failed.
func (f *fakeCluster) AwaitClusterState(expected disco.ClusterState) (err error) {
state := disco.ClusterState("")
for state != expected {
state, err = f.nodes[0].ClusterState(context.Background())
if err != nil {
return err
}
}
return nil
}
// MustAwaitClusterState verifies that node 0 thinks the cluster is in the
// requested state, or fails a test.
func (f *fakeCluster) MustAwaitClusterState(expected disco.ClusterState) {
err := f.AwaitClusterState(expected)
if err != nil {
f.tb.Fatalf("awaiting cluster state %s: %v", expected, err)
}
}
func (f *fakeCluster) Stop() error {
for i, node := range f.nodes {
err := node.Close()
if err != nil {
return fmt.Errorf("failure closing node %d: %v", i, err)
}
}
return nil
}
func (f *fakeCluster) MustNodeStates() []disco.NodeState {
nodes := f.nodes[0].Nodes()
states := make([]disco.NodeState, len(nodes))
for i, node := range nodes {
states[i] = node.State
}
return states
}
func (f *fakeCluster) MustClusterState() disco.ClusterState {
state, err := f.nodes[0].ClusterState(context.TODO())
if err != nil {
f.tb.Fatalf("getting cluster state: %v", err)
}
return state
}
// Elect tries to force a leader election by identifying a leader
// and then stopping it. The cluster, if it has at least 3 members,
// should still stay running.
func (f *fakeCluster) Elect() (oldleader *Etcd, err error) {
var leader *Etcd
for i, node := range f.nodes {
if node.IsLeader() {
f.tb.Logf("leader is node %d, stopping it", i)
leader = node
copy(f.nodes[i:], f.nodes[i+1:])
f.nodes = f.nodes[:len(f.nodes)-1]
}
}
// close the leader
err = leader.Close()
if err != nil {
return leader, err
}
_, err = f.nodes[0].ClusterState(context.TODO())
return leader, err
}
// NewFakeCluster creates a fakeCluster of n nodes, providing only the
// etcd objects, not the rest of a featurebase install.
func NewFakeCluster(tb testing.TB, n int, replicas int) *fakeCluster {
logger := logger.NewLogfLogger(tb)
_, opts := GenEtcdConfigs(tb, n)
fc := &fakeCluster{nodes: make([]*Etcd, n)}
for i := range fc.nodes {
fc.nodes[i] = NewEtcd(opts[i], logger, replicas, "foo")
}
fc.tb = tb
return fc
}

37
etcd/leasedkv.go
View file

@ -17,10 +17,11 @@ import (
// It will try to renew the lease at any cost after losing it.
// It will recreate the previous existing value for the key again.
type leasedKV struct {
e *Etcd
cancel context.CancelFunc
done <-chan struct{}
leaseID clientv3.LeaseID
e *Etcd
parentContext context.Context
cancel context.CancelFunc
done <-chan struct{}
leaseID clientv3.LeaseID
key string
ttlSeconds int64
@ -30,11 +31,12 @@ type leasedKV struct {
stopped bool // protected by mu
}
func newLeasedKV(e *Etcd, key string, ttlSeconds int64) *leasedKV {
func newLeasedKV(e *Etcd, ctx context.Context, key string, ttlSeconds int64) *leasedKV {
return &leasedKV{
e: e,
key: key,
ttlSeconds: ttlSeconds,
e: e,
parentContext: ctx,
key: key,
ttlSeconds: ttlSeconds,
}
}
@ -53,8 +55,12 @@ func (l *leasedKV) Start(initValue string) error {
return nil
}
// create creates the lease and yields a KeepAlive channel. It also stashes a cancel
// function for our local context that we use in case of internal issues, and the
// done channel for the internal context, which will be readable as soon as either
// our context or the parent context is done.
func (l *leasedKV) create(initValue string) (<-chan *clientv3.LeaseKeepAliveResponse, error) {
ctx, cancel := context.WithCancel(context.Background())
ctx, cancel := context.WithCancel(l.parentContext)
if l.cancel != nil {
l.cancel()
@ -99,8 +105,6 @@ func (l *leasedKV) create(initValue string) (<-chan *clientv3.LeaseKeepAliveResp
func (l *leasedKV) consumeLease(ch <-chan *clientv3.LeaseKeepAliveResponse) {
for {
select {
case <-l.e.closeWatch:
return
case _, ok := <-ch:
if ok {
continue
@ -146,7 +150,9 @@ func (l *leasedKV) Stop() {
l.cancel()
}
// low-effort attempt to cancel existing lease. if the cluster is
// shutting down, we don't want this to take long.
// shutting down, we don't want this to take long. Note that we don't
// use the parent context for this -- if we got cancelled, we still
// want this attempt to run.
ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
err := l.e.retryClient(func(cli *clientv3.Client) (err error) {
_, err = cli.Revoke(ctx, l.leaseID)
@ -213,6 +219,13 @@ func (l *leasedKV) Get(ctx context.Context) (string, error) {
return l.value, nil
}
// retry retries a function at a given interval until it succeeds, or until it
// returns context.DeadlineExceeded, at which point we return the last other
// error it returned, or DeadlineExceeded if we didn't have another previous
// error. So other errors (connection failures, etcetera) get retried, but
// DeadlineExceeded means we're done trying. But, if we failed due to a
// connection error, then got a DeadlineExceeded on a retry, we want to report
// the connection error, which is a lot more informative.
func retry(desc string, sleep time.Duration, f func() error) (err error) {
for {
lastErr := f()

57
etcd/leasedkv_test.go
View file

@ -3,13 +3,10 @@ package etcd
import (
"context"
"fmt"
"net"
"os"
"testing"
"time"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/testhook"
@ -23,11 +20,46 @@ import (
const initVal = "test"
const newVal = "newValue"
func TestClusterKv(t *testing.T) {
if !AllowCluster() {
t.Skip("only testing clusters when clustering is allowed")
}
c := NewFakeCluster(t, 3, 2)
err := c.Start()
if err != nil {
t.Fatalf("starting cluster: %v", err)
}
c.MustAwaitClusterState(disco.ClusterStateDown)
err = c.BringUp()
if err != nil {
t.Fatalf("bringing up cluster: %v", err)
}
c.MustAwaitClusterState(disco.ClusterStateNormal)
_, err = c.Elect()
if err != nil {
t.Fatalf("trying to cause election: %v", err)
}
ctx := context.TODO()
c.nodes[0].SetState(ctx, disco.NodeStateStarting)
c.nodes[1].SetState(ctx, disco.NodeStateStarting)
c.MustAwaitClusterState(disco.ClusterStateStarting)
c.nodes[0].SetState(ctx, disco.NodeStateStarted)
c.nodes[1].SetState(ctx, disco.NodeStateStarted)
// Two of three nodes are up, one is down, we have 2 replicas, so
// we should be able to handle reads but not writes, so we're in
// a Degraded state.
c.MustAwaitClusterState(disco.ClusterStateDegraded)
err = c.Stop()
if err != nil {
t.Fatalf("stopping cluster: %v", err)
}
}
func TestLeasedKv(t *testing.T) {
cfg := embed.NewConfig()
clientURL := pilosa.EtcdUnixSocket(t)
peerURL := pilosa.EtcdUnixSocket(t)
clientURL := unixSocket(t)
peerURL := unixSocket(t)
cfg.LPUrls = types.MustNewURLs([]string{peerURL})
cfg.APUrls = types.MustNewURLs([]string{peerURL})
cfg.LCUrls = types.MustNewURLs([]string{clientURL})
@ -53,7 +85,7 @@ func TestLeasedKv(t *testing.T) {
}()
wrapper := &Etcd{e: etcd, cli: cli, logger: logger.NewLogfLogger(t)}
lkv := newLeasedKV(wrapper, "/test", 1)
lkv := newLeasedKV(wrapper, context.TODO(), "/test", 1)
ctx := context.Background()
@ -105,16 +137,3 @@ func TestLeasedKv(t *testing.T) {
t.Fatal("expected error:", disco.ErrNoResults, "obtained:", err)
}
}
// listenerWithURL builds a TCP listener and corresponding http://localhost:%d
// URL, and returns those.
func listenerWithURL() (listener *net.TCPListener, url string, err error) {
l, err := net.Listen("tcp", "localhost:0")
if err != nil {
return listener, url, err
}
listener = l.(*net.TCPListener)
port := listener.Addr().(*net.TCPAddr).Port
url = fmt.Sprintf("http://localhost:%d", port)
return listener, url, err
}

4
event.go
View file

@ -1,7 +1,7 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package pilosa
import "github.com/molecula/featurebase/v3/topology"
import "github.com/molecula/featurebase/v3/disco"
// NodeEventType are the types of node events.
type NodeEventType int
@ -16,5 +16,5 @@ const (
// NodeEvent is a single event related to node activity in the cluster.
type NodeEvent struct {
Event NodeEventType
Node *topology.Node
Node *disco.Node
}

27
executor.go
View file

@ -24,7 +24,6 @@ import (
"github.com/molecula/featurebase/v3/shardwidth"
"github.com/molecula/featurebase/v3/task"
"github.com/molecula/featurebase/v3/testhook"
"github.com/molecula/featurebase/v3/topology"
"github.com/molecula/featurebase/v3/tracing"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
@ -49,7 +48,7 @@ type executor struct {
Holder *Holder
// Local hostname & cluster configuration.
Node *topology.Node
Node *disco.Node
Cluster *cluster
// Client used for remote requests.
@ -5921,7 +5920,7 @@ func (e *executor) executeClearValueField(ctx context.Context, qcx *Qcx, index s
}
// remoteExec executes a PQL query remotely for a set of shards on a node.
func (e *executor) remoteExec(ctx context.Context, node *topology.Node, index string, q *pql.Query, shards []uint64, embed []*Row, maxMemory int64) (results []interface{}, err error) { // nolint: interfacer
func (e *executor) remoteExec(ctx context.Context, node *disco.Node, index string, q *pql.Query, shards []uint64, embed []*Row, maxMemory int64) (results []interface{}, err error) { // nolint: interfacer
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeExec")
defer span.Finish()
@ -5944,14 +5943,14 @@ func (e *executor) remoteExec(ctx context.Context, node *topology.Node, index st
// shardsByNode returns a mapping of nodes to shards.
// Returns errShardUnavailable if a shard cannot be allocated to a node.
func (e *executor) shardsByNode(nodes []*topology.Node, index string, shards []uint64) (map[*topology.Node][]uint64, error) {
m := make(map[*topology.Node][]uint64)
func (e *executor) shardsByNode(nodes []*disco.Node, index string, shards []uint64) (map[*disco.Node][]uint64, error) {
m := make(map[*disco.Node][]uint64)
// Create a snapshot of the cluster to use for node/partition calculations.
// We use e.Cluster.Nodes() here instead of e.Cluster.noder because we need
// the node states in order to ensure that we don't include an unavailable
// node in the map of nodes to which we distribute the query.
snap := topology.NewClusterSnapshot(topology.NewLocalNoder(e.Cluster.Nodes()), e.Cluster.Hasher, e.Cluster.partitionAssigner, e.Cluster.ReplicaN)
snap := disco.NewClusterSnapshot(disco.NewLocalNoder(e.Cluster.Nodes()), e.Cluster.Hasher, e.Cluster.partitionAssigner, e.Cluster.ReplicaN)
loop:
for _, shard := range shards {
@ -5959,7 +5958,7 @@ loop:
// If the node being considered is in any state other than STARTED,
// then exclude it from the map. This way, one of that node's
// healthy replicas will be included instead.
if topology.Nodes(nodes).ContainsID(node.ID) && (node.State == disco.NodeStateStarted || node.State == disco.NodeStateUnknown) {
if disco.Nodes(nodes).ContainsID(node.ID) && (node.State == disco.NodeStateStarted || node.State == disco.NodeStateUnknown) {
m[node] = append(m[node], shard)
continue loop
}
@ -6003,11 +6002,11 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
//
// However, if this request is being sent from the primary then all
// processing should be done locally so we start with just the local node.
var nodes []*topology.Node
var nodes []*disco.Node
if !opt.Remote {
nodes = topology.Nodes(e.Cluster.Nodes()).Clone()
nodes = disco.Nodes(e.Cluster.Nodes()).Clone()
} else {
nodes = []*topology.Node{e.Cluster.nodeByID(e.Node.ID)}
nodes = []*disco.Node{e.Cluster.nodeByID(e.Node.ID)}
}
// Start mapping across all primary owners.
@ -6033,7 +6032,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
// the error from the healthy node and return that immediately.
if resp.err != nil && strings.Contains(resp.err.Error(), errConnectionRefused) {
// Filter out unavailable nodes.
nodes = topology.Nodes(nodes).FilterID(resp.node.ID)
nodes = disco.Nodes(nodes).FilterID(resp.node.ID)
// Begin mapper against secondary nodes.
if err := e.mapper(ctx, eg, ch, nodes, index, resp.shards, c, opt, true, mapFn, reduceFn); errors.Cause(err) == errShardUnavailable {
@ -6107,7 +6106,7 @@ func makeEmbeddedDataForShards(allRows []*Row, shards []uint64) []*Row {
return newRows
}
func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapResponse, nodes []*topology.Node, index string, shards []uint64, c *pql.Call, opt *execOptions, lastAttempt bool, mapFn mapFunc, reduceFn reduceFunc) (reterr error) {
func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapResponse, nodes []*disco.Node, index string, shards []uint64, c *pql.Call, opt *execOptions, lastAttempt bool, mapFn mapFunc, reduceFn reduceFunc) (reterr error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapper")
defer span.Finish()
@ -7702,7 +7701,7 @@ type mapOptions struct {
type reduceFunc func(ctx context.Context, prev, v interface{}) interface{}
type mapResponse struct {
node *topology.Node
node *disco.Node
shards []uint64
result interface{}
@ -8774,7 +8773,7 @@ func (c *NopCommitor) Commit() error {
func deleteKeyTranslation(ctx context.Context, idx *Index, shard uint64, records *roaring.Bitmap) (Commitor, error) {
// ShardToShardParition ...
paritionID := topology.ShardToShardPartition(idx.name, shard, idx.holder.partitionN)
paritionID := disco.ShardToShardPartition(idx.name, shard, idx.holder.partitionN)
return idx.TranslateStore(paritionID).Delete(records)
}

4
fragment.go
View file

@ -23,6 +23,7 @@ import (
"github.com/cespare/xxhash"
"github.com/gogo/protobuf/proto"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/logger"
pnet "github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/pb"
@ -31,7 +32,6 @@ import (
"github.com/molecula/featurebase/v3/shardwidth"
"github.com/molecula/featurebase/v3/stats"
"github.com/molecula/featurebase/v3/testhook"
"github.com/molecula/featurebase/v3/topology"
"github.com/molecula/featurebase/v3/tracing"
"github.com/molecula/featurebase/v3/vprint"
"github.com/pkg/errors"
@ -3162,7 +3162,7 @@ func (h *blockHasher) WriteValue(v uint64) {
type fragmentSyncer struct {
Fragment *fragment
Node *topology.Node
Node *disco.Node
Cluster *cluster
// FieldType helps determine which method of syncing to use.

21
holder.go
View file

@ -19,7 +19,6 @@ import (
"github.com/molecula/featurebase/v3/stats"
"github.com/molecula/featurebase/v3/storage"
"github.com/molecula/featurebase/v3/testhook"
"github.com/molecula/featurebase/v3/topology"
"github.com/molecula/featurebase/v3/vprint"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
@ -211,7 +210,7 @@ type HolderConfig struct {
func DefaultHolderConfig() *HolderConfig {
return &HolderConfig{
PartitionN: topology.DefaultPartitionN,
PartitionN: disco.DefaultPartitionN,
OpenTranslateStore: OpenInMemTranslateStore,
OpenTranslateReader: nil,
OpenTransactionStore: OpenInMemTransactionStore,
@ -1197,7 +1196,7 @@ type holderSyncer struct {
Holder *Holder
Node *topology.Node
Node *disco.Node
Cluster *cluster
// Translation sync handling.
@ -1447,7 +1446,7 @@ func (s *holderSyncer) stopTranslationSync() error {
// setTranslateReadOnlyFlags updates all translation stores to enable or disable
// writing new translation keys. Index stores are writable if the node owns the
// partition. Field stores are writable if the node is the primary.
func (s *holderSyncer) setTranslateReadOnlyFlags(snap *topology.ClusterSnapshot) {
func (s *holderSyncer) setTranslateReadOnlyFlags(snap *disco.ClusterSnapshot) {
s.Cluster.mu.RLock()
isPrimaryFieldTranslator := snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID)
@ -1490,7 +1489,7 @@ func (s *holderSyncer) setTranslateReadOnlyFlags(snap *topology.ClusterSnapshot)
// any key translation, whether that's field keys (every node replicates these
// from the primary) or index keys (only the replica nodes for each partition
// replicate these from whichever node is primary for that partition).
func (s *holderSyncer) initializeReplication(snap *topology.ClusterSnapshot) error {
func (s *holderSyncer) initializeReplication(snap *disco.ClusterSnapshot) error {
nodeMaps := make(map[string]TranslateOffsetMap)
if snap.ReplicaN > 1 {
if err := s.populateIndexReplication(nodeMaps, snap); err != nil {
@ -1502,7 +1501,7 @@ func (s *holderSyncer) initializeReplication(snap *topology.ClusterSnapshot) err
}
// filter out empty nodes
nodes := make(map[*topology.Node]bool)
nodes := make(map[*disco.Node]bool)
for _, node := range snap.Nodes {
m := nodeMaps[node.ID]
if !m.Empty() {
@ -1599,7 +1598,7 @@ func (s *holderSyncer) initializeReplication(snap *topology.ClusterSnapshot) err
// populateFieldReplication populates a map from node IDs to TranslateOffsetMaps
// to record that we need to translate fields which have key translation
// from the primary node.
func (s *holderSyncer) populateFieldReplication(nodeMaps map[string]TranslateOffsetMap, snap *topology.ClusterSnapshot) error {
func (s *holderSyncer) populateFieldReplication(nodeMaps map[string]TranslateOffsetMap, snap *disco.ClusterSnapshot) error {
// Set up field translation
if !snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID) {
primaryID := snap.PrimaryFieldTranslationNode().ID
@ -1633,7 +1632,7 @@ func (s *holderSyncer) populateFieldReplication(nodeMaps map[string]TranslateOff
// to record which nodes we need to replicate index key translation for.
// That means nodes which are the primary for a partition that we're a
// non-primary replica for.
func (s *holderSyncer) populateIndexReplication(nodeMaps map[string]TranslateOffsetMap, snap *topology.ClusterSnapshot) error {
func (s *holderSyncer) populateIndexReplication(nodeMaps map[string]TranslateOffsetMap, snap *disco.ClusterSnapshot) error {
for _, node := range snap.Nodes {
if node.ID == s.Node.ID {
continue
@ -1647,8 +1646,8 @@ func (s *holderSyncer) populateIndexReplication(nodeMaps map[string]TranslateOff
}
for partitionID := 0; partitionID < snap.PartitionN; partitionID++ {
partitionNodes := snap.PartitionNodes(partitionID)
isPrimary := partitionNodes[0].ID == node.ID // remote is primary?
isReplica := topology.Nodes(partitionNodes[1:]).ContainsID(s.Node.ID) // local is replica?
isPrimary := partitionNodes[0].ID == node.ID // remote is primary?
isReplica := disco.Nodes(partitionNodes[1:]).ContainsID(s.Node.ID) // local is replica?
if !isPrimary || !isReplica {
continue
}
@ -1677,7 +1676,7 @@ func (s *holderSyncer) populateIndexReplication(nodeMaps map[string]TranslateOff
// readBothTranslateReader reads key translation for field keys or
// index keys from a remote node. Both field and index keys may be sent,
// the distinction is that field keys have a non-empty field name.
func (s *holderSyncer) readBothTranslateReader(rd TranslateEntryReader, snap *topology.ClusterSnapshot) {
func (s *holderSyncer) readBothTranslateReader(rd TranslateEntryReader, snap *disco.ClusterSnapshot) {
for {
var entry TranslateEntry
if err := rd.ReadEntry(&entry); err != nil {

10
http_handler.go
View file

@ -31,13 +31,13 @@ import (
"github.com/gorilla/mux"
"github.com/molecula/featurebase/v3/authn"
"github.com/molecula/featurebase/v3/authz"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/ingest"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/monitor"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/rbf"
"github.com/molecula/featurebase/v3/storage"
"github.com/molecula/featurebase/v3/topology"
"github.com/molecula/featurebase/v3/tracing"
"github.com/pkg/errors"
"github.com/prometheus/client_golang/prometheus/promhttp"
@ -1227,10 +1227,10 @@ type getSchemaResponse struct {
}
type getStatusResponse struct {
State string `json:"state"`
Nodes []*topology.Node `json:"nodes"`
LocalID string `json:"localID"`
ClusterName string `json:"clusterName"`
State string `json:"state"`
Nodes []*disco.Node `json:"nodes"`
LocalID string `json:"localID"`
ClusterName string `json:"clusterName"`
}
func httpHash(s string) string {

13
internal/clustertests/pause_node_test.go
View file

@ -20,7 +20,6 @@ import (
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
)
@ -136,7 +135,7 @@ func openTranslateStores(dirPath, index string) (map[int]pilosa.TranslateStore,
}
// in case of error, close any translateStore that has been opened
rollback := make([]pilosa.TranslateStore, 0, topology.DefaultPartitionN)
rollback := make([]pilosa.TranslateStore, 0, disco.DefaultPartitionN)
defer func() {
for _, ts := range rollback {
_ = ts.Close()
@ -162,7 +161,7 @@ func openTranslateStores(dirPath, index string) (map[int]pilosa.TranslateStore,
return nil, err
}
// open bolt db
ts, err := boltdb.OpenTranslateStore(filePath, index, "", partition, topology.DefaultPartitionN, false)
ts, err := boltdb.OpenTranslateStore(filePath, index, "", partition, disco.DefaultPartitionN, false)
ts.SetReadOnly(true)
if err != nil {
return nil, err
@ -193,7 +192,7 @@ func verifyNodeHasGivenKeys(ctx context.Context, node, index, dirPath string, ke
}
// read in all the translate stores for each partition
for partition := 0; partition < topology.DefaultPartitionN; partition++ {
for partition := 0; partition < disco.DefaultPartitionN; partition++ {
err := readIndexTranslateData(ctx, client, nodeDirPath, index, partition)
if err != nil {
return err
@ -325,8 +324,8 @@ func TestPauseReplica(t *testing.T) {
}
// generate mapping from partition to primary node
partitionToNode := make([]string, topology.DefaultPartitionN)
for partition := 0; partition < topology.DefaultPartitionN; partition++ {
partitionToNode := make([]string, disco.DefaultPartitionN)
for partition := 0; partition < disco.DefaultPartitionN; partition++ {
nodes, err := cli.PartitionNodes(ctx, partition)
if err != nil {
t.Fatal(err)
@ -345,7 +344,7 @@ func TestPauseReplica(t *testing.T) {
h := fnv.New64a()
_, _ = h.Write([]byte(index))
_, _ = h.Write([]byte(key))
partition := int(h.Sum64() % uint64(topology.DefaultPartitionN))
partition := int(h.Sum64() % uint64(disco.DefaultPartitionN))
// get node for this partition
return partitionToNode[partition]
}

34
internal_client.go
View file

@ -21,10 +21,10 @@ import (
"github.com/hashicorp/go-retryablehttp"
"github.com/molecula/featurebase/v3/authn"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/ingest"
"github.com/molecula/featurebase/v3/logger"
pnet "github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/topology"
"github.com/molecula/featurebase/v3/tracing"
"github.com/pkg/errors"
"golang.org/x/oauth2"
@ -541,7 +541,7 @@ func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt Inde
}
// FragmentNodes returns a list of nodes that own a shard.
func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*topology.Node, error) {
func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*disco.Node, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FragmentNodes")
defer span.Finish()
@ -566,7 +566,7 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard
}
defer resp.Body.Close()
var a []*topology.Node
var a []*disco.Node
if err := json.NewDecoder(resp.Body).Decode(&a); err != nil {
return nil, fmt.Errorf("json decode: %s", err)
}
@ -574,7 +574,7 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard
}
// Nodes returns a list of all nodes.
func (c *InternalClient) Nodes(ctx context.Context) ([]*topology.Node, error) {
func (c *InternalClient) Nodes(ctx context.Context) ([]*disco.Node, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Nodes")
defer span.Finish()
@ -598,7 +598,7 @@ func (c *InternalClient) Nodes(ctx context.Context) ([]*topology.Node, error) {
}
defer resp.Body.Close()
var a []*topology.Node
var a []*disco.Node
if err := json.NewDecoder(resp.Body).Decode(&a); err != nil {
return nil, fmt.Errorf("json decode: %s", err)
}
@ -665,7 +665,7 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pnet.URI, index str
return qresp, nil
}
func getPrimaryNode(nodes []*topology.Node) *topology.Node {
func getPrimaryNode(nodes []*disco.Node) *disco.Node {
for _, node := range nodes {
if node.IsPrimary {
return node
@ -702,7 +702,7 @@ func (c *InternalClient) EnsureFieldWithOptions(ctx context.Context, indexName s
}
// importNode sends a pre-marshaled import request to a node.
func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, index, field string, buf []byte, opts *ImportOptions) error {
func (c *InternalClient) importNode(ctx context.Context, node *disco.Node, index, field string, buf []byte, opts *ImportOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.importNode")
defer span.Finish()
@ -765,7 +765,7 @@ func (c *InternalClient) importHelper(ctx context.Context, req Message, process
// If we don't actually know what shards we're sending to, and we have
// a local API and a qcx, we'll have a process function that uses the local
// API. Otherwise, even if we have an API
var nodes []*topology.Node
var nodes []*disco.Node
var err error
if shard != ^uint64(0) {
// we need a list of nodes specific to this shard.
@ -795,8 +795,8 @@ func (c *InternalClient) importHelper(ctx context.Context, req Message, process
// "us" is a usable local node if any, "them" is every node that we need
// to process which isn't that node. We start out with us == nil and
// them = the whole set of nodes.
var us *topology.Node
var them []*topology.Node = nodes
var us *disco.Node
var them []*disco.Node = nodes
// If we have an API, we know what node we are. Even if we don't have
// a Qcx, we still care, because looping back to the local node will
@ -1005,7 +1005,7 @@ func (c *InternalClient) ExportCSV(ctx context.Context, index, field string, sha
}
// exportNode copies a CSV export from a node to w.
func (c *InternalClient) exportNodeCSV(ctx context.Context, node *topology.Node, index, field string, shard uint64, w io.Writer) error {
func (c *InternalClient) exportNodeCSV(ctx context.Context, node *disco.Node, index, field string, shard uint64, w io.Writer) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.exportNodeCSV")
defer span.Finish()
@ -1048,7 +1048,7 @@ func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field,
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RetrieveShardFromURI")
defer span.Finish()
node := &topology.Node{
node := &disco.Node{
URI: uri,
}
@ -2141,7 +2141,7 @@ func uriPathToURL(uri *pnet.URI, path string) url.URL {
}
}
func nodePathToURL(node *topology.Node, path string) url.URL {
func nodePathToURL(node *disco.Node, path string) url.URL {
return url.URL{
Scheme: node.URI.Scheme,
Host: node.URI.HostPort(),
@ -2156,7 +2156,7 @@ func (c *InternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context,
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RetrieveTranslatePartitionFromURI")
defer span.Finish()
node := &topology.Node{
node := &disco.Node{
URI: uri,
}
@ -2303,7 +2303,7 @@ func (c *InternalClient) IDAllocDataReader(ctx context.Context) (io.ReadCloser,
return resp.Body, nil
}
func (c *InternalClient) IDAllocDataWriter(ctx context.Context, f io.Reader, primary *topology.Node) error {
func (c *InternalClient) IDAllocDataWriter(ctx context.Context, f io.Reader, primary *disco.Node) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.IDAllocDataWriter")
defer span.Finish()
@ -2419,7 +2419,7 @@ func (c *InternalClient) Status(ctx context.Context) (string, error) {
return rsp.State, nil
}
func (c *InternalClient) PartitionNodes(ctx context.Context, partitionID int) ([]*topology.Node, error) {
func (c *InternalClient) PartitionNodes(ctx context.Context, partitionID int) ([]*disco.Node, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.PartitionNodes")
defer span.Finish()
@ -2444,7 +2444,7 @@ func (c *InternalClient) PartitionNodes(ctx context.Context, partitionID int) ([
}
defer resp.Body.Close()
var a []*topology.Node
var a []*disco.Node
if err := json.NewDecoder(resp.Body).Decode(&a); err != nil {
return nil, fmt.Errorf("json decode: %s", err)
}

6
internal_client_test.go
View file

@ -16,11 +16,11 @@ import (
"github.com/davecgh/go-spew/spew"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/authn"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/test"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
)
@ -285,7 +285,7 @@ func TestClient_Export(t *testing.T) {
bw := bufio.NewWriter(buf)
// Send export request for every partition.
for i := 0; i < topology.DefaultPartitionN; i++ {
for i := 0; i < disco.DefaultPartitionN; i++ {
if err := c.ExportCSV(context.Background(), "keyed", "unkeyedf", uint64(i), bw); err != nil {
t.Fatal(err)
}
@ -326,7 +326,7 @@ func TestClient_Export(t *testing.T) {
bw := bufio.NewWriter(buf)
// Send export request.
for i := 0; i < topology.DefaultPartitionN; i++ {
for i := 0; i < disco.DefaultPartitionN; i++ {
if err := c.ExportCSV(context.Background(), "keyed", "keyedf", uint64(i), bw); err != nil {
t.Fatal(err)
}

36
server.go
View file

@ -3,7 +3,6 @@ package pilosa
import (
"context"
"encoding/json"
"fmt"
"log"
"os"
@ -25,7 +24,6 @@ import (
"github.com/molecula/featurebase/v3/sql2"
"github.com/molecula/featurebase/v3/stats"
"github.com/molecula/featurebase/v3/storage"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
@ -57,9 +55,7 @@ type Server struct { // nolint: maligned
// Distributed Consensus
disCo disco.DisCo
stator disco.Stator
metadator disco.Metadator
noder topology.Noder
noder disco.Noder
sharder disco.Sharder
schemator disco.Schemator
@ -322,7 +318,7 @@ func OptServerNodeID(nodeID string) ServerOption {
// OptServerClusterHasher is a functional option on Server
// used to specify the consistent hash algorithm for data
// location within the cluster.
func OptServerClusterHasher(h topology.Hasher) ServerOption {
func OptServerClusterHasher(h disco.Hasher) ServerOption {
return func(s *Server) error {
s.cluster.Hasher = h
return nil
@ -399,16 +395,12 @@ func OptServerMaxQueryMemory(v int64) ServerOption {
// OptServerDisCo is a functional option on Server
// used to set the Distributed Consensus implementation.
func OptServerDisCo(disCo disco.DisCo,
stator disco.Stator,
metadator disco.Metadator,
noder topology.Noder,
noder disco.Noder,
sharder disco.Sharder,
schemator disco.Schemator) ServerOption {
return func(s *Server) error {
s.disCo = disCo
s.stator = stator
s.metadator = metadator
s.noder = noder
s.sharder = sharder
s.schemator = schemator
@ -450,9 +442,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
viewsRemovalInterval: time.Hour,
disCo: disco.NopDisCo,
stator: disco.NopStator,
metadator: disco.NopMetadator,
noder: topology.NewEmptyLocalNoder(),
noder: disco.NewEmptyLocalNoder(),
sharder: disco.NopSharder,
schemator: disco.NopSchemator,
serializer: NopSerializer,
@ -520,7 +510,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.cluster.logger = s.logger
s.cluster.holder = s.holder
s.cluster.disCo = s.disCo
s.cluster.stator = s.stator
s.cluster.noder = s.noder
s.cluster.sharder = s.sharder
@ -600,7 +589,7 @@ func (s *Server) Open() error {
// Set node ID.
s.nodeID = s.disCo.ID()
node := &topology.Node{
node := &disco.Node{
ID: s.nodeID,
URI: s.uri,
GRPCURI: s.grpcURI,
@ -608,12 +597,7 @@ func (s *Server) Open() error {
IsPrimary: s.IsPrimary(),
}
// Set metadata for this node.
data, err := json.Marshal(node)
if err != nil {
return errors.Wrap(err, "marshaling json metadata")
}
if err := s.metadator.SetMetadata(context.Background(), data); err != nil {
if err := s.noder.SetMetadata(context.Background(), node); err != nil {
return errors.Wrap(err, "setting metadata")
}
@ -649,7 +633,7 @@ func (s *Server) Open() error {
// bring up the background tasks for the holder.
s.holder.Activate()
if err := s.stator.Started(context.Background()); err != nil {
if err := s.noder.SetState(context.Background(), disco.NodeStateStarted); err != nil {
return errors.Wrap(err, "setting nodeState")
}
@ -697,7 +681,7 @@ func (s *Server) Open() error {
<-timer.C
}
for {
state, err := s.stator.ClusterState(ctx)
state, err := s.noder.ClusterState(ctx)
if err != nil {
s.logger.Printf("failed to check cluster state: %v", err)
timer.Reset(time.Second)
@ -1126,7 +1110,7 @@ func (s *Server) SendAsync(m Message) error {
}
// SendTo represents an implementation of Broadcaster.
func (s *Server) SendTo(node *topology.Node, m Message) error {
func (s *Server) SendTo(node *disco.Node, m Message) error {
msg, err := s.serializer.Marshal(m)
if err != nil {
return fmt.Errorf("marshaling message: %v", err)
@ -1140,7 +1124,7 @@ func (s *Server) SendTo(node *topology.Node, m Message) error {
// node returns the pilosa.node object. It is used by membership protocols to
// get this node's name(ID), location(URI), and primary status.
func (s *Server) node() *topology.Node {
func (s *Server) node() *disco.Node {
return s.cluster.Node.Clone()
}

3
server/server.go
View file

@ -459,7 +459,6 @@ func (m *Command) SetupServer() error {
}
e := petcd.NewEtcd(m.Config.Etcd, m.logger, m.Config.Cluster.ReplicaN, version)
discoOpt := pilosa.OptServerDisCo(e, e, e, e, e, e)
serverOptions := []pilosa.ServerOption{
pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)),
@ -487,7 +486,7 @@ func (m *Command) SetupServer() error {
pilosa.OptServerMaxQueryMemory(m.Config.MaxQueryMemory),
pilosa.OptServerQueryHistoryLength(m.Config.QueryHistoryLength),
pilosa.OptServerPartitionAssigner(m.Config.Cluster.PartitionToNodeAssignment),
discoOpt,
pilosa.OptServerDisCo(e, e, e, e),
}
if m.Config.LookupDBDSN != "" {

102
test/disco.go
View file

@ -7,10 +7,8 @@ import (
"strings"
"testing"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/etcd"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/testhook"
"github.com/pkg/errors"
)
@ -51,22 +49,19 @@ func listenerWithURL() (listener *net.TCPListener, url string, err error) {
return listener, url, err
}
// GetPortsGenConfigs creates listener ports, and updates the configurations
// of servers to match these created ports, including cross-references
// like updating the InitCluster values in the Etcd configs.
// GetPortsGenConfigs generates etcd configs for a number
// of nodes, including cross-references so that every node gets
// an initial cluster list pointing it at the other nodes,
// and modifies the configs of the provided Command objects
// to point to these etcd configs. It uses etcd.GenEtcdConfigs,
// which in turn creates temporary directories and the like.
func GetPortsGenConfigs(tb testing.TB, nodes []*Command) error {
peerUrls := make([]string, len(nodes))
clusterName, cfgs := etcd.GenEtcdConfigs(tb, len(nodes))
for i := range nodes {
if nodes[i].Config == nil {
nodes[i].Config = &server.Config{}
}
config := nodes[i].Config
name := fmt.Sprintf("server%d", i)
clusterName := fmt.Sprintf("cluster-%s", tb.Name())
discoDir, err := testhook.TempDir(tb, "disco.")
if err != nil {
return errors.Wrap(err, "creating temp directory")
}
grpcListener, grpcUrl, err := listenerWithURL()
if err != nil {
return errors.Wrap(err, "creating gRPC listener")
@ -76,90 +71,11 @@ func GetPortsGenConfigs(tb testing.TB, nodes []*Command) error {
if colon != -1 {
grpcUrl = grpcUrl[colon:]
}
config.Name = name
config.Name = cfgs[i].Name
config.Cluster.Name = clusterName
config.BindGRPC = grpcUrl
config.GRPCListener = grpcListener
clientURL := pilosa.EtcdUnixSocket(tb)
peerURL := pilosa.EtcdUnixSocket(tb)
config.Etcd = etcd.Options{
Dir: discoDir,
LClientURL: clientURL,
AClientURL: clientURL,
LPeerURL: peerURL,
APeerURL: peerURL,
HeartbeatTTL: 60,
UnsafeNoFsync: true,
}
peerUrls[i] = fmt.Sprintf("%s=%s", name, peerURL)
}
allPeerUrls := strings.Join(peerUrls, ",")
for i := range nodes {
nodes[i].Config.Etcd.InitCluster = allPeerUrls
config.Etcd = cfgs[i]
}
return nil
}
//GenPortsConfig creates specific configuration for etcd.
func GenPortsConfig(tb testing.TB, ports []Ports) []*server.Config {
cfgs := make([]*server.Config, len(ports))
clusterURLs := make([]string, len(ports))
for i := range cfgs {
name := fmt.Sprintf("server%d", i)
clusterName := "cluster-abc123"
lClientURL := pilosa.EtcdUnixSocket(tb)
lPeerURL := pilosa.EtcdUnixSocket(tb)
discoDir := ""
if d, err := testhook.TempDir(tb, "disco."); err == nil {
discoDir = d
}
cfgs[i] = &server.Config{
Name: name,
BindGRPC: fmt.Sprintf(":%d", ports[i].Grpc),
GRPCListener: ports[i].LsnG,
Etcd: etcd.Options{
Dir: discoDir,
LClientURL: lClientURL,
AClientURL: lClientURL,
LPeerURL: lPeerURL,
APeerURL: lPeerURL,
HeartbeatTTL: 5,
UnsafeNoFsync: true,
},
}
cfgs[i].Cluster.Name = clusterName
clusterURLs[i] = fmt.Sprintf("%s=%s", name, lPeerURL)
}
for i := range cfgs {
cfgs[i].Etcd.InitCluster = strings.Join(clusterURLs, ",")
}
return cfgs
}
func NewPorts(lsn []*net.TCPListener) []Ports {
var out []Ports
n := len(lsn)
ports := make([]int, n)
for i := 0; i < n; i++ {
ports[i] = lsn[i].Addr().(*net.TCPAddr).Port
}
for i := 0; i < n; i = i + 3 {
out = append(out, Ports{
LsnC: lsn[i],
PortC: ports[i],
LsnP: lsn[i+1],
PortP: ports[i+1],
Grpc: ports[i+2],
LsnG: lsn[i+2],
})
}
return out
}

4
translate.go
View file

@ -10,9 +10,9 @@ import (
"sort"
"sync"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/ingest"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
)
@ -199,7 +199,7 @@ func GenerateNextPartitionedID(index string, prev uint64, partitionID, partition
// Try to use the next ID if it is in the same partition.
// Otherwise find ID in next shard that has a matching partition.
for id := prev + 1; ; id += ShardWidth {
if topology.ShardToShardPartition(index, id/ShardWidth, partitionN) == partitionID {
if disco.ShardToShardPartition(index, id/ShardWidth, partitionN) == partitionID {
return id
}
}

6
translator_test.go
View file

@ -13,16 +13,16 @@ import (
"github.com/google/go-cmp/cmp"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/mock"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/test"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
func TestInMemTranslateStore_TranslateID(t *testing.T) {
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, topology.DefaultPartitionN)
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, disco.DefaultPartitionN)
// Setup initial keys.
if _, err := s.CreateKeys("foo"); err != nil {
@ -272,7 +272,7 @@ func TestTranslation_KeyNotFound(t *testing.T) {
}
func TestInMemTranslateStore_ReadKey(t *testing.T) {
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, topology.DefaultPartitionN)
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, disco.DefaultPartitionN)
ids, err := s.FindKeys("foo")
if err != nil {

23
util.go
View file

@ -9,16 +9,11 @@ import (
"path/filepath"
"reflect"
"strings"
"sync"
"testing"
"time"
"github.com/shirou/gopsutil/v3/mem"
)
var clientPort = os.Getpid()
var muClientPort = sync.Mutex{}
// LeftShifted16MaxContainerKey is 0xffffffffffff0000. It is similar
// to the roaring.maxContainerKey 0x0000ffffffffffff, but
// shifted 16 bits to the left so its domain is the full [0, 2^64) bit space.
@ -107,24 +102,6 @@ func GetDiskUsage(path string) (DiskUsage, error) {
return DiskUsage{size}, err
}
// EtcdUnixSocket returns a url for use in test etcd clusters.
func EtcdUnixSocket(tb testing.TB) string {
muClientPort.Lock()
defer func() {
clientPort++
muClientPort.Unlock()
}()
addr := fmt.Sprintf("fake:%d", clientPort)
tb.Cleanup(func() {
err := os.Remove(addr)
if err != nil && !os.IsNotExist(err) { //not an error if the socket is not present
tb.Logf("could not remove '%s', %v", addr, err)
}
})
return fmt.Sprintf("unix://%s", addr)
}
// Rev reverses a string
func Rev(input string) string {
n := 0

8
utils_internal_test.go
View file

@ -6,10 +6,10 @@ import (
"testing"
"time"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/etcd"
pnet "github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/testhook"
"github.com/molecula/featurebase/v3/topology"
)
// utilities used by tests
@ -30,15 +30,15 @@ func NewTestCluster(tb testing.TB, n int) *cluster {
c.Hasher = NewTestModHasher()
c.Path = path
nodes := make([]*topology.Node, 0, n)
nodes := make([]*disco.Node, 0, n)
for i := 0; i < n; i++ {
nodes = append(nodes, &topology.Node{
nodes = append(nodes, &disco.Node{
ID: fmt.Sprintf("node%d", i),
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})
}
c.noder = topology.NewLocalNoder(nodes)
c.noder = disco.NewLocalNoder(nodes)
cNodes := c.noder.Nodes()