join_command.go 2.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788
  1. package server
  2. import (
  3. "bytes"
  4. "encoding/binary"
  5. "github.com/coreos/etcd/log"
  6. "github.com/coreos/etcd/third_party/github.com/coreos/raft"
  7. )
  8. func init() {
  9. raft.RegisterCommand(&JoinCommand{})
  10. }
  11. // The JoinCommand adds a node to the cluster.
  12. type JoinCommand struct {
  13. MinVersion int `json:"minVersion"`
  14. MaxVersion int `json:"maxVersion"`
  15. Name string `json:"name"`
  16. RaftURL string `json:"raftURL"`
  17. EtcdURL string `json:"etcdURL"`
  18. }
  19. func NewJoinCommand(minVersion int, maxVersion int, name, raftUrl, etcdUrl string) *JoinCommand {
  20. return &JoinCommand{
  21. MinVersion: minVersion,
  22. MaxVersion: maxVersion,
  23. Name: name,
  24. RaftURL: raftUrl,
  25. EtcdURL: etcdUrl,
  26. }
  27. }
  28. // The name of the join command in the log
  29. func (c *JoinCommand) CommandName() string {
  30. return "etcd:join"
  31. }
  32. // Join a server to the cluster
  33. func (c *JoinCommand) Apply(context raft.Context) (interface{}, error) {
  34. ps, _ := context.Server().Context().(*PeerServer)
  35. var buf bytes.Buffer
  36. b := make([]byte, 8)
  37. n := binary.PutUvarint(b, context.CommitIndex())
  38. buf.Write(b[:n])
  39. // Make sure we're not getting a cached value from the registry.
  40. ps.registry.Invalidate(c.Name)
  41. // Check if the join command is from a previous peer, who lost all its previous log.
  42. if _, ok := ps.registry.ClientURL(c.Name); ok {
  43. binary.Write(&buf, binary.BigEndian, uint8(0)) // Mark as peer.
  44. return buf.Bytes(), nil
  45. }
  46. // Check peer number in the cluster
  47. if ps.registry.PeerCount() >= ps.ClusterConfig().ActiveSize {
  48. log.Debug("Join as proxy ", c.Name)
  49. ps.registry.RegisterProxy(c.Name, c.RaftURL, c.EtcdURL)
  50. binary.Write(&buf, binary.BigEndian, uint8(1)) // Mark as proxy.
  51. return buf.Bytes(), nil
  52. }
  53. // Remove it as a proxy if it is one.
  54. if ps.registry.ProxyExists(c.Name) {
  55. ps.registry.UnregisterProxy(c.Name)
  56. }
  57. // Add to shared peer registry.
  58. ps.registry.RegisterPeer(c.Name, c.RaftURL, c.EtcdURL)
  59. // Add peer in raft
  60. err := context.Server().AddPeer(c.Name, "")
  61. // Add peer stats
  62. if c.Name != ps.RaftServer().Name() {
  63. ps.followersStats.Followers[c.Name] = &raftFollowerStats{}
  64. ps.followersStats.Followers[c.Name].Latency.Minimum = 1 << 63
  65. }
  66. binary.Write(&buf, binary.BigEndian, uint8(0)) // Mark as peer.
  67. return buf.Bytes(), err
  68. }
  69. func (c *JoinCommand) NodeName() string {
  70. return c.Name
  71. }