Merge pull request #738 from unihorn/68

feat(peer_server): forbid rejoining with different name

server/join_command.go

@@ -63,6 +63,15 @@ func (c *JoinCommandV1) Apply(context raft.Context) (interface{}, error) {
 		return b, nil
 	}
 
+	// Check if the join command adds an instance that collides with existing one on peer URL.
+	peerURLs := ps.registry.PeerURLs(ps.raftServer.Leader(), c.Name)
+	for _, peerURL := range peerURLs {
+		if peerURL == c.RaftURL {
+			log.Warnf("%v tries to join the cluster with existing URL %v", c.Name, c.EtcdURL)
+			return []byte{0}, etcdErr.NewError(etcdErr.EcodeExistingPeerAddr, c.EtcdURL, context.CommitIndex())
+		}
+	}
+
 	// Check peer number in the cluster
 	if ps.registry.PeerCount() >= ps.ClusterConfig().ActiveSize {
 		log.Debug("Reject join request from ", c.Name)
@@ -137,6 +146,15 @@ func (c *JoinCommandV2) Apply(context raft.Context) (interface{}, error) {
 		return json.Marshal(msg)
 	}
 
+	// Check if the join command adds an instance that collides with existing one on peer URL.
+	peerURLs := ps.registry.PeerURLs(ps.raftServer.Leader(), c.Name)
+	for _, peerURL := range peerURLs {
+		if peerURL == c.PeerURL {
+			log.Warnf("%v tries to join the cluster with existing URL %v", c.Name, c.PeerURL)
+			return []byte{0}, etcdErr.NewError(etcdErr.EcodeExistingPeerAddr, c.PeerURL, context.CommitIndex())
+		}
+	}
+
 	// Check peer number in the cluster.
 	if ps.registry.PeerCount() >= ps.ClusterConfig().ActiveSize {
 		log.Debug("Join as standby ", c.Name)

server/peer_server.go

@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"sort"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -187,6 +188,12 @@ func (s *PeerServer) findCluster(discoverURL string, peers []string) {
 
 	// Try its best to find possible peers, and connect with them.
 	if !isNewNode {
+		// It is not allowed to join the cluster with existing peer address
+		// This prevents old node joining with different name by mistake.
+		if !s.checkPeerAddressNonconflict() {
+			log.Fatalf("%v is not allowed to join the cluster with existing URL %v", s.Config.Name, s.Config.URL)
+		}
+
 		// Take old nodes into account.
 		allPeers := s.getKnownPeers()
 		// Discover registered peers.
@@ -426,6 +433,25 @@ func (s *PeerServer) Upgradable() error {
 	return nil
 }
 
+// checkPeerAddressNonconflict checks whether the peer address has existed with different name.
+func (s *PeerServer) checkPeerAddressNonconflict() bool {
+	// there exists the (name, peer address) pair
+	if peerURL, ok := s.registry.PeerURL(s.Config.Name); ok {
+		if peerURL == s.Config.URL {
+			return true
+		}
+	}
+
+	// check all existing peer addresses
+	peerURLs := s.registry.PeerURLs(s.raftServer.Leader(), s.Config.Name)
+	for _, peerURL := range peerURLs {
+		if peerURL == s.Config.URL {
+			return false
+		}
+	}
+	return true
+}
+
 // Helper function to do discovery and return results in expected format
 func (s *PeerServer) handleDiscovery(discoverURL string) (peers []string, err error) {
 	peers, err = discovery.Do(discoverURL, s.Config.Name, s.Config.URL)
@@ -455,6 +481,8 @@ func (s *PeerServer) handleDiscovery(discoverURL string) (peers []string, err er
 // getKnownPeers gets the previous peers from log
 func (s *PeerServer) getKnownPeers() []string {
 	peers := s.registry.PeerURLs(s.raftServer.Leader(), s.Config.Name)
+	log.Infof("Peer URLs in log: %s / %s (%s)", s.raftServer.Leader(), s.Config.Name, strings.Join(peers, ","))
+
 	for i := range peers {
 		u, err := url.Parse(peers[i])
 		if err != nil {

server/registry.go

@@ -300,8 +300,7 @@ func (r *Registry) urls(key, leaderName, selfName string, url func(key, name str
 		}
 	}
 
-	log.Infof("URLs: %s: %s / %s (%s)", key, leaderName, selfName, strings.Join(urls, ","))
-
+	log.Debugf("URLs: %s: %s / %s (%s)", key, leaderName, selfName, strings.Join(urls, ","))
 	return urls
 }