seaweedfs/weed/server/raft_server.go

218 lines
5.4 KiB
Go
Raw Normal View History

package weed_server
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"io/ioutil"
"math/rand"
"net/http"
"net/url"
"os"
"path"
"reflect"
"sort"
"strings"
"time"
2015-05-03 19:37:49 +00:00
"github.com/chrislusf/raft"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/topology"
"github.com/gorilla/mux"
)
type RaftServer struct {
peers []string // initial peers to join with
raftServer raft.Server
dataDir string
httpAddr string
router *mux.Router
topo *topology.Topology
}
func NewRaftServer(r *mux.Router, peers []string, httpAddr string, dataDir string, topo *topology.Topology, pulseSeconds int) *RaftServer {
s := &RaftServer{
peers: peers,
httpAddr: httpAddr,
dataDir: dataDir,
router: r,
topo: topo,
}
2014-02-04 09:11:28 +00:00
if glog.V(4) {
raft.SetLogLevel(2)
}
raft.RegisterCommand(&topology.MaxVolumeIdCommand{})
var err error
2014-04-16 00:56:47 +00:00
transporter := raft.NewHTTPTransporter("/cluster", 0)
transporter.Transport.MaxIdleConnsPerHost = 1024
2014-10-20 03:03:00 +00:00
glog.V(1).Infof("Starting RaftServer with IP:%v:", httpAddr)
// Clear old cluster configurations if peers are changed
if oldPeers, changed := isPeersChanged(s.dataDir, httpAddr, s.peers); changed {
glog.V(0).Infof("Peers Change: %v => %v", oldPeers, s.peers)
os.RemoveAll(path.Join(s.dataDir, "conf"))
os.RemoveAll(path.Join(s.dataDir, "log"))
os.RemoveAll(path.Join(s.dataDir, "snapshot"))
}
s.raftServer, err = raft.NewServer(s.httpAddr, s.dataDir, transporter, nil, topo, "")
if err != nil {
glog.V(0).Infoln(err)
return nil
}
transporter.Install(s.raftServer, s)
2017-01-18 17:54:43 +00:00
s.raftServer.SetHeartbeatInterval(500 * time.Millisecond)
s.raftServer.SetElectionTimeout(time.Duration(pulseSeconds) * 500 * time.Millisecond)
s.raftServer.Start()
s.router.HandleFunc("/cluster/join", s.joinHandler).Methods("POST")
2014-02-04 09:11:28 +00:00
s.router.HandleFunc("/cluster/status", s.statusHandler).Methods("GET")
if len(s.peers) > 0 {
// Join to leader if specified.
for {
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)
} else {
break
}
} else {
break
}
}
} else if s.raftServer.IsLogEmpty() {
// Initialize the server by joining itself.
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 {
glog.V(0).Infoln("Old conf,log,snapshot should have been removed.")
}
return s
}
func (s *RaftServer) Peers() (members []string) {
peers := s.raftServer.Peers()
for _, p := range peers {
members = append(members, strings.TrimPrefix(p.ConnectionString, "http://"))
}
return
}
func isPeersChanged(dir string, self string, peers []string) (oldPeers []string, changed bool) {
confPath := path.Join(dir, "conf")
// open conf file
b, err := ioutil.ReadFile(confPath)
if err != nil {
return oldPeers, true
}
conf := &raft.Config{}
if err = json.Unmarshal(b, conf); err != nil {
return oldPeers, true
}
for _, p := range conf.Peers {
oldPeers = append(oldPeers, strings.TrimPrefix(p.ConnectionString, "http://"))
}
oldPeers = append(oldPeers, self)
sort.Strings(peers)
sort.Strings(oldPeers)
2016-11-03 07:46:59 +00:00
return oldPeers, !reflect.DeepEqual(peers, oldPeers)
}
// Join joins an existing cluster.
func (s *RaftServer) Join(peers []string) error {
command := &raft.DefaultJoinCommand{
Name: s.raftServer.Name(),
ConnectionString: "http://" + s.httpAddr,
}
var err error
var b bytes.Buffer
json.NewEncoder(&b).Encode(command)
for _, m := range peers {
if m == s.httpAddr {
continue
}
target := fmt.Sprintf("http://%s/cluster/join", strings.TrimSpace(m))
glog.V(0).Infoln("Attempting to connect to:", target)
2016-11-08 03:00:46 +00:00
err = postFollowingOneRedirect(target, "application/json", b)
if err != nil {
2014-03-12 13:23:07 +00:00
glog.V(0).Infoln("Post returned error: ", err.Error())
if _, ok := err.(*url.Error); ok {
// If we receive a network error try the next member
continue
}
} else {
return nil
}
}
return errors.New("Could not connect to any cluster peers")
}
// a workaround because http POST following redirection misses request body
2016-11-08 03:00:46 +00:00
func postFollowingOneRedirect(target string, contentType string, b bytes.Buffer) error {
backupReader := bytes.NewReader(b.Bytes())
2016-11-08 03:00:46 +00:00
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 == "" {
2015-03-10 07:20:31 +00:00
return fmt.Errorf("%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
}