seaweedfs/weed/server/raft_server.go

137 lines
3.1 KiB
Go
Raw Normal View History

package weed_server
import (
"encoding/json"
"io/ioutil"
"os"
"path"
"reflect"
"sort"
"time"
2020-03-04 08:39:47 +00:00
"google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/pb"
2015-05-03 19:37:49 +00:00
"github.com/chrislusf/raft"
2020-03-04 08:39:47 +00:00
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/topology"
)
type RaftServer struct {
peers []string // initial peers to join with
raftServer raft.Server
dataDir string
serverAddr string
topo *topology.Topology
*raft.GrpcServer
}
func NewRaftServer(grpcDialOption grpc.DialOption, peers []string, serverAddr, dataDir string, topo *topology.Topology, pulseSeconds int) *RaftServer {
s := &RaftServer{
peers: peers,
serverAddr: serverAddr,
dataDir: dataDir,
topo: topo,
}
2014-02-04 09:11:28 +00:00
if glog.V(4) {
raft.SetLogLevel(2)
}
raft.RegisterCommand(&topology.MaxVolumeIdCommand{})
var err error
transporter := raft.NewGrpcTransporter(grpcDialOption)
glog.V(0).Infof("Starting RaftServer with %v", serverAddr)
// Clear old cluster configurations if peers are changed
if oldPeers, changed := isPeersChanged(s.dataDir, serverAddr, 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.serverAddr, s.dataDir, transporter, nil, topo, "")
if err != nil {
glog.V(0).Infoln(err)
return nil
}
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()
for _, peer := range s.peers {
2020-03-04 08:39:47 +00:00
s.raftServer.AddPeer(peer, pb.ServerToGrpcAddress(peer))
}
s.GrpcServer = raft.NewGrpcServer(s.raftServer)
if s.raftServer.IsLogEmpty() && isTheFirstOne(serverAddr, s.peers) {
// Initialize the server by joining itself.
glog.V(0).Infoln("Initializing new cluster")
_, err := s.raftServer.Do(&raft.DefaultJoinCommand{
Name: s.raftServer.Name(),
2020-03-04 08:39:47 +00:00
ConnectionString: pb.ServerToGrpcAddress(s.serverAddr),
})
if err != nil {
glog.V(0).Infoln(err)
return nil
}
}
glog.V(0).Infof("current cluster leader: %v", s.raftServer.Leader())
return s
}
func (s *RaftServer) Peers() (members []string) {
peers := s.raftServer.Peers()
for _, p := range peers {
members = append(members, p.Name)
}
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, p.Name)
}
oldPeers = append(oldPeers, self)
if len(peers) == 0 && len(oldPeers) <= 1 {
return oldPeers, false
}
sort.Strings(peers)
sort.Strings(oldPeers)
2016-11-03 07:46:59 +00:00
return oldPeers, !reflect.DeepEqual(peers, oldPeers)
}
func isTheFirstOne(self string, peers []string) bool {
sort.Strings(peers)
2019-02-15 08:09:48 +00:00
if len(peers) <= 0 {
return true
}
return self == peers[0]
}