2013-12-10 05:34:05 +08:00
|
|
|
package weed_server
|
|
|
|
|
|
|
|
import (
|
|
|
|
"bytes"
|
|
|
|
"encoding/json"
|
|
|
|
"errors"
|
|
|
|
"fmt"
|
2014-02-05 16:25:23 +08:00
|
|
|
"io/ioutil"
|
2014-04-12 07:23:58 +08:00
|
|
|
"math/rand"
|
2013-12-10 05:34:05 +08:00
|
|
|
"net/http"
|
|
|
|
"net/url"
|
2014-10-27 16:09:45 +08:00
|
|
|
"os"
|
|
|
|
"path"
|
2013-12-10 05:34:05 +08:00
|
|
|
"strings"
|
|
|
|
"time"
|
2014-10-27 02:34:55 +08:00
|
|
|
|
2014-12-14 16:20:21 +08:00
|
|
|
"github.com/mcqueenorama/weed-fs/go/glog"
|
|
|
|
"github.com/mcqueenorama/weed-fs/go/topology"
|
2014-10-27 02:34:55 +08:00
|
|
|
"github.com/goraft/raft"
|
|
|
|
"github.com/gorilla/mux"
|
2013-12-10 05:34:05 +08:00
|
|
|
)
|
|
|
|
|
|
|
|
type RaftServer struct {
|
|
|
|
peers []string // initial peers to join with
|
|
|
|
raftServer raft.Server
|
|
|
|
dataDir string
|
|
|
|
httpAddr string
|
|
|
|
router *mux.Router
|
2014-03-16 14:03:49 +08:00
|
|
|
topo *topology.Topology
|
2013-12-10 05:34:05 +08:00
|
|
|
}
|
|
|
|
|
2014-03-26 04:46:59 +08:00
|
|
|
func NewRaftServer(r *mux.Router, peers []string, httpAddr string, dataDir string, topo *topology.Topology, pulseSeconds int) *RaftServer {
|
2013-12-10 05:34:05 +08:00
|
|
|
s := &RaftServer{
|
|
|
|
peers: peers,
|
|
|
|
httpAddr: httpAddr,
|
|
|
|
dataDir: dataDir,
|
|
|
|
router: r,
|
2014-03-16 14:03:49 +08:00
|
|
|
topo: topo,
|
2013-12-10 05:34:05 +08:00
|
|
|
}
|
|
|
|
|
2014-02-04 17:11:28 +08:00
|
|
|
if glog.V(4) {
|
|
|
|
raft.SetLogLevel(2)
|
|
|
|
}
|
2013-12-10 05:34:05 +08:00
|
|
|
|
2014-03-16 14:03:49 +08:00
|
|
|
raft.RegisterCommand(&topology.MaxVolumeIdCommand{})
|
|
|
|
|
2013-12-10 05:34:05 +08:00
|
|
|
var err error
|
2014-04-16 08:56:47 +08:00
|
|
|
transporter := raft.NewHTTPTransporter("/cluster", 0)
|
2014-03-17 04:49:49 +08:00
|
|
|
transporter.Transport.MaxIdleConnsPerHost = 1024
|
2014-10-20 11:03:00 +08:00
|
|
|
glog.V(1).Infof("Starting RaftServer with IP:%v:", httpAddr)
|
2014-03-17 04:49:49 +08:00
|
|
|
|
2014-10-27 16:09:45 +08:00
|
|
|
// Clear old cluster configurations if peers are set
|
|
|
|
if len(s.peers) > 0 {
|
|
|
|
os.RemoveAll(path.Join(s.dataDir, "conf"))
|
|
|
|
os.RemoveAll(path.Join(s.dataDir, "log"))
|
|
|
|
os.RemoveAll(path.Join(s.dataDir, "snapshot"))
|
|
|
|
}
|
|
|
|
|
2014-03-16 14:03:49 +08:00
|
|
|
s.raftServer, err = raft.NewServer(s.httpAddr, s.dataDir, transporter, nil, topo, "")
|
2013-12-10 05:34:05 +08:00
|
|
|
if err != nil {
|
|
|
|
glog.V(0).Infoln(err)
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
transporter.Install(s.raftServer, s)
|
2014-01-29 03:15:24 +08:00
|
|
|
s.raftServer.SetHeartbeatInterval(1 * time.Second)
|
2014-10-27 16:09:45 +08:00
|
|
|
s.raftServer.SetElectionTimeout(time.Duration(pulseSeconds) * 3450 * time.Millisecond)
|
2013-12-10 05:34:05 +08:00
|
|
|
s.raftServer.Start()
|
|
|
|
|
2014-02-04 17:10:07 +08:00
|
|
|
s.router.HandleFunc("/cluster/join", s.joinHandler).Methods("POST")
|
2014-02-04 17:11:28 +08:00
|
|
|
s.router.HandleFunc("/cluster/status", s.statusHandler).Methods("GET")
|
2013-12-10 05:34:05 +08:00
|
|
|
|
|
|
|
if len(s.peers) > 0 {
|
2014-10-27 16:09:45 +08:00
|
|
|
// Join to leader if specified.
|
|
|
|
glog.V(0).Infoln("Joining cluster:", strings.Join(s.peers, ","))
|
|
|
|
time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
|
|
|
|
firstJoinError := s.Join(s.peers)
|
|
|
|
if firstJoinError != nil {
|
|
|
|
glog.V(0).Infoln("No existing server found. Starting as leader in the new cluster.")
|
|
|
|
_, err := s.raftServer.Do(&raft.DefaultJoinCommand{
|
|
|
|
Name: s.raftServer.Name(),
|
|
|
|
ConnectionString: "http://" + s.httpAddr,
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
glog.V(0).Infoln(err)
|
|
|
|
return nil
|
2014-04-12 07:23:58 +08:00
|
|
|
}
|
2013-12-10 05:34:05 +08:00
|
|
|
}
|
|
|
|
} else if s.raftServer.IsLogEmpty() {
|
2014-10-27 16:09:45 +08:00
|
|
|
// Initialize the server by joining itself.
|
2013-12-10 05:34:05 +08:00
|
|
|
glog.V(0).Infoln("Initializing new cluster")
|
|
|
|
|
|
|
|
_, err := s.raftServer.Do(&raft.DefaultJoinCommand{
|
|
|
|
Name: s.raftServer.Name(),
|
|
|
|
ConnectionString: "http://" + s.httpAddr,
|
|
|
|
})
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
glog.V(0).Infoln(err)
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
} else {
|
2014-10-27 16:09:45 +08:00
|
|
|
glog.V(0).Infoln("Old conf,log,snapshot should have been removed.")
|
2013-12-10 05:34:05 +08:00
|
|
|
}
|
|
|
|
|
|
|
|
return s
|
|
|
|
}
|
|
|
|
|
2014-02-05 16:25:23 +08:00
|
|
|
func (s *RaftServer) Peers() (members []string) {
|
2013-12-10 05:34:05 +08:00
|
|
|
peers := s.raftServer.Peers()
|
|
|
|
|
|
|
|
for _, p := range peers {
|
|
|
|
members = append(members, strings.TrimPrefix(p.ConnectionString, "http://"))
|
|
|
|
}
|
|
|
|
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
// Join joins an existing cluster.
|
|
|
|
func (s *RaftServer) Join(peers []string) error {
|
|
|
|
command := &raft.DefaultJoinCommand{
|
|
|
|
Name: s.raftServer.Name(),
|
|
|
|
ConnectionString: "http://" + s.httpAddr,
|
|
|
|
}
|
|
|
|
|
2014-04-17 14:43:27 +08:00
|
|
|
var err error
|
2013-12-10 05:34:05 +08:00
|
|
|
var b bytes.Buffer
|
|
|
|
json.NewEncoder(&b).Encode(command)
|
|
|
|
for _, m := range peers {
|
2014-04-17 14:43:27 +08:00
|
|
|
if m == s.httpAddr {
|
|
|
|
continue
|
|
|
|
}
|
2014-02-05 16:25:23 +08:00
|
|
|
target := fmt.Sprintf("http://%s/cluster/join", strings.TrimSpace(m))
|
|
|
|
glog.V(0).Infoln("Attempting to connect to:", target)
|
2013-12-10 05:34:05 +08:00
|
|
|
|
2014-04-17 14:43:27 +08:00
|
|
|
err = postFollowingOneRedirect(target, "application/json", &b)
|
2013-12-10 05:34:05 +08:00
|
|
|
|
|
|
|
if err != nil {
|
2014-03-12 21:23:07 +08:00
|
|
|
glog.V(0).Infoln("Post returned error: ", err.Error())
|
2013-12-10 05:34:05 +08:00
|
|
|
if _, ok := err.(*url.Error); ok {
|
|
|
|
// If we receive a network error try the next member
|
|
|
|
continue
|
|
|
|
}
|
2014-04-17 14:43:27 +08:00
|
|
|
} else {
|
|
|
|
return nil
|
2013-12-10 05:34:05 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return errors.New("Could not connect to any cluster peers")
|
|
|
|
}
|
2014-02-05 16:25:23 +08:00
|
|
|
|
|
|
|
// a workaround because http POST following redirection misses request body
|
|
|
|
func postFollowingOneRedirect(target string, contentType string, b *bytes.Buffer) error {
|
|
|
|
backupReader := bytes.NewReader(b.Bytes())
|
|
|
|
resp, err := http.Post(target, contentType, b)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
defer resp.Body.Close()
|
|
|
|
reply, _ := ioutil.ReadAll(resp.Body)
|
|
|
|
statusCode := resp.StatusCode
|
|
|
|
|
|
|
|
if statusCode == http.StatusMovedPermanently {
|
|
|
|
var urlStr string
|
|
|
|
if urlStr = resp.Header.Get("Location"); urlStr == "" {
|
|
|
|
return errors.New(fmt.Sprintf("%d response missing Location header", resp.StatusCode))
|
|
|
|
}
|
|
|
|
|
|
|
|
glog.V(0).Infoln("Post redirected to ", urlStr)
|
|
|
|
resp2, err2 := http.Post(urlStr, contentType, backupReader)
|
|
|
|
if err2 != nil {
|
|
|
|
return err2
|
|
|
|
}
|
|
|
|
defer resp2.Body.Close()
|
|
|
|
reply, _ = ioutil.ReadAll(resp2.Body)
|
|
|
|
statusCode = resp2.StatusCode
|
|
|
|
}
|
|
|
|
|
|
|
|
glog.V(0).Infoln("Post returned status: ", statusCode, string(reply))
|
|
|
|
if statusCode != http.StatusOK {
|
|
|
|
return errors.New(string(reply))
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|