join_command.go 2.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283
  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. // Add to shared peer registry.
  54. ps.registry.RegisterPeer(c.Name, c.RaftURL, c.EtcdURL)
  55. // Add peer in raft
  56. err := context.Server().AddPeer(c.Name, "")
  57. // Add peer stats
  58. if c.Name != ps.RaftServer().Name() {
  59. ps.followersStats.Followers[c.Name] = &raftFollowerStats{}
  60. ps.followersStats.Followers[c.Name].Latency.Minimum = 1 << 63
  61. }
  62. binary.Write(&buf, binary.BigEndian, uint8(0)) // Mark as peer.
  63. return buf.Bytes(), err
  64. }
  65. func (c *JoinCommand) NodeName() string {
  66. return c.Name
  67. }