From cad5b8b28103df5a094146a9c9bf4b6d79cefb50 Mon Sep 17 00:00:00 2001 From: chrislu Date: Thu, 18 Dec 2025 22:28:27 -0800 Subject: [PATCH] master: enable dual-write to both seaweedfs/raft and hashicorp/raft When using seaweedfs/raft (-raftHashicorp=false), also initialize hashicorp raft in parallel. All state changes are written to both raft implementations, keeping hashicorp raft in sync. This enables seamless migration to hashicorp/raft: - No manual steps required for existing deployments - When ready, switch to -raftHashicorp=true and hashicorp raft already has all the state - The old seaweedfs/raft can be removed in a future version Changes: - NewRaftServer now initializes hashicorp raft in a subdirectory - NextVolumeId writes to both raft implementations when both exist - SetRaftServer sets both RaftServer and HashicorpRaft in topology - Transport manager for hashicorp raft is registered for gRPC --- weed/command/master.go | 5 +++ weed/server/master_server.go | 5 +++ weed/server/raft_server.go | 82 ++++++++++++++++++++++++++++++++++++ weed/topology/topology.go | 9 ++++ 4 files changed, 101 insertions(+) diff --git a/weed/command/master.go b/weed/command/master.go index 543e5f199..64c4481ba 100644 --- a/weed/command/master.go +++ b/weed/command/master.go @@ -244,6 +244,11 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) { raftServer.TransportManager.Register(grpcS) } else { protobuf.RegisterRaftServer(grpcS, raftServer) + // Also register hashicorp transport for dual-write migration + if raftServer.TransportManager != nil { + raftServer.TransportManager.Register(grpcS) + glog.V(0).Infof("Dual-write: hashicorp raft transport registered") + } } reflection.Register(grpcS) glog.V(0).Infof("Start Seaweed Master %s grpc server at %s:%d", version.Version(), *masterOption.ipBind, grpcPort) diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 10b54d58f..496f226fa 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -212,6 +212,11 @@ func (ms *MasterServer) SetRaftServer(raftServer *RaftServer) { } }) raftServerName = fmt.Sprintf("[%s]", ms.Topo.RaftServer.Name()) + // Also set HashicorpRaft if available for dual-write migration + if raftServer.RaftHashicorp != nil { + ms.Topo.HashicorpRaft = raftServer.RaftHashicorp + glog.V(0).Infof("Dual-write mode: hashicorp raft enabled alongside seaweedfs/raft") + } } else if raftServer.RaftHashicorp != nil { ms.Topo.HashicorpRaft = raftServer.RaftHashicorp raftServerName = ms.Topo.HashicorpRaft.String() diff --git a/weed/server/raft_server.go b/weed/server/raft_server.go index 4d2209dc0..0222b728f 100644 --- a/weed/server/raft_server.go +++ b/weed/server/raft_server.go @@ -2,6 +2,7 @@ package weed_server import ( "encoding/json" + "fmt" "io" "math/rand/v2" "os" @@ -9,6 +10,7 @@ import ( "time" transport "github.com/Jille/raft-grpc-transport" + boltdb "github.com/hashicorp/raft-boltdb/v2" "google.golang.org/grpc" @@ -167,6 +169,13 @@ func NewRaftServer(option *RaftServerOption) (*RaftServer, error) { glog.V(0).Infof("current cluster leader: %v", s.raftServer.Leader()) + // Also initialize hashicorp raft in parallel for seamless future migration + if err := s.initHashicorpRaftForDualWrite(option); err != nil { + glog.Warningf("failed to initialize hashicorp raft for dual-write, migration will require manual steps: %v", err) + } else { + glog.V(0).Infof("hashicorp raft initialized for dual-write migration") + } + return s, nil } @@ -197,3 +206,76 @@ func (s *RaftServer) DoJoinCommand() { } } + +// initHashicorpRaftForDualWrite initializes hashicorp raft alongside seaweedfs/raft +// for seamless future migration. State changes are written to both implementations. +func (s *RaftServer) initHashicorpRaftForDualWrite(option *RaftServerOption) error { + c := hashicorpRaft.DefaultConfig() + c.LocalID = hashicorpRaft.ServerID(s.serverAddr) + c.HeartbeatTimeout = time.Duration(float64(option.HeartbeatInterval) * (rand.Float64()*0.25 + 1)) + c.ElectionTimeout = option.ElectionTimeout + if c.LeaderLeaseTimeout > c.HeartbeatTimeout { + c.LeaderLeaseTimeout = c.HeartbeatTimeout + } + if glog.V(4) { + c.LogLevel = "Debug" + } else if glog.V(2) { + c.LogLevel = "Info" + } else if glog.V(1) { + c.LogLevel = "Warn" + } else if glog.V(0) { + c.LogLevel = "Error" + } + + if err := hashicorpRaft.ValidateConfig(c); err != nil { + return fmt.Errorf("raft.ValidateConfig: %w", err) + } + + // Use a subdirectory to keep hashicorp raft data separate from old raft + hashicorpDataDir := path.Join(s.dataDir, "hashicorp") + if option.RaftBootstrap { + os.RemoveAll(path.Join(hashicorpDataDir, ldbFile)) + os.RemoveAll(path.Join(hashicorpDataDir, sdbFile)) + os.RemoveAll(path.Join(hashicorpDataDir, "snapshots")) + } + if err := os.MkdirAll(path.Join(hashicorpDataDir, "snapshots"), os.ModePerm); err != nil { + return err + } + + ldb, err := boltdb.NewBoltStore(path.Join(hashicorpDataDir, ldbFile)) + if err != nil { + return fmt.Errorf("boltdb.NewBoltStore(%q): %v", path.Join(hashicorpDataDir, ldbFile), err) + } + + sdb, err := boltdb.NewBoltStore(path.Join(hashicorpDataDir, sdbFile)) + if err != nil { + return fmt.Errorf("boltdb.NewBoltStore(%q): %v", path.Join(hashicorpDataDir, sdbFile), err) + } + + fss, err := hashicorpRaft.NewFileSnapshotStore(hashicorpDataDir, 3, os.Stderr) + if err != nil { + return fmt.Errorf("raft.NewFileSnapshotStore(%q, ...): %v", hashicorpDataDir, err) + } + + s.TransportManager = transport.New(hashicorpRaft.ServerAddress(s.serverAddr), []grpc.DialOption{option.GrpcDialOption}) + + stateMachine := StateMachine{topo: option.Topo} + s.RaftHashicorp, err = hashicorpRaft.NewRaft(c, &stateMachine, ldb, sdb, fss, s.TransportManager.Transport()) + if err != nil { + return fmt.Errorf("raft.NewRaft: %w", err) + } + + if option.RaftBootstrap || len(s.RaftHashicorp.GetConfiguration().Configuration().Servers) == 0 { + cfg := s.AddPeersConfiguration() + peerIdx := getPeerIdx(s.serverAddr, s.peers) + timeSleep := time.Duration(float64(c.LeaderLeaseTimeout) * (rand.Float64()*0.25 + 1) * float64(peerIdx)) + glog.V(0).Infof("Dual-write: Bootstrapping hashicorp raft idx: %d sleep: %v cluster: %+v", peerIdx, timeSleep, cfg) + time.Sleep(timeSleep) + f := s.RaftHashicorp.BootstrapCluster(cfg) + if err := f.Error(); err != nil { + return fmt.Errorf("raft.Raft.BootstrapCluster: %w", err) + } + } + + return nil +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index bbae97d72..c79adaa16 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -237,6 +237,15 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) { if _, err := t.RaftServer.Do(NewMaxVolumeIdCommand(next)); err != nil { return 0, err } + // Dual-write to hashicorp raft if available for seamless migration + if t.HashicorpRaft != nil { + b, err := json.Marshal(NewMaxVolumeIdCommand(next)) + if err != nil { + glog.Warningf("failed to marshal MaxVolumeIdCommand for dual-write: %v", err) + } else if future := t.HashicorpRaft.Apply(b, time.Second); future.Error() != nil { + glog.Warningf("failed to apply MaxVolumeIdCommand to hashicorp raft for dual-write: %v", future.Error()) + } + } } else if t.HashicorpRaft != nil { b, err := json.Marshal(NewMaxVolumeIdCommand(next)) if err != nil {