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

Merge pull request #912 from travisturner/copy-fragment-data

Browse source 
Copy fragment data
This commit is contained in:
Travis Turner 2017-10-30 14:41:20 -05:00 • committed by GitHub
commit c360ca2975
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 377 additions and 274 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
broadcast.go
View file

@ -22,37 +22,37 @@ import (
"github.com/pilosa/pilosa/internal"
)
// NodeSet represents an interface for Node membership and inter-node communication.
type NodeSet interface {
// 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 NodeSet
// Open starts any network activity implemented by the MemberSet
Open() error
}
// StaticNodeSet represents a basic NodeSet for testing.
type StaticNodeSet struct {
// StaticMemberSet represents a basic MemberSet for testing.
type StaticMemberSet struct {
nodes []*Node
}
// NewStaticNodeSet creates a statically defined NodeSet.
func NewStaticNodeSet() *StaticNodeSet {
return &StaticNodeSet{}
// NewStaticMemberSet creates a statically defined MemberSet.
func NewStaticMemberSet() *StaticMemberSet {
return &StaticMemberSet{}
}
// Nodes implements the NodeSet interface and returns a list of nodes in the cluster.
func (s *StaticNodeSet) Nodes() []*Node {
// 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 NodeSet interface to start network activity, but for a static NodeSet it does nothing.
func (s *StaticNodeSet) Open() error {
// Open implements the MemberSet interface to start network activity, but for a static MemberSet it does nothing.
func (s *StaticMemberSet) Open() error {
return nil
}
// Join sets the NodeSet nodes to the slice of Nodes passed in.
func (s *StaticNodeSet) Join(nodes []*Node) error {
// Join sets the MemberSet nodes to the slice of Nodes passed in.
func (s *StaticMemberSet) Join(nodes []*Node) error {
s.nodes = nodes
return nil
}

7
client.go
View file

@ -682,6 +682,13 @@ func (c *Client) BackupSlice(ctx context.Context, index, frame, view string, sli
return nil, fmt.Errorf("unable to connect to any owner")
}
func (c *Client) RetrieveSliceFromURI(ctx context.Context, index, frame, view string, slice uint64, uri URI) (io.ReadCloser, error) {
node := &Node{
URI: uri,
}
return c.backupSliceNode(ctx, index, frame, view, slice, node)
}
func (c *Client) backupSliceNode(ctx context.Context, index, frame, view string, slice uint64, node *Node) (io.ReadCloser, error) {
u := nodePathToURL(node, "/fragment/data")
u.RawQuery = url.Values{

199
cluster.go
View file

@ -15,7 +15,9 @@
package pilosa
import (
"context"
"encoding/binary"
"errors"
"fmt"
"hash/fnv"
"io"
@ -39,10 +41,10 @@ const (
// DefaultReplicaN is the default number of replicas per partition.
DefaultReplicaN = 1
// NodeState represents node state returned in /status endpoint for a node in the cluster.
NodeStateStarting = "STARTING"
NodeStateNormal = "NORMAL"
NodeStateResizing = "RESIZING"
// ClusterState represents the state returned in the /status endpoint.
ClusterStateStarting = "STARTING"
ClusterStateNormal = "NORMAL"
ClusterStateResizing = "RESIZING"
// ResizeJob states.
ResizeJobStateRunning = "RUNNING"
@ -127,9 +129,9 @@ func (h ByHost) Less(i, j int) bool { return h[i].URI.String() < h[j].URI.String
// Cluster represents a collection of nodes.
type Cluster struct {
URI URI
Nodes []*Node // TODO phase this out?
NodeSet NodeSet
URI URI
Nodes []*Node // TODO phase this out?
MemberSet MemberSet
// Hashing algorithm used to assign partitions to nodes.
Hasher Hasher
@ -151,10 +153,10 @@ type Cluster struct {
Topology *Topology
// Required for cluster Resize.
State string
Coordinator URI
IndexReporter IndexReporter
Broadcaster Broadcaster
State string
Coordinator URI
Holder *Holder
Broadcaster Broadcaster
joiningURIs chan URI
@ -196,12 +198,12 @@ func (c *Cluster) IsCoordinator() bool {
return c.Coordinator == c.URI
}
// AddHost adds a node to the Cluster and updates and saves the
// AddNode adds a node to the Cluster and updates and saves the
// new topology.
func (c *Cluster) AddHost(uri URI) error {
func (c *Cluster) AddNode(uri URI) error {
// add to cluster
_, added := c.AddNode(uri)
_, added := c.AddNodeBasicSorted(uri)
if !added {
return nil
}
@ -218,8 +220,8 @@ func (c *Cluster) AddHost(uri URI) error {
return c.saveTopology()
}
// URISet returns the list of uris in the cluster.
func (c *Cluster) URISet() []URI {
// NodeSet returns the list of uris in the cluster.
func (c *Cluster) NodeSet() []URI {
return Nodes(c.Nodes).URIs()
}
@ -235,8 +237,8 @@ func (c *Cluster) setState(state string) {
// Status returns the internal ClusterStatus representation.
func (c *Cluster) Status() *internal.ClusterStatus {
return &internal.ClusterStatus{
State: c.State,
URISet: encodeURIs(c.URISet()),
State: c.State,
NodeSet: encodeURIs(c.NodeSet()),
}
}
@ -250,9 +252,9 @@ func (c *Cluster) NodeByURI(uri URI) *Node {
return nil
}
// AddNode adds a node to the cluster, sorted by uri.
// AddNodeBasicSorted adds a node to the cluster, sorted by uri.
// Returns a pointer to the node and true if the node was added.
func (c *Cluster) AddNode(uri URI) (*Node, bool) {
func (c *Cluster) AddNodeBasicSorted(uri URI) (*Node, bool) {
n := c.NodeByURI(uri)
if n != nil {
return n, false
@ -497,7 +499,7 @@ func (h *jmphasher) Hash(key uint64, n int) int {
func (c *Cluster) Open() error {
// Cluster always comes up in state STARTING until cluster membership is determined.
c.State = NodeStateStarting
c.State = ClusterStateStarting
// Load topology file if it exists.
if err := c.loadTopology(); err != nil {
@ -511,13 +513,11 @@ func (c *Cluster) Open() error {
return fmt.Errorf("considerTopology: %v", err)
}
// Add the local node to the cluster and update state.
fmt.Println("IS Coord")
c.AddHost(c.URI)
c.AddNode(c.URI)
c.setState(state)
} else {
// Add the local node to the cluster.
fmt.Println("NOT Coord")
c.AddHost(c.URI)
c.AddNode(c.URI)
}
// Start the EventReceiver.
@ -525,9 +525,9 @@ func (c *Cluster) Open() error {
return fmt.Errorf("starting EventReceiver: %v", err)
}
// Open NodeSet communication.
if err := c.NodeSet.Open(); err != nil {
return fmt.Errorf("opening NodeSet: %v", err)
// Open MemberSet communication.
if err := c.MemberSet.Open(); err != nil {
return fmt.Errorf("opening MemberSet: %v", err)
}
// Listen for cluster-resize events.
@ -546,11 +546,11 @@ func (c *Cluster) Close() error {
}
func (c *Cluster) needTopologyAgreement() bool {
return c.State == NodeStateStarting && !URISlicesAreEqual(c.Topology.URISet, c.URISet())
return c.State == ClusterStateStarting && !URISlicesAreEqual(c.Topology.NodeSet, c.NodeSet())
}
func (c *Cluster) haveTopologyAgreement() bool {
return URISlicesAreEqual(c.Topology.URISet, c.URISet())
return URISlicesAreEqual(c.Topology.NodeSet, c.NodeSet())
}
func (c *Cluster) handleJoiningHost(uri URI) error {
@ -571,7 +571,7 @@ func (c *Cluster) handleJoiningHost(uri URI) error {
case ResizeJobStateDone:
c.CompleteCurrentJob(ResizeJobStateDone)
// Add uri to the cluster.
return c.AddHost(uri)
return c.AddNode(uri)
case ResizeJobStateAborted:
c.CompleteCurrentJob(ResizeJobStateAborted)
}
@ -605,7 +605,7 @@ func (c *Cluster) listenForJoins() {
// Only change state to NORMAL if we have successfully added at least one host.
if uriJoined {
// Put the cluster back to state NORMAL and broadcast.
if err := c.setStateAndBroadcast(NodeStateNormal); err != nil {
if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil {
c.logger().Printf("setStateAndBroadcast error: err=%s", err)
}
}
@ -662,10 +662,10 @@ func (c *Cluster) generateResizeJob(addURI URI) *ResizeJob {
toCluster.Hasher = c.Hasher
toCluster.PartitionN = c.PartitionN
toCluster.ReplicaN = c.ReplicaN
toCluster.AddNode(addURI)
toCluster.AddNodeBasicSorted(addURI)
// Add to the ResizeJob the instructions for each index.
for _, idx := range c.IndexReporter.Indexes() {
for _, idx := range c.Holder.Indexes() {
// dataDiff is map[string][]*internal.ResizeSource, where string is
// a host in toCluster.
dataDiff := c.DataDiff(toCluster, idx)
@ -704,22 +704,72 @@ func (c *Cluster) CompleteCurrentJob(state string) {
// followResizeInstruction is run by any node that receives a ResizeInstruction.
func (c *Cluster) followResizeInstruction(instr *internal.ResizeInstruction) {
go func() {
// Request each source file in ResizeSources.
for _, src := range instr.Sources {
/************************************************************/
// TODO travis: get the data files from other nodes.
fmt.Printf("\n**** Get slice %d for index %s from host %s ****\n\n", src.Slice, src.Index, src.URI)
for i := 0; i <= 4; i++ {
fmt.Printf(" %d", i)
time.Sleep(1 * time.Second)
}
fmt.Println("")
/************************************************************/
}
// Prepare the return message.
complete := &internal.ResizeInstructionComplete{
JobID: instr.JobID,
URI: instr.URI,
Error: "",
}
// Stop processing on any error.
if err := func() error {
// Create a client for calling remote nodes.
client, err := NewClientFromURI(&c.URI, nil) // TODO: ClientOptions
if err != nil {
return err
}
// Request each source file in ResizeSources.
for _, src := range instr.Sources {
fmt.Printf("\n**** Get slice %d for index %s from host %s ****\n\n", src.Slice, src.Index, src.URI)
srcURI := decodeURI(src.URI)
// TODO: there's a possible race condition here;
// if NodeStatus has not been shared with the joining
// node (and the schema created locally), then
// the following Frame() lookup could fail.
// Retrieve frame.
f := c.Holder.Frame(src.Index, src.Frame)
if f == nil {
return ErrFrameNotFound
}
// Create view.
v, err := f.CreateViewIfNotExists(src.View)
if err != nil {
return err
}
// Create the local fragment.
frag, err := v.CreateFragmentIfNotExists(src.Slice)
if err != nil {
return err
}
// Stream slice from remote node.
rd, err := client.RetrieveSliceFromURI(context.Background(), src.Index, src.Frame, src.View, src.Slice, srcURI)
if err != nil {
return err
} else if rd == nil {
return fmt.Errorf("slice %v doesn't exist on host: %s", src.Slice, src.URI)
}
// Write to local frame and always close reader.
if err := func() error {
defer rd.Close()
if _, err := frag.ReadFrom(rd); err != nil {
return err
}
return nil
}(); err != nil {
return err
}
}
return nil
}(); err != nil {
complete.Error = err.Error()
}
node := &Node{
@ -732,8 +782,15 @@ func (c *Cluster) followResizeInstruction(instr *internal.ResizeInstruction) {
}
func (c *Cluster) MarkResizeInstructionComplete(complete *internal.ResizeInstructionComplete) error {
j := c.Job(complete.JobID)
// Abort the job if an error exists in the complete object.
if complete.Error != "" {
j.result <- ResizeJobStateAborted
return errors.New(complete.Error)
}
j.mu.Lock()
defer j.mu.Unlock()
@ -870,9 +927,9 @@ func (j *ResizeJob) distributeResizeInstructions() error {
return nil
}
type URISet []URI
type NodeSet []URI
func (u URISet) ToHostPortStrings() []string {
func (u NodeSet) ToHostPortStrings() []string {
other := make([]string, 0, len(u))
for _, uri := range u {
other = append(other, uri.HostPort())
@ -882,8 +939,8 @@ func (u URISet) ToHostPortStrings() []string {
// Topology represents the list of hosts in the cluster.
type Topology struct {
mu sync.RWMutex
URISet []URI
mu sync.RWMutex
NodeSet []URI
}
func NewTopology() *Topology {
@ -898,7 +955,7 @@ func (t *Topology) ContainsURI(uri URI) bool {
}
func (t *Topology) containsURI(uri URI) bool {
for _, turi := range t.URISet {
for _, turi := range t.NodeSet {
if turi == uri {
return true
}
@ -906,14 +963,14 @@ func (t *Topology) containsURI(uri URI) bool {
return false
}
// AddHost adds the uri to the topology and returns true if added.
// AddNode adds the uri to the topology and returns true if added.
func (t *Topology) AddURI(uri URI) bool {
t.mu.Lock()
defer t.mu.Unlock()
if t.containsURI(uri) {
return false
}
t.URISet = append(t.URISet, uri)
t.NodeSet = append(t.NodeSet, uri)
return true
}
@ -955,7 +1012,7 @@ func encodeTopology(topology *Topology) *internal.Topology {
return nil
}
return &internal.Topology{
URISet: encodeURIs(topology.URISet),
NodeSet: encodeURIs(topology.NodeSet),
}
}
@ -965,30 +1022,30 @@ func decodeTopology(topology *internal.Topology) (*Topology, error) {
}
t := &Topology{
URISet: decodeURIs(topology.URISet),
NodeSet: decodeURIs(topology.NodeSet),
}
return t, nil
}
func (c *Cluster) considerTopology() (string, error) {
// If there is no .topology file, it's safe to go to state NORMAL.
if len(c.Topology.URISet) == 0 {
return NodeStateNormal, nil
if len(c.Topology.NodeSet) == 0 {
return ClusterStateNormal, nil
}
// The local node (coordinator) must be in the .topology.
if !c.Topology.ContainsURI(c.Coordinator) {
return "", fmt.Errorf("coordinator %s is not in topology: %v", c.Coordinator, c.Topology.URISet)
return "", fmt.Errorf("coordinator %s is not in topology: %v", c.Coordinator, c.Topology.NodeSet)
}
// If local node is the only thing in .topology, continue to state NORMAL.
if len(c.Topology.URISet) == 1 {
return NodeStateNormal, nil
if len(c.Topology.NodeSet) == 1 {
return ClusterStateNormal, nil
}
// Keep the cluster in state "STARTING" until hearing from all nodes.
// Topology contains 2+ hosts.
return NodeStateStarting, nil
return ClusterStateStarting, nil
}
// ReceiveEvent represents an implementation of EventHandler.
@ -1012,14 +1069,14 @@ func (c *Cluster) ReceiveEvent(e *NodeEvent) error {
}
uri := e.URI
if err := c.AddHost(uri); err != nil {
if err := c.AddNode(uri); err != nil {
return err
}
// If the result of the previous AddHost completed the joining of nodes
// If the result of the previous AddNode completed the joining of nodes
// in the topology, then change the state to NORMAL.
if c.haveTopologyAgreement() {
return c.setStateAndBroadcast(NodeStateNormal)
return c.setStateAndBroadcast(ClusterStateNormal)
}
return nil
@ -1031,17 +1088,17 @@ func (c *Cluster) ReceiveEvent(e *NodeEvent) error {
}
// If the index does not yet have data, go ahead and add the node.
if !c.IndexReporter.HasData() {
if !c.Holder.HasData() {
uri := e.URI
if err := c.AddHost(uri); err != nil {
if err := c.AddNode(uri); err != nil {
return err
}
return c.setStateAndBroadcast(NodeStateNormal)
return c.setStateAndBroadcast(ClusterStateNormal)
}
// If the cluster has data, we need to change to RESIZING and
// kick off the resizing process.
if err := c.setStateAndBroadcast(NodeStateResizing); err != nil {
if err := c.setStateAndBroadcast(ClusterStateResizing); err != nil {
return err
}
c.joiningURIs <- e.URI
@ -1061,8 +1118,8 @@ func (c *Cluster) mergeClusterStatus(cs *internal.ClusterStatus) error {
return nil
}
for _, uri := range decodeURIs(cs.URISet) {
c.AddHost(uri)
for _, uri := range decodeURIs(cs.NodeSet) {
c.AddNode(uri)
}
c.setState(cs.State)

4
cluster_internal_test.go
View file

@ -31,8 +31,8 @@ func TestFragCombos(t *testing.T) {
if err != nil {
t.Fatal(err)
}
c.AddNode(*uri0)
c.AddNode(*uri1)
c.AddNodeBasicSorted(*uri0)
c.AddNodeBasicSorted(*uri1)
tests := []struct {
idx string

11
cluster_test.go
View file

@ -114,7 +114,7 @@ func TestCluster_Nodes(t *testing.T) {
{URI: uri2},
}
t.Run("URISet", func(t *testing.T) {
t.Run("NodeSet", func(t *testing.T) {
actual := pilosa.Nodes(nodes).URIs()
expected := []pilosa.URI{uri0, uri1, uri2}
if !reflect.DeepEqual(actual, expected) {
@ -198,17 +198,17 @@ func TestCluster_Topology(t *testing.T) {
base := test.NewURIFromHostPort("host0", 0)
invalid := test.NewURIFromHostPort("invalid", 0)
t.Run("AddHost", func(t *testing.T) {
err := c1.AddHost(uri1)
t.Run("AddNode", func(t *testing.T) {
err := c1.AddNode(uri1)
if err != nil {
t.Fatal(err)
}
// add the same host.
err = c1.AddHost(uri1)
err = c1.AddNode(uri1)
if err != nil {
t.Fatal(err)
}
err = c1.AddHost(uri2)
err = c1.AddNode(uri2)
if err != nil {
t.Fatal(err)
}
@ -261,7 +261,6 @@ func TestCluster_Resize(t *testing.T) {
// Cluster 1
c1 := test.NewCluster(3)
c1.IndexReporter = h1
c1.ReplicaN = 2
// Cluster 2

52
gossip/gossip.go
View file

@ -28,10 +28,10 @@ import (
"github.com/pilosa/pilosa/internal"
)
// GossipNodeSet represents a gossip implementation of NodeSet using memberlist
// GossipNodeSet also represents a gossip implementation of pilosa.Broadcaster
// GossipNodeSet also represents an implementation of memberlist.Delegate
type GossipNodeSet struct {
// GossipMemberSet represents a gossip implementation of MemberSet using memberlist
// GossipMemberSet also represents a gossip implementation of pilosa.Broadcaster
// GossipMemberSet also represents an implementation of memberlist.Delegate
type GossipMemberSet struct {
memberlist *memberlist.Memberlist
handler pilosa.BroadcastHandler
@ -44,8 +44,8 @@ type GossipNodeSet struct {
LogOutput io.Writer
}
// Nodes implements the NodeSet interface and returns a list of nodes in the cluster.
func (g *GossipNodeSet) Nodes() []*pilosa.Node {
// Nodes implements the MemberSet interface and returns a list of nodes in the cluster.
func (g *GossipMemberSet) Nodes() []*pilosa.Node {
a := make([]*pilosa.Node, 0, g.memberlist.NumMembers())
for _, n := range g.memberlist.Members() {
uri, _ := pilosa.NewURIFromAddress(n.Name)
@ -56,15 +56,15 @@ func (g *GossipNodeSet) Nodes() []*pilosa.Node {
}
// Start implements the BroadcastReceiver interface and sets the BroadcastHandler.
func (g *GossipNodeSet) Start(h pilosa.BroadcastHandler) error {
func (g *GossipMemberSet) Start(h pilosa.BroadcastHandler) error {
g.handler = h
return nil
}
// Open implements the NodeSet interface to start network activity.
func (g *GossipNodeSet) Open() error {
// Open implements the MemberSet interface to start network activity.
func (g *GossipMemberSet) Open() error {
if g.handler == nil {
return fmt.Errorf("opening GossipNodeSet: you must call Start(pilosa.BroadcastHandler) before calling Open()")
return fmt.Errorf("opening GossipMemberSet: you must call Start(pilosa.BroadcastHandler) before calling Open()")
}
err := error(nil)
@ -88,7 +88,7 @@ func (g *GossipNodeSet) Open() error {
// attach to gossip seed node
nodes := []*pilosa.Node{&pilosa.Node{URI: *uri}} //TODO: support a list of seeds
err = g.joinWithRetry(pilosa.URISet(pilosa.Nodes(nodes).URIs()).ToHostPortStrings())
err = g.joinWithRetry(pilosa.NodeSet(pilosa.Nodes(nodes).URIs()).ToHostPortStrings())
if err != nil {
return err
}
@ -96,7 +96,7 @@ func (g *GossipNodeSet) Open() error {
}
// joinWithRetry wraps the standard memberlist Join function in a retry.
func (g *GossipNodeSet) joinWithRetry(hosts []string) error {
func (g *GossipMemberSet) joinWithRetry(hosts []string) error {
err := retry(60, 2*time.Second, func() error {
_, err := g.memberlist.Join(hosts)
return err
@ -120,8 +120,8 @@ func retry(attempts int, sleep time.Duration, fn func() error) (err error) {
return fmt.Errorf("after %d attempts, last error: %s", attempts, err)
}
// logger returns a logger for the GossipNodeSet.
func (g *GossipNodeSet) logger() *log.Logger {
// logger returns a logger for the GossipMemberSet.
func (g *GossipMemberSet) logger() *log.Logger {
return log.New(g.LogOutput, "", log.LstdFlags)
}
@ -132,9 +132,9 @@ type gossipConfig struct {
memberlistConfig *memberlist.Config
}
// NewGossipNodeSet returns a new instance of GossipNodeSet.
func NewGossipNodeSet(name string, gossipHost string, gossipPort int, gossipSeed string, server *pilosa.Server, secretKey []byte) *GossipNodeSet {
g := &GossipNodeSet{
// NewGossipMemberSet returns a new instance of GossipMemberSet.
func NewGossipMemberSet(name string, gossipHost string, gossipPort int, gossipSeed string, server *pilosa.Server, secretKey []byte) *GossipMemberSet {
g := &GossipMemberSet{
LogOutput: server.LogOutput,
}
@ -148,7 +148,7 @@ func NewGossipNodeSet(name string, gossipHost string, gossipPort int, gossipSeed
g.config.memberlistConfig.BindPort = gossipPort
g.config.memberlistConfig.AdvertiseAddr = pilosa.HostToIP(gossipHost)
g.config.memberlistConfig.AdvertisePort = gossipPort
//g.config.memberlistConfig.PushPullInterval = 15 * time.Second // Default is 15s in DefaultLocalConfig.
g.config.memberlistConfig.PushPullInterval = 0 * time.Second // Default is 15s in DefaultLocalConfig. // TODO travis: change this from 0
g.config.memberlistConfig.Delegate = g
g.config.memberlistConfig.SecretKey = secretKey
g.config.memberlistConfig.Events = server.Cluster.EventReceiver.(memberlist.EventDelegate)
@ -159,7 +159,7 @@ func NewGossipNodeSet(name string, gossipHost string, gossipPort int, gossipSeed
}
// SendSync implementation of the Broadcaster interface.
func (g *GossipNodeSet) SendSync(pb proto.Message) error {
func (g *GossipMemberSet) SendSync(pb proto.Message) error {
msg, err := pilosa.MarshalMessage(pb)
if err != nil {
return err
@ -187,7 +187,7 @@ func (g *GossipNodeSet) SendSync(pb proto.Message) error {
}
// SendAsync implementation of the Broadcaster interface.
func (g *GossipNodeSet) SendAsync(pb proto.Message) error {
func (g *GossipMemberSet) SendAsync(pb proto.Message) error {
msg, err := pilosa.MarshalMessage(pb)
if err != nil {
return err
@ -202,7 +202,7 @@ func (g *GossipNodeSet) SendAsync(pb proto.Message) error {
}
// SendTo implementation of the Broadcaster interface.
func (g *GossipNodeSet) SendTo(to *pilosa.Node, pb proto.Message) error {
func (g *GossipMemberSet) SendTo(to *pilosa.Node, pb proto.Message) error {
msg, err := pilosa.MarshalMessage(pb)
if err != nil {
return err
@ -221,13 +221,13 @@ func (g *GossipNodeSet) SendTo(to *pilosa.Node, pb proto.Message) error {
}
// NodeMeta implementation of the memberlist.Delegate interface.
func (g *GossipNodeSet) NodeMeta(limit int) []byte {
func (g *GossipMemberSet) NodeMeta(limit int) []byte {
return []byte{}
}
// NotifyMsg implementation of the memberlist.Delegate interface
// called when a user-data message is received.
func (g *GossipNodeSet) NotifyMsg(b []byte) {
func (g *GossipMemberSet) NotifyMsg(b []byte) {
m, err := pilosa.UnmarshalMessage(b)
if err != nil {
g.logger().Printf("unmarshal message error: %s", err)
@ -241,13 +241,13 @@ func (g *GossipNodeSet) NotifyMsg(b []byte) {
// GetBroadcasts implementation of the memberlist.Delegate interface
// called when user data messages can be broadcast.
func (g *GossipNodeSet) GetBroadcasts(overhead, limit int) [][]byte {
func (g *GossipMemberSet) GetBroadcasts(overhead, limit int) [][]byte {
return g.broadcasts.GetBroadcasts(overhead, limit)
}
// LocalState implementation of the memberlist.Delegate interface
// sends this Node's state data.
func (g *GossipNodeSet) LocalState(join bool) []byte {
func (g *GossipMemberSet) LocalState(join bool) []byte {
pb, err := g.statusHandler.LocalStatus()
if err != nil {
g.logger().Printf("error getting local state, err=%s", err)
@ -265,7 +265,7 @@ func (g *GossipNodeSet) LocalState(join bool) []byte {
// MergeRemoteState implementation of the memberlist.Delegate interface
// receive and process the remote side's LocalState.
func (g *GossipNodeSet) MergeRemoteState(buf []byte, join bool) {
func (g *GossipMemberSet) MergeRemoteState(buf []byte, join bool) {
// Unmarshal nodestate data.
var pb internal.NodeStatus
if err := proto.Unmarshal(buf, &pb); err != nil {

14
handler.go
View file

@ -224,8 +224,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,
URISet: decodeURIs(cs.URISet),
State: cs.State,
NodeSet: decodeURIs(cs.NodeSet),
}); err != nil {
h.logger().Printf("write status response error: %s", err)
}
@ -236,8 +236,8 @@ type getSchemaResponse struct {
}
type getStatusResponse struct {
State string `json:"state"`
URISet []URI `json:"uri-set"`
State string `json:"state"`
NodeSet []URI `json:"nodes"`
}
// handlePostQuery handles /query requests.
@ -1347,7 +1347,7 @@ func (h *Handler) handleGetFragmentNodes(w http.ResponseWriter, r *http.Request)
}
}
// handleGetFragmentBackup handles GET /fragment/data requests.
// handleGetFragmentData handles GET /fragment/data requests.
func (h *Handler) handleGetFragmentData(w http.ResponseWriter, r *http.Request) {
// Read slice parameter.
q := r.URL.Query()
@ -1370,7 +1370,7 @@ func (h *Handler) handleGetFragmentData(w http.ResponseWriter, r *http.Request)
}
}
// handlePostFragmentRestore handles POST /fragment/data requests.
// handlePostFragmentData handles POST /fragment/data requests.
func (h *Handler) handlePostFragmentData(w http.ResponseWriter, r *http.Request) {
// Read slice parameter.
q := r.URL.Query()
@ -1408,7 +1408,7 @@ func (h *Handler) handlePostFragmentData(w http.ResponseWriter, r *http.Request)
}
}
// handleGetFragmentData handles GET /fragment/block/data requests.
// handleGetFragmentBlockData handles GET /fragment/block/data requests.
func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Request) {
// Read request object.
var req internal.BlockDataRequest

2
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","uri-set":[{"scheme":"http","host":"localhost","port":10101}]}`+"\n" {
} else if body := w.Body.String(); body != `{"state":"NORMAL","nodes":[{"scheme":"http","host":"localhost","port":10101}]}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}

5
holder.go
View file

@ -649,8 +649,3 @@ func (s *HolderSyncer) syncFragment(index, frame, view string, slice uint64) err
return nil
}
type IndexReporter interface {
HasData() bool
Indexes() []*Index
}

246
internal/private.pb.go
View file

@ -450,11 +450,8 @@ func (m *Schema) GetIndexes() []*Index {
}
type Index struct {
Name string `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"`
// IndexMeta Meta = 2;
// uint64 MaxSlice = 3;
Frames []*Frame `protobuf:"bytes,4,rep,name=Frames" json:"Frames,omitempty"`
// repeated uint64 Slices = 5;
Name string `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"`
Frames []*Frame `protobuf:"bytes,4,rep,name=Frames" json:"Frames,omitempty"`
InputDefinitions []*InputDefinition `protobuf:"bytes,6,rep,name=InputDefinitions" json:"InputDefinitions,omitempty"`
}
@ -705,8 +702,8 @@ func (m *NodeStatus) GetSchema() *Schema {
}
type ClusterStatus struct {
State string `protobuf:"bytes,1,opt,name=State,proto3" json:"State,omitempty"`
URISet []*URI `protobuf:"bytes,2,rep,name=URISet" json:"URISet,omitempty"`
State string `protobuf:"bytes,1,opt,name=State,proto3" json:"State,omitempty"`
NodeSet []*URI `protobuf:"bytes,2,rep,name=NodeSet" json:"NodeSet,omitempty"`
}
func (m *ClusterStatus) Reset() { *m = ClusterStatus{} }
@ -721,9 +718,9 @@ func (m *ClusterStatus) GetState() string {
return ""
}
func (m *ClusterStatus) GetURISet() []*URI {
func (m *ClusterStatus) GetNodeSet() []*URI {
if m != nil {
return m.URISet
return m.NodeSet
}
return nil
}
@ -889,8 +886,9 @@ func (m *ResizeSource) GetSlice() uint64 {
}
type ResizeInstructionComplete struct {
JobID int64 `protobuf:"varint,1,opt,name=JobID,proto3" json:"JobID,omitempty"`
URI *URI `protobuf:"bytes,2,opt,name=URI" json:"URI,omitempty"`
JobID int64 `protobuf:"varint,1,opt,name=JobID,proto3" json:"JobID,omitempty"`
URI *URI `protobuf:"bytes,2,opt,name=URI" json:"URI,omitempty"`
Error string `protobuf:"bytes,3,opt,name=Error,proto3" json:"Error,omitempty"`
}
func (m *ResizeInstructionComplete) Reset() { *m = ResizeInstructionComplete{} }
@ -914,8 +912,15 @@ func (m *ResizeInstructionComplete) GetURI() *URI {
return nil
}
func (m *ResizeInstructionComplete) GetError() string {
if m != nil {
return m.Error
}
return ""
}
type Topology struct {
URISet []*URI `protobuf:"bytes,1,rep,name=URISet" json:"URISet,omitempty"`
NodeSet []*URI `protobuf:"bytes,1,rep,name=NodeSet" json:"NodeSet,omitempty"`
}
func (m *Topology) Reset() { *m = Topology{} }
@ -923,9 +928,9 @@ func (m *Topology) String() string { return proto.CompactTextString(m
func (*Topology) ProtoMessage() {}
func (*Topology) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{28} }
func (m *Topology) GetURISet() []*URI {
func (m *Topology) GetNodeSet() []*URI {
if m != nil {
return m.URISet
return m.NodeSet
}
return nil
}
@ -1878,8 +1883,8 @@ func (m *ClusterStatus) MarshalTo(dAtA []byte) (int, error) {
i = encodeVarintPrivate(dAtA, i, uint64(len(m.State)))
i += copy(dAtA[i:], m.State)
}
if len(m.URISet) > 0 {
for _, msg := range m.URISet {
if len(m.NodeSet) > 0 {
for _, msg := range m.NodeSet {
dAtA[i] = 0x12
i++
i = encodeVarintPrivate(dAtA, i, uint64(msg.Size()))
@ -2105,6 +2110,12 @@ func (m *ResizeInstructionComplete) MarshalTo(dAtA []byte) (int, error) {
}
i += n17
}
if len(m.Error) > 0 {
dAtA[i] = 0x1a
i++
i = encodeVarintPrivate(dAtA, i, uint64(len(m.Error)))
i += copy(dAtA[i:], m.Error)
}
return i, nil
}
@ -2123,8 +2134,8 @@ func (m *Topology) MarshalTo(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if len(m.URISet) > 0 {
for _, msg := range m.URISet {
if len(m.NodeSet) > 0 {
for _, msg := range m.NodeSet {
dAtA[i] = 0xa
i++
i = encodeVarintPrivate(dAtA, i, uint64(msg.Size()))
@ -2545,8 +2556,8 @@ func (m *ClusterStatus) Size() (n int) {
if l > 0 {
n += 1 + l + sovPrivate(uint64(l))
}
if len(m.URISet) > 0 {
for _, e := range m.URISet {
if len(m.NodeSet) > 0 {
for _, e := range m.NodeSet {
l = e.Size()
n += 1 + l + sovPrivate(uint64(l))
}
@ -2650,14 +2661,18 @@ func (m *ResizeInstructionComplete) Size() (n int) {
l = m.URI.Size()
n += 1 + l + sovPrivate(uint64(l))
}
l = len(m.Error)
if l > 0 {
n += 1 + l + sovPrivate(uint64(l))
}
return n
}
func (m *Topology) Size() (n int) {
var l int
_ = l
if len(m.URISet) > 0 {
for _, e := range m.URISet {
if len(m.NodeSet) > 0 {
for _, e := range m.NodeSet {
l = e.Size()
n += 1 + l + sovPrivate(uint64(l))
}
@ -5799,7 +5814,7 @@ func (m *ClusterStatus) Unmarshal(dAtA []byte) error {
iNdEx = postIndex
case 2:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field URISet", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field NodeSet", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
@ -5823,8 +5838,8 @@ func (m *ClusterStatus) Unmarshal(dAtA []byte) error {
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.URISet = append(m.URISet, &URI{})
if err := m.URISet[len(m.URISet)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
m.NodeSet = append(m.NodeSet, &URI{})
if err := m.NodeSet[len(m.NodeSet)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
@ -6568,6 +6583,35 @@ func (m *ResizeInstructionComplete) Unmarshal(dAtA []byte) error {
return err
}
iNdEx = postIndex
case 3:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Error", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthPrivate
}
postIndex := iNdEx + intStringLen
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Error = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipPrivate(dAtA[iNdEx:])
@ -6620,7 +6664,7 @@ func (m *Topology) Unmarshal(dAtA []byte) error {
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field URISet", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field NodeSet", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
@ -6644,8 +6688,8 @@ func (m *Topology) Unmarshal(dAtA []byte) error {
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.URISet = append(m.URISet, &URI{})
if err := m.URISet[len(m.URISet)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
m.NodeSet = append(m.NodeSet, &URI{})
if err := m.NodeSet[len(m.NodeSet)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
@ -6778,77 +6822,77 @@ var (
func init() { proto.RegisterFile("private.proto", fileDescriptorPrivate) }
var fileDescriptorPrivate = []byte{
// 1137 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x57, 0x4f, 0x6f, 0x1b, 0x45,
0x14, 0x67, 0xbd, 0x8e, 0x6b, 0x3f, 0xc7, 0x8d, 0x33, 0x94, 0xc8, 0x89, 0x22, 0xd7, 0x8c, 0x04,
0x0d, 0x95, 0x08, 0x34, 0x95, 0x10, 0x04, 0x21, 0x41, 0xe3, 0x54, 0x5d, 0x68, 0x4a, 0x19, 0x27,
0x45, 0xe2, 0x80, 0x34, 0xb1, 0x87, 0x74, 0x95, 0xf5, 0x8e, 0xd9, 0x1d, 0x27, 0x71, 0x0f, 0xdc,
0xe0, 0x00, 0x5f, 0x80, 0x3b, 0x67, 0xbe, 0x07, 0x47, 0x3e, 0x02, 0x0a, 0x1f, 0x02, 0x89, 0x0b,
0x68, 0xde, 0xce, 0xec, 0xae, 0xff, 0xc5, 0x4a, 0x6e, 0xfb, 0xde, 0xbc, 0xf7, 0xe6, 0x37, 0xbf,
0xf7, 0x67, 0x66, 0xa1, 0x36, 0x88, 0xfc, 0x33, 0xae, 0xc4, 0xf6, 0x20, 0x92, 0x4a, 0x92, 0xb2,
0x1f, 0x2a, 0x11, 0x85, 0x3c, 0xa0, 0x5f, 0x42, 0xc5, 0x0b, 0x7b, 0xe2, 0xe2, 0x40, 0x28, 0x4e,
0x5a, 0x50, 0xdd, 0x93, 0xc1, 0xb0, 0x1f, 0x3e, 0xe5, 0xc7, 0x22, 0x68, 0x38, 0x2d, 0x67, 0xab,
0xc2, 0xf2, 0x2a, 0x6d, 0x71, 0xe8, 0xf7, 0xc5, 0x57, 0x43, 0x1e, 0xaa, 0x61, 0xbf, 0x51, 0x48,
0x2c, 0x72, 0x2a, 0xfa, 0xaf, 0x03, 0x95, 0xc7, 0x11, 0xef, 0x0b, 0x8c, 0xb8, 0x01, 0x65, 0x26,
0xcf, 0xf3, 0xe1, 0x52, 0x99, 0xbc, 0x0d, 0xb7, 0xbd, 0xf0, 0x4c, 0x44, 0xb1, 0xd8, 0x0f, 0xf9,
0x71, 0x20, 0x7a, 0x18, 0xae, 0xcc, 0x26, 0xb4, 0x64, 0x13, 0x2a, 0x7b, 0xbc, 0xfb, 0x52, 0x1c,
0x8e, 0x06, 0xa2, 0xe1, 0x62, 0x90, 0x4c, 0x91, 0xae, 0x76, 0xfc, 0x57, 0xa2, 0x51, 0x6c, 0x39,
0x5b, 0x35, 0x96, 0x29, 0x26, 0xf1, 0x2e, 0x4d, 0xe1, 0x25, 0x14, 0x96, 0x19, 0x0f, 0x4f, 0x52,
0x0c, 0x25, 0xc4, 0x30, 0xa6, 0x23, 0xf7, 0xa0, 0xf4, 0xd8, 0x17, 0x41, 0x2f, 0x6e, 0xdc, 0x6a,
0xb9, 0x5b, 0xd5, 0x9d, 0x95, 0x6d, 0xcb, 0xdf, 0x36, 0xea, 0x99, 0x59, 0xa6, 0x14, 0x6e, 0x7b,
0xfd, 0x81, 0x8c, 0x14, 0x13, 0xf1, 0x40, 0x86, 0xb1, 0x20, 0x75, 0x70, 0xf7, 0xa3, 0xc8, 0x9c,
0x5d, 0x7f, 0xd2, 0x1f, 0xa0, 0xfe, 0x28, 0x90, 0xdd, 0xd3, 0x36, 0x57, 0x9c, 0x89, 0xef, 0x87,
0x22, 0x56, 0xe4, 0x0e, 0x2c, 0x61, 0x16, 0x8c, 0x5d, 0x22, 0x68, 0x2d, 0x32, 0x69, 0x68, 0x4e,
0x04, 0xad, 0x45, 0x7f, 0xa4, 0xa2, 0xc8, 0x12, 0x41, 0x6b, 0x3b, 0x81, 0xdf, 0x4d, 0x28, 0x28,
0xb2, 0x44, 0x20, 0x04, 0x8a, 0x2f, 0x7c, 0x71, 0x6e, 0xce, 0x8d, 0xdf, 0xd4, 0x83, 0xd5, 0xdc,
0xfe, 0x06, 0xe6, 0x1a, 0x94, 0x98, 0x3c, 0xf7, 0xda, 0x71, 0xc3, 0x69, 0xb9, 0x5b, 0x45, 0x66,
0x24, 0x64, 0x17, 0xd3, 0xaf, 0x97, 0x0a, 0xb8, 0x94, 0x29, 0xe8, 0x3a, 0x2c, 0x21, 0xd5, 0xfa,
0x94, 0x99, 0xaf, 0xfe, 0xa4, 0xff, 0x39, 0x50, 0x39, 0xe0, 0x17, 0x08, 0x23, 0x26, 0x9f, 0x40,
0xb9, 0xa3, 0x78, 0xd8, 0xe3, 0x51, 0x0f, 0x8d, 0xaa, 0x3b, 0x6f, 0x66, 0x14, 0xa6, 0x66, 0xdb,
0xd6, 0x66, 0x3f, 0x54, 0xd1, 0x88, 0xa5, 0x2e, 0x64, 0x17, 0x6e, 0x99, 0x9a, 0x40, 0x0c, 0xd5,
0x9d, 0xd6, 0x2c, 0xef, 0xb4, 0x6c, 0xb4, 0xb3, 0x75, 0xd8, 0xf8, 0x18, 0x6a, 0x63, 0x61, 0x35,
0xd6, 0x53, 0x31, 0xb2, 0x19, 0x39, 0x15, 0x23, 0xcd, 0xdd, 0x19, 0x0f, 0x86, 0x09, 0xcf, 0x45,
0x96, 0x08, 0xbb, 0x85, 0x0f, 0x9d, 0x8d, 0x5d, 0x58, 0xce, 0x47, 0xbd, 0x8e, 0x2f, 0xfd, 0x16,
0xc8, 0x5e, 0x24, 0xb8, 0x12, 0x08, 0xef, 0x40, 0xc4, 0x31, 0x3f, 0x11, 0xf3, 0x33, 0x9d, 0x64,
0xaf, 0x90, 0xcf, 0xde, 0x26, 0x54, 0xbc, 0xd8, 0x1e, 0xdc, 0xc5, 0xba, 0xcc, 0x14, 0xf4, 0x3e,
0x90, 0xb6, 0x08, 0x84, 0x12, 0xa6, 0x7f, 0xaf, 0x88, 0x4f, 0x3b, 0x16, 0xcb, 0x62, 0x5b, 0x72,
0x0f, 0x8a, 0xba, 0x75, 0x11, 0x4a, 0x75, 0xe7, 0xf5, 0x8c, 0xe9, 0x74, 0x4e, 0x30, 0x34, 0xa0,
0xbe, 0x0d, 0x6a, 0xda, 0x7d, 0xc1, 0x01, 0x67, 0x94, 0xb2, 0xdd, 0xca, 0x9d, 0xdc, 0x2a, 0x1d,
0x20, 0x66, 0xab, 0x4f, 0xed, 0x59, 0x6f, 0xba, 0x15, 0xfd, 0xc6, 0x68, 0x75, 0x4b, 0x3c, 0xd3,
0xab, 0x89, 0x0f, 0x7e, 0xcf, 0x3f, 0xf2, 0x04, 0x0e, 0x1d, 0x5b, 0xf7, 0x50, 0xdc, 0x70, 0x5b,
0xae, 0x8e, 0x8d, 0x02, 0x7d, 0x08, 0xa5, 0x4e, 0xf7, 0xa5, 0xe8, 0x73, 0xf2, 0x8e, 0x2e, 0xd4,
0x9e, 0xb8, 0x10, 0xb1, 0x29, 0xf3, 0x95, 0x09, 0xfa, 0x98, 0x5d, 0xa7, 0xbf, 0x38, 0x06, 0xfd,
0x1c, 0x44, 0x25, 0xdc, 0x3b, 0x6e, 0x14, 0xa7, 0x26, 0x8e, 0xd6, 0x33, 0xb3, 0x4c, 0xf6, 0xa1,
0xee, 0x85, 0x83, 0xa1, 0x6a, 0x8b, 0xef, 0xfc, 0xd0, 0x57, 0xbe, 0x0c, 0xe3, 0x46, 0x09, 0x5d,
0xd6, 0xf3, 0x5b, 0x8f, 0x59, 0xb0, 0x29, 0x17, 0xfa, 0x93, 0x03, 0x2b, 0x13, 0xca, 0x05, 0xb8,
0x0a, 0x57, 0xe3, 0xfa, 0x20, 0x1d, 0x99, 0x2e, 0x1a, 0x36, 0xe7, 0xa2, 0x19, 0x9f, 0xa0, 0xbf,
0x39, 0x70, 0x67, 0x96, 0xc1, 0x4c, 0x34, 0x4d, 0x80, 0xe7, 0x91, 0xdf, 0xe7, 0xd1, 0xe8, 0x0b,
0x31, 0x32, 0xb7, 0x47, 0x4e, 0x43, 0xbe, 0x86, 0xb5, 0x89, 0x58, 0x9f, 0x75, 0x13, 0x8a, 0x12,
0x50, 0x77, 0xe7, 0x82, 0x4a, 0xec, 0xd8, 0x1c, 0x77, 0xfa, 0x8f, 0x03, 0x6f, 0xcc, 0x5c, 0xca,
0xaa, 0xcf, 0xc9, 0x17, 0xfa, 0x7d, 0xa8, 0xbf, 0xd0, 0x83, 0xa1, 0x2d, 0x62, 0xe5, 0x87, 0x5c,
0x5b, 0x9a, 0xf2, 0x9c, 0xd2, 0x13, 0x0f, 0xca, 0xa8, 0x3b, 0xe0, 0x03, 0x03, 0xf3, 0xdd, 0x05,
0x30, 0xb7, 0xad, 0xbd, 0x99, 0x9b, 0x56, 0xd4, 0x60, 0x70, 0x8e, 0xdb, 0x4b, 0x01, 0x05, 0x3d,
0x11, 0xc7, 0x1c, 0xae, 0x35, 0xd5, 0x24, 0x6c, 0xda, 0x49, 0x32, 0x86, 0xe4, 0xea, 0x9e, 0xfc,
0x08, 0x20, 0x33, 0x35, 0xed, 0x7e, 0x45, 0x7d, 0xe6, 0x8c, 0xe9, 0x13, 0xd8, 0xb4, 0x63, 0xee,
0x1a, 0x1b, 0xda, 0x6a, 0x29, 0x64, 0xd5, 0x42, 0xf7, 0xc1, 0x3d, 0x62, 0x9e, 0xbe, 0xea, 0xb0,
0x5b, 0x6d, 0x8a, 0x8c, 0xa4, 0x5d, 0x9e, 0xc8, 0x58, 0x59, 0x17, 0xfd, 0xad, 0x75, 0xcf, 0x65,
0xa4, 0x10, 0x71, 0x8d, 0xe1, 0x37, 0xfd, 0xd9, 0x01, 0x78, 0x26, 0x7b, 0xa2, 0xa3, 0xb8, 0x1a,
0xc6, 0xe4, 0x2e, 0x46, 0xc5, 0x58, 0xd5, 0x9d, 0x5a, 0x76, 0xa6, 0x23, 0xe6, 0x31, 0xdc, 0xef,
0x41, 0xee, 0x22, 0x9c, 0x9e, 0x30, 0xe9, 0x12, 0xcb, 0x5d, 0x97, 0x5b, 0x76, 0xa0, 0x18, 0xaa,
0xea, 0x99, 0x7d, 0xa2, 0x37, 0xa0, 0x39, 0x7d, 0x0a, 0xb5, 0xbd, 0x60, 0x18, 0x2b, 0x11, 0x19,
0x38, 0xfa, 0x26, 0x51, 0x5c, 0xa5, 0xf5, 0x87, 0x02, 0x79, 0x0b, 0x4a, 0x47, 0xcc, 0xeb, 0x08,
0x65, 0xda, 0x76, 0x02, 0xa7, 0x59, 0xa4, 0x1d, 0x58, 0x9a, 0xdf, 0x6c, 0x04, 0x8a, 0xf8, 0x02,
0x33, 0xfc, 0xe0, 0xe3, 0xab, 0x0e, 0xee, 0x81, 0x9f, 0x24, 0xd4, 0x65, 0xfa, 0x13, 0x35, 0xfc,
0x02, 0x0b, 0x4e, 0x6b, 0xb8, 0xbe, 0x7b, 0x56, 0x93, 0x04, 0xea, 0x61, 0x79, 0x93, 0x5b, 0xc2,
0x3e, 0x62, 0xdc, 0xdc, 0x23, 0xe6, 0x77, 0x07, 0x56, 0x99, 0x88, 0xfd, 0x57, 0xc2, 0x0b, 0x63,
0x15, 0x0d, 0xd3, 0xe6, 0xfb, 0x5c, 0x1e, 0x7b, 0x6d, 0x8c, 0xea, 0xb2, 0x44, 0xb0, 0x19, 0x2a,
0xcc, 0xcd, 0xd0, 0x7b, 0xfa, 0xd9, 0x2b, 0xa3, 0x9e, 0xee, 0x40, 0x19, 0x19, 0xce, 0x27, 0x0c,
0xf3, 0x16, 0xe4, 0x7d, 0xb8, 0xd5, 0x91, 0xc3, 0xa8, 0x9b, 0x8e, 0xe7, 0xb5, 0xcc, 0x38, 0x41,
0x95, 0x2c, 0x33, 0x6b, 0x46, 0x7f, 0x74, 0x60, 0x39, 0xbf, 0xb2, 0xb8, 0x6c, 0x52, 0x86, 0x0a,
0x33, 0x19, 0x72, 0x67, 0x31, 0x54, 0xcc, 0x18, 0xca, 0x9e, 0x14, 0x4b, 0xb9, 0x27, 0x05, 0x65,
0xb0, 0x3e, 0x45, 0xdb, 0x9e, 0xec, 0x0f, 0x74, 0x7e, 0x6e, 0x48, 0x1f, 0x7d, 0x00, 0xe5, 0x43,
0x39, 0x90, 0x81, 0x3c, 0x19, 0xe5, 0x0a, 0xcd, 0xb9, 0xa2, 0xd0, 0x1e, 0xd5, 0xff, 0xb8, 0x6c,
0x3a, 0x7f, 0x5e, 0x36, 0x9d, 0xbf, 0x2e, 0x9b, 0xce, 0xaf, 0x7f, 0x37, 0x5f, 0x3b, 0x2e, 0xe1,
0x8f, 0xc9, 0xc3, 0xff, 0x03, 0x00, 0x00, 0xff, 0xff, 0xb2, 0x75, 0xc6, 0x9e, 0xa9, 0x0c, 0x00,
0x00,
// 1144 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x57, 0xcd, 0x6e, 0x23, 0x45,
0x10, 0x66, 0x3c, 0xb6, 0x63, 0x97, 0xe3, 0x8d, 0xd3, 0x84, 0xc8, 0x89, 0x22, 0xaf, 0xe9, 0x03,
0x09, 0x2b, 0x11, 0x20, 0x91, 0x10, 0x04, 0x21, 0xc1, 0xc6, 0x5e, 0xed, 0x00, 0x09, 0x4b, 0x3b,
0xbb, 0x48, 0x1c, 0x90, 0x3a, 0x76, 0x93, 0x8c, 0x32, 0x9e, 0x31, 0x33, 0xed, 0x24, 0xde, 0x03,
0x37, 0x38, 0xc0, 0x0b, 0x70, 0xe7, 0xcc, 0x7b, 0x70, 0xe4, 0x11, 0x50, 0x78, 0x08, 0x24, 0x2e,
0xac, 0xba, 0xa6, 0x7b, 0x66, 0xfc, 0x17, 0x2b, 0xb9, 0x4d, 0x55, 0x57, 0x55, 0x7f, 0xfd, 0xd5,
0x4f, 0xf7, 0x40, 0x75, 0x10, 0xba, 0x97, 0x5c, 0x8a, 0xdd, 0x41, 0x18, 0xc8, 0x80, 0x94, 0x5c,
0x5f, 0x8a, 0xd0, 0xe7, 0x1e, 0xfd, 0x0a, 0xca, 0x8e, 0xdf, 0x13, 0xd7, 0x47, 0x42, 0x72, 0xd2,
0x84, 0xca, 0x61, 0xe0, 0x0d, 0xfb, 0xfe, 0x97, 0xfc, 0x54, 0x78, 0x75, 0xab, 0x69, 0xed, 0x94,
0x59, 0x56, 0xa5, 0x2c, 0x4e, 0xdc, 0xbe, 0xf8, 0x7a, 0xc8, 0x7d, 0x39, 0xec, 0xd7, 0x73, 0xb1,
0x45, 0x46, 0x45, 0xff, 0xb3, 0xa0, 0xfc, 0x24, 0xe4, 0x7d, 0x81, 0x11, 0x37, 0xa1, 0xc4, 0x82,
0xab, 0x6c, 0xb8, 0x44, 0x26, 0x6f, 0xc1, 0x03, 0xc7, 0xbf, 0x14, 0x61, 0x24, 0xda, 0x3e, 0x3f,
0xf5, 0x44, 0x0f, 0xc3, 0x95, 0xd8, 0x84, 0x96, 0x6c, 0x41, 0xf9, 0x90, 0x77, 0xcf, 0xc5, 0xc9,
0x68, 0x20, 0xea, 0x36, 0x06, 0x49, 0x15, 0xc9, 0x6a, 0xc7, 0x7d, 0x29, 0xea, 0xf9, 0xa6, 0xb5,
0x53, 0x65, 0xa9, 0x62, 0x12, 0x6f, 0x61, 0x0a, 0x2f, 0xa1, 0xb0, 0xcc, 0xb8, 0x7f, 0x96, 0x60,
0x28, 0x22, 0x86, 0x31, 0x1d, 0xd9, 0x86, 0xe2, 0x13, 0x57, 0x78, 0xbd, 0xa8, 0xbe, 0xd4, 0xb4,
0x77, 0x2a, 0x7b, 0x2b, 0xbb, 0x86, 0xbf, 0x5d, 0xd4, 0x33, 0xbd, 0x4c, 0x29, 0x3c, 0x70, 0xfa,
0x83, 0x20, 0x94, 0x4c, 0x44, 0x83, 0xc0, 0x8f, 0x04, 0xa9, 0x81, 0xdd, 0x0e, 0x43, 0x7d, 0x76,
0xf5, 0x49, 0x7f, 0x84, 0xda, 0x63, 0x2f, 0xe8, 0x5e, 0xb4, 0xb8, 0xe4, 0x4c, 0xfc, 0x30, 0x14,
0x91, 0x24, 0x6b, 0x50, 0xc0, 0x2c, 0x68, 0xbb, 0x58, 0x50, 0x5a, 0x64, 0x52, 0xd3, 0x1c, 0x0b,
0x4a, 0x8b, 0xfe, 0x48, 0x45, 0x9e, 0xc5, 0x82, 0xd2, 0x76, 0x3c, 0xb7, 0x1b, 0x53, 0x90, 0x67,
0xb1, 0x40, 0x08, 0xe4, 0x5f, 0xb8, 0xe2, 0x4a, 0x9f, 0x1b, 0xbf, 0xa9, 0x03, 0xab, 0x99, 0xfd,
0x35, 0xcc, 0x75, 0x28, 0xb2, 0xe0, 0xca, 0x69, 0x45, 0x75, 0xab, 0x69, 0xef, 0xe4, 0x99, 0x96,
0x90, 0x5d, 0x4c, 0xbf, 0x5a, 0xca, 0xe1, 0x52, 0xaa, 0xa0, 0x1b, 0x50, 0x40, 0xaa, 0xd5, 0x29,
0x53, 0x5f, 0xf5, 0x49, 0xff, 0xb7, 0xa0, 0x7c, 0xc4, 0xaf, 0x11, 0x46, 0x44, 0x3e, 0x81, 0x52,
0x47, 0x72, 0xbf, 0xc7, 0xc3, 0x1e, 0x1a, 0x55, 0xf6, 0xde, 0x4c, 0x29, 0x4c, 0xcc, 0x76, 0x8d,
0x4d, 0xdb, 0x97, 0xe1, 0x88, 0x25, 0x2e, 0xe4, 0x00, 0x96, 0x74, 0x4d, 0x20, 0x86, 0xca, 0x5e,
0x73, 0x96, 0x77, 0x52, 0x36, 0xca, 0xd9, 0x38, 0x6c, 0x7e, 0x0c, 0xd5, 0xb1, 0xb0, 0x0a, 0xeb,
0x85, 0x18, 0x99, 0x8c, 0x5c, 0x88, 0x91, 0xe2, 0xee, 0x92, 0x7b, 0xc3, 0x98, 0xe7, 0x3c, 0x8b,
0x85, 0x83, 0xdc, 0x87, 0xd6, 0xe6, 0x01, 0x2c, 0x67, 0xa3, 0xde, 0xc5, 0x97, 0x7e, 0x07, 0xe4,
0x30, 0x14, 0x5c, 0x0a, 0x84, 0x77, 0x24, 0xa2, 0x88, 0x9f, 0x89, 0xf9, 0x99, 0x8e, 0xb3, 0x97,
0xcb, 0x66, 0x6f, 0x0b, 0xca, 0x4e, 0x64, 0x0e, 0x6e, 0x63, 0x5d, 0xa6, 0x0a, 0xfa, 0x08, 0x48,
0x4b, 0x78, 0x42, 0x0a, 0xdd, 0xbf, 0xb7, 0xc4, 0xa7, 0x1d, 0x83, 0x65, 0xb1, 0x2d, 0xd9, 0x86,
0xbc, 0x6a, 0x5d, 0x84, 0x52, 0xd9, 0x7b, 0x3d, 0x65, 0x3a, 0x99, 0x13, 0x0c, 0x0d, 0xa8, 0x6b,
0x82, 0xea, 0x76, 0x5f, 0x70, 0xc0, 0x19, 0xa5, 0x6c, 0xb6, 0xb2, 0x27, 0xb7, 0x4a, 0x06, 0x88,
0xde, 0xea, 0x53, 0x73, 0xd6, 0xfb, 0x6e, 0x45, 0xbf, 0xd5, 0x5a, 0xd5, 0x12, 0xc7, 0x6a, 0x35,
0xf6, 0xc1, 0xef, 0xf9, 0x47, 0x9e, 0xc0, 0xa1, 0x62, 0xab, 0x1e, 0x8a, 0xea, 0x76, 0xd3, 0x56,
0xb1, 0x51, 0xa0, 0xfb, 0x50, 0xec, 0x74, 0xcf, 0x45, 0x9f, 0x93, 0xb7, 0x55, 0xa1, 0xf6, 0xc4,
0xb5, 0x88, 0x74, 0x99, 0xaf, 0x4c, 0xd0, 0xc7, 0xcc, 0x3a, 0xfd, 0xd5, 0xd2, 0xe8, 0xe7, 0x20,
0x2a, 0xe2, 0xde, 0x51, 0x3d, 0x3f, 0x35, 0x71, 0x94, 0x9e, 0xe9, 0x65, 0xd2, 0x86, 0x9a, 0xe3,
0x0f, 0x86, 0xb2, 0x25, 0xbe, 0x77, 0x7d, 0x57, 0xba, 0x81, 0x1f, 0xd5, 0x8b, 0xe8, 0xb2, 0x91,
0xdd, 0x7a, 0xcc, 0x82, 0x4d, 0xb9, 0xd0, 0x9f, 0x2d, 0x58, 0x99, 0x50, 0x2e, 0xc0, 0x95, 0xbb,
0x1d, 0xd7, 0x07, 0xc9, 0xc8, 0xb4, 0xd1, 0xb0, 0x31, 0x17, 0xcd, 0xf8, 0x04, 0xfd, 0xdd, 0x82,
0xb5, 0x59, 0x06, 0x33, 0xd1, 0x34, 0x00, 0x9e, 0x85, 0x6e, 0x9f, 0x87, 0xa3, 0x2f, 0xc4, 0x48,
0xdf, 0x1e, 0x19, 0x0d, 0xf9, 0x06, 0xd6, 0x27, 0x62, 0x7d, 0xd6, 0x8d, 0x29, 0x8a, 0x41, 0x3d,
0x9c, 0x0b, 0x2a, 0xb6, 0x63, 0x73, 0xdc, 0xe9, 0xbf, 0x16, 0xbc, 0x31, 0x73, 0x29, 0xad, 0x3e,
0x2b, 0x5b, 0xe8, 0x8f, 0xa0, 0xf6, 0x42, 0x0d, 0x86, 0x96, 0x88, 0xa4, 0xeb, 0x73, 0x65, 0xa9,
0xcb, 0x73, 0x4a, 0x4f, 0x1c, 0x28, 0xa1, 0xee, 0x88, 0x0f, 0x34, 0xcc, 0x77, 0x16, 0xc0, 0xdc,
0x35, 0xf6, 0x7a, 0x6e, 0x1a, 0x51, 0x81, 0xc1, 0x39, 0x6e, 0x2e, 0x05, 0x14, 0xd4, 0x44, 0x1c,
0x73, 0xb8, 0xd3, 0x54, 0x0b, 0x60, 0xcb, 0x4c, 0x92, 0x31, 0x24, 0xb7, 0xf7, 0xe4, 0x47, 0x00,
0xa9, 0xa9, 0x6e, 0xf7, 0x5b, 0xea, 0x33, 0x63, 0x4c, 0x9f, 0xc2, 0x96, 0x19, 0x73, 0x77, 0xd8,
0xd0, 0x54, 0x4b, 0x2e, 0xad, 0x16, 0xda, 0x06, 0xfb, 0x39, 0x73, 0xd4, 0x55, 0x87, 0xdd, 0x6a,
0x52, 0xa4, 0x25, 0xe5, 0xf2, 0x34, 0x88, 0xa4, 0x71, 0x51, 0xdf, 0x4a, 0xf7, 0x2c, 0x08, 0x25,
0x22, 0xae, 0x32, 0xfc, 0xa6, 0xbf, 0x58, 0x00, 0xc7, 0x41, 0x4f, 0x74, 0x24, 0x97, 0xc3, 0x88,
0x3c, 0xc4, 0xa8, 0x18, 0xab, 0xb2, 0x57, 0x4d, 0xcf, 0xf4, 0x9c, 0x39, 0x0c, 0xf7, 0x7b, 0x3f,
0x73, 0x11, 0x4e, 0x4f, 0x98, 0x64, 0x89, 0x65, 0xae, 0xcb, 0x1d, 0x33, 0x50, 0x34, 0x55, 0xb5,
0xd4, 0x3e, 0xd6, 0x6b, 0xd0, 0x9c, 0x1e, 0x43, 0xf5, 0xd0, 0x1b, 0x46, 0x52, 0x84, 0x1a, 0x8e,
0xba, 0x49, 0x24, 0x97, 0x49, 0xfd, 0xa1, 0x40, 0xb6, 0x61, 0x09, 0x21, 0x0b, 0xa9, 0xfb, 0x76,
0x02, 0xa8, 0x59, 0xa5, 0x1d, 0x28, 0xcc, 0x6f, 0x37, 0x02, 0x79, 0x7c, 0x83, 0x69, 0x86, 0xf0,
0xf9, 0x55, 0x03, 0xfb, 0xc8, 0x8d, 0x53, 0x6a, 0x33, 0xf5, 0x89, 0x1a, 0x7e, 0x8d, 0x25, 0xa7,
0x34, 0x5c, 0xdd, 0x3e, 0xab, 0x71, 0x0a, 0xd5, 0xb8, 0xbc, 0xcf, 0x3d, 0x61, 0x9e, 0x31, 0x76,
0xe6, 0x19, 0xf3, 0x87, 0x05, 0xab, 0x4c, 0x44, 0xee, 0x4b, 0xe1, 0xf8, 0x91, 0x0c, 0x87, 0x49,
0xfb, 0x7d, 0x1e, 0x9c, 0x3a, 0x2d, 0x8c, 0x6a, 0xb3, 0x58, 0x30, 0x39, 0xca, 0xcd, 0xcd, 0xd1,
0xbb, 0xea, 0xe1, 0x1b, 0x84, 0x3d, 0xd5, 0x83, 0x41, 0xa8, 0x59, 0x9f, 0x30, 0xcc, 0x5a, 0x90,
0xf7, 0x60, 0xa9, 0x13, 0x0c, 0xc3, 0x6e, 0x32, 0xa0, 0xd7, 0x53, 0xe3, 0x18, 0x55, 0xbc, 0xcc,
0x8c, 0x19, 0xfd, 0xc9, 0x82, 0xe5, 0xec, 0xca, 0xe2, 0xc2, 0x49, 0x18, 0xca, 0xcd, 0x64, 0xc8,
0x9e, 0xc5, 0x50, 0x3e, 0x65, 0x28, 0x7d, 0x54, 0x14, 0x32, 0x8f, 0x0a, 0x7a, 0x0e, 0x1b, 0x53,
0xb4, 0x1d, 0x06, 0xfd, 0x81, 0xca, 0xcf, 0x7d, 0xe9, 0x5b, 0x83, 0x42, 0x3b, 0x0c, 0x35, 0x71,
0x65, 0x16, 0x0b, 0x74, 0x1f, 0x4a, 0x27, 0xc1, 0x20, 0xf0, 0x82, 0xb3, 0x51, 0xb6, 0x00, 0xad,
0xdb, 0x0a, 0xf0, 0x71, 0xed, 0xcf, 0x9b, 0x86, 0xf5, 0xd7, 0x4d, 0xc3, 0xfa, 0xfb, 0xa6, 0x61,
0xfd, 0xf6, 0x4f, 0xe3, 0xb5, 0xd3, 0x22, 0xfe, 0xb2, 0xec, 0xbf, 0x0a, 0x00, 0x00, 0xff, 0xff,
0x18, 0x9c, 0xa1, 0x75, 0xc3, 0x0c, 0x00, 0x00,
}

5
internal/private.proto
View file

@ -129,7 +129,7 @@ message NodeStatus {
message ClusterStatus {
string State = 1;
repeated URI URISet = 2;
repeated URI NodeSet = 2;
}
message Field {
@ -163,9 +163,10 @@ message ResizeSource {
message ResizeInstructionComplete {
int64 JobID = 1;
URI URI = 2;
string Error = 3;
}
message Topology {
repeated URI URISet = 1;
repeated URI NodeSet = 1;
}

42
server.go
View file

@ -380,7 +380,7 @@ func (s *Server) LocalStatus() (proto.Message, error) {
return &ns, nil
}
// ClusterStatus returns the ClusterState and URISet for the cluster.
// ClusterStatus returns the ClusterState and NodeSet for the cluster.
func (s *Server) ClusterStatus() (proto.Message, error) {
return s.Cluster.Status(), nil
}
@ -396,6 +396,25 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
return nil
}
// Sync schema.
// Create indexes that don't exist.
for _, index := range ns.Schema.Indexes {
opt := IndexOptions{}
idx, err := s.Holder.CreateIndexIfNotExists(index.Name, opt)
if err != nil {
return err
}
// Create frames that don't exist.
for _, f := range index.Frames {
opt := decodeFrameOptions(f.Meta)
_, err := idx.CreateFrameIfNotExists(f.Name, *opt)
if err != nil {
return err
}
}
// TODO: Create inputDefinitions that don't exist.
}
// Sync maxSlices (standard).
oldmaxslices := s.Holder.MaxSlices()
for index, newMax := range ns.MaxSlices.Standard {
@ -428,25 +447,6 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
}
}
// Sync schema.
// Create indexes that don't exist.
for _, index := range ns.Schema.Indexes {
opt := IndexOptions{}
idx, err := s.Holder.CreateIndexIfNotExists(index.Name, opt)
if err != nil {
return err
}
// Create frames that don't exist.
for _, f := range index.Frames {
opt := decodeFrameOptions(f.Meta)
_, err := idx.CreateFrameIfNotExists(f.Name, *opt)
if err != nil {
return err
}
}
// TODO: Create inputDefinitions that don't exist.
}
return nil
}
@ -525,7 +525,7 @@ func CountOpenFiles() int {
}
// StatusHandler specifies the methods which an object must implement to share
// state in the cluster. These are used by the GossipNodeSet to implement the
// state in the cluster. These are used by the GossipMemberSet to implement the
// LocalState and MergeRemoteState methods of memberlist.Delegate
type StatusHandler interface {
LocalStatus() (proto.Message, error)

14
server/server.go
View file

@ -122,7 +122,7 @@ func (m *Command) SetupServer() error {
cluster := pilosa.NewCluster()
cluster.ReplicaN = m.Config.Cluster.ReplicaN
cluster.IndexReporter = m.Server.Holder
cluster.Holder = m.Server.Holder
m.Server.Cluster = cluster
@ -209,15 +209,15 @@ func (m *Command) SetupServer() error {
// get the host portion of addr to use for binding
gossipHost := uri.Host()
m.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver()
gossipNodeSet := gossip.NewGossipNodeSet(uri.String(), gossipHost, gossipPort, gossipSeed, m.Server, gossipKey)
m.Server.Cluster.NodeSet = gossipNodeSet
m.Server.Broadcaster = gossipNodeSet
m.Server.BroadcastReceiver = gossipNodeSet
gossipMemberSet := gossip.NewGossipMemberSet(uri.String(), gossipHost, gossipPort, gossipSeed, m.Server, gossipKey)
m.Server.Cluster.MemberSet = gossipMemberSet
m.Server.Broadcaster = gossipMemberSet
m.Server.BroadcastReceiver = gossipMemberSet
case pilosa.ClusterStatic, pilosa.ClusterNone:
m.Server.Broadcaster = pilosa.NopBroadcaster
m.Server.Cluster.NodeSet = pilosa.NewStaticNodeSet()
m.Server.Cluster.MemberSet = pilosa.NewStaticMemberSet()
m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver
err := m.Server.Cluster.NodeSet.(*pilosa.StaticNodeSet).Join(m.Server.Cluster.Nodes)
err := m.Server.Cluster.MemberSet.(*pilosa.StaticMemberSet).Join(m.Server.Cluster.Nodes)
if err != nil {
return err
}

18
server/server_test.go
View file

@ -445,18 +445,18 @@ func TestMain_SendReceiveMessage(t *testing.T) {
}
gossipSeed := gossipHost + ":" + freePorts[0]
topology := &pilosa.Topology{URISet: []pilosa.URI{m0.Server.URI, m1.Server.URI}}
topology := &pilosa.Topology{NodeSet: []pilosa.URI{m0.Server.URI, m1.Server.URI}}
m0.Server.Cluster.Coordinator = m0.Server.URI
m0.Server.Cluster.Topology = topology
m0.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver()
gossipNodeSet0 := gossip.NewGossipNodeSet(m0.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m0.Server, nil)
m0.Server.Cluster.NodeSet = gossipNodeSet0
m0.Server.Broadcaster = gossipNodeSet0
gossipMemberSet0 := gossip.NewGossipMemberSet(m0.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m0.Server, nil)
m0.Server.Cluster.MemberSet = gossipMemberSet0
m0.Server.Broadcaster = gossipMemberSet0
m0.Server.Handler.Broadcaster = m0.Server.Broadcaster
m0.Server.Holder.Broadcaster = m0.Server.Broadcaster
m0.Server.BroadcastReceiver = gossipNodeSet0
m0.Server.BroadcastReceiver = gossipMemberSet0
if err := m0.Server.BroadcastReceiver.Start(m0.Server); err != nil {
t.Fatal(err)
@ -481,12 +481,12 @@ func TestMain_SendReceiveMessage(t *testing.T) {
m1.Server.Cluster.Coordinator = m0.Server.URI
m1.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver()
gossipNodeSet1 := gossip.NewGossipNodeSet(m1.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m1.Server, nil)
m1.Server.Cluster.NodeSet = gossipNodeSet1
m1.Server.Broadcaster = gossipNodeSet1
gossipMemberSet1 := gossip.NewGossipMemberSet(m1.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m1.Server, nil)
m1.Server.Cluster.MemberSet = gossipMemberSet1
m1.Server.Broadcaster = gossipMemberSet1
m1.Server.Handler.Broadcaster = m1.Server.Broadcaster
m1.Server.Holder.Broadcaster = m1.Server.Broadcaster
m1.Server.BroadcastReceiver = gossipNodeSet1
m1.Server.BroadcastReceiver = gossipMemberSet1
if err := m1.Server.BroadcastReceiver.Start(m1.Server); err != nil {
t.Fatal(err)

4
test/handler.go
View file

@ -85,8 +85,8 @@ func (s *Server) LocalStatus() (proto.Message, error) {
func (s *Server) ClusterStatus() (proto.Message, error) {
uri := pilosa.DefaultURI()
return &internal.ClusterStatus{
State: pilosa.NodeStateNormal,
URISet: []*internal.URI{uri.Encode()},
State: pilosa.ClusterStateNormal,
NodeSet: []*internal.URI{uri.Encode()},
}, nil
}