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

get rid of unecessary server stuff and export node and uri

Browse source
This commit is contained in:
Matt Jaffee 2018-07-03 13:26:51 -05:00
parent f6aef32093
commit 5ff77c816a
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
8 changed files with 33 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
api.go
View file

@ -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

2
ctl/export_test.go
View file

@ -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("")))

8
ctl/import_test.go
View file

@ -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 }}`)))

6
http/translator_test.go
View file

@ -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)
}

69
server.go
View file

@ -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
}

4
server/server.go
View file

@ -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")

2
server/server_test.go
View file

@ -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)
}

4
test/pilosa.go
View file

@ -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)
}