From b3a4e52a13185e769e720398a732bc5caf348805 Mon Sep 17 00:00:00 2001 From: Seebs Date: Thu, 23 Jun 2022 10:58:41 -0500 Subject: [PATCH] simplify, streamline, and possibly debug embedded etcd 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. --- api.go | 13 +- boltdb/translate_test.go | 4 +- broadcast.go | 6 +- cluster.go | 55 ++-- cluster_internal_test.go | 36 +-- ctl/backup.go | 6 +- ctl/restore.go | 12 +- disco/disco.go | 48 --- {topology => disco}/hasher.go | 2 +- {topology => disco}/node.go | 13 +- {topology => disco}/noder.go | 31 +- {topology => disco}/snapshot.go | 2 +- encoding/proto/proto.go | 21 +- etcd/config_gen.go | 71 +++++ etcd/embed.go | 356 +++++++++++------------ etcd/embed_test.go | 25 +- etcd/fake_test.go | 126 ++++++++ etcd/leasedkv.go | 37 ++- etcd/leasedkv_test.go | 57 ++-- event.go | 4 +- executor.go | 27 +- fragment.go | 4 +- holder.go | 21 +- http_handler.go | 10 +- internal/clustertests/pause_node_test.go | 13 +- internal_client.go | 34 +-- internal_client_test.go | 6 +- server.go | 36 +-- server/server.go | 3 +- test/disco.go | 102 +------ translate.go | 4 +- translator_test.go | 6 +- util.go | 23 -- utils_internal_test.go | 8 +- 34 files changed, 648 insertions(+), 574 deletions(-) rename {topology => disco}/hasher.go (98%) rename {topology => disco}/node.go (90%) rename {topology => disco}/noder.go (59%) rename {topology => disco}/snapshot.go (99%) create mode 100644 etcd/config_gen.go create mode 100644 etcd/fake_test.go diff --git a/api.go b/api.go index 66fdfb8f5..b78eb9760 100644 --- a/api.go +++ b/api.go @@ -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() diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go index cd74244eb..4c7a79139 100644 --- a/boltdb/translate_test.go +++ b/boltdb/translate_test.go @@ -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 } diff --git a/broadcast.go b/broadcast.go index ecab7b946..c88453b9c 100644 --- a/broadcast.go +++ b/broadcast.go @@ -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 ( diff --git a/cluster.go b/cluster.go index b37d3eca0..7c60d6916 100644 --- a/cluster.go +++ b/cluster.go @@ -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 } diff --git a/cluster_internal_test.go b/cluster_internal_test.go index ba70740bc..ccd53026e 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -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) diff --git a/ctl/backup.go b/ctl/backup.go index 6bdd7a83a..920ba9492 100644 --- a/ctl/backup.go +++ b/ctl/backup.go @@ -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++ { diff --git a/ctl/restore.go b/ctl/restore.go index 59b7ca73f..8791f7dac 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -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) diff --git a/disco/disco.go b/disco/disco.go index 585c7a43a..e653c8f02 100644 --- a/disco/disco.go +++ b/disco/disco.go @@ -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{} diff --git a/topology/hasher.go b/disco/hasher.go similarity index 98% rename from topology/hasher.go rename to disco/hasher.go index 4f718c6bb..59eb26ba9 100644 --- a/topology/hasher.go +++ b/disco/hasher.go @@ -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 { diff --git a/topology/node.go b/disco/node.go similarity index 90% rename from topology/node.go rename to disco/node.go index e5c33df5f..e52b00f4d 100644 --- a/topology/node.go +++ b/disco/node.go @@ -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 { diff --git a/topology/noder.go b/disco/noder.go similarity index 59% rename from topology/noder.go rename to disco/noder.go index b1c3bccd4..2fb8b6052 100644 --- a/topology/noder.go +++ b/disco/noder.go @@ -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 +} diff --git a/topology/snapshot.go b/disco/snapshot.go similarity index 99% rename from topology/snapshot.go rename to disco/snapshot.go index 313a11882..cb92e4fe6 100644 --- a/topology/snapshot.go +++ b/disco/snapshot.go @@ -1,5 +1,5 @@ // Copyright 2021 Molecula Corp. All rights reserved. -package topology +package disco import ( "encoding/binary" diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 8c165da1c..0f96c0457 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -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) diff --git a/etcd/config_gen.go b/etcd/config_gen.go new file mode 100644 index 000000000..6f2c01c03 --- /dev/null +++ b/etcd/config_gen.go @@ -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 +} diff --git a/etcd/embed.go b/etcd/embed.go index 7d5a0db11..6b5f5fd3d 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -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. diff --git a/etcd/embed_test.go b/etcd/embed_test.go index c737c62f9..ae2d5f8f2 100644 --- a/etcd/embed_test.go +++ b/etcd/embed_test.go @@ -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") } diff --git a/etcd/fake_test.go b/etcd/fake_test.go new file mode 100644 index 000000000..42600e4d4 --- /dev/null +++ b/etcd/fake_test.go @@ -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 +} diff --git a/etcd/leasedkv.go b/etcd/leasedkv.go index 70a592358..44cea686f 100644 --- a/etcd/leasedkv.go +++ b/etcd/leasedkv.go @@ -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() diff --git a/etcd/leasedkv_test.go b/etcd/leasedkv_test.go index 6a9ad947c..f80791064 100644 --- a/etcd/leasedkv_test.go +++ b/etcd/leasedkv_test.go @@ -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 -} diff --git a/event.go b/event.go index 811e6c7d9..5bebe82d7 100644 --- a/event.go +++ b/event.go @@ -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 } diff --git a/executor.go b/executor.go index 5ac0a38db..b3feb40aa 100644 --- a/executor.go +++ b/executor.go @@ -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) } diff --git a/fragment.go b/fragment.go index 97dcf1be6..cca6ad568 100644 --- a/fragment.go +++ b/fragment.go @@ -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. diff --git a/holder.go b/holder.go index bfa2fcb5d..f7dfe953f 100644 --- a/holder.go +++ b/holder.go @@ -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 { diff --git a/http_handler.go b/http_handler.go index 2f6af1288..a60ea5e25 100644 --- a/http_handler.go +++ b/http_handler.go @@ -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 { diff --git a/internal/clustertests/pause_node_test.go b/internal/clustertests/pause_node_test.go index f4b49078f..a88002fe1 100644 --- a/internal/clustertests/pause_node_test.go +++ b/internal/clustertests/pause_node_test.go @@ -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] } diff --git a/internal_client.go b/internal_client.go index dae937f9a..0056e4f31 100644 --- a/internal_client.go +++ b/internal_client.go @@ -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) } diff --git a/internal_client_test.go b/internal_client_test.go index c0f4b55c8..d768b465c 100644 --- a/internal_client_test.go +++ b/internal_client_test.go @@ -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) } diff --git a/server.go b/server.go index a187133ca..ce2ce13d7 100644 --- a/server.go +++ b/server.go @@ -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() } diff --git a/server/server.go b/server/server.go index c2972e272..52ae12d3b 100644 --- a/server/server.go +++ b/server/server.go @@ -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 != "" { diff --git a/test/disco.go b/test/disco.go index 325a626eb..285c5b4f7 100644 --- a/test/disco.go +++ b/test/disco.go @@ -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 -} diff --git a/translate.go b/translate.go index 9d5909f28..326562ba0 100644 --- a/translate.go +++ b/translate.go @@ -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 } } diff --git a/translator_test.go b/translator_test.go index 5df16b4a9..79bc55ca9 100644 --- a/translator_test.go +++ b/translator_test.go @@ -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 { diff --git a/util.go b/util.go index 35c5974b9..7a8bc52e9 100644 --- a/util.go +++ b/util.go @@ -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 diff --git a/utils_internal_test.go b/utils_internal_test.go index 797253dcf..a7dcf4044 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -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()