mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 09:05:55 +00:00
Merge pull request #1420 from jaffee/remove-cluster-refs
Remove cluster refs
This commit is contained in:
Matthew Jaffee
GitHub
commit
7f908fcf38
7 changed files with 103 additions and 114 deletions
2
api.go
2
api.go
|
|
@ -49,7 +49,7 @@ func OptAPIServer(s *Server) APIOption {
|
|||
a.server = s
|
||||
a.Holder = s.holder
|
||||
a.Broadcaster = s
|
||||
a.Cluster = s.Cluster
|
||||
a.Cluster = s.cluster
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -188,7 +188,7 @@ func TestImportCommand_BugOverwriteValue(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
cm.Host = cmd.Server.Addr().String()
|
||||
cm.Host = cmd.Server.URI.HostPort()
|
||||
|
||||
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i", strings.NewReader("")))
|
||||
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i/field/f", strings.NewReader(`{"options":{"type": "int", "min": 0, "max":2147483648 }}`)))
|
||||
|
|
|
|||
|
|
@ -220,7 +220,7 @@ func TestClient_MultiNode(t *testing.T) {
|
|||
// Ensure client can bulk import data.
|
||||
func TestClient_Import(t *testing.T) {
|
||||
cmd := test.MustRunMainWithCluster(t, 1)[0]
|
||||
host := cmd.Server.Addr().String()
|
||||
host := cmd.URL()
|
||||
holder := cmd.Server.Holder()
|
||||
hldr := test.Holder{Holder: holder}
|
||||
|
||||
|
|
@ -250,7 +250,7 @@ func TestClient_Import(t *testing.T) {
|
|||
// Ensure client can bulk import value data.
|
||||
func TestClient_ImportValue(t *testing.T) {
|
||||
cmd := test.MustRunMainWithCluster(t, 1)[0]
|
||||
host := cmd.Server.Addr().String()
|
||||
host := cmd.URL()
|
||||
holder := cmd.Server.Holder()
|
||||
hldr := test.Holder{Holder: holder}
|
||||
|
||||
|
|
@ -330,7 +330,7 @@ func TestClient_FragmentBlocks(t *testing.T) {
|
|||
|
||||
// Set a bit on a different slice.
|
||||
hldr.SetBit("i", "f", 0, 1)
|
||||
c := MustNewClient(cmd.Server.Addr().String(), defaultClient)
|
||||
c := MustNewClient(cmd.URL(), defaultClient)
|
||||
blocks, err := c.FragmentBlocks(context.Background(), nil, "i", "f", 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
|
|||
120
server.go
120
server.go
|
|
@ -18,7 +18,6 @@ import (
|
|||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
|
|
@ -53,7 +52,7 @@ type Server struct {
|
|||
|
||||
// Internal
|
||||
holder *Holder
|
||||
Cluster *Cluster
|
||||
cluster *Cluster
|
||||
translateFile *TranslateFile
|
||||
diagnostics *DiagnosticsCollector
|
||||
executor *Executor
|
||||
|
|
@ -65,7 +64,7 @@ type Server struct {
|
|||
gcNotifier GCNotifier
|
||||
logger Logger
|
||||
|
||||
NodeID string
|
||||
nodeID string
|
||||
URI URI
|
||||
antiEntropyInterval time.Duration
|
||||
metricInterval time.Duration
|
||||
|
|
@ -96,7 +95,7 @@ func OptServerLogger(l Logger) ServerOption {
|
|||
|
||||
func OptServerReplicaN(n int) ServerOption {
|
||||
return func(s *Server) error {
|
||||
s.Cluster.ReplicaN = n
|
||||
s.cluster.ReplicaN = n
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
|
@ -124,7 +123,7 @@ func OptServerAntiEntropyInterval(interval time.Duration) ServerOption {
|
|||
|
||||
func OptServerLongQueryTime(dur time.Duration) ServerOption {
|
||||
return func(s *Server) error {
|
||||
s.Cluster.LongQueryTime = dur
|
||||
s.cluster.LongQueryTime = dur
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
|
@ -161,7 +160,7 @@ func OptServerInternalClient(c InternalClient) ServerOption {
|
|||
return func(s *Server) error {
|
||||
s.executor = NewExecutor(OptExecutorInternalQueryClient(c))
|
||||
s.defaultClient = c
|
||||
s.Cluster.InternalClient = c
|
||||
s.cluster.InternalClient = c
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
|
@ -215,7 +214,7 @@ func OptServerIsCoordinator(is bool) ServerOption {
|
|||
func NewServer(opts ...ServerOption) (*Server, error) {
|
||||
s := &Server{
|
||||
closing: make(chan struct{}),
|
||||
Cluster: NewCluster(),
|
||||
cluster: NewCluster(),
|
||||
holder: NewHolder(),
|
||||
diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer),
|
||||
systemInfo: NewNopSystemInfo(),
|
||||
|
|
@ -246,9 +245,9 @@ func NewServer(opts ...ServerOption) (*Server, error) {
|
|||
s.holder.Logger = s.logger
|
||||
s.holder.Stats.SetLogger(s.logger)
|
||||
|
||||
s.Cluster.Path = path
|
||||
s.Cluster.Logger = s.logger
|
||||
s.Cluster.Holder = s.holder
|
||||
s.cluster.Path = path
|
||||
s.cluster.Logger = s.logger
|
||||
s.cluster.Holder = s.holder
|
||||
|
||||
// Initialize translation database.
|
||||
s.translateFile = NewTranslateFile()
|
||||
|
|
@ -256,38 +255,38 @@ func NewServer(opts ...ServerOption) (*Server, error) {
|
|||
s.translateFile.PrimaryTranslateStore = s.primaryTranslateStore
|
||||
|
||||
// Get or create NodeID.
|
||||
s.NodeID = s.LoadNodeID()
|
||||
s.nodeID = s.loadNodeID()
|
||||
if s.isCoordinator {
|
||||
s.Cluster.Coordinator = s.NodeID
|
||||
s.cluster.Coordinator = s.nodeID
|
||||
}
|
||||
|
||||
// Set Cluster Node.
|
||||
node := &Node{
|
||||
ID: s.NodeID,
|
||||
ID: s.nodeID,
|
||||
URI: s.URI,
|
||||
IsCoordinator: s.Cluster.Coordinator == s.NodeID,
|
||||
IsCoordinator: s.cluster.Coordinator == s.nodeID,
|
||||
}
|
||||
s.Cluster.Node = node
|
||||
s.cluster.Node = node
|
||||
if s.clusterDisabled {
|
||||
err := s.Cluster.setStatic(s.hosts)
|
||||
err := s.cluster.setStatic(s.hosts)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "setting cluster static")
|
||||
}
|
||||
}
|
||||
|
||||
// Append the NodeID tag to stats.
|
||||
s.holder.Stats = s.holder.Stats.WithTags(fmt.Sprintf("NodeID:%s", s.NodeID))
|
||||
s.holder.Stats = s.holder.Stats.WithTags(fmt.Sprintf("NodeID:%s", s.nodeID))
|
||||
|
||||
s.executor.Holder = s.holder
|
||||
s.executor.Node = node
|
||||
s.executor.Cluster = s.Cluster
|
||||
s.executor.Cluster = s.cluster
|
||||
s.executor.TranslateStore = s.translateFile
|
||||
s.executor.MaxWritesPerRequest = s.maxWritesPerRequest
|
||||
s.Cluster.Broadcaster = s
|
||||
s.Cluster.MaxWritesPerRequest = s.maxWritesPerRequest
|
||||
s.cluster.Broadcaster = s
|
||||
s.cluster.MaxWritesPerRequest = s.maxWritesPerRequest
|
||||
s.holder.Broadcaster = s
|
||||
|
||||
err = s.Cluster.setup()
|
||||
err = s.cluster.setup()
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "setting up cluster")
|
||||
}
|
||||
|
|
@ -311,7 +310,7 @@ func (s *Server) Open() error {
|
|||
}
|
||||
|
||||
// Open Cluster management.
|
||||
if err := s.Cluster.waitForStarted(); err != nil {
|
||||
if err := s.cluster.waitForStarted(); err != nil {
|
||||
return fmt.Errorf("opening Cluster: %v", err)
|
||||
}
|
||||
|
||||
|
|
@ -319,7 +318,7 @@ func (s *Server) Open() error {
|
|||
if err := s.holder.Open(); err != nil {
|
||||
return fmt.Errorf("opening Holder: %v", err)
|
||||
}
|
||||
if err := s.Cluster.setNodeState(NodeStateReady); err != nil {
|
||||
if err := s.cluster.setNodeState(NodeStateReady); err != nil {
|
||||
return fmt.Errorf("setting nodeState: %v", err)
|
||||
}
|
||||
|
||||
|
|
@ -328,7 +327,7 @@ func (s *Server) Open() error {
|
|||
// the cluster without waiting for data to load on the coordinator. Before
|
||||
// this starts, the joins are queued up in the Cluster.joiningLeavingNodes
|
||||
// buffered channel.
|
||||
s.Cluster.listenForJoins()
|
||||
s.cluster.listenForJoins()
|
||||
|
||||
// Start background monitoring.
|
||||
s.wg.Add(3)
|
||||
|
|
@ -345,8 +344,8 @@ func (s *Server) Close() error {
|
|||
close(s.closing)
|
||||
s.wg.Wait()
|
||||
|
||||
if s.Cluster != nil {
|
||||
s.Cluster.close()
|
||||
if s.cluster != nil {
|
||||
s.cluster.close()
|
||||
}
|
||||
if s.holder != nil {
|
||||
s.holder.Close()
|
||||
|
|
@ -358,37 +357,20 @@ func (s *Server) Close() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// LoadNodeID gets NodeID from disk, or creates a new value.
|
||||
// loadNodeID gets NodeID from disk, or creates a new value.
|
||||
// If server.NodeID is already set, a new ID is not created.
|
||||
func (s *Server) LoadNodeID() string {
|
||||
if s.NodeID != "" {
|
||||
return s.NodeID
|
||||
func (s *Server) loadNodeID() string {
|
||||
if s.nodeID != "" {
|
||||
return s.nodeID
|
||||
}
|
||||
nodeID, err := s.holder.loadNodeID()
|
||||
if err != nil {
|
||||
s.logger.Printf("loading NodeID: %v", err)
|
||||
return s.NodeID
|
||||
return s.nodeID
|
||||
}
|
||||
return nodeID
|
||||
}
|
||||
|
||||
type pilosaAddr URI
|
||||
|
||||
func (p pilosaAddr) String() string {
|
||||
uri := URI(p)
|
||||
return uri.HostPort()
|
||||
|
||||
}
|
||||
|
||||
func (pilosaAddr) Network() string {
|
||||
return "tcp"
|
||||
}
|
||||
|
||||
// Addr returns the address of the listener.
|
||||
func (s *Server) Addr() net.Addr {
|
||||
return pilosaAddr(s.URI)
|
||||
}
|
||||
|
||||
func (s *Server) monitorAntiEntropy() {
|
||||
ticker := time.NewTicker(s.antiEntropyInterval)
|
||||
defer ticker.Stop()
|
||||
|
|
@ -409,8 +391,8 @@ func (s *Server) monitorAntiEntropy() {
|
|||
// Initialize syncer with local holder and remote client.
|
||||
var syncer HolderSyncer
|
||||
syncer.Holder = s.holder
|
||||
syncer.Node = s.Cluster.Node
|
||||
syncer.Cluster = s.Cluster
|
||||
syncer.Node = s.cluster.Node
|
||||
syncer.Cluster = s.cluster
|
||||
syncer.Closing = s.closing
|
||||
syncer.Stats = s.holder.Stats.WithTags("HolderSyncer")
|
||||
|
||||
|
|
@ -480,33 +462,33 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
|
|||
return err
|
||||
}
|
||||
case *internal.ClusterStatus:
|
||||
err := s.Cluster.mergeClusterStatus(obj)
|
||||
err := s.cluster.mergeClusterStatus(obj)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
case *internal.ResizeInstruction:
|
||||
err := s.Cluster.followResizeInstruction(obj)
|
||||
err := s.cluster.followResizeInstruction(obj)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
case *internal.ResizeInstructionComplete:
|
||||
err := s.Cluster.markResizeInstructionComplete(obj)
|
||||
err := s.cluster.markResizeInstructionComplete(obj)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
case *internal.SetCoordinatorMessage:
|
||||
s.Cluster.setCoordinator(DecodeNode(obj.New))
|
||||
s.cluster.setCoordinator(DecodeNode(obj.New))
|
||||
case *internal.UpdateCoordinatorMessage:
|
||||
s.Cluster.updateCoordinator(DecodeNode(obj.New))
|
||||
s.cluster.updateCoordinator(DecodeNode(obj.New))
|
||||
case *internal.NodeStateMessage:
|
||||
err := s.Cluster.receiveNodeState(obj.NodeID, obj.State)
|
||||
err := s.cluster.receiveNodeState(obj.NodeID, obj.State)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
case *internal.RecalculateCaches:
|
||||
s.holder.RecalculateCaches()
|
||||
case *internal.NodeEventMessage:
|
||||
s.Cluster.ReceiveEvent(DecodeNodeEvent(obj))
|
||||
s.cluster.ReceiveEvent(DecodeNodeEvent(obj))
|
||||
}
|
||||
|
||||
return nil
|
||||
|
|
@ -515,7 +497,7 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
|
|||
// SendSync represents an implementation of Broadcaster.
|
||||
func (s *Server) SendSync(pb proto.Message) error {
|
||||
var eg errgroup.Group
|
||||
for _, node := range s.Cluster.Nodes {
|
||||
for _, node := range s.cluster.Nodes {
|
||||
node := node
|
||||
s.logger.Printf("SendSync to: %s", node.URI)
|
||||
// Don't forward the message to ourselves.
|
||||
|
|
@ -545,7 +527,7 @@ func (s *Server) SendTo(to *Node, pb proto.Message) error {
|
|||
// Node returns the pilosa.Node object. It is used by membership protocols to
|
||||
// get this node's name(ID), location(URI), and coordinator status.
|
||||
func (s *Server) Node() *Node {
|
||||
return s.Cluster.Node
|
||||
return s.cluster.Node
|
||||
}
|
||||
|
||||
// Server implements StatusHandler.
|
||||
|
|
@ -559,7 +541,7 @@ func (s *Server) Node() *Node {
|
|||
// - Schema
|
||||
// In a gossip implementation, memberlist.Delegate.LocalState() uses this.
|
||||
func (s *Server) LocalStatus() (proto.Message, error) {
|
||||
if s.Cluster == nil {
|
||||
if s.cluster == nil {
|
||||
return nil, errors.New("Server.Cluster is nil")
|
||||
}
|
||||
if s.holder == nil {
|
||||
|
|
@ -567,7 +549,7 @@ func (s *Server) LocalStatus() (proto.Message, error) {
|
|||
}
|
||||
|
||||
ns := internal.NodeStatus{
|
||||
Node: EncodeNode(s.Cluster.Node),
|
||||
Node: EncodeNode(s.cluster.Node),
|
||||
MaxSlices: s.holder.EncodeMaxSlices(),
|
||||
Schema: s.holder.EncodeSchema(),
|
||||
}
|
||||
|
|
@ -577,13 +559,13 @@ func (s *Server) LocalStatus() (proto.Message, error) {
|
|||
|
||||
// ClusterStatus returns the ClusterState and NodeSet for the cluster.
|
||||
func (s *Server) ClusterStatus() (proto.Message, error) {
|
||||
return s.Cluster.Status(), nil
|
||||
return s.cluster.Status(), nil
|
||||
}
|
||||
|
||||
// HandleRemoteStatus receives incoming NodeStatus from remote nodes.
|
||||
func (s *Server) HandleRemoteStatus(pb proto.Message) error {
|
||||
// Ignore NodeStatus messages until the cluster is in a Normal state.
|
||||
if s.Cluster.State() != ClusterStateNormal {
|
||||
if s.cluster.State() != ClusterStateNormal {
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -602,7 +584,7 @@ func (s *Server) HandleRemoteStatus(pb proto.Message) error {
|
|||
|
||||
func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
|
||||
// Ignore status updates from self.
|
||||
if s.NodeID == DecodeNode(ns.Node).ID {
|
||||
if s.nodeID == DecodeNode(ns.Node).ID {
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -643,11 +625,11 @@ func (s *Server) monitorDiagnostics() {
|
|||
s.diagnostics.Logger = s.logger
|
||||
s.diagnostics.SetVersion(Version)
|
||||
s.diagnostics.Set("Host", s.URI.host)
|
||||
s.diagnostics.Set("Cluster", strings.Join(s.Cluster.nodeIDs(), ","))
|
||||
s.diagnostics.Set("NumNodes", len(s.Cluster.Nodes))
|
||||
s.diagnostics.Set("Cluster", strings.Join(s.cluster.nodeIDs(), ","))
|
||||
s.diagnostics.Set("NumNodes", len(s.cluster.Nodes))
|
||||
s.diagnostics.Set("NumCPU", runtime.NumCPU())
|
||||
s.diagnostics.Set("NodeID", s.NodeID)
|
||||
s.diagnostics.Set("ClusterID", s.Cluster.ID)
|
||||
s.diagnostics.Set("NodeID", s.nodeID)
|
||||
s.diagnostics.Set("ClusterID", s.cluster.ID)
|
||||
s.diagnostics.EnrichWithOSInfo()
|
||||
|
||||
// Flush the diagnostics metrics at startup, then on each tick interval
|
||||
|
|
@ -727,7 +709,7 @@ func (s *Server) monitorRuntime() {
|
|||
|
||||
// ReceiveEvent implements the EventHandler interface.
|
||||
func (s *Server) ReceiveEvent(e *NodeEvent) error {
|
||||
return s.Cluster.ReceiveEvent(e)
|
||||
return s.cluster.ReceiveEvent(e)
|
||||
}
|
||||
|
||||
// countOpenFiles on operating systems that support lsof.
|
||||
|
|
|
|||
|
|
@ -36,11 +36,6 @@ func TestMain_SendReceiveMessage(t *testing.T) {
|
|||
defer m0.Close()
|
||||
defer m1.Close()
|
||||
|
||||
m0.Server.Cluster.SetState(pilosa.ClusterStateNormal)
|
||||
m1.Server.Cluster.SetState(pilosa.ClusterStateNormal)
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
|
||||
// Expected indexes and Fields
|
||||
expected := map[string][]string{
|
||||
"i": []string{"f"},
|
||||
|
|
@ -124,8 +119,8 @@ func TestClusterResize_EmptyNode(t *testing.T) {
|
|||
m0 := test.MustRunMain()
|
||||
defer m0.Close()
|
||||
|
||||
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected cluster state: %s", m0.Server.Cluster.State())
|
||||
if m0.API.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected cluster state: %s", m0.API.State())
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -135,10 +130,10 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
|
|||
defer clus[0].Close()
|
||||
defer clus[1].Close()
|
||||
|
||||
if clus[0].Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", clus[0].Server.Cluster.State())
|
||||
} else if clus[1].Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", clus[1].Server.Cluster.State())
|
||||
if clus[0].API.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", clus[0].API.State())
|
||||
} else if clus[1].API.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", clus[1].API.State())
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -147,10 +142,10 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
t.Run("NoData", func(t *testing.T) {
|
||||
clus := test.MustRunMainWithCluster(t, 2)
|
||||
|
||||
if !checkClusterState(clus[0].Server.Cluster, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", clus[0].Server.Cluster.State())
|
||||
} else if !checkClusterState(clus[1].Server.Cluster, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", clus[1].Server.Cluster.State())
|
||||
if !checkClusterState(clus[0], pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", clus[0].API.State())
|
||||
} else if !checkClusterState(clus[1], pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", clus[1].API.State())
|
||||
}
|
||||
})
|
||||
t.Run("WithIndex", func(t *testing.T) {
|
||||
|
|
@ -180,10 +175,10 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
}
|
||||
defer m1.Close()
|
||||
|
||||
if !checkClusterState(m0.Server.Cluster, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
|
||||
} else if !checkClusterState(m1.Server.Cluster, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
|
||||
if !checkClusterState(m0, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.API.State())
|
||||
} else if !checkClusterState(m1, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.API.State())
|
||||
}
|
||||
})
|
||||
t.Run("ContinuousSlices", func(t *testing.T) {
|
||||
|
|
@ -221,10 +216,10 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
}
|
||||
defer m1.Close()
|
||||
|
||||
if !checkClusterState(m0.Server.Cluster, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
|
||||
} else if !checkClusterState(m1.Server.Cluster, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
|
||||
if !checkClusterState(m0, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.API.State())
|
||||
} else if !checkClusterState(m1, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.API.State())
|
||||
}
|
||||
})
|
||||
t.Run("SkippedSlice", func(t *testing.T) {
|
||||
|
|
@ -262,10 +257,10 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
}
|
||||
defer m1.Close()
|
||||
|
||||
if !checkClusterState(m0.Server.Cluster, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
|
||||
} else if !checkClusterState(m1.Server.Cluster, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
|
||||
if !checkClusterState(m0, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.API.State())
|
||||
} else if !checkClusterState(m1, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.API.State())
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
@ -313,15 +308,15 @@ func TestCluster_GossipMembership(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if !checkClusterState(m0.Server.Cluster, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
|
||||
} else if !checkClusterState(m1.Server.Cluster, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
|
||||
} else if !checkClusterState(m2.Server.Cluster, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node2 cluster state: %s", m2.Server.Cluster.State())
|
||||
if !checkClusterState(m0, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.API.State())
|
||||
} else if !checkClusterState(m1, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.API.State())
|
||||
} else if !checkClusterState(m2, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node2 cluster state: %s", m2.API.State())
|
||||
}
|
||||
|
||||
numNodes := len(m0.Server.Cluster.Status().Nodes)
|
||||
numNodes := len(m0.API.Hosts(context.Background()))
|
||||
if numNodes != 3 {
|
||||
t.Fatalf("Expected 3 nodes, got %d", numNodes)
|
||||
}
|
||||
|
|
@ -415,9 +410,9 @@ func TestClusterResize_RemoveNode(t *testing.T) {
|
|||
|
||||
// checkClusterState polls a given cluster for its state until it
|
||||
// receives a matching state. It polls up to n times before returning.
|
||||
func checkClusterState(c *pilosa.Cluster, state string, n int) bool {
|
||||
func checkClusterState(m *test.Main, state string, n int) bool {
|
||||
for i := 0; i < n; i++ {
|
||||
if c.State() == state {
|
||||
if m.API.State() == state {
|
||||
return true
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
|
|
|||
|
|
@ -180,7 +180,7 @@ func (m *Main) Reopen() error {
|
|||
}
|
||||
|
||||
// URL returns the base URL string for accessing the running program.
|
||||
func (m *Main) URL() string { return "http://" + m.Server.Addr().String() }
|
||||
func (m *Main) URL() string { return m.Server.URI.String() }
|
||||
|
||||
// Client returns a client to connect to the program.
|
||||
func (m *Main) Client() *http.InternalClient {
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@
|
|||
package test_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
|
@ -27,15 +28,16 @@ import (
|
|||
func TestNewCluster(t *testing.T) {
|
||||
numNodes := 3
|
||||
cluster := test.MustRunMainWithCluster(t, numNodes)
|
||||
coordinator := cluster[0].Server.Cluster.Coordinator
|
||||
|
||||
coordinator := getCoordinator(cluster[0])
|
||||
for i := 1; i < numNodes; i++ {
|
||||
if coordi := cluster[i].Server.Cluster.Coordinator; coordi != coordinator {
|
||||
if coordi := getCoordinator(cluster[i]); coordi != coordinator {
|
||||
t.Fatalf("node %d does not have the same coordinator as node 0. '%v' and '%v' respectively", i, coordi, coordinator)
|
||||
}
|
||||
}
|
||||
req, err := http.NewRequest(
|
||||
"GET",
|
||||
"http://"+cluster[0].Server.Addr().String()+"/status",
|
||||
cluster[0].URL()+"/status",
|
||||
strings.NewReader(""),
|
||||
)
|
||||
|
||||
|
|
@ -75,3 +77,13 @@ func TestNewCluster(t *testing.T) {
|
|||
t.Fatalf("cluster state should be %s but is %s", pilosa.ClusterStateNormal, body.State)
|
||||
}
|
||||
}
|
||||
|
||||
func getCoordinator(m *test.Main) string {
|
||||
hosts := m.API.Hosts(context.Background())
|
||||
for _, host := range hosts {
|
||||
if host.IsCoordinator {
|
||||
return host.ID
|
||||
}
|
||||
}
|
||||
panic("no coordinator in cluster")
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue