You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

219 lines
5.4 KiB

11 years ago
11 years ago
11 years ago
7 years ago
  1. package weed_server
  2. import (
  3. "bytes"
  4. "encoding/json"
  5. "errors"
  6. "fmt"
  7. "io/ioutil"
  8. "math/rand"
  9. "net/http"
  10. "net/url"
  11. "os"
  12. "path"
  13. "reflect"
  14. "sort"
  15. "strings"
  16. "time"
  17. "github.com/chrislusf/raft"
  18. "github.com/chrislusf/seaweedfs/weed/glog"
  19. "github.com/chrislusf/seaweedfs/weed/topology"
  20. "github.com/gorilla/mux"
  21. )
  22. type RaftServer struct {
  23. peers []string // initial peers to join with
  24. raftServer raft.Server
  25. dataDir string
  26. httpAddr string
  27. router *mux.Router
  28. topo *topology.Topology
  29. }
  30. func NewRaftServer(r *mux.Router, peers []string, httpAddr string, dataDir string, topo *topology.Topology, pulseSeconds int) *RaftServer {
  31. s := &RaftServer{
  32. peers: peers,
  33. httpAddr: httpAddr,
  34. dataDir: dataDir,
  35. router: r,
  36. topo: topo,
  37. }
  38. if glog.V(4) {
  39. raft.SetLogLevel(2)
  40. }
  41. raft.RegisterCommand(&topology.MaxVolumeIdCommand{})
  42. var err error
  43. transporter := raft.NewHTTPTransporter("/cluster", 0)
  44. transporter.Transport.MaxIdleConnsPerHost = 1024
  45. glog.V(1).Infof("Starting RaftServer with IP:%v:", httpAddr)
  46. // Clear old cluster configurations if peers are changed
  47. if oldPeers, changed := isPeersChanged(s.dataDir, httpAddr, s.peers); changed {
  48. glog.V(0).Infof("Peers Change: %v => %v", oldPeers, s.peers)
  49. os.RemoveAll(path.Join(s.dataDir, "conf"))
  50. os.RemoveAll(path.Join(s.dataDir, "log"))
  51. os.RemoveAll(path.Join(s.dataDir, "snapshot"))
  52. }
  53. s.raftServer, err = raft.NewServer(s.httpAddr, s.dataDir, transporter, nil, topo, "")
  54. if err != nil {
  55. glog.V(0).Infoln(err)
  56. return nil
  57. }
  58. transporter.Install(s.raftServer, s)
  59. s.raftServer.SetHeartbeatInterval(500 * time.Millisecond)
  60. s.raftServer.SetElectionTimeout(time.Duration(pulseSeconds) * 500 * time.Millisecond)
  61. s.raftServer.Start()
  62. s.router.HandleFunc("/cluster/join", s.joinHandler).Methods("POST")
  63. s.router.HandleFunc("/cluster/status", s.statusHandler).Methods("GET")
  64. if len(s.peers) > 0 {
  65. // Join to leader if specified.
  66. for {
  67. glog.V(0).Infoln("Joining cluster:", strings.Join(s.peers, ","))
  68. time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
  69. firstJoinError := s.Join(s.peers)
  70. if firstJoinError != nil {
  71. glog.V(0).Infoln("No existing server found. Starting as leader in the new cluster.")
  72. _, err := s.raftServer.Do(&raft.DefaultJoinCommand{
  73. Name: s.raftServer.Name(),
  74. ConnectionString: "http://" + s.httpAddr,
  75. })
  76. if err != nil {
  77. glog.V(0).Infoln(err)
  78. } else {
  79. break
  80. }
  81. } else {
  82. break
  83. }
  84. }
  85. } else if s.raftServer.IsLogEmpty() {
  86. // Initialize the server by joining itself.
  87. glog.V(0).Infoln("Initializing new cluster")
  88. _, err := s.raftServer.Do(&raft.DefaultJoinCommand{
  89. Name: s.raftServer.Name(),
  90. ConnectionString: "http://" + s.httpAddr,
  91. })
  92. if err != nil {
  93. glog.V(0).Infoln(err)
  94. return nil
  95. }
  96. } else {
  97. glog.V(0).Infoln("Old conf,log,snapshot should have been removed.")
  98. }
  99. return s
  100. }
  101. func (s *RaftServer) Peers() (members []string) {
  102. peers := s.raftServer.Peers()
  103. for _, p := range peers {
  104. members = append(members, strings.TrimPrefix(p.ConnectionString, "http://"))
  105. }
  106. return
  107. }
  108. func isPeersChanged(dir string, self string, peers []string) (oldPeers []string, changed bool) {
  109. confPath := path.Join(dir, "conf")
  110. // open conf file
  111. b, err := ioutil.ReadFile(confPath)
  112. if err != nil {
  113. return oldPeers, true
  114. }
  115. conf := &raft.Config{}
  116. if err = json.Unmarshal(b, conf); err != nil {
  117. return oldPeers, true
  118. }
  119. for _, p := range conf.Peers {
  120. oldPeers = append(oldPeers, strings.TrimPrefix(p.ConnectionString, "http://"))
  121. }
  122. oldPeers = append(oldPeers, self)
  123. if len(peers) == 0 && len(oldPeers) <= 1 {
  124. return oldPeers, false
  125. }
  126. sort.Strings(peers)
  127. sort.Strings(oldPeers)
  128. return oldPeers, !reflect.DeepEqual(peers, oldPeers)
  129. }
  130. // Join joins an existing cluster.
  131. func (s *RaftServer) Join(peers []string) error {
  132. command := &raft.DefaultJoinCommand{
  133. Name: s.raftServer.Name(),
  134. ConnectionString: "http://" + s.httpAddr,
  135. }
  136. var err error
  137. var b bytes.Buffer
  138. json.NewEncoder(&b).Encode(command)
  139. for _, m := range peers {
  140. if m == s.httpAddr {
  141. continue
  142. }
  143. target := fmt.Sprintf("http://%s/cluster/join", strings.TrimSpace(m))
  144. glog.V(0).Infoln("Attempting to connect to:", target)
  145. err = postFollowingOneRedirect(target, "application/json", b)
  146. if err != nil {
  147. glog.V(0).Infoln("Post returned error: ", err.Error())
  148. if _, ok := err.(*url.Error); ok {
  149. // If we receive a network error try the next member
  150. continue
  151. }
  152. } else {
  153. return nil
  154. }
  155. }
  156. return errors.New("Could not connect to any cluster peers")
  157. }
  158. // a workaround because http POST following redirection misses request body
  159. func postFollowingOneRedirect(target string, contentType string, b bytes.Buffer) error {
  160. backupReader := bytes.NewReader(b.Bytes())
  161. resp, err := http.Post(target, contentType, &b)
  162. if err != nil {
  163. return err
  164. }
  165. defer resp.Body.Close()
  166. statusCode := resp.StatusCode
  167. data, _ := ioutil.ReadAll(resp.Body)
  168. reply := string(data)
  169. if strings.HasPrefix(reply, "\"http") {
  170. urlStr := reply[1 : len(reply)-1]
  171. glog.V(0).Infoln("Post redirected to ", urlStr)
  172. resp2, err2 := http.Post(urlStr, contentType, backupReader)
  173. if err2 != nil {
  174. return err2
  175. }
  176. defer resp2.Body.Close()
  177. data, _ = ioutil.ReadAll(resp2.Body)
  178. statusCode = resp2.StatusCode
  179. }
  180. glog.V(0).Infoln("Post returned status: ", statusCode, string(data))
  181. if statusCode != http.StatusOK {
  182. return errors.New(string(data))
  183. }
  184. return nil
  185. }