diff --git a/api.go b/api.go index 02ffa91dc..9f7356cb8 100644 --- a/api.go +++ b/api.go @@ -1053,8 +1053,8 @@ func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode return nil, nil, errors.Wrap(err, "validate api method") } - oldNode = api.Cluster.nodeByID(api.Cluster.Coordinator) - newNode = api.Cluster.nodeByID(id) + oldNode = api.Cluster.NodeByID(api.Cluster.Coordinator) + newNode = api.Cluster.NodeByID(id) if newNode == nil { return nil, nil, errors.Wrap(ErrNodeIDNotExists, "getting new node") } @@ -1102,9 +1102,6 @@ func (api *API) ResizeAbort() error { return errors.Wrap(err, "validate api method") } - if !api.Cluster.IsCoordinator() { - return ErrNodeNotCoordinator - } err := api.Cluster.CompleteCurrentJob(ResizeJobStateAborted) return errors.Wrap(err, "complete current job") } diff --git a/cluster.go b/cluster.go index bbf6f799e..b02bd50d2 100644 --- a/cluster.go +++ b/cluster.go @@ -292,6 +292,12 @@ func (c *Cluster) CoordinatorNode() *Node { // IsCoordinator is true if this node is the coordinator. func (c *Cluster) IsCoordinator() bool { + c.mu.RLock() + defer c.mu.RUnlock() + return c.isCoordinator() +} + +func (c *Cluster) isCoordinator() bool { return c.Coordinator == c.Node.ID } @@ -300,6 +306,8 @@ func (c *Cluster) IsCoordinator() bool { // will consider itself coordinator and update the other // nodes with its version of Cluster.Status. func (c *Cluster) SetCoordinator(n *Node) error { + c.mu.Lock() + defer c.mu.Unlock() // Verify that the new Coordinator value matches // this node. if c.Node.ID != n.ID { @@ -307,7 +315,7 @@ func (c *Cluster) SetCoordinator(n *Node) error { } // Update IsCoordinator on all nodes (locally). - _ = c.UpdateCoordinator(n) + _ = c.updateCoordinator(n) // Send the update coordinator message to all nodes. err := c.Broadcaster.SendSync( @@ -327,6 +335,12 @@ func (c *Cluster) SetCoordinator(n *Node) error { // to true, and sets all other nodes to false. Returns true if the value // changed. func (c *Cluster) UpdateCoordinator(n *Node) bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.updateCoordinator(n) +} + +func (c *Cluster) updateCoordinator(n *Node) bool { var changed bool if c.Coordinator != n.ID { c.Coordinator = n.ID @@ -511,6 +525,12 @@ func (c *Cluster) Status() *internal.ClusterStatus { } } +func (c *Cluster) NodeByID(id string) *Node { + c.mu.RLock() + defer c.mu.RUnlock() + return c.nodeByID(id) +} + // nodeByID returns a node reference by ID. func (c *Cluster) nodeByID(id string) *Node { for _, n := range c.Nodes { @@ -1186,6 +1206,9 @@ func (c *Cluster) generateResizeJobByAction(nodeAction nodeAction) (*ResizeJob, func (c *Cluster) CompleteCurrentJob(state string) error { c.mu.Lock() defer c.mu.Unlock() + if !c.isCoordinator() { + return ErrNodeNotCoordinator + } if c.currentJob == nil { return ErrResizeNotRunning } @@ -1797,9 +1820,11 @@ func (c *Cluster) nodeLeave(node *Node) error { } func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error { + c.mu.Lock() + defer c.mu.Unlock() c.Logger.Printf("merge cluster status: %v", cs) // Ignore status updates from self (coordinator). - if c.IsCoordinator() { + if c.isCoordinator() { return nil } @@ -1836,7 +1861,7 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error { } } - c.SetState(cs.State) + c.setState(cs.State) c.markAsJoined() diff --git a/cluster_test.go b/cluster_test.go index 6c5d41829..b09820ac8 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package pilosa_test +package pilosa import ( "bytes" @@ -22,29 +22,27 @@ import ( "testing/quick" "github.com/davecgh/go-spew/spew" - "github.com/pilosa/pilosa" - "github.com/pilosa/pilosa/test" ) // Ensure the cluster can fairly distribute partitions across the nodes. func TestCluster_Owners(t *testing.T) { - c := pilosa.Cluster{ - Nodes: []*pilosa.Node{ - {URI: test.NewURIFromHostPort("serverA", 1000)}, - {URI: test.NewURIFromHostPort("serverB", 1000)}, - {URI: test.NewURIFromHostPort("serverC", 1000)}, + c := Cluster{ + Nodes: []*Node{ + {URI: NewTestURIFromHostPort("serverA", 1000)}, + {URI: NewTestURIFromHostPort("serverB", 1000)}, + {URI: NewTestURIFromHostPort("serverC", 1000)}, }, - Hasher: test.NewModHasher(), + Hasher: NewTestModHasher(), ReplicaN: 2, } // Verify nodes are distributed. - if a := c.PartitionNodes(0); !reflect.DeepEqual(a, []*pilosa.Node{c.Nodes[0], c.Nodes[1]}) { + if a := c.PartitionNodes(0); !reflect.DeepEqual(a, []*Node{c.Nodes[0], c.Nodes[1]}) { t.Fatalf("unexpected owners: %s", spew.Sdump(a)) } // Verify nodes go around the ring. - if a := c.PartitionNodes(2); !reflect.DeepEqual(a, []*pilosa.Node{c.Nodes[2], c.Nodes[0]}) { + if a := c.PartitionNodes(2); !reflect.DeepEqual(a, []*Node{c.Nodes[2], c.Nodes[0]}) { t.Fatalf("unexpected owners: %s", spew.Sdump(a)) } } @@ -52,7 +50,7 @@ func TestCluster_Owners(t *testing.T) { // Ensure the partitioner can assign a fragment to a partition. func TestCluster_Partition(t *testing.T) { if err := quick.Check(func(index string, slice uint64, partitionN int) bool { - c := pilosa.NewCluster() + c := NewCluster() c.PartitionN = partitionN partitionID := c.Partition(index, slice) @@ -85,7 +83,7 @@ func TestHasher(t *testing.T) { {0x0ddc0ffeebadf00d, []int{0, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 15, 15, 15, 15}}, } { for i, v := range tt.bucket { - if got := pilosa.NewHasher().Hash(tt.key, i+1); got != v { + if got := NewHasher().Hash(tt.key, i+1); got != v { t.Errorf("hash(%v,%v)=%v, want %v", tt.key, i+1, got, v) } } @@ -94,8 +92,8 @@ func TestHasher(t *testing.T) { // Ensure OwnsSlices can find the actual slice list for node and index. func TestCluster_OwnsSlices(t *testing.T) { - c := test.NewCluster(5) - slices := c.OwnsSlices("test", 10, test.NewURIFromHostPort("host2", 0)) + c := NewTestCluster(5) + slices := c.OwnsSlices("test", 10, NewTestURIFromHostPort("host2", 0)) if !reflect.DeepEqual(slices, []uint64{0, 3, 6, 10}) { t.Fatalf("unexpected slices for node's index: %v", slices) @@ -104,7 +102,7 @@ func TestCluster_OwnsSlices(t *testing.T) { // Ensure ContainsSlices can find the actual slice list for node and index. func TestCluster_ContainsSlices(t *testing.T) { - c := test.NewCluster(5) + c := NewTestCluster(5) c.ReplicaN = 3 slices := c.ContainsSlices("test", 10, c.Nodes[2]) @@ -114,20 +112,20 @@ func TestCluster_ContainsSlices(t *testing.T) { } func TestCluster_Nodes(t *testing.T) { - uri0 := test.NewURIFromHostPort("node0", 0) - uri1 := test.NewURIFromHostPort("node1", 0) - uri2 := test.NewURIFromHostPort("node2", 0) - uri3 := test.NewURIFromHostPort("node3", 0) + uri0 := NewTestURIFromHostPort("node0", 0) + uri1 := NewTestURIFromHostPort("node1", 0) + uri2 := NewTestURIFromHostPort("node2", 0) + uri3 := NewTestURIFromHostPort("node3", 0) - 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} + node0 := &Node{ID: "node0", URI: uri0} + node1 := &Node{ID: "node1", URI: uri1} + node2 := &Node{ID: "node2", URI: uri2} + node3 := &Node{ID: "node3", URI: uri3} - nodes := []*pilosa.Node{node0, node1, node2} + nodes := []*Node{node0, node1, node2} t.Run("NodeIDs", func(t *testing.T) { - actual := pilosa.Nodes(nodes).IDs() + actual := 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) @@ -135,24 +133,24 @@ func TestCluster_Nodes(t *testing.T) { }) t.Run("Filter", func(t *testing.T) { - actual := pilosa.Nodes(pilosa.Nodes(nodes).Filter(nodes[1])).URIs() - expected := []pilosa.URI{uri0, uri2} + actual := Nodes(Nodes(nodes).Filter(nodes[1])).URIs() + expected := []URI{uri0, uri2} if !reflect.DeepEqual(actual, expected) { t.Errorf("expected: %v, but got: %v", expected, actual) } }) t.Run("FilterURI", func(t *testing.T) { - actual := pilosa.Nodes(pilosa.Nodes(nodes).FilterURI(uri1)).URIs() - expected := []pilosa.URI{uri0, uri2} + actual := Nodes(Nodes(nodes).FilterURI(uri1)).URIs() + expected := []URI{uri0, uri2} if !reflect.DeepEqual(actual, expected) { t.Errorf("expected: %v, but got: %v", expected, actual) } }) t.Run("Contains", func(t *testing.T) { - actualTrue := pilosa.Nodes(nodes).Contains(node1) - actualFalse := pilosa.Nodes(nodes).Contains(node3) + actualTrue := Nodes(nodes).Contains(node1) + actualFalse := Nodes(nodes).Contains(node3) if !reflect.DeepEqual(actualTrue, true) { t.Errorf("expected: %v, but got: %v", true, actualTrue) } @@ -162,9 +160,9 @@ func TestCluster_Nodes(t *testing.T) { }) t.Run("Clone", func(t *testing.T) { - clone := pilosa.Nodes(nodes).Clone() - actual := pilosa.Nodes(clone).URIs() - expected := []pilosa.URI{uri0, uri1, uri2} + clone := Nodes(nodes).Clone() + actual := Nodes(clone).URIs() + expected := []URI{uri0, uri1, uri2} if !reflect.DeepEqual(actual, expected) { t.Errorf("expected: %v, but got: %v", expected, actual) } @@ -172,16 +170,16 @@ func TestCluster_Nodes(t *testing.T) { } func TestCluster_Coordinator(t *testing.T) { - uri1 := test.NewURIFromHostPort("node1", 0) - uri2 := test.NewURIFromHostPort("node2", 0) + uri1 := NewTestURIFromHostPort("node1", 0) + uri2 := NewTestURIFromHostPort("node2", 0) - node1 := &pilosa.Node{ID: "node1", URI: uri1} - node2 := &pilosa.Node{ID: "node2", URI: uri2} + node1 := &Node{ID: "node1", URI: uri1} + node2 := &Node{ID: "node2", URI: uri2} - c1 := *pilosa.NewCluster() + c1 := *NewCluster() c1.Node = node1 c1.Coordinator = node1.ID - c2 := *pilosa.NewCluster() + c2 := *NewCluster() c2.Node = node2 c2.Coordinator = node1.ID @@ -195,17 +193,17 @@ func TestCluster_Coordinator(t *testing.T) { } func TestCluster_Topology(t *testing.T) { - c1 := test.NewCluster(1) // automatically creates Node{ID: "node0"} + c1 := NewTestCluster(1) // automatically creates Node{ID: "node0"} - uri0 := test.NewURIFromHostPort("host0", 0) - uri1 := test.NewURIFromHostPort("host1", 0) - uri2 := test.NewURIFromHostPort("host2", 0) - invalid := test.NewURIFromHostPort("invalid", 0) + uri0 := NewTestURIFromHostPort("host0", 0) + uri1 := NewTestURIFromHostPort("host1", 0) + uri2 := NewTestURIFromHostPort("host2", 0) + invalid := NewTestURIFromHostPort("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} + node0 := &Node{ID: "node0", URI: uri0} + node1 := &Node{ID: "node1", URI: uri1} + node2 := &Node{ID: "node2", URI: uri2} + nodeinvalid := &Node{ID: "nodeinvalid", URI: invalid} t.Run("AddNode", func(t *testing.T) { err := c1.AddNode(node1) @@ -243,7 +241,7 @@ func TestCluster_Topology(t *testing.T) { func TestCluster_ResizeStates(t *testing.T) { t.Run("Single node, no data", func(t *testing.T) { - tc := test.NewTestCluster(1) + tc := NewClusterCluster(1) // Open TestCluster. if err := tc.Open(); err != nil { @@ -253,11 +251,11 @@ func TestCluster_ResizeStates(t *testing.T) { node := tc.Clusters[0] // Ensure that node comes up in state NORMAL. - if node.State() != pilosa.ClusterStateNormal { - t.Errorf("expected state: %v, but got: %v", pilosa.ClusterStateNormal, node.State()) + if node.State() != ClusterStateNormal { + t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, node.State()) } - expectedTop := &pilosa.Topology{ + expectedTop := &Topology{ NodeIDs: []string{node.Node.ID}, } @@ -273,13 +271,13 @@ func TestCluster_ResizeStates(t *testing.T) { }) t.Run("Single node, in topology", func(t *testing.T) { - tc := test.NewTestCluster(0) + tc := NewClusterCluster(0) tc.AddNode(false) node := tc.Clusters[0] // write topology to data file - top := &pilosa.Topology{ + top := &Topology{ NodeIDs: []string{node.Node.ID}, } tc.WriteTopology(node.Path, top) @@ -290,8 +288,8 @@ func TestCluster_ResizeStates(t *testing.T) { } // Ensure that node comes up in state NORMAL. - if node.State() != pilosa.ClusterStateNormal { - t.Errorf("expected state: %v, but got: %v", pilosa.ClusterStateNormal, node.State()) + if node.State() != ClusterStateNormal { + t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, node.State()) } // Close TestCluster. @@ -301,13 +299,13 @@ func TestCluster_ResizeStates(t *testing.T) { }) t.Run("Single node, not in topology", func(t *testing.T) { - tc := test.NewTestCluster(0) + tc := NewClusterCluster(0) tc.AddNode(false) node := tc.Clusters[0] // write topology to data file - top := &pilosa.Topology{ + top := &Topology{ NodeIDs: []string{"some-other-host"}, } tc.WriteTopology(node.Path, top) @@ -326,7 +324,7 @@ func TestCluster_ResizeStates(t *testing.T) { }) t.Run("Multiple nodes, no data", func(t *testing.T) { - tc := test.NewTestCluster(0) + tc := NewClusterCluster(0) tc.AddNode(false) // Open TestCluster. @@ -340,13 +338,13 @@ func TestCluster_ResizeStates(t *testing.T) { node1 := tc.Clusters[1] // Ensure that nodes comes up in state NORMAL. - if node0.State() != pilosa.ClusterStateNormal { - t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State()) - } else if node1.State() != pilosa.ClusterStateNormal { - t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node1.State()) + if node0.State() != ClusterStateNormal { + t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State()) + } else if node1.State() != ClusterStateNormal { + t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, node1.State()) } - expectedTop := &pilosa.Topology{ + expectedTop := &Topology{ NodeIDs: []string{node0.Node.ID, node1.Node.ID}, } @@ -364,12 +362,12 @@ func TestCluster_ResizeStates(t *testing.T) { }) t.Run("Multiple nodes, in/not in topology", func(t *testing.T) { - tc := test.NewTestCluster(0) + tc := NewClusterCluster(0) tc.AddNode(false) node0 := tc.Clusters[0] // write topology to data file - top := &pilosa.Topology{ + top := &Topology{ NodeIDs: []string{"node0", "node2"}, } tc.WriteTopology(node0.Path, top) @@ -380,8 +378,8 @@ func TestCluster_ResizeStates(t *testing.T) { } // Ensure that node is in state STARTING before the other node joins. - if node0.State() != pilosa.ClusterStateStarting { - t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateStarting, node0.State()) + if node0.State() != ClusterStateStarting { + t.Errorf("expected node0 state: %v, but got: %v", ClusterStateStarting, node0.State()) } // Expect an error by adding a node not in the topology. @@ -395,10 +393,10 @@ func TestCluster_ResizeStates(t *testing.T) { node2 := tc.Clusters[2] // Ensure that node comes up in state NORMAL. - if node0.State() != pilosa.ClusterStateNormal { - t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State()) - } else if node2.State() != pilosa.ClusterStateNormal { - t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node2.State()) + if node0.State() != ClusterStateNormal { + t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State()) + } else if node2.State() != ClusterStateNormal { + t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, node2.State()) } // Close TestCluster. @@ -408,7 +406,7 @@ func TestCluster_ResizeStates(t *testing.T) { }) t.Run("Multiple nodes, with data", func(t *testing.T) { - tc := test.NewTestCluster(0) + tc := NewClusterCluster(0) tc.AddNode(false) node0 := tc.Clusters[0] @@ -418,20 +416,20 @@ func TestCluster_ResizeStates(t *testing.T) { } // Add Bit Data to node0. - if err := tc.CreateFrame("i", "f", pilosa.FrameOptions{}); err != nil { + if err := tc.CreateFrame("i", "f", FrameOptions{}); err != nil { t.Fatal(err) } tc.SetBit("i", "f", "standard", 1, 101, nil) tc.SetBit("i", "f", "standard", 1, 1300000, nil) // Add Field Data to node0. - if err := tc.CreateFrame("i", "fields", pilosa.FrameOptions{ + if err := tc.CreateFrame("i", "fields", FrameOptions{ InverseEnabled: false, - //CacheType: pilosa.CacheTypeNone, - Fields: []*pilosa.Field{ + //CacheType: CacheTypeNone, + Fields: []*Field{ { Name: "fld0", - Type: pilosa.FieldTypeInt, + Type: FieldTypeInt, Min: -100, Max: 100, }, @@ -461,13 +459,13 @@ func TestCluster_ResizeStates(t *testing.T) { node1 := tc.Clusters[1] // Ensure that nodes come up in state NORMAL. - if node0.State() != pilosa.ClusterStateNormal { - t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State()) - } else if node1.State() != pilosa.ClusterStateNormal { - t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node1.State()) + if node0.State() != ClusterStateNormal { + t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State()) + } else if node1.State() != ClusterStateNormal { + t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, node1.State()) } - expectedTop := &pilosa.Topology{ + expectedTop := &Topology{ NodeIDs: []string{node0.Node.ID, node1.Node.ID}, } @@ -510,7 +508,7 @@ func TestCluster_ResizeStates(t *testing.T) { // Ensures that coordinator can be changed. func TestCluster_UpdateCoordinator(t *testing.T) { t.Run("UpdateCoordinator", func(t *testing.T) { - c := test.NewCluster(2) + c := NewTestCluster(2) oldNode := c.Nodes[0] newNode := c.Nodes[1] diff --git a/server/cluster_test.go b/server/cluster_test.go index 1a64ce305..326753a3d 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -396,7 +396,7 @@ func TestCluster_GossipMembership(t *testing.T) { var eg errgroup.Group eg.Go(func() error { // Pass invalid seed as first in list - _, err = m1.RunWithTransport("localhost", 0, []string{"http://localhost:8765", seed}) + _, err := m1.RunWithTransport("localhost", 0, []string{"http://localhost:8765", seed}) if err != nil { return err } @@ -409,7 +409,7 @@ func TestCluster_GossipMembership(t *testing.T) { eg.Go(func() error { // Pass invalid seed as last in list - _, err = m2.RunWithTransport("localhost", 0, []string{seed, "http://localhost:8765"}) + _, err := m2.RunWithTransport("localhost", 0, []string{seed, "http://localhost:8765"}) if err != nil { return err } diff --git a/test/pilosa.go b/test/pilosa.go index 24838d5de..f6e3a3457 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -66,7 +66,6 @@ func NewMain(opts ...MainOpt) *Main { m.Command.Stdin = &m.Stdin m.Command.Stdout = &m.Stdout m.Command.Stderr = &m.Stderr - for _, opt := range opts { err := opt(m) if err != nil { @@ -74,8 +73,10 @@ func NewMain(opts ...MainOpt) *Main { } } - - m.SetupServer() + err = m.SetupServer() + if err != nil { + panic(err) + } if testing.Verbose() { m.Command.Stdout = io.MultiWriter(os.Stdout, m.Command.Stdout) diff --git a/utils_test.go b/utils_test.go new file mode 100644 index 000000000..05afeac19 --- /dev/null +++ b/utils_test.go @@ -0,0 +1,425 @@ +package pilosa + +import ( + "bufio" + "bytes" + "fmt" + "io/ioutil" + "path/filepath" + "sync" + "time" + + "github.com/gogo/protobuf/proto" + "github.com/pilosa/pilosa/internal" +) + +// NewTestCluster returns a cluster with n nodes and uses a mod-based hasher. +func NewTestCluster(n int) *Cluster { + path, err := ioutil.TempDir("", "pilosa-cluster-") + if err != nil { + panic(err) + } + + c := NewCluster() + c.ReplicaN = 1 + c.Hasher = NewTestModHasher() + c.Path = path + c.Topology = NewTopology() + + for i := 0; i < n; i++ { + c.Nodes = append(c.Nodes, &Node{ + ID: fmt.Sprintf("node%d", i), + URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)), + }) + } + + c.Node = c.Nodes[0] + c.Coordinator = c.Nodes[0].ID + c.SetState(ClusterStateNormal) + + return c +} + +// NewTestURI is a test URI creator that intentionally swallows errors. +func NewTestURI(scheme, host string, port uint16) URI { + uri := DefaultURI() + uri.SetScheme(scheme) + uri.SetHost(host) + uri.SetPort(port) + return *uri +} + +func NewTestURIFromHostPort(host string, port uint16) URI { + uri := DefaultURI() + uri.SetHost(host) + uri.SetPort(port) + return *uri +} + +// ModHasher represents a simple, mod-based hashing. +type TestModHasher struct{} + +// NewTestModHasher returns a new instance of ModHasher with n buckets. +func NewTestModHasher() *TestModHasher { return &TestModHasher{} } + +func (*TestModHasher) Hash(key uint64, n int) int { return int(key) % n } + +// ClusterCluster represents a cluster of test nodes, each of which +// has a Cluster. +// ClusterCluster implements Broadcaster interface. +type ClusterCluster struct { + Clusters []*Cluster + + common *commonClusterSettings + + mu sync.RWMutex + resizing bool + resizeDone chan struct{} +} + +type commonClusterSettings struct { + Nodes []*Node +} + +func (t *ClusterCluster) CreateIndex(name string) error { + for _, c := range t.Clusters { + if _, err := c.Holder.CreateIndexIfNotExists(name, IndexOptions{}); err != nil { + return err + } + } + return nil +} + +func (t *ClusterCluster) CreateFrame(index, frame string, opt FrameOptions) error { + for _, c := range t.Clusters { + idx, err := c.Holder.CreateIndexIfNotExists(index, IndexOptions{}) + if err != nil { + return err + } + if _, err := idx.CreateFrame(frame, opt); err != nil { + return err + } + } + return nil +} + +func (t *ClusterCluster) SetBit(index, frame, view string, rowID, colID uint64, x *time.Time) error { + // Determine which node should receive the SetBit. + c0 := t.Clusters[0] // use the first node's cluster to determine slice location. + slice := colID / SliceWidth + nodes := c0.SliceNodes(index, slice) + + for _, node := range nodes { + c := t.clusterByID(node.ID) + if c == nil { + continue + } + f := c.Holder.Frame(index, frame) + if f == nil { + return fmt.Errorf("index/frame does not exist: %s/%s", index, frame) + } + _, err := f.SetBit(view, rowID, colID, x) + if err != nil { + return err + } + } + + return nil +} + +func (t *ClusterCluster) SetFieldValue(index, frame string, columnID uint64, name string, value int64) error { + // Determine which node should receive the SetFieldValue. + c0 := t.Clusters[0] // use the first node's cluster to determine slice location. + slice := columnID / SliceWidth + nodes := c0.SliceNodes(index, slice) + + for _, node := range nodes { + c := t.clusterByID(node.ID) + if c == nil { + continue + } + f := c.Holder.Frame(index, frame) + if f == nil { + return fmt.Errorf("index/frame does not exist: %s/%s", index, frame) + } + _, err := f.SetFieldValue(columnID, name, value) + if err != nil { + return err + } + } + + return nil +} + +func (t *ClusterCluster) clusterByID(id string) *Cluster { + for _, c := range t.Clusters { + if c.Node.ID == id { + return c + } + } + return nil +} + +// AddNode adds a node to the cluster and (potentially) starts a resize job. +func (t *ClusterCluster) AddNode(saveTopology bool) error { + id := len(t.Clusters) + + c, err := t.addCluster(id, saveTopology) + if err != nil { + return err + } + + // Send NodeJoin event to coordinator. + if id > 0 { + coord := t.Clusters[0] + ev := &NodeEvent{ + Event: NodeJoin, + Node: c.Node, + } + + if err := coord.ReceiveEvent(ev); err != nil { + return err + } + + // Wait for the AddNode job to finish. + if c.State() != ClusterStateNormal { + t.resizeDone = make(chan struct{}) + t.mu.Lock() + t.resizing = true + t.mu.Unlock() + <-t.resizeDone + } + } + + return nil +} + +// WriteTopology writes the given topology to disk. +func (t *ClusterCluster) WriteTopology(path string, top *Topology) error { + if buf, err := proto.Marshal(top.Encode()); err != nil { + return err + } else if err := ioutil.WriteFile(filepath.Join(path, ".topology"), buf, 0666); err != nil { + return err + } + return nil +} + +func (t *ClusterCluster) addCluster(i int, saveTopology bool) (*Cluster, error) { + + id := fmt.Sprintf("node%d", i) + uri := NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)) + + node := &Node{ + ID: id, + URI: uri, + } + + // add URI to common + //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)) + if err != nil { + return nil, err + } + + // holder + h := NewHolder() + h.Path = path + + // cluster + c := NewCluster() + c.ReplicaN = 1 + c.Hasher = NewTestModHasher() + c.Path = path + c.Topology = NewTopology() + c.Holder = h + c.MemberSet = NewStaticMemberSet(c.Nodes) + c.Node = node + c.Coordinator = t.common.Nodes[0].ID // the first node is the coordinator + c.Broadcaster = t + + // add nodes + if saveTopology { + for _, n := range t.common.Nodes { + c.AddNode(n) + } + } + + // Add this node to the ClusterCluster. + t.Clusters = append(t.Clusters, c) + + return c, nil +} + +// NewClusterCluster returns a new instance of test.Cluster. +func NewClusterCluster(n int) *ClusterCluster { + + tc := &ClusterCluster{ + common: &commonClusterSettings{}, + } + + // add clusters + for i := 0; i < n; i++ { + _, err := tc.addCluster(i, true) + if err != nil { + panic(err) + } + } + return tc +} + +// SetState sets the state of the cluster on each node. +func (t *ClusterCluster) SetState(state string) { + for _, c := range t.Clusters { + c.SetState(state) + } +} + +// Open opens all clusters in the test cluster. +func (t *ClusterCluster) Open() error { + for _, c := range t.Clusters { + if err := c.Open(); err != nil { + return err + } + if err := c.Holder.Open(); err != nil { + return err + } + if err := c.SetNodeState(NodeStateReady); err != nil { + return err + } + } + + // Start the listener on the coordinator. + if len(t.Clusters) == 0 { + return nil + } + t.Clusters[0].ListenForJoins() + + return nil +} + +// Close closes all clusters in the test cluster. +func (t *ClusterCluster) Close() error { + for _, c := range t.Clusters { + err := c.Close() + if err != nil { + return err + } + } + return nil +} + +// SendSync is a test implemenetation of Broadcaster SendSync method. +func (t *ClusterCluster) SendSync(pb proto.Message) error { + switch obj := pb.(type) { + case *internal.ClusterStatus: + // Apply the send message to all nodes (except the coordinator). + for _, c := range t.Clusters { + c.MergeClusterStatus(obj) + } + t.mu.RLock() + if obj.State == ClusterStateNormal && t.resizing { + close(t.resizeDone) + } + t.mu.RUnlock() + } + + return nil +} + +// SendAsync is a test implemenetation of Broadcaster SendAsync method. +func (t *ClusterCluster) SendAsync(pb proto.Message) error { + return nil +} + +// SendTo is a test implemenetation of Broadcaster SendTo method. +func (t *ClusterCluster) SendTo(to *Node, pb proto.Message) error { + switch obj := pb.(type) { + case *internal.ResizeInstruction: + err := t.FollowResizeInstruction(obj) + if err != nil { + return err + } + case *internal.ResizeInstructionComplete: + coord := t.clusterByID(to.ID) + go coord.MarkResizeInstructionComplete(obj) + } + return nil +} + +// FollowResizeInstruction is a version of cluster.FollowResizeInstruction used for testing. +func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstruction) error { + + // Prepare the return message. + complete := &internal.ResizeInstructionComplete{ + JobID: instr.JobID, + Node: instr.Node, + Error: "", + } + + // Stop processing on any error. + if err := func() error { + + // 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) + instrNode := 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 { + return err + } + + for _, src := range instr.Sources { + srcNode := 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) + if destFragment == nil { + // Create fragment on destination if it doesn't exist. + f := destCluster.Holder.Frame(src.Index, src.Frame) + v := f.View(src.View) + var err error + destFragment, err = v.CreateFragmentIfNotExists(src.Slice) + if err != nil { + return err + } + } + + buf := bytes.NewBuffer(nil) + + bw := bufio.NewWriter(buf) + br := bufio.NewReader(buf) + + // Get the fragment from source. + if _, err := srcFragment.WriteTo(bw); err != nil { + return err + } + + // Flush the bufio.buf to the io.Writer (buf). + bw.Flush() + + // Write data to destination. + if _, err := destFragment.ReadFrom(br); err != nil { + return err + } + } + + return nil + }(); err != nil { + complete.Error = err.Error() + } + + node := DecodeNode(instr.Coordinator) + if err := t.SendTo(node, complete); err != nil { + return err + } + + return nil +}