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

Remove most of the resize-related logic

Browse source 
We had two different, incompatible-with-each-other, and both
individually broken, partial implementations of resizing logic.
There's the original pre-etcd resize, and then the etcd resize,
and neither works, but there's conflicts between the ways they
don't work.

No attempt to fix this is likely to yield decent results, so
instead, we yank them both out entirely, so if we decide to
implement resizing (which we will) we won't be confused by
stray code pertaining to resizing that's not really hooked
up to anything.

We're leaving the resize messages in protobuf to avoid renumbering
protobuf messages. We rename some of our message types to UNUSED0,
etcetera, so that any code still using the old names won't
compile, to make sure we get rid of it, but we can't just drop
the numbers without breaking rolling restart.

The Resize_AddNode tests are removed not just because we don't
have resizing, but because they were completely broken anyway
and never worked at all. But there's no reason to fix them because
they exist to fix the functionality we didn't have and are now
removing the vestigial remains of.

We also drop the one usage of the AddNode function of Noder, because
it was used only by one test code fragment that was creatincg clusters,
and that can be done more correctly. There were no other call sites
at all.

We mark the monitorAntiEntropy function to be ignored by
code coverage because it's not actually being covered. There's
a separate ticket for removing that entirely.
This commit is contained in:
Seebs 2022-06-10 14:22:45 -05:00 • committed by seebs
parent c04fd9734e
commit fd9d4de31d
21 changed files with 112 additions and 2626 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
api.go
View file

@ -125,7 +125,6 @@ var validAPIMethods = map[disco.ClusterState]map[apiMethod]struct{}{
// has on a node under load, this is set to effectively allow all requests
// in a DEGRADED state.
disco.ClusterStateDegraded: appendMap(methodsCommon, methodsNormal),
disco.ClusterStateResizing: appendMap(methodsCommon, methodsResizing),
// Ideally, this would be just `methodsCommon`, but in an attempt to reduce
// the influence that state (determined by etcd) has on a node under load,
// this is set to effectively allow all requests in a DOWN state.
@ -2272,40 +2271,8 @@ func (api *API) indexField(indexName string, fieldName string, shard uint64) (*I
return index, field, nil
}
// RemoveNode puts the cluster into the "RESIZING" state and begins the job of
// removing the given node.
func (api *API) RemoveNode(id string) (*topology.Node, error) {
if err := api.validate(apiRemoveNode); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
if api.cluster.disCo.ID() == id {
return nil, errors.Wrapf(ErrPreconditionFailed, "cannot issue node removal request to the node being removed, id=%s", id)
}
removeNode := api.cluster.nodeByID(id)
if removeNode == nil {
return nil, errors.Wrap(ErrNodeIDNotExists, "finding node to remove")
}
if err := api.cluster.removeNode(id); err != nil {
return nil, errors.Wrapf(err, "removing node %s", id)
}
return removeNode, nil
}
// ResizeAbort stops the current resize job.
func (api *API) ResizeAbort() error {
if err := api.validate(apiResizeAbort); err != nil {
return errors.Wrap(err, "validating api method")
}
return api.cluster.resizeAbortAndBroadcast()
}
// State returns the cluster state which is usually "NORMAL", but could be
// "STARTING", "RESIZING", or potentially others. See disco.go for more
// "STARTING", or potentially others. See disco.go for more
// details.
func (api *API) State() (disco.ClusterState, error) {
if err := api.validate(apiState); err != nil {
@ -3257,8 +3224,6 @@ const (
//apiMaxShards // not implemented
apiQuery
apiRecalculateCaches
apiRemoveNode
apiResizeAbort
apiSchema
apiShardNodes
apiState
@ -3286,14 +3251,6 @@ var methodsCommon = map[apiMethod]struct{}{
apiState: {},
}
var methodsResizing = map[apiMethod]struct{}{
apiFragmentData: {},
apiTranslateData: {},
apiFieldTranslateData: {},
apiResizeAbort: {},
apiSchema: {},
}
// var methodsDegraded = map[apiMethod]struct{}{
// apiExportCSV: {},
// apiFragmentBlockData: {},
@ -3330,7 +3287,6 @@ var methodsNormal = map[apiMethod]struct{}{
apiIndex: {},
apiQuery: {},
apiRecalculateCaches: {},
apiRemoveNode: {},
apiShardNodes: {},
apiSchema: {},
apiViews: {},

42
apimethod_string.go
View file

@ -27,31 +27,29 @@ func _() {
_ = x[apiIndex-16]
_ = x[apiQuery-17]
_ = x[apiRecalculateCaches-18]
_ = x[apiRemoveNode-19]
_ = x[apiResizeAbort-20]
_ = x[apiSchema-21]
_ = x[apiShardNodes-22]
_ = x[apiState-23]
_ = x[apiViews-24]
_ = x[apiApplySchema-25]
_ = x[apiStartTransaction-26]
_ = x[apiFinishTransaction-27]
_ = x[apiTransactions-28]
_ = x[apiGetTransaction-29]
_ = x[apiActiveQueries-30]
_ = x[apiPastQueries-31]
_ = x[apiIDReserve-32]
_ = x[apiIDCommit-33]
_ = x[apiIDReset-34]
_ = x[apiPartitionNodes-35]
_ = x[apiIngestOperations-36]
_ = x[apiIngestNodeOperations-37]
_ = x[apiMutexCheck-38]
_ = x[apiSchema-19]
_ = x[apiShardNodes-20]
_ = x[apiState-21]
_ = x[apiViews-22]
_ = x[apiApplySchema-23]
_ = x[apiStartTransaction-24]
_ = x[apiFinishTransaction-25]
_ = x[apiTransactions-26]
_ = x[apiGetTransaction-27]
_ = x[apiActiveQueries-28]
_ = x[apiPastQueries-29]
_ = x[apiIDReserve-30]
_ = x[apiIDCommit-31]
_ = x[apiIDReset-32]
_ = x[apiPartitionNodes-33]
_ = x[apiIngestOperations-34]
_ = x[apiIngestNodeOperations-35]
_ = x[apiMutexCheck-36]
}
const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldTranslateDataapiFieldapiImportapiImportValueapiIndexapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSchemaapiShardNodesapiStateapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransactionapiActiveQueriesapiPastQueriesapiIDReserveapiIDCommitapiIDResetapiPartitionNodesapiIngestOperationsapiIngestNodeOperationsapiMutexCheck"
const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldTranslateDataapiFieldapiImportapiImportValueapiIndexapiQueryapiRecalculateCachesapiSchemaapiShardNodesapiStateapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransactionapiActiveQueriesapiPastQueriesapiIDReserveapiIDCommitapiIDResetapiPartitionNodesapiIngestOperationsapiIngestNodeOperationsapiMutexCheck"
var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 210, 218, 227, 241, 249, 257, 277, 290, 304, 313, 326, 334, 342, 356, 375, 395, 410, 427, 443, 457, 469, 480, 490, 507, 526, 549, 562}
var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 210, 218, 227, 241, 249, 257, 277, 286, 299, 307, 315, 329, 348, 368, 383, 400, 416, 430, 442, 453, 463, 480, 499, 522, 535}
func (i apiMethod) String() string {
if i < 0 || i >= apiMethod(len(_apiMethod_index)-1) {

24
broadcast.go
View file

@ -60,16 +60,16 @@ const (
messageTypeCreateView
messageTypeDeleteView
messageTypeClusterStatus
messageTypeResizeInstruction
messageTypeResizeInstructionComplete
messageTypeUNUSED0 // used to be ResizeInstruction
messageTypeUNUSED1 // used to be ResizeInstructionComplete
messageTypeNodeState
messageTypeRecalculateCaches
messageTypeLoadSchemaMessage
messageTypeNodeEvent
messageTypeNodeStatus
messageTypeTransaction
messageTypeResizeNodeMessage
messageTypeResizeAbortMessage
messageTypeUNUSED2 // used to be ResizeNodeMessage
messageTypeUNUSED3 // used to be ResizeAbortMessage
messageTypeUpdateField
)
@ -102,10 +102,6 @@ func getMessage(typ byte) Message {
return &DeleteViewMessage{}
case messageTypeClusterStatus:
return &ClusterStatus{}
case messageTypeResizeInstruction:
return &ResizeInstruction{}
case messageTypeResizeInstructionComplete:
return &ResizeInstructionComplete{}
case messageTypeNodeState:
return &NodeStateMessage{}
case messageTypeRecalculateCaches:
@ -118,10 +114,6 @@ func getMessage(typ byte) Message {
return &NodeStatus{}
case messageTypeTransaction:
return &TransactionMessage{}
case messageTypeResizeNodeMessage:
return &ResizeNodeMessage{}
case messageTypeResizeAbortMessage:
return &ResizeAbortMessage{}
case messageTypeUpdateField:
return &UpdateFieldMessage{}
default:
@ -147,10 +139,6 @@ func getMessageType(m Message) byte {
return messageTypeDeleteView
case *ClusterStatus:
return messageTypeClusterStatus
case *ResizeInstruction:
return messageTypeResizeInstruction
case *ResizeInstructionComplete:
return messageTypeResizeInstructionComplete
case *NodeStateMessage:
return messageTypeNodeState
case *RecalculateCaches:
@ -163,10 +151,6 @@ func getMessageType(m Message) byte {
return messageTypeNodeStatus
case *TransactionMessage:
return messageTypeTransaction
case *ResizeNodeMessage:
return messageTypeResizeNodeMessage
case *ResizeAbortMessage:
return messageTypeResizeAbortMessage
case *UpdateFieldMessage:
return messageTypeUpdateField
default:

1053
cluster.go
View file

File diff suppressed because it is too large Load diff

365
cluster_internal_test.go
View file

@ -2,11 +2,9 @@
package pilosa
import (
"context"
"fmt"
"math/rand"
"reflect"
"strings"
"testing"
"testing/quick"
"time"
@ -19,59 +17,6 @@ import (
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
)
// Ensure that fragCombos creates the correct fragment mapping.
func TestFragCombos(t *testing.T) {
t.Skip() // skipping due to change partitioning strategy
uri0, err := pnet.NewURIFromAddress("host0")
if err != nil {
t.Fatal(err)
}
uri1, err := pnet.NewURIFromAddress("host1")
if err != nil {
t.Fatal(err)
}
node0 := &topology.Node{ID: "node0", URI: *uri0}
node1 := &topology.Node{ID: "node1", URI: *uri1}
c := newCluster()
c.addNodeBasicSorted(node0)
c.addNodeBasicSorted(node1)
tests := []struct {
idx string
availableShards *roaring.Bitmap
fieldViews viewsByField
expected fragsByHost
}{
{
idx: "i",
availableShards: roaring.NewBitmap(0, 1, 2),
fieldViews: viewsByField{"f": []string{"v1", "v2"}},
expected: fragsByHost{
"node0": []frag{{"f", "v1", uint64(0)}, {"f", "v2", uint64(0)}},
"node1": []frag{{"f", "v1", uint64(1)}, {"f", "v2", uint64(1)}, {"f", "v1", uint64(2)}, {"f", "v2", uint64(2)}},
},
},
{
idx: "foo",
availableShards: roaring.NewBitmap(0, 1, 2, 3),
fieldViews: viewsByField{"f": []string{"v0"}},
expected: fragsByHost{
"node0": []frag{{"f", "v0", uint64(1)}, {"f", "v0", uint64(2)}},
"node1": []frag{{"f", "v0", uint64(0)}, {"f", "v0", uint64(3)}},
},
},
}
for _, test := range tests {
actual := c.fragCombos(test.idx, test.availableShards, test.fieldViews)
if !reflect.DeepEqual(actual, test.expected) {
t.Errorf("expected: %v, but got: %v", test.expected, actual)
}
}
}
// newHolderWithTempPath returns a new instance of Holder.
func newHolderWithTempPath(tb testing.TB, backend string) *Holder {
path, err := testhook.TempDirInDir(tb, *TempDir, "pilosa-holder-")
@ -109,251 +54,6 @@ func newIndexWithTempPath(tb testing.TB, name string) *Index {
return index
}
// Ensure that fragSources creates the correct fragment mapping.
func TestFragSources(t *testing.T) {
t.Skip() // skipping due to change partitioning strategy
uri0, err := pnet.NewURIFromAddress("host0")
if err != nil {
t.Fatal(err)
}
uri1, err := pnet.NewURIFromAddress("host1")
if err != nil {
t.Fatal(err)
}
uri2, err := pnet.NewURIFromAddress("host2")
if err != nil {
t.Fatal(err)
}
uri3, err := pnet.NewURIFromAddress("host3")
if err != nil {
t.Fatal(err)
}
node0 := &topology.Node{ID: "node0", URI: *uri0}
node1 := &topology.Node{ID: "node1", URI: *uri1}
node2 := &topology.Node{ID: "node2", URI: *uri2}
node3 := &topology.Node{ID: "node3", URI: *uri3}
c1 := newCluster()
c1.ReplicaN = 1
c1.addNodeBasicSorted(node0)
c1.addNodeBasicSorted(node1)
c2 := newCluster()
c2.ReplicaN = 1
c2.addNodeBasicSorted(node0)
c2.addNodeBasicSorted(node1)
c2.addNodeBasicSorted(node2)
c3 := newCluster()
c3.ReplicaN = 2
c3.addNodeBasicSorted(node0)
c3.addNodeBasicSorted(node1)
c4 := newCluster()
c4.ReplicaN = 2
c4.addNodeBasicSorted(node0)
c4.addNodeBasicSorted(node1)
c4.addNodeBasicSorted(node2)
c5 := newCluster()
c5.ReplicaN = 2
c5.addNodeBasicSorted(node0)
c5.addNodeBasicSorted(node1)
c5.addNodeBasicSorted(node2)
c5.addNodeBasicSorted(node3)
idx := newIndexWithTempPath(t, "i")
field, err := idx.CreateFieldIfNotExists("f", OptFieldTypeDefault())
if err != nil {
t.Fatal(err)
}
// Obtain transaction.
var shard uint64
tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
defer tx.Rollback()
_, err = field.SetBit(tx, 1, 101, nil)
if err != nil {
t.Fatal(err)
}
PanicOn(tx.Commit())
shard = 1
tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
defer tx.Rollback()
_, err = field.SetBit(tx, 1, ShardWidth*shard+1, nil)
if err != nil {
t.Fatal(err)
}
PanicOn(tx.Commit())
shard = 2
tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
defer tx.Rollback()
_, err = field.SetBit(tx, 1, ShardWidth*shard+1, nil)
if err != nil {
t.Fatal(err)
}
PanicOn(tx.Commit())
shard = 3
tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
defer tx.Rollback()
_, err = field.SetBit(tx, 1, ShardWidth*shard+1, nil)
if err != nil {
t.Fatal(err)
}
PanicOn(tx.Commit())
tests := []struct {
from *cluster
to *cluster
idx *Index
expected map[string][]*ResizeSource
err string
}{
{
from: c1,
to: c2,
idx: idx,
expected: map[string][]*ResizeSource{
"node0": {},
"node1": {},
"node2": {
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(0)},
{&topology.Node{ID: "node1", URI: pnet.URI{Scheme: "http", Host: "host1", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(2)},
},
},
err: "",
},
{
from: c4,
to: c3,
idx: idx,
expected: map[string][]*ResizeSource{
"node0": {
{&topology.Node{ID: "node1", URI: pnet.URI{Scheme: "http", Host: "host1", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(1)},
},
"node1": {
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(0)},
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(2)},
},
},
err: "",
},
{
from: c5,
to: c4,
idx: idx,
expected: map[string][]*ResizeSource{
"node0": {
{&topology.Node{ID: "node2", URI: pnet.URI{Scheme: "http", Host: "host2", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(0)},
{&topology.Node{ID: "node2", URI: pnet.URI{Scheme: "http", Host: "host2", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(2)},
},
"node1": {
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(3)},
},
"node2": {},
},
err: "",
},
{
from: c2,
to: c4,
idx: idx,
expected: nil,
err: "clusters are the same size",
},
{
from: c1,
to: c5,
idx: idx,
expected: nil,
err: "adding more than one node at a time is not supported",
},
{
from: c5,
to: c1,
idx: idx,
expected: nil,
err: "removing more than one node at a time is not supported",
},
}
for _, test := range tests {
actual, err := (test.from).fragSources(test.to, test.idx)
if test.err != "" {
if !strings.Contains(err.Error(), test.err) {
t.Fatalf("expected error: %s, got: %s", test.err, err.Error())
}
} else {
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(actual, test.expected) {
t.Errorf("expected: %v, but got: %v", test.expected, actual)
}
}
}
}
// Ensure that fragSources creates the correct fragment mapping.
func TestResizeJob(t *testing.T) {
uri0, err := pnet.NewURIFromAddress("host0")
if err != nil {
t.Fatal(err)
}
uri1, err := pnet.NewURIFromAddress("host1")
if err != nil {
t.Fatal(err)
}
uri2, err := pnet.NewURIFromAddress("host2")
if err != nil {
t.Fatal(err)
}
node0 := &topology.Node{ID: "node0", URI: *uri0}
node1 := &topology.Node{ID: "node1", URI: *uri1}
node2 := &topology.Node{ID: "node2", URI: *uri2}
tests := []struct {
existingNodes []*topology.Node
node *topology.Node
action string
expectedIDs map[string]bool
}{
{
existingNodes: []*topology.Node{node0, node1},
node: node2,
action: resizeJobActionAdd,
expectedIDs: map[string]bool{node0.ID: false, node1.ID: false, node2.ID: false},
},
{
existingNodes: []*topology.Node{node0, node1, node2},
node: node2,
action: resizeJobActionRemove,
expectedIDs: map[string]bool{node0.ID: false, node1.ID: false},
},
}
for _, test := range tests {
actual := newResizeJob(test.existingNodes, test.node, test.action)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(actual.IDs, test.expectedIDs) {
t.Errorf("expected: %v, but got: %v", test.expectedIDs, actual.IDs)
}
}
}
// Ensure the cluster can fairly distribute partitions across the nodes.
func TestCluster_Owners(t *testing.T) {
c := cluster{
@ -502,60 +202,6 @@ func TestCluster_Nodes(t *testing.T) {
})
}
func TestCluster_PreviousNode(t *testing.T) {
node0 := &topology.Node{ID: "node0"}
node1 := &topology.Node{ID: "node1"}
node2 := &topology.Node{ID: "node2"}
t.Run("OneNode", func(t *testing.T) {
c := newCluster()
c.addNodeBasicSorted(node0)
c.Node = node0
if prev := c.unprotectedPreviousNode(); prev != nil {
t.Errorf("expected: nil, but got: %v", prev)
}
})
t.Run("TwoNode", func(t *testing.T) {
c := newCluster()
c.addNodeBasicSorted(node0)
c.addNodeBasicSorted(node1)
c.Node = node0
if prev := c.unprotectedPreviousNode(); prev != node1 {
t.Errorf("expected: node1, but got: %v", prev)
}
c.Node = node1
if prev := c.unprotectedPreviousNode(); prev != node0 {
t.Errorf("expected: node0, but got: %v", prev)
}
})
t.Run("ThreeNode", func(t *testing.T) {
c := newCluster()
c.addNodeBasicSorted(node0)
c.addNodeBasicSorted(node1)
c.addNodeBasicSorted(node2)
c.Node = node0
if prev := c.unprotectedPreviousNode(); prev != node2 {
t.Errorf("expected: node2, but got: %v", prev)
}
c.Node = node1
if prev := c.unprotectedPreviousNode(); prev != node0 {
t.Errorf("expected: node0, but got: %v", prev)
}
c.Node = node2
if prev := c.unprotectedPreviousNode(); prev != node1 {
t.Errorf("expected: node1, but got: %v", prev)
}
})
}
func TestAE(t *testing.T) {
t.Run("AbortDoesn'tBlockUninitialized", func(t *testing.T) {
c := newCluster()
@ -615,14 +261,3 @@ func TestAE(t *testing.T) {
}
})
}
func TestTranslateIndexKey(t *testing.T) {
c := newCluster()
node0 := &topology.Node{ID: "node0"}
c.addNodeBasicSorted(node0)
c.holder = newHolderWithTempPath(t, "rbf")
_, e := c.translateIndexKey(context.Background(), "i", "a", false)
if e == nil {
t.Fatal("expecting error")
}
}

43
disco/disco.go
View file

@ -57,7 +57,6 @@ const (
ClusterStateStarting ClusterState = "STARTING" // cluster is starting and some internal services are not ready yet.
ClusterStateDegraded ClusterState = "DEGRADED" // cluster is running but we've lost some # of hosts >0 but < replicaN. Only read queries are allowed.
ClusterStateNormal ClusterState = "NORMAL" // cluster is up and running.
ClusterStateResizing ClusterState = "RESIZING" // cluster is replicating data to other nodes.
ClusterStateDown ClusterState = "DOWN" // cluster is unable to serve queries.
)
@ -67,11 +66,9 @@ const (
NodeStateUnknown NodeState = "UNKNOWN"
NodeStateStarting NodeState = "STARTING"
NodeStateStarted NodeState = "STARTED"
NodeStateResizing NodeState = "RESIZING"
)
type Stator interface {
// Started will mark the actual node as already started.
// It must be called after all initialization processes
// are up and running.
@ -83,15 +80,7 @@ type Stator interface {
// - If all nodes are up and running: "NORMAL"
// - If number of DOWN nodes is lower than number of replicas: "DEGRADED"
// - If number of unresponsive nodes is greater than (or equal to) the number of replicas: "DOWN"
// - If any of the nodes started a resize operation, or a new
// node was specifically added or removed from the cluster: "RESIZING"
ClusterState(context.Context) (ClusterState, error)
// NodeState returns the specific state of a node given its ID.
NodeState(context.Context, string) (NodeState, error)
// NodeStates will return all the states by node ID of the actual nodes in the cluster.
NodeStates(context.Context) (map[string]NodeState, error)
}
// Schema is a map of all indexes, each of those being a map of fields, then
@ -141,21 +130,6 @@ type Metadator interface {
SetMetadata(ctx context.Context, metadata []byte) error
}
// Resizer triggers resizing the node and changes cluster state into RESIZING.
// We can also return some kind of handler from Resize function (e.g. key-value)
type Resizer interface {
// Resize will trigger a resize event. Node state will change to RESIZE state.
// The returned function can be used to send info about the resize process to other nodes.
Resize(ctx context.Context) (func([]byte) error, error)
// DoneResize will mark the resize event as done. This will be called when all the resize actions are done.
DoneResize() error
// Watch will give information about a resize event in another node, using its peerID.
// onUpdate function will be called per each event sent by the node in RESIZE state.
Watch(ctx context.Context, peerID string, onUpdate func([]byte) error) error
}
// Sharder is an interface used to maintain the set of availableShards bitmaps
// per field.
type Sharder interface {
@ -217,14 +191,6 @@ func (n *nopStator) Started(ctx context.Context) error {
return nil
}
func (n *nopStator) NodeState(context.Context, string) (NodeState, error) {
return NodeStateUnknown, nil
}
func (n *nopStator) NodeStates(context.Context) (map[string]NodeState, error) {
return nil, nil
}
// NopMetadator represents a Metadator that doesn't do anything.
var NopMetadator Metadator = &nopMetadator{}
@ -237,15 +203,6 @@ func (*nopMetadator) SetMetadata(context.Context, []byte) error {
return nil
}
// NopResizer represents a Resizer that doesn't do anything.
var NopResizer Resizer = &nopResizer{}
type nopResizer struct{}
func (*nopResizer) Resize(context.Context) (func([]byte) error, error) { return nil, nil }
func (*nopResizer) DoneResize() error { return nil }
func (*nopResizer) Watch(context.Context, string, func([]byte) error) error { return nil }
// NopSharder represents a Sharder that doesn't do anything.
var NopSharder Sharder = &nopSharder{}

168
encoding/proto/proto.go
View file

@ -119,22 +119,6 @@ func (s Serializer) Unmarshal(buf []byte, m pilosa.Message) error {
}
s.decodeClusterStatus(msg, mt)
return nil
case *pilosa.ResizeInstruction:
msg := &pb.ResizeInstruction{}
err := proto.Unmarshal(buf, msg)
if err != nil {
return errors.Wrap(err, "unmarshaling ResizeInstruction")
}
s.decodeResizeInstruction(msg, mt)
return nil
case *pilosa.ResizeInstructionComplete:
msg := &pb.ResizeInstructionComplete{}
err := proto.Unmarshal(buf, msg)
if err != nil {
return errors.Wrap(err, "unmarshaling ResizeInstructionComplete")
}
s.decodeResizeInstructionComplete(msg, mt)
return nil
case *pilosa.NodeStateMessage:
msg := &pb.NodeStateMessage{}
err := proto.Unmarshal(buf, msg)
@ -312,23 +296,6 @@ func (s Serializer) Unmarshal(buf []byte, m pilosa.Message) error {
*mt = s.decodeRowMatrix(msg)
return nil
case *pilosa.ResizeNodeMessage:
msg := &pb.ResizeNodeMessage{}
err := proto.Unmarshal(buf, msg)
if err != nil {
return errors.Wrap(err, "unmarshaling ResizeNodeMessage")
}
decodeResizeNodeMessage(msg, mt)
return nil
case *pilosa.ResizeAbortMessage:
msg := &pb.ResizeAbortMessage{}
err := proto.Unmarshal(buf, msg)
if err != nil {
return errors.Wrap(err, "unmarshaling ResizeAbortMessage")
}
decodeResizeAbortMessage(msg, mt)
return nil
case *ingest.ShardedRequest:
msg := &pb.ShardedIngestRequest{}
err := proto.Unmarshal(buf, msg)
@ -368,10 +335,6 @@ func (s Serializer) encodeToProto(m pilosa.Message) proto.Message {
return s.encodeDeleteViewMessage(mt)
case *pilosa.ClusterStatus:
return s.encodeClusterStatus(mt)
case *pilosa.ResizeInstruction:
return s.encodeResizeInstruction(mt)
case *pilosa.ResizeInstructionComplete:
return s.encodeResizeInstructionComplete(mt)
case *pilosa.NodeStateMessage:
return s.encodeNodeStateMessage(mt)
case *pilosa.RecalculateCaches:
@ -414,10 +377,6 @@ func (s Serializer) encodeToProto(m pilosa.Message) proto.Message {
return s.encodeTransactionMessage(mt)
case *pilosa.AtomicRecord:
return s.encodeAtomicRecord(mt)
case *pilosa.ResizeNodeMessage:
return s.encodeResizeNodeMessage(mt)
case *pilosa.ResizeAbortMessage:
return s.encodeResizeAbortMessage(mt)
case *ingest.ShardedRequest:
return s.encodeShardedIngestRequest(mt)
}
@ -606,52 +565,6 @@ func (s Serializer) encodeQueryResponse(m *pilosa.QueryResponse) *pb.QueryRespon
return resp
}
func (s Serializer) encodeResizeInstruction(m *pilosa.ResizeInstruction) *pb.ResizeInstruction {
return &pb.ResizeInstruction{
JobID: m.JobID,
Node: s.encodeNode(m.Node),
Primary: s.encodeNode(m.Primary),
Sources: s.encodeResizeSources(m.Sources),
TranslationSources: s.encodeTranslationResizeSources(m.TranslationSources),
NodeStatus: s.encodeNodeStatus(m.NodeStatus),
ClusterStatus: s.encodeClusterStatus(m.ClusterStatus),
}
}
func (s Serializer) encodeResizeSources(srcs []*pilosa.ResizeSource) []*pb.ResizeSource {
new := make([]*pb.ResizeSource, 0, len(srcs))
for _, src := range srcs {
new = append(new, s.encodeResizeSource(src))
}
return new
}
func (s Serializer) encodeResizeSource(m *pilosa.ResizeSource) *pb.ResizeSource {
return &pb.ResizeSource{
Node: s.encodeNode(m.Node),
Index: m.Index,
Field: m.Field,
View: m.View,
Shard: m.Shard,
}
}
func (s Serializer) encodeTranslationResizeSources(srcs []*pilosa.TranslationResizeSource) []*pb.TranslationResizeSource {
new := make([]*pb.TranslationResizeSource, 0, len(srcs))
for _, src := range srcs {
new = append(new, s.encodeTranslationResizeSource(src))
}
return new
}
func (s Serializer) encodeTranslationResizeSource(m *pilosa.TranslationResizeSource) *pb.TranslationResizeSource {
return &pb.TranslationResizeSource{
Node: s.encodeNode(m.Node),
Index: m.Index,
PartitionID: int32(m.PartitionID),
}
}
func (s Serializer) encodeSchema(m *pilosa.Schema) *pb.Schema {
return &pb.Schema{
Indexes: s.encodeIndexInfos(m.Indexes),
@ -838,14 +751,6 @@ func (s Serializer) encodeDeleteViewMessage(m *pilosa.DeleteViewMessage) *pb.Del
}
}
func (s Serializer) encodeResizeInstructionComplete(m *pilosa.ResizeInstructionComplete) *pb.ResizeInstructionComplete {
return &pb.ResizeInstructionComplete{
JobID: m.JobID,
Node: s.encodeNode(m.Node),
Error: m.Error,
}
}
func (s Serializer) encodeNodeStateMessage(m *pilosa.NodeStateMessage) *pb.NodeStateMessage {
return &pb.NodeStateMessage{
NodeID: m.NodeID,
@ -1034,52 +939,6 @@ func (s Serializer) encodeShardIngestOperation(op *ingest.Operation) *pb.ShardIn
return out
}
func (s Serializer) decodeResizeInstruction(ri *pb.ResizeInstruction, m *pilosa.ResizeInstruction) {
m.JobID = ri.JobID
m.Node = &topology.Node{}
s.decodeNode(ri.Node, m.Node)
m.Primary = &topology.Node{}
s.decodeNode(ri.Primary, m.Primary)
m.Sources = make([]*pilosa.ResizeSource, len(ri.Sources))
s.decodeResizeSources(ri.Sources, m.Sources)
m.TranslationSources = make([]*pilosa.TranslationResizeSource, len(ri.TranslationSources))
s.decodeTranslationResizeSources(ri.TranslationSources, m.TranslationSources)
m.NodeStatus = &pilosa.NodeStatus{}
s.decodeNodeStatus(ri.NodeStatus, m.NodeStatus)
m.ClusterStatus = &pilosa.ClusterStatus{}
s.decodeClusterStatus(ri.ClusterStatus, m.ClusterStatus)
}
func (s Serializer) decodeResizeSources(srcs []*pb.ResizeSource, m []*pilosa.ResizeSource) {
for i := range srcs {
m[i] = &pilosa.ResizeSource{}
s.decodeResizeSource(srcs[i], m[i])
}
}
func (s Serializer) decodeResizeSource(rs *pb.ResizeSource, m *pilosa.ResizeSource) {
m.Node = &topology.Node{}
s.decodeNode(rs.Node, m.Node)
m.Index = rs.Index
m.Field = rs.Field
m.View = rs.View
m.Shard = rs.Shard
}
func (s Serializer) decodeTranslationResizeSources(srcs []*pb.TranslationResizeSource, m []*pilosa.TranslationResizeSource) {
for i := range srcs {
m[i] = &pilosa.TranslationResizeSource{}
s.decodeTranslationResizeSource(srcs[i], m[i])
}
}
func (s Serializer) decodeTranslationResizeSource(rs *pb.TranslationResizeSource, m *pilosa.TranslationResizeSource) {
m.Node = &topology.Node{}
s.decodeNode(rs.Node, m.Node)
m.Index = rs.Index
m.PartitionID = int(rs.PartitionID)
}
func (s Serializer) decodeSchema(sc *pb.Schema, m *pilosa.Schema) {
m.Indexes = make([]*pilosa.IndexInfo, len(sc.Indexes))
s.decodeIndexes(sc.Indexes, m.Indexes)
@ -1252,13 +1111,6 @@ func (s Serializer) decodeDeleteViewMessage(pb *pb.DeleteViewMessage, m *pilosa.
m.View = pb.View
}
func (s Serializer) decodeResizeInstructionComplete(pb *pb.ResizeInstructionComplete, m *pilosa.ResizeInstructionComplete) {
m.JobID = pb.JobID
m.Node = &topology.Node{}
s.decodeNode(pb.Node, m.Node)
m.Error = pb.Error
}
func (s Serializer) decodeNodeStateMessage(pb *pb.NodeStateMessage, m *pilosa.NodeStateMessage) {
m.NodeID = pb.NodeID
m.State = pb.State
@ -1986,26 +1838,6 @@ func (s Serializer) encodeDecimal(p *pql.Decimal) *pb.Decimal {
return retval
}
func (s Serializer) encodeResizeNodeMessage(m *pilosa.ResizeNodeMessage) *pb.ResizeNodeMessage {
return &pb.ResizeNodeMessage{
NodeID: m.NodeID,
Action: m.Action,
}
}
func (s Serializer) encodeResizeAbortMessage(*pilosa.ResizeAbortMessage) *pb.ResizeAbortMessage {
return &pb.ResizeAbortMessage{}
}
func decodeResizeNodeMessage(pb *pb.ResizeNodeMessage, m *pilosa.ResizeNodeMessage) {
m.NodeID = pb.NodeID
m.Action = pb.Action
}
func decodeResizeAbortMessage(pb *pb.ResizeAbortMessage, m *pilosa.ResizeAbortMessage) {
}
func (s Serializer) decodeShardedIngestRequest(req *pb.ShardedIngestRequest) (*ingest.ShardedRequest, error) {
if req == nil || len(req.Ops) == 0 {
return &ingest.ShardedRequest{}, nil

151
etcd/embed.go
View file

@ -6,9 +6,7 @@ import (
"context"
"encoding/json"
"fmt"
"log"
"net"
"os"
"path"
"sort"
"strings"
@ -59,7 +57,6 @@ var (
_ disco.Schemator = &Etcd{}
_ disco.Stator = &Etcd{}
_ disco.Metadator = &Etcd{}
_ disco.Resizer = &Etcd{}
_ disco.Sharder = &Etcd{}
)
@ -69,7 +66,6 @@ const (
nodePrefix = "/node/"
heartbeatPrefix = nodePrefix + "heartbeat/"
schemaPrefix = "/schema/"
resizePrefix = nodePrefix + "resize/"
metadataPrefix = nodePrefix + "metadata/"
shardPrefix = "/shard/"
)
@ -93,15 +89,11 @@ var (
// reach us out of order.
type nodeData struct {
heartbeatState string
resizeState string
metadata []byte
topologyNode *topology.Node
}
func (n *nodeData) computedState() disco.NodeState {
if n.resizeState != "" {
return disco.NodeStateResizing
}
if n.heartbeatState != "" {
return disco.NodeState(n.heartbeatState)
}
@ -116,7 +108,7 @@ type Etcd struct {
cli *clientv3.Client
cliMu sync.Mutex
heartbeatLeasedKV, resizeLeasedKV *leasedKV
heartbeatLeasedKV *leasedKV
// We have a watcher running. watchCancel() cancels its context.
watchCancel func()
@ -164,10 +156,6 @@ func (e *Etcd) Close() error {
e.watchCancel()
}
if e.e != nil {
if e.resizeLeasedKV != nil {
e.resizeLeasedKV.Stop()
e.resizeLeasedKV = nil
}
if e.heartbeatLeasedKV != nil {
e.heartbeatLeasedKV.Stop()
}
@ -262,7 +250,7 @@ func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) {
return errors.Wrap(err, "exhausted all retries")
}
func (e *Etcd) parseOptions() *embed.Config {
func (e *Etcd) parseOptions() (*embed.Config, error) {
cfg := embed.NewConfig()
cfg.LogLevel = "error"
cfg.Logger = "zap"
@ -293,8 +281,7 @@ func (e *Etcd) parseOptions() *embed.Config {
//check for multiple nodes in the cluster and error if present
nodes := strings.Split(e.options.InitCluster, ",")
if len(nodes) > 1 {
e.logger.Errorf("Multiple cluster nodes detected - this version of FeatureBase only supports single node. %+v", e.options.InitCluster)
os.Exit(1)
return nil, fmt.Errorf("multiple cluster nodes detected - this version of FeatureBase only supports single node. %+v", e.options.InitCluster)
}
// %% end sonarcloud ignore %%
}
@ -305,24 +292,7 @@ func (e *Etcd) parseOptions() *embed.Config {
}
if e.options.ClusterURL != "" {
cfg.ClusterState = embed.ClusterStateFlagExisting
cli, err := clientv3.NewFromURL(e.options.ClusterURL)
if err != nil {
panic(err)
}
defer cli.Close()
log.Println("Cluster Members:")
mIDs, mNames, mURLs := memberList(cli)
for i, id := range mIDs {
log.Printf("\tid: %d, name: %s, url: %s\n", id, mNames[i], mURLs[i])
cfg.InitialCluster += "," + mNames[i] + "=" + mURLs[i]
}
log.Println("Joining Cluster:")
id, name := memberAdd(cli, e.options.APeerURL)
log.Printf("\tid: %d, name: %s\n", id, name)
return nil, errors.New("joining an existing cluster is unsupported")
}
// can only use tls if not using pre-configured listeners
cfg.ClientTLSInfo = transport.TLSInfo{
@ -336,12 +306,15 @@ func (e *Etcd) parseOptions() *embed.Config {
KeyFile: e.options.PeerKeyFile,
}
return cfg
return cfg, nil
}
// Start starts etcd and hearbeat
func (e *Etcd) Start(ctx context.Context) (_ disco.InitialClusterState, err error) {
opts := e.parseOptions()
opts, err := e.parseOptions()
if err != nil {
return disco.InitialClusterStateNew, err
}
state := disco.InitialClusterState(opts.ClusterState)
e.e, err = embed.StartEtcd(opts)
@ -403,24 +376,6 @@ func (e *Etcd) startHeartbeatAndWatcher(ctx context.Context) error {
return nil
}
func (e *Etcd) NodeState(ctx context.Context, peerID string) (disco.NodeState, error) {
return e.nodeState(ctx, peerID)
}
func (e *Etcd) nodeState(ctx context.Context, peerID string) (disco.NodeState, error) {
e.nodeMu.Lock()
defer e.nodeMu.Unlock()
err := e.populateNodeStates(ctx)
return e.nodeStates[peerID], err
}
func (e *Etcd) NodeStates(ctx context.Context) (map[string]disco.NodeState, error) {
e.nodeMu.Lock()
defer e.nodeMu.Unlock()
err := e.populateNodeStates(ctx)
return e.nodeStates, err
}
func (e *Etcd) Started(ctx context.Context) (err error) {
return e.heartbeatLeasedKV.Set(ctx, string(disco.NodeStateStarted))
}
@ -465,7 +420,6 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er
var (
heartbeats int = 0
resize bool
starting bool
)
e.nodeMu.Lock()
@ -480,8 +434,6 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er
switch state {
case disco.NodeStateStarting:
starting = true
case disco.NodeStateResizing:
resize = true
case disco.NodeStateUnknown:
continue
}
@ -489,10 +441,6 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er
heartbeats++
}
if resize {
return disco.ClusterStateResizing, nil
}
if starting {
return disco.ClusterStateStarting, nil
}
@ -508,57 +456,6 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er
return disco.ClusterStateNormal, nil
}
func (e *Etcd) Resize(ctx context.Context) (func([]byte) error, error) {
key := path.Join(resizePrefix, e.e.Server.ID().String())
if e.resizeLeasedKV == nil {
e.resizeLeasedKV = newLeasedKV(e, key, e.options.HeartbeatTTL)
}
if err := e.resizeLeasedKV.Start(""); err != nil {
return nil, errors.Wrap(err, "Resize: creates a new hearbeat")
}
return func(value []byte) error {
log.Println("Update progress:", key, string(value))
return e.putKey(ctx, key, string(value), clientv3.WithIgnoreLease())
}, nil
}
func (e *Etcd) DoneResize() error {
if e.resizeLeasedKV != nil {
e.resizeLeasedKV.Stop()
}
e.resizeLeasedKV = nil
return nil
}
func (e *Etcd) Watch(ctx context.Context, peerID string, onUpdate func([]byte) error) error {
key := path.Join(resizePrefix, peerID)
for resp := range e.cli.Watch(ctx, key) {
if err := resp.Err(); err != nil {
return errors.Wrapf(err, "Watch: key (%s) response", key)
}
for _, ev := range resp.Events {
switch ev.Type {
case mvccpb.PUT:
if onUpdate != nil && ev.Kv.Value != nil {
if err := onUpdate(ev.Kv.Value); err != nil {
return err
}
}
case mvccpb.DELETE:
// nothing to watch - key was deleted
return errors.WithMessagef(disco.ErrKeyDeleted, "Watch key %s", key)
}
}
}
return nil
}
// parseNodeKey reads heartbeatPrefix + "23" and yields (heartbeatPrefix, "23", nil).
func parseNodeKey(key []byte) (prefix string, peerID string, err error) {
// we're looking for things starting with nodePrefix
@ -596,12 +493,6 @@ func (e *Etcd) deleteNodeData(key []byte, revision int64) error {
e.knownNodes[peerID].metadata = nil
e.knownNodes[peerID].topologyNode = &topology.Node{}
e.nodeStatesDirty = true
case resizePrefix:
if e.knownNodes[peerID] == nil {
e.knownNodes[peerID] = &nodeData{}
}
e.knownNodes[peerID].resizeState = ""
e.nodeStatesDirty = true
default:
return fmt.Errorf("node watch: invalid prefix %q", prefix)
}
@ -609,7 +500,7 @@ func (e *Etcd) deleteNodeData(key []byte, revision int64) error {
}
// putNodeData does the actual updating of the node state maps, etc,
// given an incoming heartbeat, metadata, or resizing change. It requires
// given an incoming heartbeat or metadata change. It requires
// that you already hold the node mutex.
func (e *Etcd) putNodeData(key []byte, value []byte, revision int64) (err error) {
prefix, peerID, err := parseNodeKey(key)
@ -640,12 +531,6 @@ func (e *Etcd) putNodeData(key []byte, value []byte, revision int64) (err error)
// This saves us one remake of the node later, probably.
e.knownNodes[peerID].topologyNode.State = e.knownNodes[peerID].computedState()
e.nodeStatesDirty = true
case resizePrefix:
if e.knownNodes[peerID] == nil {
e.knownNodes[peerID] = &nodeData{}
}
e.knownNodes[peerID].resizeState = string(value)
e.nodeStatesDirty = true
default:
return fmt.Errorf("node watch: invalid prefix %q", prefix)
}
@ -728,7 +613,7 @@ func (e *Etcd) watchNodesOnce(ctx context.Context, cli *clientv3.Client) (err er
}
}
// WatchNodes monitors changes to /heartbeat/, /resizing/, 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())
@ -1170,17 +1055,3 @@ func (e *Etcd) NodeIDs() []string {
}
return ids
}
// SetNodes implements the Noder interface as NOP
// (because we can't force to set nodes for etcd).
func (e *Etcd) SetNodes(nodes []*topology.Node) {}
// AppendNode implements the Noder interface as NOP
// (because resizer is responsible for adding new nodes).
func (e *Etcd) AppendNode(node *topology.Node) {}
// RemoveNode implements the Noder interface as NOP
// (because resizer is responsible for removing existing nodes)
func (e *Etcd) RemoveNode(nodeID string) bool {
return false
}

39
etcd/embed_test.go
View file

@ -1,11 +1,14 @@
package etcd
import (
"context"
"net/url"
"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"
)
@ -43,3 +46,39 @@ func TestRestartEtcd(t *testing.T) {
}
e.Close()
}
func TestParseOptions(t *testing.T) {
var e = &Etcd{options: Options{ClusterURL: "http://foo"}, logger: logger.NewLogfLogger(t)}
curl, _ := url.Parse(pilosa.EtcdUnixSocket(t))
e.options.LClientURL = curl.String()
curl, _ = url.Parse(pilosa.EtcdUnixSocket(t))
e.options.LPeerURL = curl.String()
e.options.ClusterURL = "http://foo"
_, err := e.parseOptions()
if err == nil {
t.Fatalf("cluster URL should be rejected")
}
e.options.ClusterURL = ""
e.options.InitCluster = "a,b"
_, err = e.parseOptions()
if AllowCluster() {
if err != nil {
t.Fatalf("expect options parsing to succeed")
}
} else {
t.Logf("no-allow: %v", err)
if err == nil {
t.Fatalf("should have failed to parse a multi-node cluster in non-clustered build")
}
}
// verify failure on start with invalid options
state, err := e.Start(context.Background())
if err == nil {
t.Fatalf("should have gotten error starting etcd with invalid options")
}
if state != disco.InitialClusterStateNew {
t.Fatalf("expected cluster state of %q, got %q", disco.InitialClusterStateNew, state)
}
}

80
http_handler.go
View file

@ -279,8 +279,6 @@ func (h *Handler) Close() error {
func (h *Handler) populateValidators() {
h.validators = map[string]*queryValidationSpec{}
h.validators["PostClusterResizeAbort"] = queryValidationSpecRequired()
h.validators["PostClusterResizeRemoveNode"] = queryValidationSpecRequired()
h.validators["GetExport"] = queryValidationSpecRequired("index", "field", "shard")
h.validators["GetIndexes"] = queryValidationSpecRequired()
h.validators["GetIndex"] = queryValidationSpecRequired()
@ -479,8 +477,6 @@ var latticeRoutes = []string{"/tables", "/query", "/querybuilder", "/signin"} //
// newRouter creates a new mux http router.
func newRouter(handler *Handler) http.Handler {
router := mux.NewRouter()
router.HandleFunc("/cluster/resize/abort", handler.chkAuthZ(handler.handlePostClusterResizeAbort, authz.Admin)).Methods("POST").Name("PostClusterResizeAbort")
router.HandleFunc("/cluster/resize/remove-node", handler.chkAuthZ(handler.handlePostClusterResizeRemoveNode, authz.Admin)).Methods("POST").Name("PostClusterResizeRemoveNode")
// TODO: figure out how to protect these if needed
router.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux).Methods("GET")
@ -2830,82 +2826,6 @@ func parseUint64Slice(s string) ([]uint64, error) {
return a, nil
}
// handlePostClusterResizeRemoveNode handles POST /cluster/resize/remove-node request.
func (h *Handler) handlePostClusterResizeRemoveNode(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
// Decode request.
var req removeNodeRequest
err := json.NewDecoder(r.Body).Decode(&req)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
removeNode, err := h.api.RemoveNode(req.ID)
if err != nil {
if errors.Cause(err) == ErrNodeIDNotExists {
http.Error(w, "removing node: "+err.Error(), http.StatusNotFound)
} else {
http.Error(w, "removing node: "+err.Error(), http.StatusInternalServerError)
}
return
}
// Encode response.
w.Header().Set("Content-Type", "application/json")
if err := json.NewEncoder(w).Encode(removeNodeResponse{
Remove: removeNode,
}); err != nil {
h.logger.Errorf("response encoding error: %s", err)
}
}
type removeNodeRequest struct {
ID string `json:"id"`
}
type removeNodeResponse struct {
Remove *topology.Node `json:"remove"`
}
// handlePostClusterResizeAbort handles POST /cluster/resize/abort request.
func (h *Handler) handlePostClusterResizeAbort(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
err := h.api.ResizeAbort()
var msg string
if err != nil {
switch errors.Cause(err) {
case ErrNodeNotPrimary:
http.Error(w, err.Error(), http.StatusBadRequest)
return
case ErrResizeNotRunning:
msg = err.Error()
default:
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
// Encode response.
w.Header().Set("Content-Type", "application/json")
if err := json.NewEncoder(w).Encode(clusterResizeAbortResponse{
Info: msg,
}); err != nil {
h.logger.Errorf("response encoding error: %s", err)
}
}
type clusterResizeAbortResponse struct {
Info string `json:"info"`
}
func (h *Handler) handleRecalculateCaches(w http.ResponseWriter, r *http.Request) {
err := h.api.RecalculateCaches(r.Context())
if err != nil {

2
internal_client.go
View file

@ -2387,7 +2387,7 @@ func (c *InternalClient) FieldTranslateDataReader(ctx context.Context, index, fi
return resp.Body, nil
}
// Status returns pilosa cluster state as a string ("NORMAL", "DEGRADED", "DOWN", "RESIZING", ...)
// Status returns pilosa cluster state as a string ("NORMAL", "DEGRADED", "DOWN", ...)
func (c *InternalClient) Status(ctx context.Context) (string, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Status")
defer span.Finish()

7
lattice/src/App/Home/ClusterHealth/clusterStatus.ts
View file

@ -1,4 +1,4 @@
type clusterStatuses = 'NORMAL' | 'DEGRADED' | 'STARTING' | 'RESIZING' | 'DOWN' | 'UNKNOWN';
type clusterStatuses = 'NORMAL' | 'DEGRADED' | 'STARTING' | 'DOWN' | 'UNKNOWN';
export const CLUSTER_STATUS: {
[key in clusterStatuses]: { label: string; status: string };
@ -16,11 +16,6 @@ export const CLUSTER_STATUS: {
label: 'Some nodes are up, but not enough to answer queries.',
status: 'error'
},
RESIZING: {
label:
'Cluster is resizing. Most endpoints are unavailable until the resize completes.',
status: 'info'
},
DOWN: {
label: 'Cluster is unable to serve queries.',
status: 'disabled'

6
pilosa.go
View file

@ -62,10 +62,8 @@ var (
// ErrPreconditionFailed is returned when specified index/field createdAt timestamps don't match
ErrPreconditionFailed = errors.New("precondition failed")
ErrNodeIDNotExists = errors.New("node with provided ID does not exist")
ErrNodeNotPrimary = errors.New("node is not the primary")
ErrResizeNotRunning = errors.New("no resize job currently running")
ErrResizeNoReplicas = errors.New("not enough data to perform resize (replica factor may need to be increased)")
ErrNodeIDNotExists = errors.New("node with provided ID does not exist")
ErrNodeNotPrimary = errors.New("node is not the primary")
ErrNotImplemented = errors.New("not implemented")
ErrFieldsArgumentRequired = errors.New("fields argument required")

62
server.go
View file

@ -59,7 +59,6 @@ type Server struct { // nolint: maligned
disCo disco.DisCo
stator disco.Stator
metadator disco.Metadator
resizer disco.Resizer
noder topology.Noder
sharder disco.Sharder
schemator disco.Schemator
@ -402,7 +401,6 @@ func OptServerMaxQueryMemory(v int64) ServerOption {
func OptServerDisCo(disCo disco.DisCo,
stator disco.Stator,
metadator disco.Metadator,
resizer disco.Resizer,
noder topology.Noder,
sharder disco.Sharder,
schemator disco.Schemator) ServerOption {
@ -411,7 +409,6 @@ func OptServerDisCo(disCo disco.DisCo,
s.disCo = disCo
s.stator = stator
s.metadator = metadator
s.resizer = resizer
s.noder = noder
s.sharder = sharder
s.schemator = schemator
@ -455,7 +452,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
disCo: disco.NopDisCo,
stator: disco.NopStator,
metadator: disco.NopMetadator,
resizer: disco.NopResizer,
noder: topology.NewEmptyLocalNoder(),
sharder: disco.NopSharder,
schemator: disco.NopSchemator,
@ -525,7 +521,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.cluster.holder = s.holder
s.cluster.disCo = s.disCo
s.cluster.stator = s.stator
s.cluster.resizer = s.resizer
s.cluster.noder = s.noder
s.cluster.sharder = s.sharder
@ -596,6 +591,11 @@ func (s *Server) Open() error {
if err != nil {
return errors.Wrap(err, "starting DisCo")
}
// I'm pretty sure this can't happen, because the path that would have led to it
// happening now generates an error already, but let's be careful.
if initState == disco.InitialClusterStateExisting {
return errors.New("disco reports existing cluster, but this is not supported")
}
// Set node ID.
s.nodeID = s.disCo.ID()
@ -648,12 +648,6 @@ func (s *Server) Open() error {
}
// bring up the background tasks for the holder.
s.holder.Activate()
// if we joined existing cluster then broadcast "resize on add" message
if initState == disco.InitialClusterStateExisting {
if err := s.cluster.addNode(s.nodeID); err != nil {
return errors.Wrap(err, "adding a node to the existing cluster")
}
}
if err := s.stator.Started(context.Background()); err != nil {
return errors.Wrap(err, "setting nodeState")
@ -904,6 +898,9 @@ func (s *Server) TTLRemoval(ctx context.Context) {
}
func (s *Server) monitorAntiEntropy() {
// %% begin sonarcloud ignore %%
// This code isn't really used anymore because of problems with the design,
// but we haven't taken it out yet. But there's no code coverage of it.
if s.antiEntropyInterval == 0 || s.cluster.ReplicaN <= 1 {
return // anti entropy disabled
}
@ -920,25 +917,23 @@ func (s *Server) monitorAntiEntropy() {
select {
case <-s.closing:
return
case <-s.cluster.abortAntiEntropyCh: // receive here so we don't block resizing
case <-s.cluster.abortAntiEntropyCh:
// receive here so we don't block resizing
// ... note that resizing is gone now, but I don't know whether we still need this.
continue
case <-ticker.C:
s.holder.Stats.Count(MetricAntiEntropy, 1, 1.0)
}
t := time.Now()
state, err := s.cluster.State()
// We used to check for resizing before doing anti-entropy, but resizing is out
// so we don't otherwise care about state.
_, err := s.cluster.State()
if err != nil {
s.logger.Printf("cluster state error: err=%s", err)
continue
}
if state == disco.ClusterStateResizing {
continue // don't launch anti-entropy during resize.
// the cluster sets its state to resizing and *then* sends to
// abortAntiEntropyCh before starting to resize
}
// Sync holders.
s.logger.Infof("holder sync beginning")
s.cluster.muAntiEntropy.Lock()
@ -966,6 +961,7 @@ func (s *Server) monitorAntiEntropy() {
break
}
}
// %% end sonarcloud ignore %%
}
// receiveMessage represents an implementation of BroadcastHandler.
@ -1027,34 +1023,6 @@ func (s *Server) receiveMessage(m Message) error {
return err
}
case *ResizeNodeMessage:
switch obj.Action {
case resizeJobActionRemove:
if err := s.cluster.resizeNodeOnRemove(obj.NodeID); err != nil {
return errors.Wrapf(err, "resizing node %s on remove %s", s.cluster.disCo.ID(), obj.NodeID)
}
case resizeJobActionAdd:
if err := s.cluster.resizeNodeOnAdd(obj.NodeID); err != nil {
return errors.Wrapf(err, "resizing node %s on remove %s", s.cluster.disCo.ID(), obj.NodeID)
}
default:
return fmt.Errorf("incorrect resizing node action: %s", obj.Action)
}
case *ResizeInstruction:
err := s.cluster.followResizeInstruction(context.Background(), obj)
if err != nil {
return err
}
case *ResizeAbortMessage:
err := s.cluster.resizeAbort()
if err != nil {
return err
}
case *RecalculateCaches:
s.holder.recalculateCaches()

461
server/cluster_test.go
View file

@ -3,16 +3,12 @@ package server_test
import (
"context"
"encoding/json"
"fmt"
"net"
"net/http"
"reflect"
"strings"
"testing"
"time"
"github.com/molecula/featurebase/v3"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/test"
@ -103,7 +99,7 @@ func TestMain_SendReceiveMessage(t *testing.T) {
}
// Ensure that an empty node comes up in a NORMAL state.
func TestClusterResize_EmptyNode(t *testing.T) {
func TestCluster_EmptyNode(t *testing.T) {
m0 := test.RunCommand(t)
defer m0.Close()
@ -114,7 +110,9 @@ func TestClusterResize_EmptyNode(t *testing.T) {
}
// Ensure that a cluster of empty nodes comes up in a NORMAL state.
func TestClusterResize_EmptyNodes(t *testing.T) {
// Do not combine this with TestCluster_EmptyNode; for non-clustering
// builds, this test gets skipped but we still want that test.
func TestCluster_EmptyNodes(t *testing.T) {
clus := test.MustRunCluster(t, 3)
defer clus.Close()
@ -127,455 +125,6 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
}
}
// Ensure that adding a node correctly resizes the cluster.
func TestClusterResize_AddNode(t *testing.T) {
t.Run("NoData", func(t *testing.T) {
clus := test.MustRunCluster(t, 3)
defer clus.Close()
clus.GetNode(0).AssertState(t, disco.ClusterStateNormal, 1*time.Second)
clus.GetNode(1).AssertState(t, disco.ClusterStateNormal, 1*time.Second)
})
t.Run("WithIndex", func(t *testing.T) {
// Configure node0
m0 := test.MustRunCluster(t, 1).GetNode(0)
defer m0.Close()
// Create a client for each node.
client0 := m0.Client()
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
// Configure node1
m1 := test.NewCommandNode(t)
lsns := make([]*net.TCPListener, 3)
for i := range lsns {
l, err := net.Listen("tcp", "localhost:0")
if err != nil {
t.Fatal(err)
}
lsns[i] = l.(*net.TCPListener)
}
portsCfg := test.GenPortsConfig(t, test.NewPorts(lsns))
m1.Config.Etcd = portsCfg[0].Etcd
m1.Config.Name = portsCfg[0].Name
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
m1.Config.BindGRPC = portsCfg[0].BindGRPC
m1.Config.GRPCListener = portsCfg[0].GRPCListener
err := m1.Start()
if err != nil {
t.Fatal(err)
}
defer m1.Close()
m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
})
t.Run("ContinuousShards", func(t *testing.T) {
// Configure node0
c := test.MustRunCluster(t, 3)
defer c.Close()
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
client0 := m0.Client()
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
col := pilosa.ShardWidth + 20
// Write data on first node.
if _, err := m0.Queryf(t, "i", "", `
Set(1, f=1)
Set(%d, f=1)
`, col); err != nil {
t.Fatal(err)
}
// exp is the expected result for the Row queries that follow.
exp := fmt.Sprintf(`{"results":[{"columns":[1,%d]}]}`, col)
// Verify the data exists on the single node.
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := c.GetNode(1)
defer m1.Close()
m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
// Verify the data exists on both nodes.
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
m1.QueryExpect(t, "i", "", `Row(f=1)`, exp)
})
t.Run("OneShard", func(t *testing.T) {
// Configure node0
c := test.MustRunCluster(t, 3)
defer c.Close()
// Configure node0
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
client0 := m0.Client()
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
// Write data on first node.
if _, err := m0.Query(t, "i", "", `
Set(1, f=1)
`); err != nil {
t.Fatal(err)
}
// exp is the expected result for the Row queries that follow.
exp := `{"results":[{"columns":[1]}]}`
// Verify the data exists on the single node.
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := c.GetNode(1)
defer m1.Close()
m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
// Verify the data exists on both nodes.
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
m1.QueryExpect(t, "i", "", `Row(f=1)`, exp)
})
t.Run("SkippedShard", func(t *testing.T) {
// same reason as the ContinuousShards test above.
c := test.MustRunCluster(t, 3)
defer c.Close()
// Configure node0
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
client0 := m0.Client()
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
col := pilosa.ShardWidth*2 + 20
// Write data on first node. Note that no data is placed on shard 1.
if _, err := m0.Queryf(t, "i", "", `
Set(1, f=1)
Set(%d, f=1)
`, col); err != nil {
t.Fatal(err)
}
// exp is the expected result for the Row queries that follow.
exp := fmt.Sprintf(`{"results":[{"columns":[1,%d]}]}`, col)
// Verify the data exists on the single node.
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := c.GetNode(1)
defer m1.Close()
m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
// Verify the data exists on both nodes.
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
m1.QueryExpect(t, "i", "", `Row(f=1)`, exp)
})
}
// Ensure that adding a node correctly resizes the cluster.
func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
t.Run("WithIndex", func(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()
// Configure node0
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
client0 := m0.Client()
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
errc := make(chan error)
go func() {
_, err := m0.API.CreateIndex(context.Background(), "blah", pilosa.IndexOptions{})
errc <- err
}()
// Configure node1
m1 := c.GetNode(1)
defer m1.Close()
m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
if err := <-errc; err != nil {
t.Fatalf("error from index creation: %v", err)
}
})
t.Run("ContinuousShards", func(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()
// Configure node0
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
client0 := m0.Client()
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
col := pilosa.ShardWidth + 20
// Write data on first node.
if _, err := m0.Queryf(t, "i", "", `
Set(1, f=1)
Set(%d, f=1)
`, col); err != nil {
t.Fatal(err)
}
// exp is the expected result for the Row queries that follow.
exp := fmt.Sprintf(`{"results":[{"columns":[1,%d]}]}`, col)
// Verify the data exists on the single node.
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := c.GetNode(1)
defer m1.Close()
m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
// Verify the data exists on both nodes.
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
m1.QueryExpect(t, "i", "", `Row(f=1)`, exp)
})
t.Run("SkippedShard", func(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()
// Configure node0
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
client0 := m0.Client()
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
col := pilosa.ShardWidth*2 + 20
// Write data on first node. Note that no data is placed on shard 1.
if _, err := m0.Queryf(t, "i", "", `
Set(1, f=1)
Set(%d, f=1)
`, col); err != nil {
t.Fatal(err)
}
// exp is the expected result for the Row queries that follow.
exp := fmt.Sprintf(`{"results":[{"columns":[1,%d]}]}`, col)
// Verify the data exists on the single node.
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := c.GetNode(1)
defer m1.Close()
m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
// Verify the data exists on both nodes.
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
m1.QueryExpect(t, "i", "", `Row(f=1)`, exp)
})
t.Run("WithIndexKeys", func(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()
// Configure node0
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
client0 := m0.Client()
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{Keys: true}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
// Write data on first node.
if _, err := m0.Query(t, "i", "", `
Set('col1', f=1)
Set('col2', f=1)
`); err != nil {
t.Fatal(err)
}
// exp is the expected result for the Row queries that follow.
exp := `{"results":[{"columns":[],"keys":["col2","col1"]}]}`
// Verify the data exists on the single node.
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := c.GetNode(1)
defer m1.Close()
m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second)
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
m1.QueryExpect(t, "i", "", `Row(f=1)`, exp)
})
}
func TestClusterResize_RemoveNode(t *testing.T) {
cluster := test.MustRunCluster(t, 3)
defer cluster.Close()
coord := cluster.GetPrimary()
other := cluster.GetNonPrimary()
mustNodeID := func(baseURL string) string {
body := test.Do(t, "GET", fmt.Sprintf("%s/status", baseURL), "").Body
var resp map[string]interface{}
err := json.Unmarshal([]byte(body), &resp)
if err != nil {
panic(err)
}
if localID, ok := resp["localID"].(string); ok {
return localID
}
panic("localID should be a string")
}
t.Run("ErrorRemoveInvalidNode", func(t *testing.T) {
resp := test.Do(t, "POST", coord.URL()+"/cluster/resize/remove-node", `{"id": "invalid-node-id"}`)
expBody := "removing node: finding node to remove: node with provided ID does not exist"
if resp.StatusCode != http.StatusNotFound {
t.Fatalf("expected StatusCode %d but got %d", http.StatusNotFound, resp.StatusCode)
} else if strings.TrimSpace(resp.Body) != expBody {
t.Fatalf("expected Body '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body))
}
})
t.Run("ErrorRemovePrimary", func(t *testing.T) {
nodeID := mustNodeID(coord.URL())
resp := test.Do(t, "POST", coord.URL()+"/cluster/resize/remove-node", fmt.Sprintf(`{"id": "%s"}`, nodeID))
expBody := fmt.Sprintf("removing node: cannot issue node removal request to the node being removed, id=%s: precondition failed", nodeID)
if resp.StatusCode != http.StatusInternalServerError {
t.Fatalf("expected StatusCode %d but got %d", http.StatusInternalServerError, resp.StatusCode)
} else if strings.TrimSpace(resp.Body) != expBody {
t.Fatalf("expected Body '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body))
}
})
t.Run("ErrorRemoveOnNonPrimary", func(t *testing.T) {
nodeID := mustNodeID(other.URL())
resp := test.Do(t, "POST", other.URL()+"/cluster/resize/remove-node", fmt.Sprintf(`{"id": "%s"}`, nodeID))
expBody := fmt.Sprintf(`removing node: cannot issue node removal request to the node being removed, id=%s: precondition failed`, nodeID)
if resp.StatusCode != http.StatusInternalServerError {
t.Fatalf("expected StatusCode %d but got %d", http.StatusInternalServerError, resp.StatusCode)
} else if strings.TrimSpace(resp.Body) != expBody {
t.Fatalf("expected Body '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body))
}
})
t.Run("ErrorRemoveWithoutReplicas", func(t *testing.T) {
t.Skip("TODO: Unskip the test if you understand it")
client0 := coord.Client()
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
// This is an attempt to ensure there is data on both nodes, but is not guaranteed.
// TODO: Deterministic node IDs would ensure consistent results
setColumns := ""
for i := 0; i < 20; i++ {
setColumns += fmt.Sprintf("Set(%d, f=1) ", i*pilosa.ShardWidth)
}
if _, err := coord.Query(t, "i", "", setColumns); err != nil {
t.Fatal(err)
}
nodeID := mustNodeID(other.URL())
resp := test.Do(t, "POST", coord.URL()+"/cluster/resize/remove-node", fmt.Sprintf(`{"id": "%s"}`, nodeID))
expBody := "not enough data to perform resize"
if resp.StatusCode != http.StatusInternalServerError {
t.Fatalf("expected StatusCode %d but got %d", http.StatusInternalServerError, resp.StatusCode)
} else if !strings.Contains(resp.Body, expBody) {
t.Fatalf("expected to contain '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body))
}
})
}
func TestClusterMutualTLS(t *testing.T) {
commandOpts := make([][]server.CommandOption, 3)
configs := make([]*server.Config, 3)

2
server/grpc.go
View file

@ -125,8 +125,6 @@ func errToStatusError(err error) error {
return status.Error(codes.Aborted, err.Error())
case pilosa.ErrClusterDoesNotOwnShard,
pilosa.ErrResizeNoReplicas,
pilosa.ErrResizeNotRunning,
pilosa.ErrNodeNotPrimary,
pilosa.ErrTooManyWrites,
pilosa.ErrNodeIDNotExists:

10
server/handler_test.go
View file

@ -526,16 +526,6 @@ func TestHandler_Endpoints(t *testing.T) {
mustJSONDecode(t, w.Body)
})
t.Run("Abort no resize job", func(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/cluster/resize/abort", nil))
if w.Code != gohttp.StatusInternalServerError {
bod, err := ioutil.ReadAll(w.Body)
t.Fatalf("unexpected status code: %d, bod: %s, readerr: %v", w.Code, bod, err)
}
// TODO need to test aborting a cluster resize job. this may not be the right place
})
hldr.SetBit("i0", "f0", 30, (1*pilosa.ShardWidth)+1)
hldr.SetBit("i0", "f0", 30, (1*pilosa.ShardWidth)+2)
hldr.SetBit("i0", "f0", 30, (3*pilosa.ShardWidth)+4)

2
server/server.go
View file

@ -459,7 +459,7 @@ 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, e)
discoOpt := pilosa.OptServerDisCo(e, e, e, e, e, e)
serverOptions := []pilosa.ServerOption{
pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)),

94
server/server_test.go
View file

@ -660,100 +660,6 @@ func TestClusteringNodesReplica2(t *testing.T) {
}
}
func TestRemoveNodeAfterItDies(t *testing.T) {
t.Skip("TestRemoveNodeAfterItDies won't be supported unless we implement resizer.")
cluster := test.MustNewCluster(t, 3)
for _, c := range cluster.Nodes {
c.Config.Cluster.ReplicaN = 2
}
err := cluster.Start()
if err != nil {
t.Fatalf("starting cluster: %v", err)
}
// The anonymous function is necessary so that the slice
// passed to Close() as a receiver is the modified value
// of cluster, because we're removing the last entry from it
// below.
defer func() {
cluster.Close()
}()
coord, others := cluster.GetPrimary(), cluster.GetNonPrimaries()
err = cluster.AwaitState(disco.ClusterStateNormal, 100*time.Millisecond)
if err != nil {
t.Fatalf("starting cluster: %v", err)
}
// prevent double-closing cluster.GetNode(2) from the deferred Close above
disabled := others[0]
if err := disabled.Close(); err != nil {
t.Fatalf("closing third node: %v", err)
}
err = cluster.AwaitPrimaryState(disco.ClusterStateDegraded, 30*time.Second)
if err != nil {
t.Fatalf("degrading cluster: %v", err)
}
if _, err := coord.API.RemoveNode(disabled.API.Node().ID); err != nil {
t.Fatalf("removing failed node: %v", err)
}
err = cluster.AwaitPrimaryState(disco.ClusterStateNormal, 30*time.Second)
if err != nil {
t.Fatalf("removing disabled node: %v", err)
}
hosts := coord.API.Hosts(context.Background())
if len(hosts) != 2 {
t.Fatalf("unexpected hosts: %v", hosts)
}
}
func TestRemoveConcurrentIndexCreation(t *testing.T) {
t.Skip("TestRemoveConcurrentIndexCreation won't be supported under etcd. Under RESIZING, creating/updating schema not allowed now.")
cluster := test.MustNewCluster(t, 3)
for _, c := range cluster.Nodes {
c.Config.Cluster.ReplicaN = 2
}
err := cluster.Start()
if err != nil {
t.Fatalf("starting cluster: %v", err)
}
defer cluster.Close()
node0 := cluster.GetNode(0)
err = cluster.AwaitState(disco.ClusterStateNormal, 100*time.Millisecond)
if err != nil {
t.Fatalf("starting cluster: %v", err)
}
errc := make(chan error)
go func() {
_, err := node0.API.CreateIndex(context.Background(), "blah", pilosa.IndexOptions{})
errc <- err
}()
if _, err := node0.API.RemoveNode(cluster.GetNode(2).API.Node().ID); err != nil {
t.Fatalf("removing node: %v", err)
}
err = cluster.AwaitPrimaryState(disco.ClusterStateNormal, 100*time.Millisecond)
if err != nil {
t.Fatalf("starting cluster: %v", err)
}
hosts := node0.API.Hosts(context.Background())
if len(hosts) != 2 {
t.Fatalf("unexpected hosts: %v", hosts)
}
if err := <-errc; err != nil {
t.Fatalf("error from index creation: %v", err)
}
}
// Ensure program imports timestamps as UTC.
func TestMain_ImportTimestamp(t *testing.T) {
m := test.RunCommand(t)

30
topology/noder.go
View file

@ -10,9 +10,6 @@ import (
type Noder interface {
Nodes() []*Node // Remember: this has to be sorted correctly!!
PrimaryNodeID(hasher Hasher) string
SetNodes([]*Node)
AppendNode(*Node)
RemoveNode(nodeID string) bool
}
// localNoder is a simple implementation of the Noder interface
@ -67,30 +64,3 @@ func (n *localNoder) PrimaryNodeID(hasher Hasher) string {
}
return primaryNode.ID
}
// SetNodes implements the Noder interface.
func (n *localNoder) SetNodes(nodes []*Node) {
n.nodes = nodes
}
// AppendNode implements the Noder interface.
func (n *localNoder) AppendNode(node *Node) {
n.nodes = append(n.nodes, node)
// All hosts must be merged in the same order on all nodes in the cluster.
sort.Sort(ByID(n.nodes))
}
// RemoveNode implements the Noder interface.
func (n *localNoder) RemoveNode(nodeID string) bool {
i := NodePositionByID(n.nodes, nodeID)
if i < 0 {
return false
}
copy(n.nodes[i:], n.nodes[i+1:])
n.nodes[len(n.nodes)-1] = nil
n.nodes = n.nodes[:len(n.nodes)-1]
return true
}

51
utils_internal_test.go
View file

@ -6,6 +6,7 @@ import (
"testing"
"time"
"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"
@ -15,6 +16,9 @@ import (
// NewTestCluster returns a cluster with n nodes and uses a mod-based hasher.
func NewTestCluster(tb testing.TB, n int) *cluster {
if n > 1 && etcd.AllowCluster() {
tb.Skipf("cluster size %d not supported in unclustered mode", n)
}
path, err := testhook.TempDir(tb, "pilosa-cluster-")
if err != nil {
panic(err)
@ -26,12 +30,15 @@ func NewTestCluster(tb testing.TB, n int) *cluster {
c.Hasher = NewTestModHasher()
c.Path = path
nodes := make([]*topology.Node, 0, n)
for i := 0; i < n; i++ {
c.noder.AppendNode(&topology.Node{
nodes = append(nodes, &topology.Node{
ID: fmt.Sprintf("node%d", i),
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})
}
c.noder = topology.NewLocalNoder(nodes)
cNodes := c.noder.Nodes()
@ -65,48 +72,6 @@ func (*TestModHasher) Hash(key uint64, n int) int { return int(key) % n }
func (*TestModHasher) Name() string { return "mod" }
var _ = NewTestClusterWithReplication // happy linter
func NewTestClusterWithReplication(tb testing.TB, nNodes, nReplicas, partitionN int) (c *cluster, cleaner func()) {
path, err := testhook.TempDir(tb, "pilosa-cluster-")
if err != nil {
panic(err)
}
// holder
h := NewHolder(path, mustHolderConfig())
// cluster
availableShardFileFlushDuration.Set(100 * time.Millisecond)
c = newCluster()
c.holder = h
c.ReplicaN = nReplicas
c.Hasher = &topology.Jmphasher{}
c.Path = path
c.partitionN = partitionN
for i := 0; i < nNodes; i++ {
nodeID := fmt.Sprintf("node%d", i)
c.noder.AppendNode(&topology.Node{
ID: nodeID,
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})
}
cNodes := c.noder.Nodes()
c.Node = cNodes[0]
if err := c.holder.Open(); err != nil {
panic(err)
}
return c, func() {
c.holder.Close()
c.close()
}
}
func TestReplaceFirstFromBack(t *testing.T) {
for name, test := range map[string]struct {
input string