diff --git a/api.go b/api.go index 3fca21cab..abf2a84ea 100644 --- a/api.go +++ b/api.go @@ -477,8 +477,10 @@ func (api *API) Hosts(ctx context.Context) []*Node { return api.cluster.Nodes } +// Node gets the ID, URI and coordinator status for this particular node. func (api *API) Node() *Node { - return api.server.Node() + node := api.server.node() + return &node } // RecalculateCaches forces all TopN caches to be updated. Used mainly for integration tests. @@ -515,7 +517,7 @@ func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error { } // Forward the error message. - if err := api.server.ReceiveMessage(pb); err != nil { + if err := api.server.receiveMessage(pb); err != nil { return errors.Wrap(err, "receiving message") } return nil diff --git a/ctl/export_test.go b/ctl/export_test.go index e959da6bc..e3189efe3 100644 --- a/ctl/export_test.go +++ b/ctl/export_test.go @@ -49,7 +49,7 @@ func TestExportCommand_Run(t *testing.T) { buf := bytes.Buffer{} stdin, stdout, stderr := GetIO(buf) cm := NewExportCommand(stdin, stdout, stderr) - hostport := cmd.Server.URI.HostPort() + hostport := cmd.API.Node().URI.HostPort() cm.Host = hostport http.DefaultClient.Do(test.MustNewHTTPRequest("POST", "http://"+hostport+"/index/i", strings.NewReader(""))) diff --git a/ctl/import_test.go b/ctl/import_test.go index 32792c02d..56970e9be 100644 --- a/ctl/import_test.go +++ b/ctl/import_test.go @@ -62,7 +62,7 @@ func TestImportCommand_Run(t *testing.T) { } cmd := test.MustRunCluster(t, 1)[0] - cm.Host = cmd.Server.URI.HostPort() + cm.Host = cmd.API.Node().URI.HostPort() cm.Index = "i" cm.Field = "f" @@ -87,7 +87,7 @@ func TestImportCommand_RunValue(t *testing.T) { } cmd := test.MustRunCluster(t, 1)[0] - cm.Host = cmd.Server.URI.HostPort() + cm.Host = cmd.API.Node().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": 100}}`))) @@ -107,7 +107,7 @@ func TestImportCommand_InvalidFile(t *testing.T) { buf := bytes.Buffer{} stdin, stdout, stderr := GetIO(buf) cm := NewImportCommand(stdin, stdout, stderr) - cm.Host = cmd.Server.URI.HostPort() + cm.Host = cmd.API.Node().URI.HostPort() cm.Index = "i" cm.Field = "f" file, err := ioutil.TempFile("", "import.csv") @@ -188,7 +188,7 @@ func TestImportCommand_BugOverwriteValue(t *testing.T) { t.Fatal(err) } - cm.Host = cmd.Server.URI.HostPort() + cm.Host = cmd.API.Node().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 }}`))) diff --git a/http/translator_test.go b/http/translator_test.go index a3a5e8603..04531dab5 100644 --- a/http/translator_test.go +++ b/http/translator_test.go @@ -75,7 +75,7 @@ func TestTranslateStore_Reader(t *testing.T) { defer main.Close() // Connect to server and stream all available data. - store := http.NewTranslateStore(main.Server.URI.String()) + store := http.NewTranslateStore(main.URL()) rc, err := store.Reader(context.Background(), 100) if err != nil { @@ -128,7 +128,7 @@ func TestTranslateStore_Reader(t *testing.T) { // Connect to server and begin streaming. ctx, cancel := context.WithCancel(context.Background()) - store := http.NewTranslateStore(main.Server.URI.String()) + store := http.NewTranslateStore(main.URL()) if _, err := store.Reader(ctx, 0); err != nil { t.Fatal(err) } @@ -155,7 +155,7 @@ func TestTranslateStore_Reader(t *testing.T) { main := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0] defer main.Close() - _, err := http.NewTranslateStore(main.Server.URI.String()).Reader(context.Background(), 0) + _, err := http.NewTranslateStore(main.URL()).Reader(context.Background(), 0) if err != pilosa.ErrNotImplemented { t.Fatalf("unexpected error: %s", err) } diff --git a/server.go b/server.go index 14f2be179..56195b427 100644 --- a/server.go +++ b/server.go @@ -36,12 +36,11 @@ import ( // Default server settings. const ( - DefaultDiagnosticServer = "https://diagnostics.pilosa.com/v0/diagnostics" + defaultDiagnosticServer = "https://diagnostics.pilosa.com/v0/diagnostics" ) // Ensure Server implements interfaces. var _ broadcaster = &Server{} -var _ MemberServer = &Server{} // Server represents a holder wrapped by a running HTTP server. type Server struct { @@ -64,7 +63,7 @@ type Server struct { logger Logger nodeID string - URI URI + uri URI antiEntropyInterval time.Duration metricInterval time.Duration diagnosticInterval time.Duration @@ -188,7 +187,7 @@ func OptServerDiagnosticsInterval(dur time.Duration) ServerOption { func OptServerURI(uri *URI) ServerOption { return func(s *Server) error { - s.URI = *uri + s.uri = *uri return nil } } @@ -230,7 +229,7 @@ func NewServer(opts ...ServerOption) (*Server, error) { closing: make(chan struct{}), cluster: newCluster(), holder: NewHolder(), - diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer), + diagnostics: NewDiagnosticsCollector(defaultDiagnosticServer), systemInfo: NewNopSystemInfo(), gcNotifier: NopGCNotifier, @@ -277,7 +276,7 @@ func NewServer(opts ...ServerOption) (*Server, error) { // Set Cluster Node. node := &Node{ ID: s.nodeID, - URI: s.URI, + URI: s.uri, IsCoordinator: s.cluster.Coordinator == s.nodeID, } s.cluster.Node = node @@ -431,8 +430,8 @@ func (s *Server) monitorAntiEntropy() { } } -// ReceiveMessage represents an implementation of BroadcastHandler. -func (s *Server) ReceiveMessage(pb proto.Message) error { +// receiveMessage represents an implementation of BroadcastHandler. +func (s *Server) receiveMessage(pb proto.Message) error { switch obj := pb.(type) { case *internal.CreateShardMessage: idx := s.holder.Index(obj.Index) @@ -512,7 +511,7 @@ func (s *Server) ReceiveMessage(pb proto.Message) error { case *internal.NodeEventMessage: s.cluster.ReceiveEvent(DecodeNodeEvent(obj)) case *internal.NodeStatus: - s.HandleRemoteStatus(pb) + s.handleRemoteStatus(pb) } return nil @@ -525,7 +524,7 @@ func (s *Server) SendSync(pb proto.Message) error { node := node s.logger.Printf("SendSync to: %s", node.URI) // Don't forward the message to ourselves. - if s.URI == node.URI { + if s.uri == node.URI { continue } @@ -548,44 +547,17 @@ func (s *Server) SendTo(to *Node, pb proto.Message) error { return s.defaultClient.SendMessage(context.Background(), &to.URI, pb) } -// Node returns the pilosa.Node object. It is used by membership protocols to +// 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 +func (s *Server) node() Node { + return *s.cluster.Node } -// Server implements StatusHandler. -// LocalStatus is used to periodically sync information -// between nodes. Under normal conditions, nodes should -// remain in sync through Broadcast messages. For cases -// where a node fails to receive a Broadcast message, or -// when a new (empty) node needs to get in sync with the -// rest of the cluster, two things are shared via gossip: -// - MaxShard by Index -// - Schema -// In a gossip implementation, memberlist.Delegate.LocalState() uses this. -func (s *Server) LocalStatus() (proto.Message, error) { - if s.cluster == nil { - return nil, errors.New("Server.Cluster is nil") - } - if s.holder == nil { - return nil, errors.New("Server.Holder is nil") - } - - ns := internal.NodeStatus{ - Node: EncodeNode(s.cluster.Node), - MaxShards: s.holder.encodeMaxShards(), - Schema: s.holder.encodeSchema(), - } - - return &ns, nil -} - -// HandleRemoteStatus receives incoming NodeStatus from remote nodes. -func (s *Server) HandleRemoteStatus(pb proto.Message) error { +// handleRemoteStatus receives incoming NodeStatus from remote nodes. +func (s *Server) handleRemoteStatus(pb proto.Message) { // Ignore NodeStatus messages until the cluster is in a Normal state. if s.cluster.State() != ClusterStateNormal { - return nil + return } go func() { @@ -597,8 +569,6 @@ func (s *Server) HandleRemoteStatus(pb proto.Message) error { s.logger.Printf("merge remote status: %s", err) } }() - - return nil } func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error { @@ -643,7 +613,7 @@ func (s *Server) monitorDiagnostics() { s.diagnostics.Logger = s.logger s.diagnostics.SetVersion(Version) - s.diagnostics.Set("Host", s.URI.host) + 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("NumCPU", runtime.NumCPU()) @@ -758,10 +728,3 @@ func expandDirName(path string) (string, error) { } return path, nil } - -type MemberServer interface { - ReceiveMessage(proto.Message) error - LocalStatus() (proto.Message, error) - HandleRemoteStatus(proto.Message) error - Node() *Node -} diff --git a/server/server.go b/server/server.go index 7781ed4bc..c8de0664c 100644 --- a/server/server.go +++ b/server/server.go @@ -139,7 +139,7 @@ func (m *Command) Start() (err error) { return errors.Wrap(err, "opening server") } - m.logger.Printf("Listening as %s\n", m.Server.URI) + m.logger.Printf("Listening as %s\n", m.API.Node().URI) return nil } @@ -309,7 +309,7 @@ func (m *Command) SetupNetworking() error { } // get the host portion of addr to use for binding - gossipHost := m.Server.URI.Host() + gossipHost := m.API.Node().URI.Host() m.gossipTransport, err = gossip.NewTransport(gossipHost, gossipPort, m.logger.Logger()) if err != nil { return errors.Wrap(err, "getting transport") diff --git a/server/server_test.go b/server/server_test.go index 8b1e29acc..506807f70 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -44,7 +44,7 @@ func TestMain_Set_Quick(t *testing.T) { defer m.Close() // Create client. - client, err := http.NewInternalClient(m.Server.URI.HostPort(), http.GetHTTPClient(nil)) + client, err := http.NewInternalClient(m.API.Node().URI.HostPort(), http.GetHTTPClient(nil)) if err != nil { t.Fatal(err) } diff --git a/test/pilosa.go b/test/pilosa.go index 4774b4cf8..3fa2a6b07 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -130,11 +130,11 @@ func (m *Command) Reopen() error { } // URL returns the base URL string for accessing the running program. -func (m *Command) URL() string { return m.Server.URI.String() } +func (m *Command) URL() string { return m.API.Node().URI.String() } // Client returns a client to connect to the program. func (m *Command) Client() *http.InternalClient { - client, err := http.NewInternalClient(m.Server.URI.HostPort(), http.GetHTTPClient(nil)) + client, err := http.NewInternalClient(m.API.Node().URI.HostPort(), http.GetHTTPClient(nil)) if err != nil { panic(err) }