From e790d74ebcf9b68642e20708f8192ee459177236 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 18 Oct 2026 20:49:38 -0700 Subject: [PATCH] master: replicate the growth freeze and gate it on veteran heartbeats (#11763) --- weed/server/master_grpc_server.go | 4 +- weed/server/raft_server.go | 19 ++++++++- weed/topology/cluster_commands.go | 6 +++ weed/topology/growth_freeze_test.go | 43 +++++++++++++++++++++ weed/topology/topology.go | 60 ++++++++++++++++++++++++++--- 5 files changed, 124 insertions(+), 8 deletions(-) diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index 38092211b0..dcba381e6d 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -184,7 +184,9 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey) ms.Topo.UpAdjustMaxVolumeId(needle.VolumeId(heartbeat.MaxVolumeIdSeen)) - ms.Topo.CheckReportedTopologyId(heartbeat.LastSeenTopologyId) + if len(heartbeat.Volumes) > 0 { + ms.Topo.CheckReportedTopologyId(heartbeat.LastSeenTopologyId) + } if dn == nil { // Skip delta heartbeat for volume server versions better than 3.28 https://github.com/seaweedfs/seaweedfs/pull/3630 if heartbeat.Ip == "" { diff --git a/weed/server/raft_server.go b/weed/server/raft_server.go index 8130b7838e..bcf04267cb 100644 --- a/weed/server/raft_server.go +++ b/weed/server/raft_server.go @@ -60,6 +60,10 @@ func (s StateMachine) Save() ([]byte, error) { MaxVolumeId: s.topo.GetMaxVolumeId(), TopologyId: s.topo.GetTopologyId(), } + if s.topo.IsGrowthFrozen() { + state.GrowthFrozen = new(bool) + *state.GrowthFrozen = true + } glog.V(1).Infof("Save raft state %+v", state) return json.Marshal(state) } @@ -79,6 +83,9 @@ func (s StateMachine) Recovery(data []byte) error { if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil { glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err) } + if state.GrowthFrozen != nil { + s.topo.SetGrowthFrozen(*state.GrowthFrozen) + } return nil } @@ -101,16 +108,24 @@ func (s *StateMachine) Apply(l *hashicorpRaft.Log) interface{} { if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil { glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err) } + if state.GrowthFrozen != nil { + s.topo.SetGrowthFrozen(*state.GrowthFrozen) + } glog.V(1).Infoln("max volume id", before, "==>", s.topo.GetMaxVolumeId()) return nil } func (s *StateMachine) Snapshot() (hashicorpRaft.FSMSnapshot, error) { - return &topology.MaxVolumeIdCommand{ + state := &topology.MaxVolumeIdCommand{ MaxVolumeId: s.topo.GetMaxVolumeId(), TopologyId: s.topo.GetTopologyId(), - }, nil + } + if s.topo.IsGrowthFrozen() { + state.GrowthFrozen = new(bool) + *state.GrowthFrozen = true + } + return state, nil } func (s *StateMachine) Restore(r io.ReadCloser) error { diff --git a/weed/topology/cluster_commands.go b/weed/topology/cluster_commands.go index 1c713201e7..0df471f367 100644 --- a/weed/topology/cluster_commands.go +++ b/weed/topology/cluster_commands.go @@ -13,6 +13,9 @@ import ( type MaxVolumeIdCommand struct { MaxVolumeId needle.VolumeId `json:"maxVolumeId"` TopologyId string `json:"topologyId"` + // GrowthFrozen replicates the volume-growth freeze through raft so a + // leader change does not silently lift it. nil means no change. + GrowthFrozen *bool `json:"growthFrozen,omitempty"` } func NewMaxVolumeIdCommand(value needle.VolumeId, topologyId string) *MaxVolumeIdCommand { @@ -46,6 +49,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) } + if c.GrowthFrozen != nil { + topo.SetGrowthFrozen(*c.GrowthFrozen) + } glog.V(1).Infoln("max volume id", before, "==>", topo.GetMaxVolumeId()) return nil, nil diff --git a/weed/topology/growth_freeze_test.go b/weed/topology/growth_freeze_test.go index 2a6d250148..2f35d4a6ac 100644 --- a/weed/topology/growth_freeze_test.go +++ b/weed/topology/growth_freeze_test.go @@ -68,6 +68,49 @@ func TestCheckReportedTopologyIdBeforeTopologyIdKnown(t *testing.T) { } } +func TestGrowthFreezeSurvivesMirrorRestore(t *testing.T) { + dir := t.TempDir() + file := dir + "/max_volume_id" + + frozen := NewTopology("topo", nil, 128, 3, false) + frozen.SetGrowthFreezeConfig(true, 0) + frozen.SetMaxVolumeIdFile(file) + frozen.SetTopologyId("cluster-a") + frozen.CheckReportedTopologyId("cluster-b") + if !frozen.IsGrowthFrozen() { + t.Fatal("expected frozen growth") + } + if err := frozen.PersistMaxVolumeId(42); err != nil { + t.Fatalf("persist: %v", err) + } + + restarted := NewTopology("topo", nil, 128, 3, false) + restarted.SetGrowthFreezeConfig(true, 0) + restarted.SetMaxVolumeIdFile(file) + if !restarted.IsGrowthFrozen() { + t.Fatal("freeze flag should survive a mirror restore") + } + if !restarted.ResumeVolumeGrowth() { + t.Fatal("restored freeze should be resumable") + } +} + +func TestGrowthFreezeCommandApply(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + topo.SetGrowthFreezeConfig(true, 0) + + frozen := true + topo.SetGrowthFrozen(frozen) + if !topo.IsGrowthFrozen() { + t.Fatal("replicated freeze should set the flag") + } + unfrozen := false + topo.SetGrowthFrozen(unfrozen) + if topo.IsGrowthFrozen() { + t.Fatal("replicated unfreeze should clear the flag") + } +} + func TestAllocationDelay(t *testing.T) { topo := NewTopology("topo", nil, 128, 3, false) diff --git a/weed/topology/topology.go b/weed/topology/topology.go index e53b4dea9f..e40c429a61 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -406,6 +406,9 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) { if wait := t.allocationDelay(); wait > 0 { glog.V(0).Infof("deferring volume id allocation for %v", wait) time.Sleep(wait) + if t.IsGrowthFrozen() { + return 0, fmt.Errorf("volume growth is frozen after master state loss; verify the cluster inventory, then run: volume.growth.resume") + } } vid := t.GetMaxVolumeId() next := vid.Next() @@ -980,10 +983,11 @@ func (t *Topology) CheckReportedTopologyId(reported string) { if own == "" || reported == own { return } + t.growthFrozenAt.Store(time.Now().UnixNano()) if t.growthFrozen.CompareAndSwap(false, true) { - t.growthFrozenAt.Store(time.Now().UnixNano()) stats.MasterVolumeGrowthFrozen.Set(1) glog.Errorf("volume growth frozen: a volume server reports topology id %s but this cluster is %s; the master state was rebuilt. Verify the cluster inventory, then run: volume.growth.resume", reported, own) + t.applyGrowthFreeze(true) } } @@ -1004,12 +1008,50 @@ func (t *Topology) IsGrowthFrozen() bool { } func (t *Topology) ResumeVolumeGrowth() bool { + if !t.growthFrozen.CompareAndSwap(true, false) { + return false + } + stats.MasterVolumeGrowthFrozen.Set(0) + glog.V(0).Infof("volume growth resumed by operator") + t.applyGrowthFreeze(false) + return true +} + +// SetGrowthFrozen applies a replicated freeze state on every master. The +// timestamp lands before the flag so a reader can never see frozen with a +// zero time and misjudge the auto-resume window. +func (t *Topology) SetGrowthFrozen(frozen bool) { + if frozen { + t.growthFrozenAt.Store(time.Now().UnixNano()) + if t.growthFrozen.CompareAndSwap(false, true) { + stats.MasterVolumeGrowthFrozen.Set(1) + glog.V(0).Infof("volume growth freeze applied from raft") + } + return + } if t.growthFrozen.CompareAndSwap(true, false) { stats.MasterVolumeGrowthFrozen.Set(0) - glog.V(0).Infof("volume growth resumed by operator") - return true } - return false +} + +// applyGrowthFreeze replicates the freeze flag so followers keep it across a +// leader change. No raft server (single-master test topologies) just sets it. +func (t *Topology) applyGrowthFreeze(frozen bool) { + cmd := NewMaxVolumeIdCommand(t.GetMaxVolumeId(), t.GetTopologyId()) + cmd.GrowthFrozen = &frozen + t.RaftServerAccessLock.RLock() + defer t.RaftServerAccessLock.RUnlock() + if t.RaftServer != nil { + if _, err := t.RaftServer.Do(cmd); err != nil { + glog.Warningf("apply growth freeze %v: %v", frozen, err) + } + } else if t.HashicorpRaft != nil { + if b, err := json.Marshal(cmd); err == nil { + t.HashicorpRaft.Apply(b, time.Second) + } + } else { + t.SetGrowthFrozen(frozen) + } } // DeferAllocation holds off new id allocation for d. Called when the leader @@ -1043,6 +1085,9 @@ func (t *Topology) SetMaxVolumeIdFile(file string) { if cmd.TopologyId != "" { t.SetTopologyId(cmd.TopologyId) } + if cmd.GrowthFrozen != nil { + t.SetGrowthFrozen(*cmd.GrowthFrozen) + } } // PersistMaxVolumeId writes the bound to the mirror file. Called by the @@ -1066,7 +1111,12 @@ func (t *Topology) PersistMaxVolumeId(vid needle.VolumeId) error { t.lastPersistedTopologyId.Store(topoId) return nil } - b, err := json.Marshal(NewMaxVolumeIdCommand(vid, topoId)) + cmd := NewMaxVolumeIdCommand(vid, topoId) + if t.growthFrozen.Load() { + cmd.GrowthFrozen = new(bool) + *cmd.GrowthFrozen = true + } + b, err := json.Marshal(cmd) if err != nil { return err }