Change Config.Coordinator from URI to bool

This commit is contained in:
Travis Turner 2018-03-05 12:14:10 -06:00
parent 5a2e1c3703
commit 1cc45b22a2
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
13 changed files with 218 additions and 180 deletions

View file

@ -66,8 +66,9 @@ const (
// Node represents a node in the cluster.
type Node struct {
ID string `json:"id"`
URI URI `json:"uri"`
ID string `json:"id"`
URI URI `json:"uri"`
IsCoordinator bool `json:"isCoordinator"`
}
func (n Node) String() string {
@ -86,8 +87,9 @@ func EncodeNodes(a []*Node) []*internal.Node {
// EncodeNode converts a Node into its internal representation.
func EncodeNode(n *Node) *internal.Node {
return &internal.Node{
ID: n.ID,
URI: n.URI.Encode(),
ID: n.ID,
URI: n.URI.Encode(),
IsCoordinator: n.IsCoordinator,
}
}
@ -106,8 +108,9 @@ func DecodeNodes(a []*internal.Node) []*Node {
// DecodeNode converts a proto message into a Node.
func DecodeNode(node *internal.Node) *Node {
return &Node{
ID: node.ID,
URI: decodeURI(node.URI),
ID: node.ID,
URI: decodeURI(node.URI),
IsCoordinator: node.IsCoordinator,
}
}
@ -238,7 +241,7 @@ type Cluster struct {
// Required for cluster Resize.
Static bool // Static is primarily used for testing in a non-gossip environment.
state string
Coordinator URI
Coordinator string
Holder *Holder
Broadcaster Broadcaster
@ -291,12 +294,12 @@ func (c *Cluster) logger() *log.Logger {
// Coordinator returns the coordinator node.
func (c *Cluster) CoordinatorNode() *Node {
return c.nodeByURI(c.Coordinator)
return c.nodeByID(c.Coordinator)
}
// IsCoordinator is true if this node is the coordinator.
func (c *Cluster) IsCoordinator() bool {
return c.Static || c.Coordinator == c.Node.URI
return c.Coordinator == c.Node.ID
}
// SetCoordinator updates the Coordinator to n.
@ -308,8 +311,8 @@ func (c *Cluster) SetCoordinator(n *Node) bool {
return false
}
if c.Coordinator != newNode.URI {
c.Coordinator = newNode.URI
if c.Coordinator != newNode.ID {
c.Coordinator = newNode.ID
return true
}
return false
@ -320,6 +323,11 @@ func (c *Cluster) SetCoordinator(n *Node) bool {
func (c *Cluster) AddNode(node *Node) error {
c.logger().Printf("add node %s to cluster on %s", node, c.Node)
// If the node being added is the coordinator, set it for this node.
if node.IsCoordinator {
c.Coordinator = node.ID
}
// add to cluster
if !c.addNodeBasicSorted(node) {
return nil

View file

@ -180,8 +180,8 @@ func TestFragSources(t *testing.T) {
"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)},
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node1", &internal.URI{"http", "host1", 10101}, false}, "i", "f", "standard", uint64(2)},
},
},
err: "",
@ -192,11 +192,11 @@ func TestFragSources(t *testing.T) {
idx: idx,
expected: map[string][]*internal.ResizeSource{
"node0": []*internal.ResizeSource{
{&internal.Node{"node1", &internal.URI{"http", "host1", 10101}}, "i", "f", "standard", uint64(1)},
{&internal.Node{"node1", &internal.URI{"http", "host1", 10101}, false}, "i", "f", "standard", uint64(1)},
},
"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)},
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(2)},
},
},
err: "",
@ -207,11 +207,11 @@ func TestFragSources(t *testing.T) {
idx: idx,
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)},
{&internal.Node{"node2", &internal.URI{"http", "host2", 10101}, false}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node2", &internal.URI{"http", "host2", 10101}, false}, "i", "f", "standard", uint64(2)},
},
"node1": []*internal.ResizeSource{
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}}, "i", "f", "standard", uint64(3)},
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(3)},
},
"node2": []*internal.ResizeSource{},
},

View file

@ -180,10 +180,10 @@ func TestCluster_Coordinator(t *testing.T) {
c1 := *pilosa.NewCluster()
c1.Node = node1
c1.Coordinator = node1.URI
c1.Coordinator = node1.ID
c2 := *pilosa.NewCluster()
c2.Node = node2
c2.Coordinator = node1.URI
c2.Coordinator = node1.ID
t.Run("IsCoordinator", func(t *testing.T) {
if !c1.IsCoordinator() {
@ -519,14 +519,14 @@ func TestCluster_SetCoordinator(t *testing.T) {
// Set coordinator to the same value.
if c.SetCoordinator(oldNode) {
t.Errorf("did not expect coordinator to change")
} else if c.Coordinator != oldNode.URI {
} else if c.Coordinator != oldNode.ID {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, oldNode.URI)
}
// Set coordinator to a new value.
if !c.SetCoordinator(newNode) {
t.Errorf("expected coordinator to change")
} else if c.Coordinator != newNode.URI {
} else if c.Coordinator != newNode.ID {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, newNode.URI)
}
})

View file

@ -139,7 +139,7 @@ type Config struct {
Cluster struct {
Disabled bool `toml:"disabled"`
Coordinator string `toml:"coordinator"`
Coordinator bool `toml:"coordinator"`
ReplicaN int `toml:"replicas"`
Hosts []string `toml:"hosts"`
LongQueryTime Duration `toml:"long-query-time"`
@ -183,7 +183,7 @@ func NewConfig() *Config {
// Cluster config.
c.Cluster.Disabled = DefaultClusterDisabled
// c.Cluster.Coordinator = ""
// c.Cluster.Coordinator = false
c.Cluster.ReplicaN = DefaultReplicaN
c.Cluster.Hosts = []string{}
c.Cluster.LongQueryTime = Duration(time.Minute)

View file

@ -34,7 +34,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
// Cluster
flags.BoolVarP(&srv.Config.Cluster.Disabled, "cluster.disabled", "", srv.Config.Cluster.Disabled, "Disabled multi-node cluster communication (used for testing)")
flags.StringVarP(&srv.Config.Cluster.Coordinator, "cluster.coordinator", "", "", "Host that will act as cluster coordinator during startup and resizing.")
flags.BoolVarP(&srv.Config.Cluster.Coordinator, "cluster.coordinator", "", srv.Config.Cluster.Coordinator, "Host that will act as cluster coordinator during startup and resizing.")
flags.IntVarP(&srv.Config.Cluster.ReplicaN, "cluster.replicas", "", 1, "Number of hosts each piece of data should be stored on.")
flags.StringSliceVarP(&srv.Config.Cluster.Hosts, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster.")
flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", "", time.Minute, "Duration that will trigger log and stat messages for slow queries.")

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":[{"id":"test-node","uri":{"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},"isCoordinator":false}]}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}
@ -1212,7 +1212,7 @@ 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 body := w.Body.String(); body != `[{"id":"node2","uri":{"scheme":"http","host":"host2"}},{"id":"node0","uri":{"scheme":"http","host":"host0"}}]`+"\n" {
} else if body := w.Body.String(); body != `[{"id":"node2","uri":{"scheme":"http","host":"host2"},"isCoordinator":false},{"id":"node0","uri":{"scheme":"http","host":"host0"},"isCoordinator":false}]`+"\n" {
t.Fatalf("unexpected body: %q", body)
}

View file

@ -742,8 +742,9 @@ func (m *URI) GetPort() uint32 {
}
type Node struct {
ID string `protobuf:"bytes,1,opt,name=ID,proto3" json:"ID,omitempty"`
URI *URI `protobuf:"bytes,2,opt,name=URI" json:"URI,omitempty"`
ID string `protobuf:"bytes,1,opt,name=ID,proto3" json:"ID,omitempty"`
URI *URI `protobuf:"bytes,2,opt,name=URI" json:"URI,omitempty"`
IsCoordinator bool `protobuf:"varint,3,opt,name=IsCoordinator,proto3" json:"IsCoordinator,omitempty"`
}
func (m *Node) Reset() { *m = Node{} }
@ -765,6 +766,13 @@ func (m *Node) GetURI() *URI {
return nil
}
func (m *Node) GetIsCoordinator() bool {
if m != nil {
return m.IsCoordinator
}
return false
}
type NodeStateMessage struct {
NodeID string `protobuf:"bytes,1,opt,name=NodeID,proto3" json:"NodeID,omitempty"`
State string `protobuf:"bytes,2,opt,name=State,proto3" json:"State,omitempty"`
@ -2136,6 +2144,16 @@ func (m *Node) MarshalTo(dAtA []byte) (int, error) {
}
i += n12
}
if m.IsCoordinator {
dAtA[i] = 0x18
i++
if m.IsCoordinator {
dAtA[i] = 1
} else {
dAtA[i] = 0
}
i++
}
return i, nil
}
@ -3068,6 +3086,9 @@ func (m *Node) Size() (n int) {
l = m.URI.Size()
n += 1 + l + sovPrivate(uint64(l))
}
if m.IsCoordinator {
n += 2
}
return n
}
@ -6575,6 +6596,26 @@ func (m *Node) Unmarshal(dAtA []byte) error {
return err
}
iNdEx = postIndex
case 3:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field IsCoordinator", wireType)
}
var v int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= (int(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
m.IsCoordinator = bool(v != 0)
default:
iNdEx = preIndex
skippy, err := skipPrivate(dAtA[iNdEx:])
@ -8313,86 +8354,87 @@ var (
func init() { proto.RegisterFile("private.proto", fileDescriptorPrivate) }
var fileDescriptorPrivate = []byte{
// 1296 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0x4f, 0x6f, 0x1b, 0x45,
0x14, 0x67, 0xbd, 0xb6, 0x63, 0x3f, 0xc7, 0x89, 0x33, 0x4d, 0x83, 0x13, 0x45, 0xae, 0x19, 0x01,
0x0d, 0x95, 0x88, 0x8a, 0x2b, 0x01, 0x0d, 0xaa, 0x54, 0x12, 0xbb, 0xea, 0x02, 0x09, 0x65, 0x9c,
0x06, 0x89, 0x03, 0xd2, 0xc4, 0x1e, 0xd2, 0x55, 0xd6, 0xbb, 0x66, 0x77, 0x9c, 0xc4, 0x3d, 0x70,
0x44, 0x48, 0x88, 0x3b, 0xe2, 0xca, 0x97, 0xe1, 0xc8, 0x47, 0x40, 0xe1, 0x43, 0x20, 0x71, 0x01,
0xcd, 0xbf, 0xdd, 0xf5, 0xdf, 0x90, 0xd0, 0xdb, 0xbe, 0xdf, 0xfb, 0x33, 0xbf, 0x79, 0xef, 0xcd,
0x9b, 0x59, 0x28, 0xf7, 0x43, 0xf7, 0x8c, 0x72, 0xb6, 0xdd, 0x0f, 0x03, 0x1e, 0xa0, 0x82, 0xeb,
0x73, 0x16, 0xfa, 0xd4, 0xc3, 0x9f, 0x43, 0xd1, 0xf1, 0xbb, 0xec, 0x62, 0x9f, 0x71, 0x8a, 0xea,
0x50, 0xda, 0x0b, 0xbc, 0x41, 0xcf, 0xff, 0x8c, 0x1e, 0x33, 0xaf, 0x6a, 0xd5, 0xad, 0xad, 0x22,
0x49, 0x43, 0xc2, 0xe2, 0xd0, 0xed, 0xb1, 0x2f, 0x06, 0xd4, 0xe7, 0x83, 0x5e, 0x35, 0xa3, 0x2c,
0x52, 0x10, 0xfe, 0xdb, 0x82, 0xe2, 0x93, 0x90, 0xf6, 0x98, 0x8c, 0xb8, 0x01, 0x05, 0x12, 0x9c,
0xa7, 0xc3, 0xc5, 0x32, 0x7a, 0x1b, 0x96, 0x1c, 0xff, 0x8c, 0x85, 0x11, 0x6b, 0xf9, 0xf4, 0xd8,
0x63, 0x5d, 0x19, 0xae, 0x40, 0xc6, 0x50, 0xb4, 0x09, 0xc5, 0x3d, 0xda, 0x79, 0xc1, 0x0e, 0x87,
0x7d, 0x56, 0xb5, 0x65, 0x90, 0x04, 0x88, 0xb5, 0x6d, 0xf7, 0x25, 0xab, 0x66, 0xeb, 0xd6, 0x56,
0x99, 0x24, 0xc0, 0x38, 0xdf, 0xdc, 0x04, 0x5f, 0x84, 0x61, 0x91, 0x50, 0xff, 0x24, 0xe6, 0x90,
0x97, 0x1c, 0x46, 0x30, 0x74, 0x17, 0xf2, 0x4f, 0x5c, 0xe6, 0x75, 0xa3, 0xea, 0x42, 0xdd, 0xde,
0x2a, 0x35, 0x96, 0xb7, 0x4d, 0xfe, 0xb6, 0x25, 0x4e, 0xb4, 0x1a, 0x63, 0x58, 0x72, 0x7a, 0xfd,
0x20, 0xe4, 0x84, 0x45, 0xfd, 0xc0, 0x8f, 0x18, 0xaa, 0x80, 0xdd, 0x0a, 0x43, 0xbd, 0x77, 0xf1,
0x89, 0xbf, 0x83, 0xca, 0xae, 0x17, 0x74, 0x4e, 0x9b, 0x94, 0x53, 0xc2, 0xbe, 0x1d, 0xb0, 0x88,
0xa3, 0x55, 0xc8, 0xc9, 0x2a, 0x68, 0x3b, 0x25, 0x08, 0x54, 0x66, 0x52, 0xa7, 0x59, 0x09, 0x02,
0x95, 0xfe, 0x32, 0x15, 0x59, 0xa2, 0x04, 0x81, 0xb6, 0x3d, 0xb7, 0xa3, 0x52, 0x90, 0x25, 0x4a,
0x40, 0x08, 0xb2, 0x47, 0x2e, 0x3b, 0xd7, 0xfb, 0x96, 0xdf, 0xd8, 0x81, 0x95, 0xd4, 0xfa, 0x9a,
0xe6, 0x1a, 0xe4, 0x49, 0x70, 0xee, 0x34, 0xa3, 0xaa, 0x55, 0xb7, 0xb7, 0xb2, 0x44, 0x4b, 0x32,
0xbb, 0xb2, 0xfc, 0x42, 0x95, 0x91, 0xaa, 0x04, 0xc0, 0xeb, 0x90, 0x93, 0xa9, 0x16, 0xbb, 0x4c,
0x7c, 0xc5, 0x27, 0xfe, 0xc7, 0x82, 0xe2, 0x3e, 0xbd, 0x90, 0x34, 0x22, 0xf4, 0x08, 0x0a, 0x6d,
0x4e, 0xfd, 0x2e, 0x0d, 0xbb, 0xd2, 0xa8, 0xd4, 0x78, 0x23, 0x49, 0x61, 0x6c, 0xb6, 0x6d, 0x6c,
0x5a, 0x3e, 0x0f, 0x87, 0x24, 0x76, 0x41, 0x3b, 0xb0, 0xa0, 0x7b, 0x42, 0x72, 0x28, 0x35, 0xea,
0xd3, 0xbc, 0xe3, 0xb6, 0x11, 0xce, 0xc6, 0x61, 0xe3, 0x23, 0x28, 0x8f, 0x84, 0x15, 0x5c, 0x4f,
0xd9, 0xd0, 0x54, 0xe4, 0x94, 0x0d, 0x45, 0xee, 0xce, 0xa8, 0x37, 0x50, 0x79, 0xce, 0x12, 0x25,
0xec, 0x64, 0x3e, 0xb4, 0x36, 0x76, 0x60, 0x31, 0x1d, 0xf5, 0x3a, 0xbe, 0xf8, 0x6b, 0x40, 0x7b,
0x21, 0xa3, 0x9c, 0x49, 0x7a, 0xfb, 0x2c, 0x8a, 0xe8, 0x09, 0x9b, 0x5d, 0x69, 0x55, 0xbd, 0x4c,
0xba, 0x7a, 0x9b, 0x50, 0x74, 0x22, 0xb3, 0x71, 0x5b, 0xf6, 0x65, 0x02, 0xe0, 0x7b, 0x80, 0x9a,
0xcc, 0x63, 0x9c, 0xe9, 0xf3, 0x3b, 0x27, 0x3e, 0x6e, 0x1b, 0x2e, 0x57, 0xdb, 0xa2, 0xbb, 0x90,
0x15, 0x47, 0x57, 0x52, 0x29, 0x35, 0x6e, 0x25, 0x99, 0x8e, 0xe7, 0x04, 0x91, 0x06, 0xd8, 0x35,
0x41, 0xf5, 0x71, 0xbf, 0x62, 0x83, 0x53, 0x5a, 0xd9, 0x2c, 0x65, 0x8f, 0x2f, 0x15, 0x0f, 0x10,
0xbd, 0xd4, 0x63, 0xb3, 0xd7, 0x9b, 0x2e, 0x85, 0x4f, 0x62, 0xb2, 0xe2, 0xa4, 0xde, 0x84, 0xec,
0x5b, 0x90, 0x93, 0xbe, 0x9a, 0xed, 0xc4, 0x0c, 0x50, 0x5a, 0x7c, 0x14, 0x53, 0xbd, 0xe9, 0x42,
0xab, 0xe9, 0x85, 0x8a, 0x26, 0xee, 0x57, 0xda, 0x56, 0x9c, 0xe9, 0x03, 0xe1, 0xa3, 0x22, 0xc9,
0xef, 0xd9, 0x35, 0x1b, 0x4b, 0xa4, 0x88, 0x2d, 0x86, 0x40, 0x54, 0xb5, 0xeb, 0xb6, 0x88, 0x2d,
0x05, 0xfc, 0x00, 0xf2, 0xed, 0xce, 0x0b, 0xd6, 0xa3, 0xe8, 0x1d, 0x71, 0xd2, 0xba, 0xec, 0x82,
0x45, 0xfa, 0x9c, 0x2e, 0x8f, 0xd5, 0x9f, 0x18, 0x3d, 0xfe, 0xd1, 0xd2, 0x7b, 0x9a, 0xc1, 0x28,
0x2f, 0xd7, 0x8e, 0xaa, 0xd9, 0x89, 0x91, 0x29, 0x70, 0xa2, 0xd5, 0xa8, 0x05, 0x15, 0xc7, 0xef,
0x0f, 0x78, 0x93, 0x7d, 0xe3, 0xfa, 0x2e, 0x77, 0x03, 0x3f, 0xaa, 0xe6, 0xa5, 0xcb, 0x7a, 0x7a,
0xe9, 0x11, 0x0b, 0x32, 0xe1, 0x82, 0xbf, 0xb7, 0x60, 0x79, 0x0c, 0xbc, 0x82, 0x57, 0x66, 0x3e,
0xaf, 0xf7, 0xe3, 0x99, 0x6f, 0x4b, 0xc3, 0xda, 0x4c, 0x36, 0xa3, 0x57, 0xc0, 0xaf, 0x16, 0xac,
0x4e, 0x33, 0x98, 0xca, 0xa6, 0x06, 0xf0, 0x2c, 0x74, 0x7b, 0x34, 0x1c, 0x7e, 0xca, 0x86, 0xfa,
0xfa, 0x4b, 0x21, 0xe8, 0x4b, 0x58, 0x1b, 0x8b, 0xf5, 0x71, 0x47, 0xa5, 0x48, 0x91, 0xba, 0x33,
0x93, 0x94, 0xb2, 0x23, 0x33, 0xdc, 0xf1, 0x5f, 0x16, 0xdc, 0x9e, 0xaa, 0x4a, 0x7a, 0xd2, 0x4a,
0xf7, 0xe4, 0x3d, 0xa8, 0x1c, 0x89, 0xc9, 0xd6, 0x64, 0x11, 0x77, 0x7d, 0x2a, 0x2c, 0x75, 0xd3,
0x4e, 0xe0, 0xc8, 0x81, 0x82, 0xc4, 0xf6, 0x69, 0x5f, 0xd3, 0x7c, 0xf7, 0x0a, 0x9a, 0xdb, 0xc6,
0x5e, 0x0f, 0x7e, 0x23, 0x0a, 0x32, 0xf2, 0x22, 0x32, 0xb7, 0x9a, 0x14, 0xc4, 0x48, 0x1f, 0x71,
0xb8, 0xd6, 0x58, 0x0e, 0x60, 0xd3, 0x8c, 0xc2, 0x11, 0x26, 0xf3, 0x4f, 0xea, 0x43, 0x80, 0xc4,
0x54, 0x4f, 0x80, 0x39, 0xfd, 0x99, 0x32, 0xc6, 0x4f, 0x61, 0xd3, 0xcc, 0xe9, 0x6b, 0x2c, 0x68,
0xba, 0x25, 0x93, 0x74, 0x0b, 0x6e, 0x81, 0xfd, 0x9c, 0x38, 0xe2, 0xae, 0x96, 0xa7, 0xd5, 0x94,
0x48, 0x4b, 0xc2, 0xe5, 0x69, 0x10, 0x71, 0xe3, 0x22, 0xbe, 0x05, 0xf6, 0x2c, 0x08, 0xb9, 0x64,
0x5c, 0x26, 0xf2, 0x1b, 0x7f, 0x00, 0xd9, 0x83, 0xa0, 0xcb, 0xd0, 0x12, 0x64, 0x9c, 0xa6, 0x8e,
0x91, 0x71, 0x9a, 0xe8, 0x8e, 0x0c, 0xaf, 0x67, 0x48, 0x39, 0xd9, 0xdc, 0x73, 0xe2, 0x10, 0xa1,
0xc1, 0x8f, 0xa1, 0x22, 0x1c, 0xdb, 0x9c, 0xf2, 0x78, 0x06, 0xaf, 0x41, 0x5e, 0x60, 0x71, 0x20,
0x2d, 0xc9, 0x1b, 0x4d, 0xd8, 0x99, 0xd1, 0x26, 0x05, 0xfc, 0x93, 0x05, 0x60, 0x42, 0x0c, 0x22,
0x84, 0x15, 0x13, 0xe9, 0x5a, 0x6a, 0x2c, 0x25, 0x4b, 0x0a, 0x94, 0x28, 0x96, 0xef, 0xa5, 0xde,
0x11, 0x93, 0xf3, 0x2d, 0x56, 0x91, 0xd4, 0x6b, 0x63, 0xcb, 0x8c, 0x33, 0x5d, 0xa8, 0x4a, 0x62,
0xaf, 0x70, 0x9d, 0x32, 0x71, 0x85, 0x95, 0xf7, 0xbc, 0x41, 0xc4, 0x59, 0xa8, 0x19, 0x89, 0xf7,
0x8e, 0x02, 0xe2, 0x1d, 0x25, 0xc0, 0xf4, 0x4d, 0xa1, 0x37, 0x21, 0x27, 0x98, 0x9a, 0x33, 0x39,
0xbe, 0x0d, 0xa5, 0xc4, 0x6d, 0x3d, 0xd5, 0xa7, 0xce, 0x01, 0x04, 0x59, 0xf9, 0xba, 0xd5, 0xa5,
0x93, 0x0f, 0xdb, 0x0a, 0xd8, 0xfb, 0xae, 0xea, 0x35, 0x9b, 0x88, 0x4f, 0x89, 0xd0, 0x0b, 0x79,
0x16, 0x04, 0x42, 0xc5, 0xbd, 0xbe, 0xa2, 0x9a, 0x59, 0xcc, 0xf1, 0x9b, 0xdc, 0x35, 0xe6, 0x81,
0x68, 0xa7, 0x1e, 0x88, 0x6d, 0x58, 0x51, 0x0d, 0xfb, 0x2a, 0x83, 0xfe, 0x92, 0x81, 0x15, 0xc2,
0x22, 0xf7, 0x25, 0x73, 0xfc, 0x88, 0x87, 0x83, 0x78, 0xd8, 0x7c, 0x12, 0x1c, 0xeb, 0x54, 0xdb,
0x44, 0x09, 0x71, 0x5b, 0x64, 0xe6, 0xb4, 0xc5, 0x7d, 0xf1, 0xab, 0x12, 0x84, 0x5d, 0x31, 0x74,
0x82, 0x50, 0x17, 0x7a, 0xdc, 0x34, 0x6d, 0x82, 0xee, 0xc3, 0x42, 0x3b, 0x18, 0x84, 0x9d, 0xf8,
0x4a, 0x5a, 0x4b, 0xac, 0x15, 0x33, 0xa5, 0x26, 0xc6, 0x2c, 0xd5, 0x47, 0xb9, 0xf9, 0x7d, 0x84,
0x1e, 0x8d, 0xf5, 0x91, 0xfc, 0x8b, 0x28, 0x35, 0x5e, 0x4f, 0x1c, 0x46, 0xd4, 0x64, 0xd4, 0x1a,
0xff, 0x60, 0xc1, 0x62, 0x9a, 0xc2, 0x7f, 0x3a, 0x18, 0x71, 0x45, 0x32, 0x53, 0x2b, 0x62, 0x4f,
0xab, 0x48, 0x36, 0xa9, 0x48, 0xf2, 0xe6, 0xcc, 0xa5, 0xde, 0x9c, 0xf8, 0x14, 0xd6, 0x27, 0xca,
0xb4, 0x17, 0xf4, 0xfa, 0xa2, 0x1f, 0xfe, 0x47, 0xb9, 0x56, 0x21, 0xd7, 0x0a, 0x43, 0x5d, 0xa8,
0x22, 0x51, 0x02, 0x7e, 0x08, 0xb7, 0xdb, 0x8c, 0xa7, 0x8a, 0x64, 0xba, 0xad, 0x0e, 0xf6, 0x01,
0x3b, 0x9f, 0xb1, 0x7d, 0xa1, 0xc2, 0xbb, 0x50, 0x38, 0x0c, 0xfa, 0x81, 0x17, 0x9c, 0x0c, 0xaf,
0x38, 0xb4, 0x55, 0x58, 0x50, 0x33, 0x49, 0x5d, 0xf9, 0x45, 0x62, 0x44, 0x7c, 0x4b, 0xb4, 0x64,
0x87, 0x7a, 0x9d, 0x81, 0x47, 0x39, 0x93, 0x7f, 0x32, 0xd1, 0x6e, 0xe5, 0xb7, 0xcb, 0x9a, 0xf5,
0xfb, 0x65, 0xcd, 0xfa, 0xe3, 0xb2, 0x66, 0xfd, 0xfc, 0x67, 0xed, 0xb5, 0xe3, 0xbc, 0xfc, 0x67,
0x7e, 0xf0, 0x6f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x14, 0x23, 0x92, 0x89, 0x44, 0x0f, 0x00, 0x00,
// 1306 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0xdd, 0x6e, 0x1b, 0x45,
0x14, 0x66, 0xbd, 0xb6, 0x63, 0x1f, 0xd7, 0xa9, 0x33, 0x4d, 0x83, 0x13, 0x45, 0xae, 0x19, 0x15,
0x1a, 0x2a, 0x11, 0x15, 0x57, 0x42, 0x34, 0xa8, 0x52, 0x89, 0xed, 0xaa, 0x0b, 0x24, 0x94, 0x71,
0x1a, 0x24, 0x24, 0x90, 0x26, 0xf6, 0x90, 0xae, 0xb2, 0xde, 0x35, 0xbb, 0xe3, 0x24, 0xee, 0x05,
0x97, 0x08, 0x09, 0x71, 0x8f, 0xb8, 0xe5, 0x65, 0xb8, 0xe4, 0x11, 0x50, 0x78, 0x08, 0x24, 0x6e,
0x40, 0xf3, 0xb7, 0xbb, 0xfe, 0x0d, 0x09, 0xdc, 0xed, 0xf9, 0xe6, 0x9c, 0x33, 0xdf, 0x9c, 0xbf,
0x99, 0x85, 0xf2, 0x20, 0x74, 0x4f, 0x29, 0x67, 0xdb, 0x83, 0x30, 0xe0, 0x01, 0x2a, 0xb8, 0x3e,
0x67, 0xa1, 0x4f, 0x3d, 0xfc, 0x29, 0x14, 0x1d, 0xbf, 0xc7, 0xce, 0xf7, 0x18, 0xa7, 0xa8, 0x0e,
0xa5, 0x66, 0xe0, 0x0d, 0xfb, 0xfe, 0x27, 0xf4, 0x88, 0x79, 0x55, 0xab, 0x6e, 0x6d, 0x15, 0x49,
0x1a, 0x12, 0x1a, 0x07, 0x6e, 0x9f, 0x7d, 0x36, 0xa4, 0x3e, 0x1f, 0xf6, 0xab, 0x19, 0xa5, 0x91,
0x82, 0xf0, 0x5f, 0x16, 0x14, 0x9f, 0x86, 0xb4, 0xcf, 0xa4, 0xc7, 0x0d, 0x28, 0x90, 0xe0, 0x2c,
0xed, 0x2e, 0x96, 0xd1, 0x5b, 0xb0, 0xec, 0xf8, 0xa7, 0x2c, 0x8c, 0x58, 0xdb, 0xa7, 0x47, 0x1e,
0xeb, 0x49, 0x77, 0x05, 0x32, 0x81, 0xa2, 0x4d, 0x28, 0x36, 0x69, 0xf7, 0x25, 0x3b, 0x18, 0x0d,
0x58, 0xd5, 0x96, 0x4e, 0x12, 0x20, 0x5e, 0xed, 0xb8, 0xaf, 0x58, 0x35, 0x5b, 0xb7, 0xb6, 0xca,
0x24, 0x01, 0x26, 0xf9, 0xe6, 0xa6, 0xf8, 0x22, 0x0c, 0x37, 0x08, 0xf5, 0x8f, 0x63, 0x0e, 0x79,
0xc9, 0x61, 0x0c, 0x43, 0xf7, 0x20, 0xff, 0xd4, 0x65, 0x5e, 0x2f, 0xaa, 0x2e, 0xd5, 0xed, 0xad,
0x52, 0xe3, 0xe6, 0xb6, 0x89, 0xdf, 0xb6, 0xc4, 0x89, 0x5e, 0xc6, 0x18, 0x96, 0x9d, 0xfe, 0x20,
0x08, 0x39, 0x61, 0xd1, 0x20, 0xf0, 0x23, 0x86, 0x2a, 0x60, 0xb7, 0xc3, 0x50, 0x9f, 0x5d, 0x7c,
0xe2, 0x6f, 0xa1, 0xb2, 0xeb, 0x05, 0xdd, 0x93, 0x16, 0xe5, 0x94, 0xb0, 0x6f, 0x86, 0x2c, 0xe2,
0x68, 0x15, 0x72, 0x32, 0x0b, 0x5a, 0x4f, 0x09, 0x02, 0x95, 0x91, 0xd4, 0x61, 0x56, 0x82, 0x40,
0xa5, 0xbd, 0x0c, 0x45, 0x96, 0x28, 0x41, 0xa0, 0x1d, 0xcf, 0xed, 0xaa, 0x10, 0x64, 0x89, 0x12,
0x10, 0x82, 0xec, 0xa1, 0xcb, 0xce, 0xf4, 0xb9, 0xe5, 0x37, 0x76, 0x60, 0x25, 0xb5, 0xbf, 0xa6,
0xb9, 0x06, 0x79, 0x12, 0x9c, 0x39, 0xad, 0xa8, 0x6a, 0xd5, 0xed, 0xad, 0x2c, 0xd1, 0x92, 0x8c,
0xae, 0x4c, 0xbf, 0x58, 0xca, 0xc8, 0xa5, 0x04, 0xc0, 0xeb, 0x90, 0x93, 0xa1, 0x16, 0xa7, 0x4c,
0x6c, 0xc5, 0x27, 0xfe, 0xdb, 0x82, 0xe2, 0x1e, 0x3d, 0x97, 0x34, 0x22, 0xf4, 0x18, 0x0a, 0x1d,
0x4e, 0xfd, 0x1e, 0x0d, 0x7b, 0x52, 0xa9, 0xd4, 0x78, 0x23, 0x09, 0x61, 0xac, 0xb6, 0x6d, 0x74,
0xda, 0x3e, 0x0f, 0x47, 0x24, 0x36, 0x41, 0x3b, 0xb0, 0xa4, 0x6b, 0x42, 0x72, 0x28, 0x35, 0xea,
0xb3, 0xac, 0xe3, 0xb2, 0x11, 0xc6, 0xc6, 0x60, 0xe3, 0x03, 0x28, 0x8f, 0xb9, 0x15, 0x5c, 0x4f,
0xd8, 0xc8, 0x64, 0xe4, 0x84, 0x8d, 0x44, 0xec, 0x4e, 0xa9, 0x37, 0x54, 0x71, 0xce, 0x12, 0x25,
0xec, 0x64, 0xde, 0xb7, 0x36, 0x76, 0xe0, 0x46, 0xda, 0xeb, 0x55, 0x6c, 0xf1, 0x57, 0x80, 0x9a,
0x21, 0xa3, 0x9c, 0x49, 0x7a, 0x7b, 0x2c, 0x8a, 0xe8, 0x31, 0x9b, 0x9f, 0x69, 0x95, 0xbd, 0x4c,
0x3a, 0x7b, 0x9b, 0x50, 0x74, 0x22, 0x73, 0x70, 0x5b, 0xd6, 0x65, 0x02, 0xe0, 0xfb, 0x80, 0x5a,
0xcc, 0x63, 0x9c, 0xe9, 0xfe, 0x5d, 0xe0, 0x1f, 0x77, 0x0c, 0x97, 0xcb, 0x75, 0xd1, 0x3d, 0xc8,
0x8a, 0xd6, 0x95, 0x54, 0x4a, 0x8d, 0x5b, 0x49, 0xa4, 0xe3, 0x39, 0x41, 0xa4, 0x02, 0x76, 0x8d,
0x53, 0xdd, 0xee, 0x97, 0x1c, 0x70, 0x46, 0x29, 0x9b, 0xad, 0xec, 0xc9, 0xad, 0xe2, 0x01, 0xa2,
0xb7, 0x7a, 0x62, 0xce, 0x7a, 0xdd, 0xad, 0xf0, 0x71, 0x4c, 0x56, 0x74, 0xea, 0x75, 0xc8, 0xbe,
0x09, 0x39, 0x69, 0xab, 0xd9, 0x4e, 0xcd, 0x00, 0xb5, 0x8a, 0x0f, 0x63, 0xaa, 0xd7, 0xdd, 0x68,
0x35, 0xbd, 0x51, 0xd1, 0xf8, 0xfd, 0x42, 0xeb, 0x8a, 0x9e, 0xde, 0x17, 0x36, 0xca, 0x93, 0xfc,
0x9e, 0x9f, 0xb3, 0x89, 0x40, 0x0a, 0xdf, 0x62, 0x08, 0x44, 0x55, 0xbb, 0x6e, 0x0b, 0xdf, 0x52,
0xc0, 0x0f, 0x21, 0xdf, 0xe9, 0xbe, 0x64, 0x7d, 0x8a, 0xde, 0x16, 0x9d, 0xd6, 0x63, 0xe7, 0x2c,
0xd2, 0x7d, 0x7a, 0x73, 0x22, 0xff, 0xc4, 0xac, 0xe3, 0x1f, 0x2c, 0x7d, 0xa6, 0x39, 0x8c, 0xf2,
0x72, 0xef, 0xa8, 0x9a, 0x9d, 0x1a, 0x99, 0x02, 0x27, 0x7a, 0x19, 0xb5, 0xa1, 0xe2, 0xf8, 0x83,
0x21, 0x6f, 0xb1, 0xaf, 0x5d, 0xdf, 0xe5, 0x6e, 0xe0, 0x47, 0xd5, 0xbc, 0x34, 0x59, 0x4f, 0x6f,
0x3d, 0xa6, 0x41, 0xa6, 0x4c, 0xf0, 0x77, 0x16, 0xdc, 0x9c, 0x00, 0x2f, 0xe1, 0x95, 0x59, 0xcc,
0xeb, 0xbd, 0x78, 0xe6, 0xdb, 0x52, 0xb1, 0x36, 0x97, 0xcd, 0xf8, 0x15, 0xf0, 0x8b, 0x05, 0xab,
0xb3, 0x14, 0x66, 0xb2, 0xa9, 0x01, 0x3c, 0x0f, 0xdd, 0x3e, 0x0d, 0x47, 0x1f, 0xb3, 0x91, 0xbe,
0xfe, 0x52, 0x08, 0xfa, 0x1c, 0xd6, 0x26, 0x7c, 0x7d, 0xd8, 0x55, 0x21, 0x52, 0xa4, 0xee, 0xcc,
0x25, 0xa5, 0xf4, 0xc8, 0x1c, 0x73, 0xfc, 0xa7, 0x05, 0xb7, 0x67, 0x2e, 0x25, 0x35, 0x69, 0xa5,
0x6b, 0xf2, 0x3e, 0x54, 0x0e, 0xc5, 0x64, 0x6b, 0xb1, 0x88, 0xbb, 0x3e, 0x15, 0x9a, 0xba, 0x68,
0xa7, 0x70, 0xe4, 0x40, 0x41, 0x62, 0x7b, 0x74, 0xa0, 0x69, 0xbe, 0x73, 0x09, 0xcd, 0x6d, 0xa3,
0xaf, 0x07, 0xbf, 0x11, 0x05, 0x19, 0x79, 0x11, 0x99, 0x5b, 0x4d, 0x0a, 0x62, 0xa4, 0x8f, 0x19,
0x5c, 0x69, 0x2c, 0x07, 0xb0, 0x69, 0x46, 0xe1, 0x18, 0x93, 0xc5, 0x9d, 0xfa, 0x08, 0x20, 0x51,
0xd5, 0x13, 0x60, 0x41, 0x7d, 0xa6, 0x94, 0xf1, 0x33, 0xd8, 0x34, 0x73, 0xfa, 0x0a, 0x1b, 0x9a,
0x6a, 0xc9, 0x24, 0xd5, 0x82, 0xdb, 0x60, 0xbf, 0x20, 0x8e, 0xb8, 0xab, 0x65, 0xb7, 0x9a, 0x14,
0x69, 0x49, 0x98, 0x3c, 0x0b, 0x22, 0x6e, 0x4c, 0xc4, 0xb7, 0xc0, 0x9e, 0x07, 0x21, 0x97, 0x8c,
0xcb, 0x44, 0x7e, 0xe3, 0x2f, 0x21, 0xbb, 0x1f, 0xf4, 0x18, 0x5a, 0x86, 0x8c, 0xd3, 0xd2, 0x3e,
0x32, 0x4e, 0x0b, 0xdd, 0x91, 0xee, 0xf5, 0x0c, 0x29, 0x27, 0x87, 0x7b, 0x41, 0x1c, 0x22, 0x37,
0xbe, 0x0b, 0x65, 0x27, 0x6a, 0x06, 0x41, 0xd8, 0x13, 0xa9, 0x0e, 0x42, 0x7d, 0x27, 0x8d, 0x83,
0xf8, 0x09, 0x54, 0x84, 0xfb, 0x0e, 0xa7, 0x3c, 0x9e, 0xd4, 0x6b, 0x90, 0x17, 0x58, 0xbc, 0x9d,
0x96, 0xe4, 0xbd, 0x27, 0xf4, 0xcc, 0x00, 0x94, 0x02, 0xfe, 0xd1, 0x02, 0x30, 0x2e, 0x86, 0x11,
0xc2, 0x8a, 0xaf, 0x34, 0x2d, 0x35, 0x96, 0x13, 0x62, 0x02, 0x25, 0xea, 0x2c, 0xef, 0xa6, 0x5e,
0x1b, 0xd3, 0x53, 0x30, 0x5e, 0x22, 0xa9, 0x37, 0xc9, 0x96, 0x19, 0x7a, 0x3a, 0x9d, 0x95, 0x44,
0x5f, 0xe1, 0x3a, 0xb0, 0xe2, 0xa2, 0x2b, 0x37, 0xbd, 0x61, 0xc4, 0x59, 0xa8, 0x19, 0x89, 0x57,
0x91, 0x02, 0xe2, 0x13, 0x25, 0xc0, 0xec, 0x43, 0xa1, 0xbb, 0x90, 0x13, 0x4c, 0x4d, 0xe7, 0x4e,
0x1e, 0x43, 0x2d, 0xe2, 0x8e, 0x9e, 0xfd, 0x33, 0xa7, 0x05, 0x82, 0xac, 0x7c, 0x03, 0xeb, 0x04,
0xcb, 0xe7, 0x6f, 0x05, 0xec, 0x3d, 0x57, 0x55, 0xa4, 0x4d, 0xc4, 0xa7, 0x44, 0xe8, 0xb9, 0xec,
0x18, 0x81, 0x50, 0x71, 0xfb, 0xaf, 0xa8, 0x92, 0x17, 0xd3, 0xfe, 0x3a, 0x37, 0x92, 0x79, 0x46,
0xda, 0xa9, 0x67, 0x64, 0x07, 0x56, 0x54, 0x59, 0xff, 0x9f, 0x4e, 0x7f, 0xce, 0xc0, 0x0a, 0x61,
0x91, 0xfb, 0x8a, 0x39, 0x7e, 0xc4, 0xc3, 0x61, 0x3c, 0x92, 0x3e, 0x0a, 0x8e, 0x74, 0xa8, 0x6d,
0xa2, 0x84, 0xb8, 0x2c, 0x32, 0x0b, 0xca, 0xe2, 0x81, 0xf8, 0xa1, 0x19, 0xaf, 0xd7, 0x69, 0xd5,
0xb4, 0x0a, 0x7a, 0x00, 0x4b, 0x9d, 0x60, 0x18, 0x76, 0xe3, 0x8b, 0x6b, 0x2d, 0xd1, 0x56, 0xcc,
0xd4, 0x32, 0x31, 0x6a, 0xa9, 0x3a, 0xca, 0x2d, 0xae, 0x23, 0xf4, 0x78, 0xa2, 0x8e, 0xe4, 0xbf,
0x46, 0xa9, 0xf1, 0x7a, 0x62, 0x30, 0xb6, 0x4c, 0xc6, 0xb5, 0xf1, 0xf7, 0x16, 0xdc, 0x48, 0x53,
0xf8, 0x57, 0x8d, 0x11, 0x67, 0x24, 0x33, 0x33, 0x23, 0xf6, 0xac, 0x8c, 0x64, 0x93, 0x8c, 0x24,
0x2f, 0xd3, 0x5c, 0xea, 0x65, 0x8a, 0x4f, 0x60, 0x7d, 0x2a, 0x4d, 0xcd, 0xa0, 0x3f, 0x10, 0xf5,
0xf0, 0x1f, 0xd2, 0xb5, 0x0a, 0xb9, 0x76, 0x18, 0xea, 0x44, 0x15, 0x89, 0x12, 0xf0, 0x23, 0xb8,
0xdd, 0x61, 0x3c, 0x95, 0x24, 0x53, 0x6d, 0x75, 0xb0, 0xf7, 0xd9, 0xd9, 0x9c, 0xe3, 0x8b, 0x25,
0xbc, 0x0b, 0x85, 0x83, 0x60, 0x10, 0x78, 0xc1, 0xf1, 0xe8, 0x92, 0xa6, 0xad, 0xc2, 0x92, 0x9a,
0x49, 0xea, 0x61, 0x50, 0x24, 0x46, 0xc4, 0xb7, 0x44, 0x49, 0x76, 0xa9, 0xd7, 0x1d, 0x7a, 0x94,
0x33, 0xf9, 0xbf, 0x13, 0xed, 0x56, 0x7e, 0xbd, 0xa8, 0x59, 0xbf, 0x5d, 0xd4, 0xac, 0xdf, 0x2f,
0x6a, 0xd6, 0x4f, 0x7f, 0xd4, 0x5e, 0x3b, 0xca, 0xcb, 0x3f, 0xeb, 0x87, 0xff, 0x04, 0x00, 0x00,
0xff, 0xff, 0xd7, 0xef, 0xfa, 0x68, 0x6a, 0x0f, 0x00, 0x00,
}

View file

@ -136,6 +136,7 @@ message URI {
message Node {
string ID = 1;
URI URI = 2;
bool IsCoordinator = 3;
}
message NodeStateMessage {

View file

@ -132,7 +132,11 @@ func (s *Server) Open() error {
s.NodeID = s.LoadNodeID()
// Set Cluster Node.
node := &Node{ID: s.NodeID, URI: s.URI}
node := &Node{
ID: s.NodeID,
URI: s.URI,
IsCoordinator: s.Cluster.Coordinator == s.NodeID,
}
s.Cluster.Node = node
// Append the NodeID tag to stats.
@ -185,11 +189,6 @@ func (s *Server) Open() error {
return fmt.Errorf("starting BroadcastReceiver: %v", err)
}
// If a Coordinator is not specified, then default to s.URI.
if s.Cluster.Coordinator.Port() == 0 {
s.Cluster.Coordinator = s.URI
}
// Open Cluster management.
if err := s.Cluster.Open(); err != nil {
return fmt.Errorf("opening Cluster: %v", err)

View file

@ -52,7 +52,7 @@ func TestMain_SendReceiveMessage(t *testing.T) {
m0.Config.Gossip.Port = "0"
m0.Config.Gossip.Seeds = []string{}
m0.Server.Cluster.Coordinator = m0.Server.URI
m0.Server.Cluster.Coordinator = m0.Server.NodeID
m0.Server.Cluster.Topology = &pilosa.Topology{NodeIDs: []string{m0.Server.NodeID, m1.Server.NodeID}}
m0.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver(m0.Server.LogOutput)
gossipMemberSet0, err := gossip.NewGossipMemberSet(m0.Server.URI.HostPort(), m0.Config, m0.Server)
@ -80,7 +80,7 @@ func TestMain_SendReceiveMessage(t *testing.T) {
m1.Config.Gossip.Port = "0"
m1.Config.Gossip.Seeds = gossipMemberSet0.Seeds()
m1.Server.Cluster.Coordinator = m0.Server.URI
m1.Server.Cluster.Coordinator = m0.Server.NodeID
m1.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver(m1.Server.LogOutput)
gossipMemberSet1, err := gossip.NewGossipMemberSet(m1.Server.URI.HostPort(), m1.Config, m1.Server)
if err != nil {
@ -220,21 +220,21 @@ func TestClusterResize_EmptyNode(t *testing.T) {
// Ensure that a cluster of empty nodes comes up in a NORMAL state.
func TestClusterResize_EmptyNodes(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
gossipHost := "localhost"
gossipPort := 0
seed, coord, err := m0.RunWithTransport(gossipHost, gossipPort, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport(gossipHost, gossipPort, []string{})
if err != nil {
t.Fatal(err)
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
seed, coord, err = m1.RunWithTransport(gossipHost, gossipPort, []string{seed}, coord)
seed, err = m1.RunWithTransport(gossipHost, gossipPort, []string{seed})
if err != nil {
t.Fatal(err)
}
@ -250,21 +250,21 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
func TestClusterResize_AddNode(t *testing.T) {
t.Run("NoData", func(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport("localhost", 0, []string{})
if err != nil {
t.Fatal(err)
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
var eg errgroup.Group
eg.Go(func() error {
_, _, err = m1.RunWithTransport("localhost", 0, []string{seed}, coord)
_, err = m1.RunWithTransport("localhost", 0, []string{seed})
if err != nil {
return err
}
@ -284,10 +284,10 @@ func TestClusterResize_AddNode(t *testing.T) {
})
t.Run("WithIndex", func(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport("localhost", 0, []string{})
if err != nil {
t.Fatal(err)
}
@ -303,12 +303,12 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
var eg errgroup.Group
eg.Go(func() error {
_, _, err = m1.RunWithTransport("localhost", 0, []string{seed}, coord)
_, err = m1.RunWithTransport("localhost", 0, []string{seed})
if err != nil {
return err
}
@ -330,10 +330,10 @@ func TestClusterResize_AddNode(t *testing.T) {
t.Run("ContinuousSlices", func(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport("localhost", 0, []string{})
if err != nil {
t.Fatal(err)
}
@ -358,12 +358,12 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
var eg errgroup.Group
eg.Go(func() error {
_, _, err = m1.RunWithTransport("localhost", 0, []string{seed}, coord)
_, err = m1.RunWithTransport("localhost", 0, []string{seed})
if err != nil {
return err
}
@ -385,10 +385,10 @@ func TestClusterResize_AddNode(t *testing.T) {
t.Run("SkippedSlice", func(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport("localhost", 0, []string{})
if err != nil {
t.Fatal(err)
}
@ -413,12 +413,12 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
var eg errgroup.Group
eg.Go(func() error {
_, _, err = m1.RunWithTransport("localhost", 0, []string{seed}, coord)
_, err = m1.RunWithTransport("localhost", 0, []string{seed})
if err != nil {
return err
}
@ -443,22 +443,22 @@ func TestClusterResize_AddNode(t *testing.T) {
func TestCluster_GossipMembership(t *testing.T) {
t.Run("Node0Down", func(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport("localhost", 0, []string{})
if err != nil {
t.Fatal(err)
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
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}, coord)
_, err = m1.RunWithTransport("localhost", 0, []string{"http://localhost:8765", seed})
if err != nil {
return err
}
@ -466,12 +466,12 @@ func TestCluster_GossipMembership(t *testing.T) {
})
// Configure node2
m2 := test.NewMainWithCluster()
m2 := test.NewMainWithCluster(false)
defer m2.Close()
eg.Go(func() error {
// Pass invalid seed as last in list
_, _, err = m2.RunWithTransport("localhost", 0, []string{seed, "http://localhost:8765"}, coord)
_, err = m2.RunWithTransport("localhost", 0, []string{seed, "http://localhost:8765"})
if err != nil {
return err
}

View file

@ -181,31 +181,13 @@ func (m *Command) SetupServer() error {
InsecureSkipVerify: m.Config.TLS.SkipVerify,
}
// TODO Review this location
TLSConfig = m.Server.TLS
}
c := pilosa.GetHTTPClient(TLSConfig)
m.Server.RemoteClient = c
m.Server.Handler.RemoteClient = c
m.Server.Cluster.RemoteClient = c
// Default coordintor to port 0 when not specified so that coordinator
// can be set to the value of server.URI after server binds to a port.
// This would only be useful in a one-node cluster.
coord := m.Config.Cluster.Coordinator
if coord == "" {
coord = ":0"
}
// Set the coordinator node.
curi, err := pilosa.AddressWithDefaults(coord)
if err != nil {
return err
}
m.Server.Cluster.Coordinator = *curi
// Set configuration options.
m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval)
m.Server.Cluster.LongQueryTime = time.Duration(m.Config.Cluster.LongQueryTime)
@ -214,8 +196,12 @@ func (m *Command) SetupServer() error {
// SetupNetworking sets up internode communication based on the configuration.
func (m *Command) SetupNetworking() error {
m.Server.NodeID = m.Server.LoadNodeID()
if m.Config.Cluster.Disabled {
m.Server.Cluster.Static = true
m.Server.Cluster.Coordinator = m.Server.NodeID
for _, address := range m.Config.Cluster.Hosts {
uri, err := pilosa.NewURIFromAddress(address)
if err != nil {
@ -256,7 +242,10 @@ func (m *Command) SetupNetworking() error {
}
}
m.Server.NodeID = m.Server.LoadNodeID()
// Set Coordinator.
if m.Config.Cluster.Coordinator || len(m.Config.Gossip.Seeds) == 0 {
m.Server.Cluster.Coordinator = m.Server.NodeID
}
m.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver(m.Server.LogOutput)
gossipMemberSet, err := gossip.NewGossipMemberSetWithTransport(m.Server.NodeID, m.Config, transport, m.Server)

View file

@ -49,7 +49,7 @@ func NewCluster(n int) *pilosa.Cluster {
}
c.Node = c.Nodes[0]
c.Coordinator = c.Nodes[0].URI
c.Coordinator = c.Nodes[0].ID
return c
}
@ -262,7 +262,7 @@ func (t *TestCluster) addCluster(i int, saveTopology bool) (*pilosa.Cluster, err
c.Holder = h
c.MemberSet = pilosa.NewStaticMemberSet(c.Nodes)
c.Node = node
c.Coordinator = t.common.Nodes[0].URI // the first node is the coordinator
c.Coordinator = t.common.Nodes[0].ID // the first node is the coordinator
c.Broadcaster = t
// add nodes

View file

@ -65,9 +65,10 @@ func NewMain() *Main {
}
// NewMainWithCluster returns a new instance of Main with clustering enabled.
func NewMainWithCluster() *Main {
func NewMainWithCluster(isCoordinator bool) *Main {
m := NewMain()
m.Config.Cluster.Disabled = false
m.Config.Cluster.Coordinator = isCoordinator
return m
}
@ -94,12 +95,11 @@ func runMainWithCluster(size int) ([]*Main, error) {
gossipPort := 0
var err error
var gossipSeeds = make([]string, size)
var coordinator pilosa.URI
for i := 0; i < size; i++ {
m := NewMainWithCluster()
m := NewMainWithCluster(i == 0)
gossipSeeds[i], coordinator, err = m.RunWithTransport(gossipHost, gossipPort, gossipSeeds[:i], coordinator)
gossipSeeds[i], err = m.RunWithTransport(gossipHost, gossipPort, gossipSeeds[:i])
if err != nil {
return nil, errors.Wrap(err, "RunWithTransport")
}
@ -146,7 +146,7 @@ func (m *Main) Reopen() error {
}
// RunWithTransport runs Main and returns the dynamically allocated gossip port.
func (m *Main) RunWithTransport(host string, bindPort int, joinSeeds []string, coordinator pilosa.URI) (seed string, coord pilosa.URI, err error) {
func (m *Main) RunWithTransport(host string, bindPort int, joinSeeds []string) (seed string, err error) {
defer close(m.Started)
/*
@ -166,19 +166,19 @@ func (m *Main) RunWithTransport(host string, bindPort int, joinSeeds []string, c
// SetupServer
err = m.SetupServer()
if err != nil {
return seed, coord, err
return seed, err
}
// Open server listener.
err = m.Server.OpenListener()
if err != nil {
return seed, coord, err
return seed, err
}
// Open gossip transport to use in SetupServer.
transport, err := gossip.NewTransport(host, bindPort)
if err != nil {
return seed, coord, err
return seed, err
}
m.GossipTransport = transport
@ -193,23 +193,22 @@ func (m *Main) RunWithTransport(host string, bindPort int, joinSeeds []string, c
// SetupNetworking
err = m.SetupNetworking()
if err != nil {
return seed, coord, err
return seed, err
}
if err = m.Server.BroadcastReceiver.Start(m.Server); err != nil {
return seed, coord, err
return seed, err
}
m.Server.Cluster.Coordinator = coordinator
m.Server.Cluster.Static = false
// Initialize server.
err = m.Server.Open()
if err != nil {
return seed, coord, err
return seed, err
}
return seed, m.Server.Cluster.Coordinator, nil
return seed, nil
}
// URL returns the base URL string for accessing the running program.