diff --git a/etcd/embed.go b/etcd/embed.go index 4c306fe5f..3118b6d4f 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -17,15 +17,18 @@ package etcd import ( "bytes" "context" + "encoding/json" "fmt" "log" "net" "path" + "sort" "strings" "time" "github.com/pilosa/pilosa/v2/disco" "github.com/pilosa/pilosa/v2/roaring" + "github.com/pilosa/pilosa/v2/topology" "github.com/pkg/errors" "go.etcd.io/etcd/clientv3" "go.etcd.io/etcd/clientv3/clientv3util" @@ -1026,3 +1029,43 @@ func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint6 return nil } + +// Nodes implements the Noder interface. +func (n *Etcd) Nodes() []*topology.Node { + // If we have looked up nodes within a certain time, then we're going to + // use the cached value for now. This is temporary and will be addressed + // correctly in #1133. + peers := n.Peers() + nodes := make([]*topology.Node, len(peers)) + for i, peer := range peers { + node := &topology.Node{} + if meta, err := n.Metadata(context.Background(), peer.ID); err != nil { + log.Println(err, "getting metadata") // TODO: handle this with a logger + } else if err := json.Unmarshal(meta, node); err != nil { + log.Println(err, "unmarshaling json metadata") + } + + node.ID = peer.ID + + nodes[i] = node + } + + // Nodes must be sorted. + sort.Sort(topology.ByID(nodes)) + + return nodes +} + +// SetNodes implements the Noder interface as NOP +// (because we can't force to set nodes for etcd). +func (n *Etcd) SetNodes(nodes []*topology.Node) {} + +// AppendNode implements the Noder interface as NOP +// (because resizer is responsible for adding new nodes). +func (n *Etcd) AppendNode(node *topology.Node) {} + +// RemoveNode implements the Noder interface as NOP +// (because resizer is responsible for removing existing nodes) +func (n *Etcd) RemoveNode(nodeID string) bool { + return false +} diff --git a/server/server.go b/server/server.go index a25830b8e..c3613ed0d 100644 --- a/server/server.go +++ b/server/server.go @@ -56,7 +56,6 @@ import ( "github.com/pilosa/pilosa/v2/stats" "github.com/pilosa/pilosa/v2/statsd" "github.com/pilosa/pilosa/v2/syswrap" - "github.com/pilosa/pilosa/v2/test/port" "github.com/pilosa/pilosa/v2/testhook" "github.com/pkg/errors" ) @@ -417,8 +416,7 @@ func (m *Command) SetupServer() error { } e := petcd.NewEtcd(m.Config.DisCo, m.Config.Cluster.ReplicaN) - n := petcd.NewNoder(m.Config.DisCo, m.Config.Cluster.ReplicaN) - discoOpt := pilosa.OptServerDisCo(e, e, e, e, n, e, e) + discoOpt := pilosa.OptServerDisCo(e, e, e, e, e, e, e) serverOptions := []pilosa.ServerOption{ pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)), @@ -504,22 +502,8 @@ func (m *Command) setupNetworking() error { // get the host portion of addr to use for binding gossipHost := m.listenURI.Host m.gossipTransport, err = gossip.NewTransport(gossipHost, gossipPort, m.logger.Logger()) - if err != nil && gossipPort >= 32768 { - // In testing, we sometimes try to reuse an ephemeral port. - // Which probably works. If it doesn't, this test will take - // about a minute longer because we'll come back in from a - // new port. See also the gossip config in gossip/gossip.go. - // TODO: Maybe make that more configurable here. - m.logger.Printf("ephemeral port %d already occupied, switching to :0 (%v)", gossipPort, err) - if err := port.GetPort(func(p int) error { - gossipPort = p - m.Config.Gossip.Port = fmt.Sprintf(":%d", gossipPort) - m.gossipTransport, err = gossip.NewTransport(gossipHost, gossipPort, m.logger.Logger()) - return err - }, 10); err != nil { - return errors.Wrap(err, "getting transport") - } - + if err != nil { + return errors.Wrap(err, "getting transport") } gossipMemberSet, err := gossip.NewMemberSet(