seaweedfs/weed/server/raft_server.go

201 lines
5 KiB
Go
Raw Normal View History

package weed_server
import (
"encoding/json"
2022-04-04 08:50:56 +00:00
transport "github.com/Jille/raft-grpc-transport"
2023-05-07 20:33:44 +00:00
"github.com/seaweedfs/raft/protobuf"
2022-04-04 08:50:56 +00:00
"io"
2020-10-24 06:06:44 +00:00
"math/rand"
"os"
"path"
"time"
2020-03-04 08:39:47 +00:00
"google.golang.org/grpc"
"github.com/seaweedfs/seaweedfs/weed/pb"
2020-03-04 08:39:47 +00:00
2022-04-04 08:50:56 +00:00
hashicorpRaft "github.com/hashicorp/raft"
"github.com/seaweedfs/raft"
2020-03-04 08:39:47 +00:00
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/topology"
)
type RaftServerOption struct {
GrpcDialOption grpc.DialOption
2022-03-26 20:13:19 +00:00
Peers map[string]pb.ServerAddress
ServerAddr pb.ServerAddress
DataDir string
Topo *topology.Topology
RaftResumeState bool
HeartbeatInterval time.Duration
ElectionTimeout time.Duration
2022-04-04 08:50:56 +00:00
RaftBootstrap bool
}
type RaftServer struct {
2023-05-07 20:33:44 +00:00
protobuf.UnimplementedRaftServer
raftGrpcServer *raft.GrpcServer
2022-04-04 08:50:56 +00:00
peers map[string]pb.ServerAddress // initial peers to join with
raftServer raft.Server
2022-04-04 12:51:51 +00:00
RaftHashicorp *hashicorpRaft.Raft
2022-04-04 08:50:56 +00:00
TransportManager *transport.Manager
dataDir string
serverAddr pb.ServerAddress
topo *topology.Topology
}
2020-10-02 18:01:20 +00:00
type StateMachine struct {
raft.StateMachine
topo *topology.Topology
}
2022-04-04 08:50:56 +00:00
var _ hashicorpRaft.FSM = &StateMachine{}
2020-10-02 18:01:20 +00:00
func (s StateMachine) Save() ([]byte, error) {
state := topology.MaxVolumeIdCommand{
MaxVolumeId: s.topo.GetMaxVolumeId(),
}
glog.V(1).Infof("Save raft state %+v", state)
return json.Marshal(state)
}
func (s StateMachine) Recovery(data []byte) error {
state := topology.MaxVolumeIdCommand{}
err := json.Unmarshal(data, &state)
if err != nil {
return err
}
glog.V(1).Infof("Recovery raft state %+v", state)
s.topo.UpAdjustMaxVolumeId(state.MaxVolumeId)
return nil
}
2022-04-04 08:50:56 +00:00
func (s *StateMachine) Apply(l *hashicorpRaft.Log) interface{} {
before := s.topo.GetMaxVolumeId()
state := topology.MaxVolumeIdCommand{}
err := json.Unmarshal(l.Data, &state)
if err != nil {
return err
}
s.topo.UpAdjustMaxVolumeId(state.MaxVolumeId)
glog.V(1).Infoln("max volume id", before, "==>", s.topo.GetMaxVolumeId())
return nil
}
func (s *StateMachine) Snapshot() (hashicorpRaft.FSMSnapshot, error) {
return &topology.MaxVolumeIdCommand{
MaxVolumeId: s.topo.GetMaxVolumeId(),
}, nil
}
func (s *StateMachine) Restore(r io.ReadCloser) error {
b, err := io.ReadAll(r)
2022-04-04 08:50:56 +00:00
if err != nil {
return err
}
if err := s.Recovery(b); err != nil {
return err
}
return nil
}
func NewRaftServer(option *RaftServerOption) (*RaftServer, error) {
s := &RaftServer{
2022-03-26 20:13:19 +00:00
peers: option.Peers,
serverAddr: option.ServerAddr,
dataDir: option.DataDir,
topo: option.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(option.GrpcDialOption)
glog.V(0).Infof("Starting RaftServer with %v", option.ServerAddr)
// always clear previous log to avoid server is promotable
os.RemoveAll(path.Join(s.dataDir, "log"))
if !option.RaftResumeState {
2020-10-02 18:01:20 +00:00
// always clear previous metadata
os.RemoveAll(path.Join(s.dataDir, "conf"))
os.RemoveAll(path.Join(s.dataDir, "snapshot"))
}
if err := os.MkdirAll(path.Join(s.dataDir, "snapshot"), os.ModePerm); err != nil {
2020-10-02 18:01:20 +00:00
return nil, err
}
stateMachine := StateMachine{topo: option.Topo}
s.raftServer, err = raft.NewServer(string(s.serverAddr), s.dataDir, transporter, stateMachine, option.Topo, "")
if err != nil {
glog.V(0).Infoln(err)
2020-10-02 18:01:20 +00:00
return nil, err
}
heartbeatInterval := time.Duration(float64(option.HeartbeatInterval) * (rand.Float64()*0.25 + 1))
s.raftServer.SetHeartbeatInterval(heartbeatInterval)
s.raftServer.SetElectionTimeout(option.ElectionTimeout)
2020-10-02 18:01:20 +00:00
if err := s.raftServer.LoadSnapshot(); err != nil {
return nil, err
}
if err := s.raftServer.Start(); err != nil {
return nil, err
}
2022-03-26 17:21:26 +00:00
for name, peer := range s.peers {
if err := s.raftServer.AddPeer(name, peer.ToGrpcAddress()); err != nil {
2020-10-02 18:01:20 +00:00
return nil, err
}
}
2020-10-02 18:01:20 +00:00
// Remove deleted peers
for existsPeerName := range s.raftServer.Peers() {
2022-03-26 17:24:05 +00:00
if existingPeer, found := s.peers[existsPeerName]; !found {
if err := s.raftServer.RemovePeer(existsPeerName); err != nil {
glog.V(0).Infoln(err)
return nil, err
} else {
glog.V(0).Infof("removing old peer: %s", existingPeer)
}
}
}
2023-05-07 20:33:44 +00:00
s.raftGrpcServer = raft.NewGrpcServer(s.raftServer)
glog.V(0).Infof("current cluster leader: %v", s.raftServer.Leader())
2020-10-02 18:01:20 +00:00
return s, nil
}
func (s *RaftServer) Peers() (members []string) {
2022-04-04 13:52:08 +00:00
if s.raftServer != nil {
peers := s.raftServer.Peers()
for _, p := range peers {
members = append(members, p.Name)
}
} else if s.RaftHashicorp != nil {
cfg := s.RaftHashicorp.GetConfiguration()
for _, p := range cfg.Configuration().Servers {
members = append(members, string(p.ID))
}
}
return
}
func (s *RaftServer) DoJoinCommand() {
glog.V(0).Infoln("Initializing new cluster")
if _, err := s.raftServer.Do(&raft.DefaultJoinCommand{
Name: s.raftServer.Name(),
ConnectionString: s.serverAddr.ToGrpcAddress(),
}); err != nil {
glog.Errorf("fail to send join command: %v", err)
}
}