diff --git a/weed/server/raft_hashicorp.go b/weed/server/raft_hashicorp.go index f64c0d26d7..9d1d601920 100644 --- a/weed/server/raft_hashicorp.go +++ b/weed/server/raft_hashicorp.go @@ -149,6 +149,7 @@ func NewHashicorpRaftServer(option *RaftServerOption) (*RaftServer, error) { if err := os.MkdirAll(path.Join(s.dataDir, "snapshots"), os.ModePerm); err != nil { return nil, err } + option.Topo.SetMaxVolumeIdFile(filepath.Join(s.dataDir, "max_volume_id")) baseDir := s.dataDir ldb, err := boltdb.NewBoltStore(filepath.Join(baseDir, ldbFile)) diff --git a/weed/server/raft_server.go b/weed/server/raft_server.go index 55d92254c1..8130b7838e 100644 --- a/weed/server/raft_server.go +++ b/weed/server/raft_server.go @@ -76,6 +76,9 @@ func (s StateMachine) Recovery(data []byte) error { s.topo.SetTopologyId(state.TopologyId) glog.V(0).Infof("Recovered TopologyId: %s", state.TopologyId) } + if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil { + glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err) + } return nil } @@ -95,6 +98,9 @@ func (s *StateMachine) Apply(l *hashicorpRaft.Log) interface{} { glog.V(0).Infof("Set TopologyId from raft log: %s", state.TopologyId) } } + if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil { + glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err) + } glog.V(1).Infoln("max volume id", before, "==>", s.topo.GetMaxVolumeId()) return nil @@ -163,6 +169,7 @@ func NewRaftServer(option *RaftServerOption) (*RaftServer, error) { if err := os.MkdirAll(path.Join(s.dataDir, "snapshot"), os.ModePerm); err != nil { return nil, err } + option.Topo.SetMaxVolumeIdFile(path.Join(s.dataDir, "max_volume_id")) stateMachine := StateMachine{topo: option.Topo} s.raftServer, err = raft.NewServer(string(s.serverAddr), s.dataDir, transporter, stateMachine, option.Topo, s.serverAddr.ToGrpcAddress()) diff --git a/weed/topology/cluster_commands.go b/weed/topology/cluster_commands.go index dfa3bf6cc5..1c713201e7 100644 --- a/weed/topology/cluster_commands.go +++ b/weed/topology/cluster_commands.go @@ -43,6 +43,9 @@ func (c *MaxVolumeIdCommand) Apply(server raft.Server) (interface{}, error) { } } } + if err := topo.PersistMaxVolumeId(c.MaxVolumeId); err != nil { + glog.Warningf("persist max volume id %d: %v", c.MaxVolumeId, err) + } glog.V(1).Infoln("max volume id", before, "==>", topo.GetMaxVolumeId()) return nil, nil diff --git a/weed/topology/max_volume_id_file_test.go b/weed/topology/max_volume_id_file_test.go new file mode 100644 index 0000000000..bdeff5c5aa --- /dev/null +++ b/weed/topology/max_volume_id_file_test.go @@ -0,0 +1,104 @@ +package topology + +import ( + "os" + "path/filepath" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +func TestMaxVolumeIdFileRestore(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "max_volume_id") + + topo := NewTopology("topo", nil, 0, 5, false) + topo.SetMaxVolumeIdFile(file) + if err := topo.PersistMaxVolumeId(needle.VolumeId(41)); err != nil { + t.Fatalf("persist: %v", err) + } + + restarted := NewTopology("topo", nil, 0, 5, false) + restarted.SetMaxVolumeIdFile(file) + if got := restarted.GetMaxVolumeId(); got != needle.VolumeId(41) { + t.Fatalf("restored max volume id = %d, want 41", got) + } +} + +func TestMaxVolumeIdFileRestoresTopologyId(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "max_volume_id") + + topo := NewTopology("topo", nil, 0, 5, false) + topo.SetTopologyId("cluster-a") + topo.SetMaxVolumeIdFile(file) + if err := topo.PersistMaxVolumeId(needle.VolumeId(7)); err != nil { + t.Fatalf("persist: %v", err) + } + + restarted := NewTopology("topo", nil, 0, 5, false) + restarted.SetMaxVolumeIdFile(file) + if got := restarted.GetTopologyId(); got != "cluster-a" { + t.Fatalf("restored topology id = %q, want cluster-a", got) + } +} + +func TestMaxVolumeIdFileRewritesOnTopologyIdChange(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "max_volume_id") + + topo := NewTopology("topo", nil, 0, 5, false) + topo.SetMaxVolumeIdFile(file) + if err := topo.PersistMaxVolumeId(needle.VolumeId(7)); err != nil { + t.Fatalf("persist: %v", err) + } + topo.SetTopologyId("cluster-b") + if err := topo.PersistMaxVolumeId(needle.VolumeId(7)); err != nil { + t.Fatalf("persist same bound with new topology id: %v", err) + } + + restarted := NewTopology("topo", nil, 0, 5, false) + restarted.SetMaxVolumeIdFile(file) + if got := restarted.GetTopologyId(); got != "cluster-b" { + t.Fatalf("restored topology id = %q, want cluster-b", got) + } +} + +func TestMaxVolumeIdFileMonotonic(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "max_volume_id") + + topo := NewTopology("topo", nil, 0, 5, false) + topo.SetMaxVolumeIdFile(file) + if err := topo.PersistMaxVolumeId(needle.VolumeId(10)); err != nil { + t.Fatalf("persist: %v", err) + } + if err := topo.PersistMaxVolumeId(needle.VolumeId(5)); err != nil { + t.Fatalf("persist lower: %v", err) + } + if got := topo.lastPersistedMaxVolumeId.Load(); got != 10 { + t.Fatalf("lastPersisted = %d, want 10", got) + } + + data, err := os.ReadFile(file) + if err != nil { + t.Fatalf("read file: %v", err) + } + if len(data) == 0 { + t.Fatal("mirror file is empty") + } +} + +func TestMaxVolumeIdFileCorrupt(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "max_volume_id") + if err := os.WriteFile(file, []byte("{not json"), 0644); err != nil { + t.Fatal(err) + } + + topo := NewTopology("topo", nil, 0, 5, false) + topo.SetMaxVolumeIdFile(file) + if got := topo.GetMaxVolumeId(); got != 0 { + t.Fatalf("corrupt file seeded bound %d, want 0", got) + } +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index a892a0bad4..cd4d92ab37 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -6,6 +6,8 @@ import ( "fmt" "math" "math/rand/v2" + "os" + "path/filepath" "slices" "sync" "sync/atomic" @@ -81,6 +83,11 @@ type Topology struct { // the same node whether or not the grpc port suffix is present. dataNodeIndex map[string]*DataNode dataNodeIndexLock sync.RWMutex + + maxVolumeIdFile string + maxVolumeIdFileLock sync.Mutex + lastPersistedMaxVolumeId atomic.Uint32 + lastPersistedTopologyId atomic.Value } func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology { @@ -390,6 +397,10 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) { vid := t.GetMaxVolumeId() next := vid.Next() + if err := t.PersistMaxVolumeId(next); err != nil { + return 0, fmt.Errorf("persist max volume id %d: %w", next, err) + } + t.RaftServerAccessLock.RLock() defer t.RaftServerAccessLock.RUnlock() @@ -938,3 +949,77 @@ func (t *Topology) SetTopologyId(topologyId string) { glog.Fatalf("Split-brain detected! Current TopologyId is %s, but received %s. Stopping to prevent data corruption.", t.topologyId, topologyId) } } + +// SetMaxVolumeIdFile mirrors the allocation bound to a file outside raft, so +// a master whose raft state is gone still restores a safe boundary. An +// existing file seeds both the bound and the topology id. +func (t *Topology) SetMaxVolumeIdFile(file string) { + t.maxVolumeIdFile = file + data, err := os.ReadFile(file) + if err != nil { + return + } + cmd := &MaxVolumeIdCommand{} + if err := json.Unmarshal(data, cmd); err != nil { + glog.Warningf("corrupt max volume id file %s: %v", file, err) + return + } + t.lastPersistedMaxVolumeId.Store(uint32(cmd.MaxVolumeId)) + t.lastPersistedTopologyId.Store(cmd.TopologyId) + t.UpAdjustMaxVolumeId(cmd.MaxVolumeId) + if cmd.TopologyId != "" { + t.SetTopologyId(cmd.TopologyId) + } +} + +// PersistMaxVolumeId writes the bound to the mirror file. Called by the +// leader before a raft apply and by both raft implementations on apply and +// recovery, so follower mirrors stay current too. +func (t *Topology) PersistMaxVolumeId(vid needle.VolumeId) error { + topoId := t.GetTopologyId() + mirrored := func() bool { + return t.lastPersistedMaxVolumeId.Load() >= uint32(vid) && t.lastPersistedTopologyId.Load() == topoId + } + if mirrored() { + return nil + } + t.maxVolumeIdFileLock.Lock() + defer t.maxVolumeIdFileLock.Unlock() + if mirrored() { + return nil + } + if t.maxVolumeIdFile == "" { + t.lastPersistedMaxVolumeId.Store(uint32(vid)) + t.lastPersistedTopologyId.Store(topoId) + return nil + } + b, err := json.Marshal(NewMaxVolumeIdCommand(vid, topoId)) + if err != nil { + return err + } + tmp := t.maxVolumeIdFile + ".tmp" + f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + return err + } + if _, err := f.Write(b); err != nil { + f.Close() + return err + } + if err := f.Sync(); err != nil { + f.Close() + return err + } + if err := f.Close(); err != nil { + return err + } + if err := os.Rename(tmp, t.maxVolumeIdFile); err != nil { + return err + } + if err := util.FsyncDir(filepath.Dir(t.maxVolumeIdFile)); err != nil { + return err + } + t.lastPersistedMaxVolumeId.Store(uint32(vid)) + t.lastPersistedTopologyId.Store(topoId) + return nil +}