Merge pull request #1340 from jaten-molecula/noder_impl

Noder impl

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

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(