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

Merge branch 'develop' into remove-setupserver-calls

Browse source
This commit is contained in:
Matt Jaffee 2018-07-03 16:33:32 -05:00
commit 7900e47f8a
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
14 changed files with 119 additions and 125 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
api.go
View file

@ -331,8 +331,8 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
}
// Validate that this handler owns the shard.
if !api.cluster.ownsShard(api.LocalID(), indexName, shard) {
api.server.logger.Printf("node %s does not own shard %d of index %s", api.LocalID(), shard, indexName)
if !api.cluster.ownsShard(api.Node().ID, indexName, shard) {
api.server.logger.Printf("node %s does not own shard %d of index %s", api.Node().ID, shard, indexName)
return ErrClusterDoesNotOwnShard
}
@ -477,6 +477,12 @@ 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 {
node := api.server.node()
return &node
}
// RecalculateCaches forces all TopN caches to be updated. Used mainly for integration tests.
func (api *API) RecalculateCaches(ctx context.Context) error {
if err := api.validate(apiRecalculateCaches); err != nil {
@ -511,21 +517,16 @@ 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
}
// LocalID returns the current node's ID.
func (api *API) LocalID() string {
return api.cluster.Node.ID
}
// Schema returns information about each index in Pilosa including which fields
// and views they contain.
func (api *API) Schema(ctx context.Context) []*IndexInfo {
return api.holder.Schema()
func (api *API) Schema(ctx context.Context) []*Index {
return api.holder.Indexes()
}
// Views returns the views in the given field.
@ -720,8 +721,8 @@ func (api *API) LongQueryTime() time.Duration {
func (api *API) indexField(indexName string, fieldName string, shard uint64) (*Index, *Field, error) {
// Validate that this handler owns the shard.
if !api.cluster.ownsShard(api.LocalID(), indexName, shard) {
api.server.logger.Printf("node %s does not own shard %d of index %s", api.LocalID(), shard, indexName)
if !api.cluster.ownsShard(api.Node().ID, indexName, shard) {
api.server.logger.Printf("node %s does not own shard %d of index %s", api.Node().ID, shard, indexName)
return nil, nil, ErrClusterDoesNotOwnShard
}
@ -755,7 +756,7 @@ func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode
}
// If the new coordinator is this node, do the SetCoordinator directly.
if newNode.ID == api.LocalID() {
if newNode.ID == api.Node().ID {
return oldNode, newNode, api.cluster.setCoordinator(newNode)
}

6
broadcast.go
View file

@ -65,6 +65,7 @@ const (
messageTypeNodeState
messageTypeRecalculateCaches
messageTypeNodeEvent
messageTypeNodeStatus
)
// MarshalMessage encodes the protobuf message into a byte slice.
@ -101,6 +102,8 @@ func MarshalMessage(m proto.Message) ([]byte, error) {
typ = messageTypeRecalculateCaches
case *internal.NodeEventMessage:
typ = messageTypeNodeEvent
case *internal.NodeStatus:
typ = messageTypeNodeStatus
default:
return nil, fmt.Errorf("message type not implemented for marshalling: %s", reflect.TypeOf(obj))
}
@ -114,7 +117,6 @@ func MarshalMessage(m proto.Message) ([]byte, error) {
// UnmarshalMessage decodes the byte slice into a protobuf message.
func UnmarshalMessage(buf []byte) (proto.Message, error) {
typ, buf := buf[0], buf[1:]
var m proto.Message
switch typ {
case messageTypeCreateShard:
@ -147,6 +149,8 @@ func UnmarshalMessage(buf []byte) (proto.Message, error) {
m = &internal.RecalculateCaches{}
case messageTypeNodeEvent:
m = &internal.NodeEventMessage{}
case messageTypeNodeStatus:
m = &internal.NodeStatus{}
default:
return nil, fmt.Errorf("invalid message type: %d", typ)
}

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

15
field.go
View file

@ -1073,6 +1073,21 @@ func (f *Field) ImportValue(columnIDs []uint64, values []int64) error {
return nil
}
func (f *Field) MarshalJSON() ([]byte, error) {
thing := struct {
Name string
Options FieldOptions
Views []*viewInfo
}{
Name: f.Name(),
Options: f.Options(),
}
for _, viewname := range f.viewNames() {
thing.Views = append(thing.Views, &viewInfo{Name: viewname})
}
return json.Marshal(thing)
}
// encodeFields converts a into its internal representation.
func encodeFields(a []*Field) []*internal.Field {
other := make([]*internal.Field, len(a))

68
gossip/gossip.go
View file

@ -15,6 +15,8 @@
package gossip
import (
"bytes"
"context"
"fmt"
"io/ioutil"
"log"
@ -42,8 +44,8 @@ type GossipMemberSet struct {
broadcasts *memberlist.TransmitLimitedQueue
pserver pilosa.MemberServer
config *gossipConfig
papi *pilosa.API
config *gossipConfig
Logger pilosa.Logger
@ -145,9 +147,10 @@ func WithLogger(logger *log.Logger) GossipMemberSetOption {
}
// NewGossipMemberSet returns a new instance of GossipMemberSet based on options.
func NewGossipMemberSet(cfg Config, s *pilosa.Server, options ...GossipMemberSetOption) (*GossipMemberSet, error) {
host := s.Node().URI.Host()
func NewGossipMemberSet(cfg Config, api *pilosa.API, options ...GossipMemberSetOption) (*GossipMemberSet, error) {
host := api.Node().URI.Host()
g := &GossipMemberSet{
papi: api,
Logger: pilosa.NopLogger,
}
@ -157,7 +160,7 @@ func NewGossipMemberSet(cfg Config, s *pilosa.Server, options ...GossipMemberSet
return nil, errors.Wrap(err, "executing option")
}
}
ger := newGossipEventReceiver(g.logger, s)
ger := newGossipEventReceiver(g.logger, api)
g.gossipEventReceiver = ger
if g.transport == nil {
@ -189,11 +192,11 @@ func NewGossipMemberSet(cfg Config, s *pilosa.Server, options ...GossipMemberSet
// memberlist config
conf := memberlist.DefaultWANConfig()
conf.Transport = g.transport.Net
conf.Name = s.Node().ID
conf.BindAddr = s.Node().URI.Host()
conf.Name = api.Node().ID
conf.BindAddr = api.Node().URI.Host()
conf.BindPort = port
conf.AdvertisePort = port
conf.AdvertiseAddr = hostToIP(s.Node().URI.Host())
conf.AdvertiseAddr = hostToIP(api.Node().URI.Host())
//
conf.TCPTimeout = time.Duration(cfg.StreamTimeout)
conf.SuspicionMult = cfg.SuspicionMult
@ -214,14 +217,12 @@ func NewGossipMemberSet(cfg Config, s *pilosa.Server, options ...GossipMemberSet
gossipSeeds: cfg.Seeds,
}
g.pserver = s
return g, nil
}
// NodeMeta implementation of the memberlist.Delegate interface.
func (g *GossipMemberSet) NodeMeta(limit int) []byte {
buf, err := proto.Marshal(pilosa.EncodeNode(g.pserver.Node()))
buf, err := proto.Marshal(pilosa.EncodeNode(g.papi.Node()))
if err != nil {
g.Logger.Printf("marshal message error: %s", err)
return []byte{}
@ -232,14 +233,9 @@ func (g *GossipMemberSet) NodeMeta(limit int) []byte {
// NotifyMsg implementation of the memberlist.Delegate interface
// called when a user-data message is received.
func (g *GossipMemberSet) NotifyMsg(b []byte) {
m, err := pilosa.UnmarshalMessage(b)
err := g.papi.ClusterMessage(context.Background(), bytes.NewBuffer(b))
if err != nil {
g.Logger.Printf("unmarshal message error: %s", err)
return
}
if err := g.pserver.ReceiveMessage(m); err != nil {
g.Logger.Printf("receive message error: %s", err)
return
g.Logger.Printf("cluster message error: %s", err)
}
}
@ -252,14 +248,14 @@ func (g *GossipMemberSet) GetBroadcasts(overhead, limit int) [][]byte {
// LocalState implementation of the memberlist.Delegate interface
// sends this Node's state data.
func (g *GossipMemberSet) LocalState(join bool) []byte {
pb, err := g.pserver.LocalStatus()
if err != nil {
g.Logger.Printf("error getting local state, err=%s", err)
return []byte{}
pb := &internal.NodeStatus{
Node: pilosa.EncodeNode(g.papi.Node()),
MaxShards: &internal.MaxShards{Standard: g.papi.MaxShards(context.Background())},
Schema: &internal.Schema{Indexes: pilosa.EncodeIndexes(g.papi.Schema(context.Background()))},
}
// Marshal nodestate data to bytes.
buf, err := proto.Marshal(pb)
buf, err := pilosa.MarshalMessage(pb)
if err != nil {
g.Logger.Printf("error marshalling nodestate data, err=%s", err)
return []byte{}
@ -270,13 +266,7 @@ func (g *GossipMemberSet) LocalState(join bool) []byte {
// MergeRemoteState implementation of the memberlist.Delegate interface
// receive and process the remote side's LocalState.
func (g *GossipMemberSet) MergeRemoteState(buf []byte, join bool) {
// Unmarshal nodestate data.
var pb internal.NodeStatus
if err := proto.Unmarshal(buf, &pb); err != nil {
g.Logger.Printf("error unmarshalling nodestate data, err=%s", err)
return
}
err := g.pserver.HandleRemoteStatus(&pb)
err := g.papi.ClusterMessage(context.Background(), bytes.NewBuffer(buf))
if err != nil {
g.Logger.Printf("merge state error: %s", err)
}
@ -288,18 +278,18 @@ func (g *GossipMemberSet) MergeRemoteState(buf []byte, join bool) {
// Care must be taken that events are processed in a timely manner from
// the channel, since this delegate will block until an event can be sent.
type gossipEventReceiver struct {
ch chan memberlist.NodeEvent
eventHandler *pilosa.Server
ch chan memberlist.NodeEvent
papi *pilosa.API
logger *log.Logger
}
// newGossipEventReceiver returns a new instance of GossipEventReceiver.
func newGossipEventReceiver(logger *log.Logger, pserver *pilosa.Server) *gossipEventReceiver {
func newGossipEventReceiver(logger *log.Logger, papi *pilosa.API) *gossipEventReceiver {
ger := &gossipEventReceiver{
ch: make(chan memberlist.NodeEvent, 1),
logger: logger,
eventHandler: pserver,
ch: make(chan memberlist.NodeEvent, 1),
logger: logger,
papi: papi,
}
go ger.listen()
return ger
@ -342,7 +332,11 @@ func (g *gossipEventReceiver) listen() {
Event: uint32(nodeEventType),
Node: &n,
}
if err := g.eventHandler.ReceiveMessage(ne); err != nil {
buf, err := pilosa.MarshalMessage(ne)
if err != nil {
panic(err)
}
if err := g.papi.ClusterMessage(context.Background(), bytes.NewBuffer(buf)); err != nil {
g.logger.Printf("receive event error: %s", err)
}
}

2
holder.go
View file

@ -267,7 +267,7 @@ func (h *Holder) encodeMaxShards() *internal.MaxShards {
// encodeSchema creates an internal representation of schema.
func (h *Holder) encodeSchema() *internal.Schema {
return &internal.Schema{
Indexes: encodeIndexes(h.Indexes()),
Indexes: EncodeIndexes(h.Indexes()),
}
}

8
http/handler.go
View file

@ -358,9 +358,7 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
}
schema := h.API.Schema(r.Context())
if err := json.NewEncoder(w).Encode(getSchemaResponse{
Indexes: schema,
}); err != nil {
if err := json.NewEncoder(w).Encode(map[string]interface{}{"indexes": schema}); err != nil {
h.Logger.Printf("write schema response error: %s", err)
}
}
@ -374,7 +372,7 @@ func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
status := getStatusResponse{
State: h.API.State(),
Nodes: h.API.Hosts(r.Context()),
LocalID: h.API.LocalID(),
LocalID: h.API.Node().ID,
}
if err := json.NewEncoder(w).Encode(status); err != nil {
h.Logger.Printf("write status response error: %s", err)
@ -467,7 +465,7 @@ func (h *Handler) handleGetIndex(w http.ResponseWriter, r *http.Request) {
}
indexName := mux.Vars(r)["index"]
for _, idx := range h.API.Schema(r.Context()) {
if idx.Name == indexName {
if idx.Name() == indexName {
if err := json.NewEncoder(w).Encode(idx); err != nil {
h.Logger.Printf("write response error: %s", err)
}

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

21
index.go
View file

@ -15,6 +15,7 @@
package pilosa
import (
"encoding/json"
"fmt"
"io/ioutil"
"os"
@ -75,6 +76,22 @@ func NewIndex(path, name string) (*Index, error) {
}, nil
}
func (i *Index) MarshalJSON() ([]byte, error) {
fields := make([]*Field, 0, len(i.fields))
for _, f := range i.fields {
fields = append(fields, f)
}
thing := struct {
Name string
Fields []*Field
}{
Name: i.name,
Fields: fields,
}
return json.Marshal(thing)
}
// Name returns name of the index.
func (i *Index) Name() string { return i.name }
@ -386,8 +403,8 @@ func (p indexInfoSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p indexInfoSlice) Len() int { return len(p) }
func (p indexInfoSlice) Less(i, j int) bool { return p[i].Name < p[j].Name }
// encodeIndexes converts a into its internal representation.
func encodeIndexes(a []*Index) []*internal.Index {
// EncodeIndexes converts a into its internal representation.
func EncodeIndexes(a []*Index) []*internal.Index {
other := make([]*internal.Index, len(a))
for i := range a {
other[i] = encodeIndex(a[i])

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)
@ -511,6 +510,8 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
s.holder.RecalculateCaches()
case *internal.NodeEventMessage:
s.cluster.ReceiveEvent(DecodeNodeEvent(obj))
case *internal.NodeStatus:
s.handleRemoteStatus(pb)
}
return nil
@ -523,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
}
@ -546,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() {
@ -595,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 {
@ -641,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())
@ -756,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
}

6
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")
@ -317,7 +317,7 @@ func (m *Command) SetupNetworking() error {
gossipMemberSet, err := gossip.NewGossipMemberSet(
m.Config.Gossip,
m.Server,
m.API,
gossip.WithLogger(m.logger.Logger()),
gossip.WithTransport(m.gossipTransport),
)

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

@ -121,11 +121,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)
}