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

Merge pull request #1077 from travisturner/cluster-resize-nodeid-as-name

Browse source 
Use NodeID instead of URI for node identification
This commit is contained in:
Travis Turner 2018-02-05 10:24:30 -06:00 • committed by GitHub
commit 20d6119906
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
27 changed files with 1134 additions and 928 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
broadcast.go
View file

@ -24,11 +24,9 @@ import (
// MemberSet represents an interface for Node membership and inter-node communication.
type MemberSet interface {
// Returns a list of all Nodes in the cluster
Nodes() []*Node
// Open starts any network activity implemented by the MemberSet
Open() error
// Node is the local node, used for membership broadcasts.
Open(n *Node) error
}
// StaticMemberSet represents a basic MemberSet for testing.
@ -41,13 +39,8 @@ func NewStaticMemberSet() *StaticMemberSet {
return &StaticMemberSet{}
}
// Nodes implements the MemberSet interface and returns a list of nodes in the cluster.
func (s *StaticMemberSet) Nodes() []*Node {
return s.nodes
}
// Open implements the MemberSet interface to start network activity, but for a static MemberSet it does nothing.
func (s *StaticMemberSet) Open() error {
func (s *StaticMemberSet) Open(n *Node) error {
return nil
}

22
client_test.go
View file

@ -36,10 +36,10 @@ func createCluster(c *pilosa.Cluster) ([]*test.Server, []*test.Holder) {
for i := 0; i < numNodes; i++ {
hldr[i] = test.MustOpenHolder()
server[i] = test.NewServer()
server[i].Handler.URI = server[i].HostURI()
server[i].Handler.Cluster = c
server[i].Handler.Cluster.Nodes[i].URI = server[i].HostURI()
server[i].Handler.Holder = hldr[i].Holder
server[i].Handler.Node = server[i].Handler.Cluster.Nodes[i]
}
return server, hldr
}
@ -64,21 +64,21 @@ func TestClient_MultiNode(t *testing.T) {
s[0].Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor(defaultClient)
e.Holder = hldr[0].Holder
e.URI = cluster.Nodes[0].URI
e.Node = cluster.Nodes[0]
e.Cluster = cluster
return e.Execute(ctx, index, query, slices, opt)
}
s[1].Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor(defaultClient)
e.Holder = hldr[1].Holder
e.URI = cluster.Nodes[1].URI
e.Node = cluster.Nodes[1]
e.Cluster = cluster
return e.Execute(ctx, index, query, slices, opt)
}
s[2].Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor(defaultClient)
e.Holder = hldr[2].Holder
e.URI = cluster.Nodes[2].URI
e.Node = cluster.Nodes[2]
e.Cluster = cluster
return e.Execute(ctx, index, query, slices, opt)
}
@ -217,10 +217,10 @@ func TestClient_Import(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.URI = s.HostURI()
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.Holder = hldr.Holder
s.Handler.Node = s.Handler.Cluster.Nodes[0]
// Send import request.
c := test.MustNewClient(s.Host(), defaultClient)
@ -268,10 +268,10 @@ func TestClient_ImportInverseEnabled(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.URI = s.HostURI()
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.Holder = hldr.Holder
s.Handler.Node = s.Handler.Cluster.Nodes[0]
// Send import request.
c := test.MustNewClient(s.Host(), defaultClient)
@ -317,10 +317,10 @@ func TestClient_ImportValue(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.URI = s.HostURI()
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.Holder = hldr.Holder
s.Handler.Node = s.Handler.Cluster.Nodes[0]
// Send import request.
c := test.MustNewClient(s.Host(), defaultClient)
@ -355,10 +355,10 @@ func TestClient_BackupRestore(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.URI = s.HostURI()
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.Holder = hldr.Holder
s.Handler.Node = s.Handler.Cluster.Nodes[0]
c := test.MustNewClient(s.Host(), defaultClient)
@ -420,10 +420,10 @@ func TestClient_BackupInverseView(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.URI = s.HostURI()
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.Holder = hldr.Holder
s.Handler.Node = s.Handler.Cluster.Nodes[0]
c := test.MustNewClient(s.Host(), defaultClient)
@ -457,10 +457,10 @@ func TestClient_BackupInvalidView(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.URI = s.HostURI()
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.Holder = hldr.Holder
s.Handler.Node = s.Handler.Cluster.Nodes[0]
c := test.MustNewClient(s.Host(), defaultClient)
@ -486,10 +486,10 @@ func TestClient_FragmentBlocks(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.URI = s.HostURI()
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.Holder = hldr.Holder
s.Handler.Node = s.Handler.Cluster.Nodes[0]
// Retrieve blocks.
c := test.MustNewClient(s.Host(), defaultClient)

519
cluster.go
View file

File diff suppressed because it is too large Load diff

122
cluster_internal_test.go
View file

@ -34,9 +34,12 @@ func TestFragCombos(t *testing.T) {
t.Fatal(err)
}
node0 := &Node{ID: "node0", URI: *uri0}
node1 := &Node{ID: "node1", URI: *uri1}
c := NewCluster()
c.addNodeBasicSorted(*uri0)
c.addNodeBasicSorted(*uri1)
c.addNodeBasicSorted(node0)
c.addNodeBasicSorted(node1)
tests := []struct {
idx string
@ -49,8 +52,8 @@ func TestFragCombos(t *testing.T) {
maxSlice: uint64(2),
frameViews: viewsByFrame{"f": []string{"v1", "v2"}},
expected: fragsByHost{
URI{"http", "host0", 10101}: []frag{{"f", "v1", uint64(0)}, {"f", "v2", uint64(0)}},
URI{"http", "host1", 10101}: []frag{{"f", "v1", uint64(1)}, {"f", "v2", uint64(1)}, {"f", "v1", uint64(2)}, {"f", "v2", uint64(2)}},
"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)}},
},
},
{
@ -58,8 +61,8 @@ func TestFragCombos(t *testing.T) {
maxSlice: uint64(3),
frameViews: viewsByFrame{"f": []string{"v0"}},
expected: fragsByHost{
URI{"http", "host0", 10101}: []frag{{"f", "v0", uint64(1)}, {"f", "v0", uint64(2)}},
URI{"http", "host1", 10101}: []frag{{"f", "v0", uint64(0)}, {"f", "v0", uint64(3)}},
"node0": []frag{{"f", "v0", uint64(1)}, {"f", "v0", uint64(2)}},
"node1": []frag{{"f", "v0", uint64(0)}, {"f", "v0", uint64(3)}},
},
},
}
@ -106,34 +109,39 @@ func TestFragSources(t *testing.T) {
t.Fatal(err)
}
node0 := &Node{ID: "node0", URI: *uri0}
node1 := &Node{ID: "node1", URI: *uri1}
node2 := &Node{ID: "node2", URI: *uri2}
node3 := &Node{ID: "node3", URI: *uri3}
c1 := NewCluster()
c1.ReplicaN = 1
c1.addNodeBasicSorted(*uri0)
c1.addNodeBasicSorted(*uri1)
c1.addNodeBasicSorted(node0)
c1.addNodeBasicSorted(node1)
c2 := NewCluster()
c2.ReplicaN = 1
c2.addNodeBasicSorted(*uri0)
c2.addNodeBasicSorted(*uri1)
c2.addNodeBasicSorted(*uri2)
c2.addNodeBasicSorted(node0)
c2.addNodeBasicSorted(node1)
c2.addNodeBasicSorted(node2)
c3 := NewCluster()
c3.ReplicaN = 2
c3.addNodeBasicSorted(*uri0)
c3.addNodeBasicSorted(*uri1)
c3.addNodeBasicSorted(node0)
c3.addNodeBasicSorted(node1)
c4 := NewCluster()
c4.ReplicaN = 2
c4.addNodeBasicSorted(*uri0)
c4.addNodeBasicSorted(*uri1)
c4.addNodeBasicSorted(*uri2)
c4.addNodeBasicSorted(node0)
c4.addNodeBasicSorted(node1)
c4.addNodeBasicSorted(node2)
c5 := NewCluster()
c5.ReplicaN = 2
c5.addNodeBasicSorted(*uri0)
c5.addNodeBasicSorted(*uri1)
c5.addNodeBasicSorted(*uri2)
c5.addNodeBasicSorted(*uri3)
c5.addNodeBasicSorted(node0)
c5.addNodeBasicSorted(node1)
c5.addNodeBasicSorted(node2)
c5.addNodeBasicSorted(node3)
idx := newIndexWithTempPath("i")
frame, err := idx.CreateFrameIfNotExists("f", FrameOptions{})
@ -161,19 +169,19 @@ func TestFragSources(t *testing.T) {
from *Cluster
to *Cluster
idx *Index
expected map[URI][]*internal.ResizeSource
expected map[string][]*internal.ResizeSource
err string
}{
{
from: c1,
to: c2,
idx: idx,
expected: map[URI][]*internal.ResizeSource{
URI{"http", "host0", 10101}: []*internal.ResizeSource{},
URI{"http", "host1", 10101}: []*internal.ResizeSource{},
URI{"http", "host2", 10101}: []*internal.ResizeSource{
{&internal.URI{"http", "host0", 10101}, "i", "f", "standard", uint64(0)},
{&internal.URI{"http", "host1", 10101}, "i", "f", "standard", uint64(2)},
expected: map[string][]*internal.ResizeSource{
"node0": []*internal.ResizeSource{},
"node1": []*internal.ResizeSource{},
"node2": []*internal.ResizeSource{
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node1", &internal.URI{"http", "host1", 10101}}, "i", "f", "standard", uint64(2)},
},
},
err: "",
@ -182,13 +190,13 @@ func TestFragSources(t *testing.T) {
from: c4,
to: c3,
idx: idx,
expected: map[URI][]*internal.ResizeSource{
URI{"http", "host0", 10101}: []*internal.ResizeSource{
{&internal.URI{"http", "host1", 10101}, "i", "f", "standard", uint64(1)},
expected: map[string][]*internal.ResizeSource{
"node0": []*internal.ResizeSource{
{&internal.Node{"node1", &internal.URI{"http", "host1", 10101}}, "i", "f", "standard", uint64(1)},
},
URI{"http", "host1", 10101}: []*internal.ResizeSource{
{&internal.URI{"http", "host0", 10101}, "i", "f", "standard", uint64(0)},
{&internal.URI{"http", "host0", 10101}, "i", "f", "standard", uint64(2)},
"node1": []*internal.ResizeSource{
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}}, "i", "f", "standard", uint64(2)},
},
},
err: "",
@ -197,15 +205,15 @@ func TestFragSources(t *testing.T) {
from: c5,
to: c4,
idx: idx,
expected: map[URI][]*internal.ResizeSource{
URI{"http", "host0", 10101}: []*internal.ResizeSource{
{&internal.URI{"http", "host2", 10101}, "i", "f", "standard", uint64(0)},
{&internal.URI{"http", "host2", 10101}, "i", "f", "standard", uint64(2)},
expected: map[string][]*internal.ResizeSource{
"node0": []*internal.ResizeSource{
{&internal.Node{"node2", &internal.URI{"http", "host2", 10101}}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node2", &internal.URI{"http", "host2", 10101}}, "i", "f", "standard", uint64(2)},
},
URI{"http", "host1", 10101}: []*internal.ResizeSource{
{&internal.URI{"http", "host0", 10101}, "i", "f", "standard", uint64(3)},
"node1": []*internal.ResizeSource{
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}}, "i", "f", "standard", uint64(3)},
},
URI{"http", "host2", 10101}: []*internal.ResizeSource{},
"node2": []*internal.ResizeSource{},
},
err: "",
},
@ -265,33 +273,37 @@ func TestResizeJob(t *testing.T) {
t.Fatal(err)
}
node0 := &Node{ID: "node0", URI: *uri0}
node1 := &Node{ID: "node1", URI: *uri1}
node2 := &Node{ID: "node2", URI: *uri2}
tests := []struct {
existingURIs []URI
uri URI
action string
expectedURIs map[URI]bool
existingNodes []*Node
node *Node
action string
expectedIDs map[string]bool
}{
{
existingURIs: []URI{*uri0, *uri1},
uri: *uri2,
action: ResizeJobActionAdd,
expectedURIs: map[URI]bool{*uri0: false, *uri1: false, *uri2: false},
existingNodes: []*Node{node0, node1},
node: node2,
action: ResizeJobActionAdd,
expectedIDs: map[string]bool{node0.ID: false, node1.ID: false, node2.ID: false},
},
{
existingURIs: []URI{*uri0, *uri1, *uri2},
uri: *uri2,
action: ResizeJobActionRemove,
expectedURIs: map[URI]bool{*uri0: false, *uri1: false},
existingNodes: []*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.existingURIs, test.uri, test.action)
actual := NewResizeJob(test.existingNodes, test.node, test.action)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(actual.URIs, test.expectedURIs) {
t.Errorf("expected: %v, but got: %v", test.expectedURIs, actual.URIs)
if !reflect.DeepEqual(actual.IDs, test.expectedIDs) {
t.Errorf("expected: %v, but got: %v", test.expectedIDs, actual.IDs)
}
}
}

151
cluster_test.go
View file

@ -106,7 +106,7 @@ func TestCluster_OwnsSlices(t *testing.T) {
func TestCluster_ContainsSlices(t *testing.T) {
c := test.NewCluster(5)
c.ReplicaN = 3
slices := c.ContainsSlices("test", 10, test.NewURIFromHostPort("host2", 0))
slices := c.ContainsSlices("test", 10, c.Nodes[2])
if !reflect.DeepEqual(slices, []uint64{0, 2, 3, 5, 6, 9, 10}) {
t.Fatalf("unexpected slices for node's index: %v", slices)
@ -119,15 +119,16 @@ func TestCluster_Nodes(t *testing.T) {
uri2 := test.NewURIFromHostPort("node2", 0)
uri3 := test.NewURIFromHostPort("node3", 0)
nodes := []*pilosa.Node{
{URI: uri0},
{URI: uri1},
{URI: uri2},
}
node0 := &pilosa.Node{ID: "node0", URI: uri0}
node1 := &pilosa.Node{ID: "node1", URI: uri1}
node2 := &pilosa.Node{ID: "node2", URI: uri2}
node3 := &pilosa.Node{ID: "node3", URI: uri3}
t.Run("NodeSet", func(t *testing.T) {
actual := pilosa.Nodes(nodes).URIs()
expected := []pilosa.URI{uri0, uri1, uri2}
nodes := []*pilosa.Node{node0, node1, node2}
t.Run("NodeIDs", func(t *testing.T) {
actual := pilosa.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)
}
@ -150,19 +151,8 @@ func TestCluster_Nodes(t *testing.T) {
})
t.Run("Contains", func(t *testing.T) {
actualTrue := pilosa.Nodes(nodes).Contains(nodes[1])
actualFalse := pilosa.Nodes(nodes).Contains(&pilosa.Node{})
if !reflect.DeepEqual(actualTrue, true) {
t.Errorf("expected: %v, but got: %v", true, actualTrue)
}
if !reflect.DeepEqual(actualFalse, false) {
t.Errorf("expected: %v, but got: %v", false, actualTrue)
}
})
t.Run("ContainsURI", func(t *testing.T) {
actualTrue := pilosa.Nodes(nodes).ContainsURI(uri1)
actualFalse := pilosa.Nodes(nodes).ContainsURI(uri3)
actualTrue := pilosa.Nodes(nodes).Contains(node1)
actualFalse := pilosa.Nodes(nodes).Contains(node3)
if !reflect.DeepEqual(actualTrue, true) {
t.Errorf("expected: %v, but got: %v", true, actualTrue)
}
@ -185,58 +175,66 @@ func TestCluster_Coordinator(t *testing.T) {
uri1 := test.NewURIFromHostPort("node1", 0)
uri2 := test.NewURIFromHostPort("node2", 0)
node1 := &pilosa.Node{ID: "node1", URI: uri1}
node2 := &pilosa.Node{ID: "node2", URI: uri2}
c1 := *pilosa.NewCluster()
c1.URI = uri1
c1.Coordinator = uri1
c1.Node = node1
c1.Coordinator = node1.URI
c2 := *pilosa.NewCluster()
c2.URI = uri2
c2.Coordinator = uri1
c2.Node = node2
c2.Coordinator = node1.URI
t.Run("IsCoordinator", func(t *testing.T) {
if !c1.IsCoordinator() {
t.Errorf("!IsCoordinator error: %v", c1.URI)
t.Errorf("!IsCoordinator error: %v", c1.Node)
} else if c2.IsCoordinator() {
t.Errorf("IsCoordinator error: %v", c2.URI)
t.Errorf("IsCoordinator error: %v", c2.Node)
}
})
}
func TestCluster_Topology(t *testing.T) {
c1 := test.NewCluster(1)
c1 := test.NewCluster(1) // automatically creates Node{ID: "node0"}
uri1 := test.NewURIFromHostPort("node1", 0)
uri2 := test.NewURIFromHostPort("node2", 0)
base := test.NewURIFromHostPort("host0", 0)
uri0 := test.NewURIFromHostPort("host0", 0)
uri1 := test.NewURIFromHostPort("host1", 0)
uri2 := test.NewURIFromHostPort("host2", 0)
invalid := test.NewURIFromHostPort("invalid", 0)
node0 := &pilosa.Node{ID: "node0", URI: uri0}
node1 := &pilosa.Node{ID: "node1", URI: uri1}
node2 := &pilosa.Node{ID: "node2", URI: uri2}
nodeinvalid := &pilosa.Node{ID: "nodeinvalid", URI: invalid}
t.Run("AddNode", func(t *testing.T) {
err := c1.AddNode(uri1)
err := c1.AddNode(node1)
if err != nil {
t.Fatal(err)
}
// add the same host.
err = c1.AddNode(uri1)
err = c1.AddNode(node1)
if err != nil {
t.Fatal(err)
}
err = c1.AddNode(uri2)
err = c1.AddNode(node2)
if err != nil {
t.Fatal(err)
}
actual := c1.NodeSet()
expected := []pilosa.URI{base, uri1, uri2}
actual := c1.NodeIDs()
expected := []string{node0.ID, node1.ID, node2.ID}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
}
})
t.Run("ContainsURI", func(t *testing.T) {
if !c1.Topology.ContainsURI(uri1) {
t.Errorf("!ContainsHost error: %v", uri1)
} else if c1.Topology.ContainsURI(invalid) {
t.Errorf("ContainsHost error: %v", invalid)
t.Run("ContainsID", func(t *testing.T) {
if !c1.Topology.ContainsID(node1.ID) {
t.Errorf("!ContainsHost error: %v", node1.ID)
} else if c1.Topology.ContainsID(nodeinvalid.ID) {
t.Errorf("ContainsHost error: %v", nodeinvalid.ID)
}
})
}
@ -260,12 +258,12 @@ func TestCluster_ResizeStates(t *testing.T) {
}
expectedTop := &pilosa.Topology{
NodeSet: []pilosa.URI{node.URI},
NodeIDs: []string{node.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node.Topology.NodeSet, expectedTop.NodeSet) {
t.Errorf("expected topology: %v, but got: %v", expectedTop.NodeSet, node.Topology.NodeSet)
if !reflect.DeepEqual(node.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected topology: %v, but got: %v", expectedTop.NodeIDs, node.Topology.NodeIDs)
}
// Close TestCluster.
@ -282,7 +280,7 @@ func TestCluster_ResizeStates(t *testing.T) {
// write topology to data file
top := &pilosa.Topology{
NodeSet: []pilosa.URI{node.URI},
NodeIDs: []string{node.Node.ID},
}
tc.WriteTopology(node.Path, top)
@ -310,14 +308,12 @@ func TestCluster_ResizeStates(t *testing.T) {
// write topology to data file
top := &pilosa.Topology{
NodeSet: []pilosa.URI{
test.NewURIFromHostPort("some-other-host", 0),
},
NodeIDs: []string{"some-other-host"},
}
tc.WriteTopology(node.Path, top)
// Open TestCluster.
expected := "considerTopology: coordinator http://host0:0 is not in topology: [http://some-other-host:0]"
expected := "considerTopology: coordinator node0 is not in topology: [some-other-host]"
err := tc.Open()
if err == nil || err.Error() != expected {
t.Errorf("did not receive expected error: %s", expected)
@ -351,14 +347,14 @@ func TestCluster_ResizeStates(t *testing.T) {
}
expectedTop := &pilosa.Topology{
NodeSet: []pilosa.URI{node0.URI, node1.URI},
NodeIDs: []string{node0.Node.ID, node1.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node0.Topology.NodeSet, expectedTop.NodeSet) {
t.Errorf("expected node0 topology: %v, but got: %v", expectedTop.NodeSet, node0.Topology.NodeSet)
} else if !reflect.DeepEqual(node1.Topology.NodeSet, expectedTop.NodeSet) {
t.Errorf("expected node1 topology: %v, but got: %v", expectedTop.NodeSet, node1.Topology.NodeSet)
if !reflect.DeepEqual(node0.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node0 topology: %v, but got: %v", expectedTop.NodeIDs, node0.Topology.NodeIDs)
} else if !reflect.DeepEqual(node1.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node1 topology: %v, but got: %v", expectedTop.NodeIDs, node1.Topology.NodeIDs)
}
// Close TestCluster.
@ -372,13 +368,9 @@ func TestCluster_ResizeStates(t *testing.T) {
tc.AddNode(false)
node0 := tc.Clusters[0]
u0 := test.NewURIFromHostPort("host0", 0)
//u1 := test.NewURIFromHostPort("host1", 0)
u2 := test.NewURIFromHostPort("host2", 0)
// write topology to data file
top := &pilosa.Topology{
NodeSet: []pilosa.URI{u0, u2},
NodeIDs: []string{"node0", "node2"},
}
tc.WriteTopology(node0.Path, top)
@ -393,7 +385,7 @@ func TestCluster_ResizeStates(t *testing.T) {
}
// Expect an error by adding a node not in the topology.
expectedError := "host is not in topology: http://host1:0"
expectedError := "host is not in topology: node1"
err := tc.AddNode(false)
if err == nil || err.Error() != expectedError {
t.Errorf("did not receive expected error: %s", expectedError)
@ -477,14 +469,14 @@ func TestCluster_ResizeStates(t *testing.T) {
}
expectedTop := &pilosa.Topology{
NodeSet: []pilosa.URI{node0.URI, node1.URI},
NodeIDs: []string{node0.Node.ID, node1.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node0.Topology.NodeSet, expectedTop.NodeSet) {
t.Errorf("expected node0 topology: %v, but got: %v", expectedTop.NodeSet, node0.Topology.NodeSet)
} else if !reflect.DeepEqual(node1.Topology.NodeSet, expectedTop.NodeSet) {
t.Errorf("expected node1 topology: %v, but got: %v", expectedTop.NodeSet, node1.Topology.NodeSet)
if !reflect.DeepEqual(node0.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node0 topology: %v, but got: %v", expectedTop.NodeIDs, node0.Topology.NodeIDs)
} else if !reflect.DeepEqual(node1.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node1 topology: %v, but got: %v", expectedTop.NodeIDs, node1.Topology.NodeIDs)
}
// Bits
@ -519,28 +511,23 @@ func TestCluster_ResizeStates(t *testing.T) {
// Ensures that coordinator can be changed.
func TestCluster_SetCoordinator(t *testing.T) {
t.Run("SetCoordinator", func(t *testing.T) {
c := test.NewCluster(1)
oldURI, err := pilosa.NewURIFromAddress("localhost:8888")
if err != nil {
t.Fatal(err)
}
c.Coordinator = *oldURI
c := test.NewCluster(2)
newURI, err := pilosa.NewURIFromAddress("localhost:9999")
if err != nil {
t.Fatal(err)
}
oldNode := c.Nodes[0]
newNode := c.Nodes[1]
// Set coordinator to the same value.
c.SetCoordinator(c.Coordinator, *oldURI)
if c.Coordinator != *oldURI {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, *oldURI)
if c.SetCoordinator(oldNode) {
t.Errorf("did not expect coordinator to change")
} else if c.Coordinator != oldNode.URI {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, oldNode.URI)
}
// Set coordinator to a new value.
c.SetCoordinator(c.Coordinator, *newURI)
if c.Coordinator != *newURI {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, *newURI)
if !c.SetCoordinator(newNode) {
t.Errorf("expected coordinator to change")
} else if c.Coordinator != newNode.URI {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, newNode.URI)
}
})
}

4
ctl/backup_test.go
View file

@ -50,7 +50,9 @@ func TestBackupCommand_Run(t *testing.T) {
if err != nil {
t.Fatal(err)
}
s.Handler.URI = *uri
node := &pilosa.Node{ID: "node", URI: *uri}
s.Handler.Node = node
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = *uri
s.Handler.Holder = hldr.Holder

6
ctl/export_test.go
View file

@ -63,9 +63,11 @@ func TestExportCommand_Run(t *testing.T) {
if err != nil {
t.Fatal(err)
}
s.Handler.URI = *uri
node := &pilosa.Node{ID: "node", URI: *uri}
s.Handler.Node = node
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = *uri
s.Handler.Cluster.Nodes[0] = node
s.Handler.Holder = hldr.Holder
cm.Host = s.Host()

12
ctl/import_test.go
View file

@ -69,9 +69,11 @@ func TestImportCommand_Run(t *testing.T) {
if err != nil {
t.Fatal(err)
}
s.Handler.URI = *uri
node := &pilosa.Node{ID: "node", URI: *uri}
s.Handler.Node = node
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = *uri
s.Handler.Cluster.Nodes[0] = node
s.Handler.Holder = hldr.Holder
cm.Host = s.Host()
@ -109,9 +111,11 @@ func TestImportCommand_RunValue(t *testing.T) {
if err != nil {
t.Fatal(err)
}
s.Handler.URI = *uri
node := &pilosa.Node{ID: "node", URI: *uri}
s.Handler.Node = node
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = *uri
s.Handler.Cluster.Nodes[0] = node
s.Handler.Holder = hldr.Holder
cm.Host = s.Host()

4
ctl/restore_test.go
View file

@ -52,7 +52,9 @@ func TestRestoreCommand_Run(t *testing.T) {
if err != nil {
t.Fatal(err)
}
s.Handler.URI = *uri
node := &pilosa.Node{ID: "node", URI: *uri}
s.Handler.Node = node
s.Handler.Cluster = test.NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = *uri
s.Handler.Holder = hldr.Holder

2
event.go
View file

@ -27,7 +27,7 @@ const (
// NodeEvent is a single event related to node activity in the cluster.
type NodeEvent struct {
Event NodeEventType
URI URI
Node *Node
}
// EventHandler is the interface for the pilosa object which knows how to

18
executor.go
View file

@ -40,7 +40,7 @@ type Executor struct {
Holder *Holder
// Local hostname & cluster configuration.
URI URI
Node *Node
Cluster *Cluster
// Client used for remote requests.
@ -956,7 +956,7 @@ func (e *Executor) executeClearBitView(ctx context.Context, index string, c *pql
ret := false
for _, node := range e.Cluster.FragmentNodes(index, slice) {
// Update locally if host matches.
if node.URI == e.URI {
if node.ID == e.Node.ID {
val, err := f.ClearBit(view, rowID, colID, nil)
if err != nil {
return false, err
@ -1061,7 +1061,7 @@ func (e *Executor) executeSetBitView(ctx context.Context, index string, c *pql.C
for _, node := range e.Cluster.FragmentNodes(index, slice) {
// Update locally if host matches.
if node.URI == e.URI {
if node.ID == e.Node.ID {
val, err := f.SetBit(view, rowID, colID, timestamp)
if err != nil {
return false, err
@ -1140,7 +1140,7 @@ func (e *Executor) executeSetFieldValue(ctx context.Context, index string, c *pq
}
// Execute on remote nodes in parallel.
nodes := Nodes(e.Cluster.Nodes).FilterURI(e.URI)
nodes := Nodes(e.Cluster.Nodes).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *Node) {
@ -1198,7 +1198,7 @@ func (e *Executor) executeSetRowAttrs(ctx context.Context, index string, c *pql.
}
// Execute on remote nodes in parallel.
nodes := Nodes(e.Cluster.Nodes).FilterURI(e.URI)
nodes := Nodes(e.Cluster.Nodes).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *Node) {
@ -1285,7 +1285,7 @@ func (e *Executor) executeBulkSetRowAttrs(ctx context.Context, index string, cal
}
// Execute on remote nodes in parallel.
nodes := Nodes(e.Cluster.Nodes).FilterURI(e.URI)
nodes := Nodes(e.Cluster.Nodes).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *Node) {
@ -1344,7 +1344,7 @@ func (e *Executor) executeSetColumnAttrs(ctx context.Context, index string, c *p
}
// Execute on remote nodes in parallel.
nodes := Nodes(e.Cluster.Nodes).FilterURI(e.URI)
nodes := Nodes(e.Cluster.Nodes).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *Node) {
@ -1451,7 +1451,7 @@ func (e *Executor) mapReduce(ctx context.Context, index string, slices []uint64,
if !opt.Remote {
nodes = Nodes(e.Cluster.Nodes).Clone()
} else {
nodes = []*Node{e.Cluster.NodeByURI(e.URI)}
nodes = []*Node{e.Cluster.nodeByID(e.Node.ID)}
}
// Start mapping across all primary owners.
@ -1507,7 +1507,7 @@ func (e *Executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Nod
resp := mapResponse{node: n, slices: nodeSlices}
// Send local slices to mapper, otherwise remote exec.
if n.URI == e.URI {
if n.ID == e.Node.ID {
resp.result, resp.err = e.mapperLocal(ctx, nodeSlices, mapFn, reduceFn)
} else if !opt.Remote {
results, err := e.remoteExec(ctx, n, index, &pql.Query{Calls: []*pql.Call{c}}, nodeSlices, opt)

6
fragment.go
View file

@ -1685,7 +1685,7 @@ func (h *blockHasher) WriteValue(v uint64) {
type FragmentSyncer struct {
Fragment *Fragment
URI URI
Node *Node
Cluster *Cluster
RemoteClient *http.Client
@ -1715,7 +1715,7 @@ func (s *FragmentSyncer) SyncFragment() error {
blockSets := make([][]FragmentBlock, 0, len(nodes))
for _, node := range nodes {
// Read local blocks.
if node.URI == s.URI {
if node.ID == s.Node.ID {
b := s.Fragment.Blocks()
blockSets = append(blockSets, b)
continue
@ -1789,7 +1789,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
var pairSets []PairSet
var clients []InternalClient
for _, node := range s.Cluster.FragmentNodes(f.Index(), f.Slice()) {
if s.URI == node.URI {
if s.Node.ID == node.ID {
continue
}

40
gossip/gossip.go
View file

@ -43,6 +43,7 @@ var _ memberlist.Delegate = &GossipMemberSet{}
// GossipMemberSet represents a gossip implementation of MemberSet using memberlist.
type GossipMemberSet struct {
mu sync.RWMutex
node *pilosa.Node
memberlist *memberlist.Memberlist
handler pilosa.BroadcastHandler
@ -55,20 +56,6 @@ type GossipMemberSet struct {
LogOutput io.Writer
}
// Nodes implements the MemberSet interface and returns a list of nodes in the cluster.
func (g *GossipMemberSet) Nodes() []*pilosa.Node {
g.mu.RLock()
defer g.mu.RUnlock()
a := make([]*pilosa.Node, 0, g.memberlist.NumMembers())
for _, n := range g.memberlist.Members() {
uri, _ := pilosa.NewURIFromAddress(n.Name)
// TODO don't swallow the error above
a = append(a, &pilosa.Node{URI: *uri})
}
return a
}
// Start implements the BroadcastReceiver interface and sets the BroadcastHandler.
func (g *GossipMemberSet) Start(h pilosa.BroadcastHandler) error {
g.handler = h
@ -81,11 +68,13 @@ func (g *GossipMemberSet) Seed() string {
}
// Open implements the MemberSet interface to start network activity.
func (g *GossipMemberSet) Open() error {
func (g *GossipMemberSet) Open(n *pilosa.Node) error {
if g.handler == nil {
return fmt.Errorf("must call Start(pilosa.BroadcastHandler) before calling Open()")
}
g.node = n
err := error(nil)
g.mu.Lock()
g.memberlist, err = memberlist.Create(g.config.memberlistConfig)
@ -112,7 +101,7 @@ func (g *GossipMemberSet) Open() error {
nodes := []*pilosa.Node{&pilosa.Node{URI: *uri}} //TODO: support a list of seeds
g.mu.RLock()
err = g.joinWithRetry(pilosa.NodeSet(pilosa.Nodes(nodes).URIs()).ToHostPortStrings())
err = g.joinWithRetry(pilosa.URIs(pilosa.Nodes(nodes).URIs()).HostPortStrings())
g.mu.RUnlock()
if err != nil {
return errors.Wrap(err, "joinWithRetry")
@ -280,7 +269,12 @@ func (g *GossipMemberSet) SendAsync(pb proto.Message) error {
// NodeMeta implementation of the memberlist.Delegate interface.
func (g *GossipMemberSet) NodeMeta(limit int) []byte {
return []byte{}
buf, err := proto.Marshal(pilosa.EncodeNode(g.node))
if err != nil {
g.logger().Printf("marshal message error: %s", err)
return []byte{}
}
return buf
}
// NotifyMsg implementation of the memberlist.Delegate interface
@ -387,14 +381,18 @@ func (g *GossipEventReceiver) listen() {
continue
}
uri, _ := pilosa.NewURIFromAddress(e.Node.Name)
// TODO: don't swallow this error
// Get the node from the event.Node meta data.
var n internal.Node
if err := proto.Unmarshal(e.Node.Meta, &n); err != nil {
panic("failed to unmarshal event node meta data")
}
node := pilosa.DecodeNode(&n)
ne := &pilosa.NodeEvent{
Event: nodeEventType,
URI: *uri,
Node: node,
}
_ = g.eventHandler.ReceiveEvent(ne) // TODO: don't swallow this error
_ = g.eventHandler.ReceiveEvent(ne)
}
}

77
handler.go
View file

@ -57,7 +57,7 @@ type Handler struct {
StatusHandler StatusHandler
// Local hostname & cluster configuration.
URI URI
Node *Node
Cluster *Cluster
RemoteClient *http.Client
@ -268,8 +268,8 @@ func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
cs := pb.(*internal.ClusterStatus)
if err := json.NewEncoder(w).Encode(getStatusResponse{
State: cs.State,
NodeSet: decodeURIs(cs.NodeSet),
State: cs.State,
Nodes: DecodeNodes(cs.Nodes),
}); err != nil {
h.logger().Printf("write status response error: %s", err)
}
@ -280,8 +280,8 @@ type getSchemaResponse struct {
}
type getStatusResponse struct {
State string `json:"state"`
NodeSet []URI `json:"nodes"`
State string `json:"state"`
Nodes []*Node `json:"nodes"`
}
// handlePostQuery handles /query requests.
@ -1206,8 +1206,8 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
}
// Validate that this handler owns the slice.
if !h.Cluster.OwnsFragment(h.URI, req.Index, req.Slice) {
msg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.URI, req.Index, req.Slice)
if !h.Cluster.OwnsFragment(h.Node.ID, req.Index, req.Slice) {
msg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Node.ID, req.Index, req.Slice)
http.Error(w, msg, http.StatusPreconditionFailed)
return
}
@ -1276,8 +1276,8 @@ func (h *Handler) handlePostImportValue(w http.ResponseWriter, r *http.Request)
}
// Validate that this handler owns the slice.
if !h.Cluster.OwnsFragment(h.URI, req.Index, req.Slice) {
msg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.URI, req.Index, req.Slice)
if !h.Cluster.OwnsFragment(h.Node.ID, req.Index, req.Slice) {
msg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Node.ID, req.Index, req.Slice)
http.Error(w, msg, http.StatusPreconditionFailed)
return
}
@ -1342,8 +1342,8 @@ func (h *Handler) handleGetExportCSV(w http.ResponseWriter, r *http.Request) {
}
// Validate that this handler owns the slice.
if !h.Cluster.OwnsFragment(h.URI, index, slice) {
msg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.URI, index, slice)
if !h.Cluster.OwnsFragment(h.Node.ID, index, slice) {
msg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Node.ID, index, slice)
http.Error(w, msg, http.StatusPreconditionFailed)
return
}
@ -1570,7 +1570,7 @@ func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request)
// Loop over each slice and import it if this node owns it.
for slice := uint64(0); slice <= maxSlices[indexName]; slice++ {
// Ignore this slice if we don't own it.
if !h.Cluster.OwnsFragment(h.URI, indexName, slice) {
if !h.Cluster.OwnsFragment(h.Node.ID, indexName, slice) {
continue
}
@ -1964,31 +1964,24 @@ func (h *Handler) handlePostClusterResizeSetCoordinator(w http.ResponseWriter, r
return
}
oldURI := h.Cluster.Coordinator
newNode := h.Cluster.nodeByID(req.ID)
if newNode == nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
var newURI *URI
if err := func() error {
newURI, err = NewURIFromAddress(req.Address)
if err != nil {
return fmt.Errorf("problem with set-coordinator address: %s", err)
}
//if !Nodes(h.Cluster.Nodes).ContainsURI(*newURI) {
// return fmt.Errorf("set-coordinator node does not exist: %s", newURI)
//}
// Send the set-coordinator message to all nodes.
err := h.Broadcaster.SendSync(
&internal.SetCoordinatorMessage{
Old: (&h.Cluster.Coordinator).Encode(),
New: newURI.Encode(),
New: EncodeNode(newNode),
})
if err != nil {
return fmt.Errorf("problem sending SetCoordinator message: %s", err)
}
// Set Coordinator on local node.
h.Cluster.SetCoordinator(oldURI, *newURI)
_ = h.Cluster.SetCoordinator(newNode)
return nil
}(); err != nil {
@ -1998,20 +1991,19 @@ func (h *Handler) handlePostClusterResizeSetCoordinator(w http.ResponseWriter, r
// Encode response.
if err := json.NewEncoder(w).Encode(setCoordinatorResponse{
Old: &oldURI,
New: newURI,
New: newNode,
}); err != nil {
h.logger().Printf("response encoding error: %s", err)
}
}
type setCoordinatorRequest struct {
Address string `json:"address"`
ID string `json:"id"`
}
type setCoordinatorResponse struct {
Old *URI `json:"old"`
New *URI `json:"new"`
Old *Node `json:"old"`
New *Node `json:"new"`
}
// handlePostClusterResizeRemoveNode handles POST /cluster/resize/remove-node request.
@ -2024,19 +2016,16 @@ func (h *Handler) handlePostClusterResizeRemoveNode(w http.ResponseWriter, r *ht
return
}
var removeURI *URI
removeNode := h.Cluster.nodeByID(req.ID)
if removeNode == nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
if err := func() error {
removeURI, err = NewURIFromAddress(req.Address)
if err != nil {
return fmt.Errorf("problem with remove node address: %s", err)
}
// TODO: make sure the address is in the cluster
// TODO: prevent removing the coordinator node
// Start the resize process (similar to NodeJoin)
err := h.Cluster.NodeLeave(*removeURI)
err := h.Cluster.NodeLeave(removeNode)
if err != nil {
return err
}
@ -2049,18 +2038,18 @@ func (h *Handler) handlePostClusterResizeRemoveNode(w http.ResponseWriter, r *ht
// Encode response.
if err := json.NewEncoder(w).Encode(removeNodeResponse{
Remove: removeURI,
Remove: removeNode,
}); err != nil {
h.logger().Printf("response encoding error: %s", err)
}
}
type removeNodeRequest struct {
Address string `json:"address"`
ID string `json:"id"`
}
type removeNodeResponse struct {
Remove *URI `json:"remove"`
Remove *Node `json:"remove"`
}
// handlePostClusterResizeAbort handles POST /cluster/resize/abort request.
@ -2221,7 +2210,7 @@ func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Reques
}
func (h *Handler) handleGetID(w http.ResponseWriter, r *http.Request) {
_, err := w.Write([]byte(h.Holder.NodeID))
_, err := w.Write([]byte(h.Cluster.Node.ID))
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
}

6
handler_test.go
View file

@ -147,7 +147,7 @@ func TestHandler_Status(t *testing.T) {
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/status", nil))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"state":"NORMAL","nodes":[{"scheme":"http","host":"localhost","port":10101}]}`+"\n" {
} else if body := w.Body.String(); body != `{"state":"NORMAL","nodes":[{"id":"test-node","uri":{"scheme":"http","host":"localhost","port":10101}}]}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}
@ -1215,8 +1215,8 @@ func TestHandler_Fragment_Nodes(t *testing.T) {
h.ServeHTTP(w, r)
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if w.Body.String() != `[{"uri":{"scheme":"http","host":"host2"}},{"uri":{"scheme":"http","host":"host0"}}]`+"\n" {
t.Fatalf("unexpected body: %q", w.Body.String())
} else if body := w.Body.String(); body != `[{"id":"node2","uri":{"scheme":"http","host":"host2"}},{"id":"node0","uri":{"scheme":"http","host":"host0"}}]`+"\n" {
t.Fatalf("unexpected body: %q", body)
}
}

54
holder.go
View file

@ -69,8 +69,6 @@ type Holder struct {
CacheFlushInterval time.Duration
LogOutput io.Writer
NodeID string
}
// NewHolder returns a new instance of Holder.
@ -528,25 +526,29 @@ func (h *Holder) setFileLimit() {
func (h *Holder) logger() *log.Logger { return log.New(h.LogOutput, "", log.LstdFlags) }
func (h *Holder) loadNodeID() error {
func (h *Holder) loadNodeID() (string, error) {
idPath := path.Join(h.Path, "ID")
nodeID := ""
nodeIDBytes, err := ioutil.ReadFile(idPath)
if err == nil {
h.NodeID = strings.TrimSpace(string(nodeIDBytes))
} else if os.IsNotExist(err) {
u := uuid.NewV4()
nodeID = u.String()
err = ioutil.WriteFile(idPath, []byte(nodeID), 0600)
if err != nil {
return err
}
h.NodeID = nodeID
} else if err != nil {
return err
h.logger().Printf("load NodeID: %s", idPath)
if err := os.MkdirAll(h.Path, 0777); err != nil {
return "", err
}
return nil
nodeIDBytes, err := ioutil.ReadFile(idPath)
if err == nil {
nodeID = strings.TrimSpace(string(nodeIDBytes))
} else if os.IsNotExist(err) {
nodeID = uuid.NewV4().String()
err = ioutil.WriteFile(idPath, []byte(nodeID), 0600)
if err != nil {
return "", err
}
} else if err != nil {
return "", err
}
return nodeID, nil
}
// HolderSyncer is an active anti-entropy tool that compares the local holder
@ -554,7 +556,7 @@ func (h *Holder) loadNodeID() error {
type HolderSyncer struct {
Holder *Holder
URI URI
Node *Node
Cluster *Cluster
RemoteClient *http.Client
@ -610,7 +612,7 @@ func (s *HolderSyncer) SyncHolder() error {
for slice := uint64(0); slice <= s.Holder.Index(di.Name).MaxSlice(); slice++ {
// Ignore slices that this host doesn't own.
if !s.Cluster.OwnsFragment(s.URI, di.Name, slice) {
if !s.Cluster.OwnsFragment(s.Node.ID, di.Name, slice) {
continue
}
@ -652,7 +654,7 @@ func (s *HolderSyncer) syncIndex(index string) error {
s.Stats.CountWithCustomTags("ColumnAttrStoreBlocks", int64(len(blks)), 1.0, []string{indexTag})
// Sync with every other host.
for _, node := range Nodes(s.Cluster.Nodes).FilterURI(s.URI) {
for _, node := range Nodes(s.Cluster.Nodes).FilterID(s.Node.ID) {
client := NewInternalHTTPClientFromURI(&node.URI, s.RemoteClient)
// Retrieve attributes from differing blocks.
@ -663,7 +665,7 @@ func (s *HolderSyncer) syncIndex(index string) error {
} else if len(m) == 0 {
continue
}
s.Stats.CountWithCustomTags("ColumnAttrDiff", int64(len(m)), 1.0, []string{indexTag, node.URI.HostPort()})
s.Stats.CountWithCustomTags("ColumnAttrDiff", int64(len(m)), 1.0, []string{indexTag, node.ID})
// Update local copy.
if err := idx.ColumnAttrStore().SetBulkAttrs(m); err != nil {
@ -698,7 +700,7 @@ func (s *HolderSyncer) syncFrame(index, name string) error {
s.Stats.CountWithCustomTags("RowAttrStoreBlocks", int64(len(blks)), 1.0, []string{indexTag, frameTag})
// Sync with every other host.
for _, node := range Nodes(s.Cluster.Nodes).FilterURI(s.URI) {
for _, node := range Nodes(s.Cluster.Nodes).FilterID(s.Node.ID) {
client := NewInternalHTTPClientFromURI(&node.URI, s.RemoteClient)
// Retrieve attributes from differing blocks.
@ -711,7 +713,7 @@ func (s *HolderSyncer) syncFrame(index, name string) error {
} else if len(m) == 0 {
continue
}
s.Stats.CountWithCustomTags("RowAttrDiff", int64(len(m)), 1.0, []string{indexTag, frameTag, node.URI.HostPort()})
s.Stats.CountWithCustomTags("RowAttrDiff", int64(len(m)), 1.0, []string{indexTag, frameTag, node.ID})
// Update local copy.
if err := f.RowAttrStore().SetBulkAttrs(m); err != nil {
@ -751,7 +753,7 @@ func (s *HolderSyncer) syncFragment(index, frame, view string, slice uint64) err
// Sync fragments together.
fs := FragmentSyncer{
Fragment: frag,
URI: s.URI,
Node: s.Node,
Cluster: s.Cluster,
Closing: s.Closing,
RemoteClient: s.RemoteClient,
@ -765,7 +767,7 @@ func (s *HolderSyncer) syncFragment(index, frame, view string, slice uint64) err
// HolderCleaner removes fragments and data files that are no longer used.
type HolderCleaner struct {
URI URI
Node *Node
Holder *Holder
Cluster *Cluster
@ -794,7 +796,7 @@ func (c *HolderCleaner) CleanHolder() error {
}
// Get the fragments that node is responsible for (based on hash(index, node)).
containedSlices := c.Cluster.ContainsSlices(index.Name(), index.MaxSlice(), c.URI)
containedSlices := c.Cluster.ContainsSlices(index.Name(), index.MaxSlice(), c.Node)
// Get the fragments registered in memory.
for _, frame := range index.Frames() {

8
holder_test.go
View file

@ -419,7 +419,7 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
s.Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor(client)
e.Holder = hldr1.Holder
e.URI = cluster.Nodes[1].URI
e.Node = cluster.Nodes[1]
e.Cluster = cluster
return e.Execute(ctx, index, query, slices, opt)
}
@ -487,7 +487,7 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
// Set up syncer.
syncer := pilosa.HolderSyncer{
Holder: hldr0.Holder,
URI: cluster.Nodes[0].URI,
Node: cluster.Nodes[0],
Cluster: cluster,
RemoteClient: pilosa.GetHTTPClient(nil),
Stats: pilosa.NopStatsClient,
@ -586,7 +586,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) {
// Set up cleaner for replication 2.
cleaner2 := pilosa.HolderCleaner{
URI: cluster.Nodes[0].URI,
Node: cluster.Nodes[0],
Holder: hldr0.Holder,
Cluster: cluster,
}
@ -629,7 +629,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) {
// Set up cleaner for replication 1.
cleaner1 := pilosa.HolderCleaner{
URI: cluster.Nodes[0].URI,
Node: cluster.Nodes[0],
Holder: hldr0.Holder,
Cluster: cluster,
}

802
internal/private.pb.go
View file

File diff suppressed because it is too large Load diff

30
internal/private.proto
View file

@ -121,21 +121,26 @@ message URI {
uint32 Port = 3;
}
message Node {
string ID = 1;
URI URI = 2;
}
message NodeStateMessage {
URI URI = 1;
string NodeID = 1;
string State = 2;
}
message NodeStatus {
URI URI = 1;
Node Node = 1;
MaxSlices MaxSlices = 2;
Schema Schema = 3;
}
message ClusterStatus {
string State = 1;
repeated URI NodeSet = 2;
string ClusterID = 3;
string ClusterID = 1;
string State = 2;
repeated Node Nodes = 3;
}
message Field {
@ -159,15 +164,15 @@ message DeleteViewMessage {
message ResizeInstruction {
int64 JobID = 1;
URI URI = 2;
URI Coordinator = 3;
Node Node = 2;
Node Coordinator = 3;
repeated ResizeSource Sources = 4;
Schema Schema = 5;
ClusterStatus ClusterStatus = 6;
}
message ResizeSource {
URI URI = 1;
Node Node = 1;
string Index = 2;
string Frame = 3;
string View = 4;
@ -176,17 +181,16 @@ message ResizeSource {
message ResizeInstructionComplete {
int64 JobID = 1;
URI URI = 2;
Node Node = 2;
string Error = 3;
}
message SetCoordinatorMessage {
URI Old = 1;
URI New = 2;
Node New = 1;
}
message Topology {
repeated URI NodeSet = 1;
string ClusterID = 2;
string ClusterID = 1;
repeated string NodeIDs = 2;
}

4
pilosa.go
View file

@ -162,8 +162,8 @@ func StringInSlice(a string, list []string) bool {
return false
}
// URISlicesAreEqual determines if two string slices are equal.
func URISlicesAreEqual(a, b []URI) bool {
// StringSlicesAreEqual determines if two string slices are equal.
func StringSlicesAreEqual(a, b []string) bool {
if a == nil && b == nil {
return true

64
server.go
View file

@ -58,9 +58,6 @@ type Server struct {
wg sync.WaitGroup
closing chan struct{}
// Unique name identifying the server.
Name string
// Data storage and HTTP interface.
Holder *Holder
Handler *Handler
@ -70,8 +67,8 @@ type Server struct {
RemoteClient *http.Client
// Cluster configuration.
// Host is replaced with actual host after opening if port is ":0".
Network string
NodeID string
URI URI
Cluster *Cluster
diagnostics *diagnostics.Diagnostics
@ -128,16 +125,15 @@ func (s *Server) Open() error {
}
}
// Set Cluster URI.
s.Cluster.URI = s.URI
// Get or create NodeID.
s.NodeID = s.LoadNodeID()
// Find the Node ID and append that tag to stats.
for i, n := range s.Cluster.Nodes {
if n.URI == s.URI {
s.Holder.Stats = s.Holder.Stats.WithTags(fmt.Sprintf("NodeID:%d", i))
break
}
}
// Set Cluster Node.
node := &Node{ID: s.NodeID, URI: s.URI}
s.Cluster.Node = node
// Append the NodeID tag to stats.
s.Holder.Stats = s.Holder.Stats.WithTags(fmt.Sprintf("NodeID:%s", s.NodeID))
// Peek at the holder to determine if there is data on disk.
// Don't actually load the data until after the Cluster
@ -151,7 +147,7 @@ func (s *Server) Open() error {
// Create executor for executing queries.
e := NewExecutor(s.RemoteClient)
e.Holder = s.Holder
e.URI = s.URI
e.Node = node
e.Cluster = s.Cluster
e.MaxWritesPerRequest = s.MaxWritesPerRequest
@ -163,7 +159,7 @@ func (s *Server) Open() error {
s.Handler.Broadcaster = s.Broadcaster
s.Handler.BroadcastHandler = s
s.Handler.StatusHandler = s
s.Handler.URI = s.URI
s.Handler.Node = node
s.Handler.Cluster = s.Cluster
s.Handler.Executor = e
s.Handler.LogOutput = s.LogOutput
@ -211,11 +207,6 @@ func (s *Server) Open() error {
// buffered channel.
s.Cluster.ListenForJoins()
// Load NodeID.
if err := s.Holder.loadNodeID(); err != nil {
s.Logger().Println(err)
}
// Start background monitoring.
s.wg.Add(3)
go func() { defer s.wg.Done(); s.monitorAntiEntropy() }()
@ -259,11 +250,6 @@ func (s *Server) OpenListener() error {
s.URI.SetPort(uint16(s.ln.Addr().(*net.TCPAddr).Port))
}
// If name is not provided in the config, default to the URI.
if s.Name == "" {
s.Name = s.URI.String()
}
return nil
}
@ -286,6 +272,20 @@ func (s *Server) Close() error {
return nil
}
// LoadNodeID gets NodeID from disk, or creates a new value.
// If server.NodeID is already set, a new ID is not created.
func (s *Server) LoadNodeID() string {
if s.NodeID != "" {
return s.NodeID
}
nodeID, err := s.Holder.loadNodeID()
if err != nil {
s.Logger().Printf("loading NodeID: %v", err)
return s.NodeID
}
return nodeID
}
// Addr returns the address of the listener.
func (s *Server) Addr() net.Addr {
if s.ln == nil {
@ -336,7 +336,7 @@ func (s *Server) monitorAntiEntropy() {
// Initialize syncer with local holder and remote client.
var syncer HolderSyncer
syncer.Holder = s.Holder
syncer.URI = s.URI
syncer.Node = s.Cluster.Node
syncer.Cluster = s.Cluster
syncer.Closing = s.closing
syncer.RemoteClient = s.RemoteClient
@ -442,9 +442,9 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
return err
}
case *internal.SetCoordinatorMessage:
s.Cluster.SetCoordinator(DecodeURI(obj.Old), DecodeURI(obj.New))
s.Cluster.SetCoordinator(DecodeNode(obj.New))
case *internal.NodeStateMessage:
err := s.Cluster.ReceiveNodeState(DecodeURI(obj.URI), obj.State)
err := s.Cluster.ReceiveNodeState(obj.NodeID, obj.State)
if err != nil {
return err
}
@ -503,7 +503,7 @@ func (s *Server) LocalStatus() (proto.Message, error) {
}
ns := internal.NodeStatus{
URI: encodeURI(s.URI),
Node: EncodeNode(s.Cluster.Node),
MaxSlices: s.Holder.EncodeMaxSlices(),
Schema: s.Holder.EncodeSchema(),
}
@ -538,7 +538,7 @@ func (s *Server) HandleRemoteStatus(pb proto.Message) error {
func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
// Ignore status updates from self.
if s.URI == decodeURI(ns.URI) {
if s.NodeID == DecodeNode(ns.Node).ID {
return nil
}
@ -594,10 +594,10 @@ func (s *Server) monitorDiagnostics() {
s.diagnostics.SetInterval(s.DiagnosticInterval)
s.diagnostics.Open()
s.diagnostics.Set("Host", s.URI.host)
s.diagnostics.Set("Cluster", strings.Join(NodeSet(s.Cluster.NodeSet()).ToStrings(), ","))
s.diagnostics.Set("Cluster", strings.Join(s.Cluster.NodeIDs(), ","))
s.diagnostics.Set("NumNodes", len(s.Cluster.Nodes))
s.diagnostics.Set("NumCPU", runtime.NumCPU())
s.diagnostics.Set("NodeID", s.Holder.NodeID)
s.diagnostics.Set("NodeID", s.NodeID)
s.diagnostics.Set("ClusterID", s.Cluster.ID)
s.diagnostics.EnrichWithOSInfo()

6
server/cluster_test.go
View file

@ -38,8 +38,8 @@ func TestMain_SendReceiveMessage(t *testing.T) {
// Update cluster config
m0.Server.Cluster.Nodes = []*pilosa.Node{
{URI: m0.Server.URI},
{URI: m1.Server.URI},
{ID: m0.Server.NodeID, URI: m0.Server.URI},
{ID: m1.Server.NodeID, URI: m1.Server.URI},
}
m1.Server.Cluster.Nodes = m0.Server.Cluster.Nodes
@ -50,7 +50,7 @@ func TestMain_SendReceiveMessage(t *testing.T) {
m0.Config.Gossip.Seed = ""
m0.Server.Cluster.Coordinator = m0.Server.URI
m0.Server.Cluster.Topology = &pilosa.Topology{NodeSet: []pilosa.URI{m0.Server.URI, m1.Server.URI}}
m0.Server.Cluster.Topology = &pilosa.Topology{NodeIDs: []string{m0.Server.NodeID, m1.Server.NodeID}}
m0.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver()
gossipMemberSet0, err := gossip.NewGossipMemberSet(m0.Server.URI.HostPort(), m0.Config, m0.Server)
if err != nil {

13
server/server.go
View file

@ -128,11 +128,6 @@ func (m *Command) SetupServer() error {
}
m.Server.URI = *uri
// If using a dynamically allocated port, server.Name will get set later.
if m.Config.Bind != "localhost:0" {
m.Server.Name = m.Server.URI.String()
}
cluster := pilosa.NewCluster()
cluster.ReplicaN = m.Config.Cluster.ReplicaN
cluster.Holder = m.Server.Holder
@ -218,7 +213,6 @@ func (m *Command) SetupServer() error {
func (m *Command) SetupNetworking() error {
switch m.Config.Cluster.Type {
case pilosa.ClusterGossip:
// Set internal port (string).
gossipPortStr := pilosa.DefaultGossipPort
// Config.GossipPort is deprecated, so Config.Gossip.Port has priority
@ -245,11 +239,10 @@ func (m *Command) SetupNetworking() error {
}
}
m.Server.NodeID = m.Server.LoadNodeID()
m.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver()
if m.Server.Name == "" {
return fmt.Errorf("must provide a valid name for gossip membership")
}
gossipMemberSet, err := gossip.NewGossipMemberSetWithTransport(m.Server.Name, m.Config, transport, m.Server)
gossipMemberSet, err := gossip.NewGossipMemberSetWithTransport(m.Server.NodeID, m.Config, transport, m.Server)
if err != nil {
return err
}

54
test/cluster.go
View file

@ -6,7 +6,6 @@ import (
"fmt"
"io/ioutil"
"path/filepath"
"sort"
"sync"
"time"
@ -30,10 +29,14 @@ func NewCluster(n int) *pilosa.Cluster {
for i := 0; i < n; i++ {
c.Nodes = append(c.Nodes, &pilosa.Node{
ID: fmt.Sprintf("node%d", i),
URI: NewURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})
}
c.Node = c.Nodes[0]
c.Coordinator = c.Nodes[0].URI
return c
}
@ -84,7 +87,7 @@ type TestCluster struct {
}
type commonClusterSettings struct {
NodeSet pilosa.NodeSet
Nodes []*pilosa.Node
}
func (t *TestCluster) CreateIndex(name string) error {
@ -115,7 +118,7 @@ func (t *TestCluster) SetBit(index, frame, view string, rowID, colID uint64, x *
nodes := c0.FragmentNodes(index, slice)
for _, node := range nodes {
c := t.clusterByURI(node.URI)
c := t.clusterByID(node.ID)
if c == nil {
continue
}
@ -139,7 +142,7 @@ func (t *TestCluster) SetFieldValue(index, frame string, columnID uint64, name s
nodes := c0.FragmentNodes(index, slice)
for _, node := range nodes {
c := t.clusterByURI(node.URI)
c := t.clusterByID(node.ID)
if c == nil {
continue
}
@ -156,9 +159,9 @@ func (t *TestCluster) SetFieldValue(index, frame string, columnID uint64, name s
return nil
}
func (t *TestCluster) clusterByURI(uri pilosa.URI) *pilosa.Cluster {
func (t *TestCluster) clusterByID(id string) *pilosa.Cluster {
for _, c := range t.Clusters {
if c.URI == uri {
if c.Node.ID == id {
return c
}
}
@ -179,7 +182,7 @@ func (t *TestCluster) AddNode(saveTopology bool) error {
coord := t.Clusters[0]
ev := &pilosa.NodeEvent{
Event: pilosa.NodeJoin,
URI: c.URI,
Node: c.Node,
}
if err := coord.ReceiveEvent(ev); err != nil {
@ -211,11 +214,20 @@ func (t *TestCluster) WriteTopology(path string, top *pilosa.Topology) error {
func (t *TestCluster) addCluster(i int, saveTopology bool) (*pilosa.Cluster, error) {
id := fmt.Sprintf("node%d", i)
uri := NewURI("http", fmt.Sprintf("host%d", i), uint16(0))
node := &pilosa.Node{
ID: id,
URI: uri,
}
// add URI to common
t.common.NodeSet = append(t.common.NodeSet, uri)
sort.Sort(t.common.NodeSet)
//t.common.NodeIDs = append(t.common.NodeIDs, id)
//sort.Sort(t.common.NodeIDs)
// add node to common
t.common.Nodes = append(t.common.Nodes, node)
// create node-specific temp directory
path, err := ioutil.TempDir("", fmt.Sprintf("pilosa-cluster-node-%d-", i))
@ -235,14 +247,14 @@ func (t *TestCluster) addCluster(i int, saveTopology bool) (*pilosa.Cluster, err
c.Topology = pilosa.NewTopology()
c.Holder = h
c.MemberSet = pilosa.NewStaticMemberSet()
c.URI = uri
c.Coordinator = t.common.NodeSet[0] // the first node is the coordinator
c.Node = node
c.Coordinator = t.common.Nodes[0].URI // the first node is the coordinator
c.Broadcaster = t
// add nodes
if saveTopology {
for _, u := range t.common.NodeSet {
c.AddNode(u)
for _, n := range t.common.Nodes {
c.AddNode(n)
}
}
@ -344,7 +356,7 @@ func (t *TestCluster) SendTo(to *pilosa.Node, pb proto.Message) error {
return err
}
case *internal.ResizeInstructionComplete:
coord := t.clusterByURI(to.URI)
coord := t.clusterByID(to.ID)
go coord.MarkResizeInstructionComplete(obj)
}
return nil
@ -356,7 +368,7 @@ func (t *TestCluster) FollowResizeInstruction(instr *internal.ResizeInstruction)
// Prepare the return message.
complete := &internal.ResizeInstructionComplete{
JobID: instr.JobID,
URI: instr.URI,
Node: instr.Node,
Error: "",
}
@ -365,8 +377,8 @@ func (t *TestCluster) FollowResizeInstruction(instr *internal.ResizeInstruction)
// figure out which node it was meant for, then call the operation on that cluster
// basically need to mimic this: client.RetrieveSliceFromURI(context.Background(), src.Index, src.Frame, src.View, src.Slice, srcURI)
instrURI := pilosa.DecodeURI(instr.URI)
destCluster := t.clusterByURI(instrURI)
instrNode := pilosa.DecodeNode(instr.Node)
destCluster := t.clusterByID(instrNode.ID)
// Sync the schema received in the resize instruction.
if err := destCluster.Holder.ApplySchema(instr.Schema); err != nil {
@ -374,8 +386,8 @@ func (t *TestCluster) FollowResizeInstruction(instr *internal.ResizeInstruction)
}
for _, src := range instr.Sources {
srcURI := pilosa.DecodeURI(src.URI)
srcCluster := t.clusterByURI(srcURI)
srcNode := pilosa.DecodeNode(src.Node)
srcCluster := t.clusterByID(srcNode.ID)
srcFragment := srcCluster.Holder.Fragment(src.Index, src.Frame, src.View, src.Slice)
destFragment := destCluster.Holder.Fragment(src.Index, src.Frame, src.View, src.Slice)
@ -414,9 +426,7 @@ func (t *TestCluster) FollowResizeInstruction(instr *internal.ResizeInstruction)
complete.Error = err.Error()
}
node := &pilosa.Node{
URI: pilosa.DecodeURI(instr.Coordinator),
}
node := pilosa.DecodeNode(instr.Coordinator)
if err := t.SendTo(node, complete); err != nil {
return err
}

2
test/executor.go
View file

@ -26,7 +26,7 @@ func NewExecutor(holder *pilosa.Holder, cluster *pilosa.Cluster) *Executor {
e := &Executor{Executor: executor}
e.Holder = holder
e.Cluster = cluster
e.URI = cluster.Nodes[0].URI
e.Node = cluster.Nodes[0]
return e
}

13
test/handler.go
View file

@ -67,7 +67,6 @@ func NewServer() *Server {
if err != nil {
panic(err)
}
s.Handler.URI = *uri
// Handler test messages can no-op.
s.Handler.Broadcaster = pilosa.NopBroadcaster
@ -75,6 +74,8 @@ func NewServer() *Server {
s.Handler.Cluster = NewCluster(1)
s.Handler.Cluster.Nodes[0].URI = *uri
s.Handler.Node = s.Handler.Cluster.Nodes[0]
return s
}
@ -85,10 +86,16 @@ func (s *Server) LocalStatus() (proto.Message, error) {
// ClusterStatus exists so that test.Server implements StatusHandler.
func (s *Server) ClusterStatus() (proto.Message, error) {
id := "test-node"
uri := pilosa.DefaultURI()
node := &pilosa.Node{
ID: id,
URI: *uri,
}
return &internal.ClusterStatus{
State: pilosa.ClusterStateNormal,
NodeSet: []*internal.URI{uri.Encode()},
ClusterID: "",
State: pilosa.ClusterStateNormal,
Nodes: pilosa.EncodeNodes([]*pilosa.Node{node}),
}, nil
}

10
uri.go
View file

@ -57,6 +57,16 @@ func DefaultURI() *URI {
}
}
type URIs []URI
func (u URIs) HostPortStrings() []string {
s := make([]string, len(u))
for i, a := range u {
s[i] = a.HostPort()
}
return s
}
// NewURIFromHostPort returns a URI with specified host and port.
func NewURIFromHostPort(host string, port uint16) (*URI, error) {
uri := DefaultURI()