mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-29 16:30:44 +01:00
master: replicate the recovery-wait growth unfreeze through raft (#12165)
* master: replicate the recovery-wait growth unfreeze through raft state_loss_recovery_wait lifted the volume-growth freeze only in memory, while the freeze itself was replicated. A restart replayed the frozen flag and growth stayed refused until the wait expired again, on every restart, unless an operator ran volume.growth.resume. The lift now applies GrowthFrozen=false like ResumeVolumeGrowth, from a goroutine since IsGrowthFrozen is a read path that cannot block on raft. The apply runs under reportedTopologyIdsLock so a freeze landing after the lift keeps its command last in the log; the stale unfreeze skips itself when the flag is frozen again. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * master: cover raft replication of the recovery-wait unfreeze Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * master: retry and re-arm a growth unfreeze raft failed to record applyGrowthFreeze swallowed the raft error, so a rejected lift during a leader change left the flag clear while the log still held the freeze; nothing retried and a restart re-froze growth anyway. The apply now returns the error, replicateGrowthUnfreeze retries it, and on persistent failure re-arms the flag so the next IsGrowthFrozen attempts the lift again instead of diverging from the log. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * master: release the topology-id lock before the freeze raft write CheckReportedTopologyId and SetTopologyInitPending held reportedTopologyIdsLock across the blocking raft apply, so a slow apply stalled every later report — including the heartbeat that should trigger the next freeze. The lock now covers only the mismatch decision; the replicate step runs after it is released. * master: bind the recovery unfreeze to the freeze instant it lifts An unfreeze submitted while a newer freeze sat committed but not yet applied — e.g. after this node won an election — could land behind that freeze in the log and erase it. The command now carries the freeze instant cluster-wide (GrowthFrozenAt) and the lift is bound to the instant its submitter observed (GrowthUnfreezeFor): the state machine no-ops a lift while a strictly newer freeze exists, which also lets the unfreeze run without holding the topology-id lock at all. Concurrent freeze writes keep their own mutex so log order cannot regress the instant. * master: a superseded unfreeze retry leaves the resumed flag alone replicateGrowthUnfreeze broke out of its retry loop when a newer\nfreeze superseded the lift, but the break fell through to the\nre-arm: an operator resume in between was silently undone. Return\ninstead — the newer freeze generation owns the flag. * master: replayed freeze entries without an instant stay ageless Freeze commands written before the instant field existed carry zero.\nSubstituting the replay time made each restart stamp the freeze as\nnew, so a committed bound unfreeze always looked older and was\nrejected — the freeze came back on every restart. Keep the zero\ninstant: the recovery window reads as elapsed and bound lifts apply. * master: pack the growth freeze flag and generation into one atomic Separate flag and timestamp atomics left three holes: a resume's\nload-and-clear could lift a generation stamped in between, a bound\nlift compared stamps numerically although masters' clocks need not\nbe monotonic, and an ageless generation had no identity a bound lift\ncould match without colliding with the unconditional encoding.\n\nOne value now carries the state — 0 unfrozen, -1 an ageless freeze,\nelse the freeze instant — so every transition is a single atomic\nswap or compare-and-swap bound to an exact generation. * master: a repeat mismatch report keeps the armed freeze generation Re-stamping while already frozen never reaches the raft log, so the local generation diverges from the replicated one and every bound unfreeze becomes a no-op on replay — the freeze returns after restart or failover. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * master: snapshots read the freeze flag and generation in one load Save, Snapshot, and PersistMaxVolumeId read IsGrowthFrozen and the generation separately, so a resume landing between them persisted frozen=true with stamp 0 — restoring an ageless freeze a committed bound lift cannot clear. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
1 parent
b62bbb99fc
commit
eec2452e30
4 files changed
+367
-56
No files matched your search
@@ -60,9 +60,10 @@ func (s StateMachine) Save() ([]byte, error) {
|
||||
MaxVolumeId: s.topo.GetMaxVolumeId(),
|
||||
TopologyId: s.topo.GetTopologyId(),
|
||||
}
|
||||
if s.topo.IsGrowthFrozen() {
|
||||
if frozen, frozenAt := s.topo.GrowthFreezeState(); frozen {
|
||||
state.GrowthFrozen = new(bool)
|
||||
*state.GrowthFrozen = true
|
||||
state.GrowthFrozenAt = frozenAt
|
||||
}
|
||||
glog.V(1).Infof("Save raft state %+v", state)
|
||||
return json.Marshal(state)
|
||||
@@ -81,7 +82,7 @@ func (s StateMachine) Recovery(data []byte) error {
|
||||
glog.V(0).Infof("Recovered TopologyId: %s", state.TopologyId)
|
||||
}
|
||||
if state.GrowthFrozen != nil {
|
||||
s.topo.SetGrowthFrozen(*state.GrowthFrozen)
|
||||
s.topo.SetGrowthFrozen(*state.GrowthFrozen, state.GrowthFrozenAt, state.GrowthUnfreezeFor)
|
||||
}
|
||||
if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil {
|
||||
glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err)
|
||||
@@ -106,7 +107,7 @@ func (s *StateMachine) Apply(l *hashicorpRaft.Log) interface{} {
|
||||
}
|
||||
}
|
||||
if state.GrowthFrozen != nil {
|
||||
s.topo.SetGrowthFrozen(*state.GrowthFrozen)
|
||||
s.topo.SetGrowthFrozen(*state.GrowthFrozen, state.GrowthFrozenAt, state.GrowthUnfreezeFor)
|
||||
}
|
||||
if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil {
|
||||
glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err)
|
||||
@@ -117,11 +118,12 @@ func (s *StateMachine) Apply(l *hashicorpRaft.Log) interface{} {
|
||||
}
|
||||
|
||||
func (s *StateMachine) Snapshot() (hashicorpRaft.FSMSnapshot, error) {
|
||||
frozen := s.topo.IsGrowthFrozen()
|
||||
frozen, frozenAt := s.topo.GrowthFreezeState()
|
||||
state := &topology.MaxVolumeIdCommand{
|
||||
MaxVolumeId: s.topo.GetMaxVolumeId(),
|
||||
TopologyId: s.topo.GetTopologyId(),
|
||||
GrowthFrozen: &frozen,
|
||||
MaxVolumeId: s.topo.GetMaxVolumeId(),
|
||||
TopologyId: s.topo.GetTopologyId(),
|
||||
GrowthFrozen: &frozen,
|
||||
GrowthFrozenAt: frozenAt,
|
||||
}
|
||||
return state, nil
|
||||
}
|
||||
|
||||
@@ -16,6 +16,17 @@ type MaxVolumeIdCommand struct {
|
||||
// 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"`
|
||||
// GrowthFrozenAt pins the freeze instant so every master measures the
|
||||
// same recovery window and can tell one freeze generation from the next.
|
||||
// Zero on entries written before the field existed.
|
||||
GrowthFrozenAt int64 `json:"growthFrozenAt,omitempty"`
|
||||
// GrowthUnfreezeFor binds GrowthFrozen=false to the freeze generation
|
||||
// the submitter observed: the lift applies only while that exact
|
||||
// generation is armed, so a stale unfreeze cannot erase a freeze that
|
||||
// reached the log first. Zero lifts unconditionally, for entries
|
||||
// written before the binding existed; -1 binds the ageless generation
|
||||
// left by freeze entries written before instants existed.
|
||||
GrowthUnfreezeFor int64 `json:"growthUnfreezeFor,omitempty"`
|
||||
}
|
||||
|
||||
func NewMaxVolumeIdCommand(value needle.VolumeId, topologyId string) *MaxVolumeIdCommand {
|
||||
@@ -47,7 +58,7 @@ func (c *MaxVolumeIdCommand) Apply(server raft.Server) (interface{}, error) {
|
||||
}
|
||||
}
|
||||
if c.GrowthFrozen != nil {
|
||||
topo.SetGrowthFrozen(*c.GrowthFrozen)
|
||||
topo.SetGrowthFrozen(*c.GrowthFrozen, c.GrowthFrozenAt, c.GrowthUnfreezeFor)
|
||||
}
|
||||
if err := topo.PersistMaxVolumeId(c.MaxVolumeId); err != nil {
|
||||
glog.Warningf("persist max volume id %d: %v", c.MaxVolumeId, err)
|
||||
|
||||
@@ -1,10 +1,36 @@
|
||||
package topology
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/raft"
|
||||
)
|
||||
|
||||
// appliedRaftServer answers as leader and records each command Do applies.
|
||||
type appliedRaftServer struct {
|
||||
raft.Server
|
||||
applied chan raft.Command
|
||||
}
|
||||
|
||||
func (appliedRaftServer) State() string { return raft.Leader }
|
||||
|
||||
func (s appliedRaftServer) Do(c raft.Command) (interface{}, error) {
|
||||
s.applied <- c
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
type failingRaftServer struct {
|
||||
raft.Server
|
||||
}
|
||||
|
||||
func (failingRaftServer) State() string { return raft.Leader }
|
||||
|
||||
func (failingRaftServer) Do(raft.Command) (interface{}, error) {
|
||||
return nil, errors.New("not the leader")
|
||||
}
|
||||
|
||||
func TestGrowthFreezeOnTopologyMismatch(t *testing.T) {
|
||||
topo := NewTopology("topo", nil, 128, 3, false)
|
||||
topo.SetGrowthFreezeConfig(true, 0)
|
||||
@@ -58,6 +84,120 @@ func TestGrowthFreezeAutoRecovery(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestGrowthFreezeAutoRecoveryReplicates(t *testing.T) {
|
||||
applied := make(chan raft.Command, 2)
|
||||
topo := NewTopology("topo", nil, 128, 3, false)
|
||||
topo.RaftServer = appliedRaftServer{applied: applied}
|
||||
topo.SetGrowthFreezeConfig(true, 50*time.Millisecond)
|
||||
topo.SetTopologyId("cluster-a")
|
||||
|
||||
topo.CheckReportedTopologyId("cluster-b")
|
||||
if !topo.IsGrowthFrozen() {
|
||||
t.Fatal("mismatching topology id should freeze growth")
|
||||
}
|
||||
cmd := (<-applied).(*MaxVolumeIdCommand)
|
||||
if cmd.GrowthFrozen == nil || !*cmd.GrowthFrozen {
|
||||
t.Fatal("the freeze should be replicated")
|
||||
}
|
||||
|
||||
time.Sleep(60 * time.Millisecond)
|
||||
if topo.IsGrowthFrozen() {
|
||||
t.Fatal("growth should auto-resume past the recovery wait")
|
||||
}
|
||||
select {
|
||||
case c := <-applied:
|
||||
lift := c.(*MaxVolumeIdCommand)
|
||||
if lift.GrowthFrozen == nil || *lift.GrowthFrozen {
|
||||
t.Fatal("the auto unfreeze should be replicated like volume.growth.resume")
|
||||
}
|
||||
if lift.GrowthUnfreezeFor != cmd.GrowthFrozenAt {
|
||||
t.Fatal("the unfreeze should be bound to the freeze instant it lifts")
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("auto unfreeze was not replicated; a restart would re-apply the freeze")
|
||||
}
|
||||
}
|
||||
|
||||
// A repeat mismatch while already frozen must keep the generation the
|
||||
// replicated freeze carries: re-stamping locally would bind every later
|
||||
// lift to a generation followers never saw, and the freeze would return on
|
||||
// replay.
|
||||
func TestRepeatMismatchKeepsFreezeGeneration(t *testing.T) {
|
||||
topo := NewTopology("topo", nil, 128, 3, false)
|
||||
topo.SetGrowthFreezeConfig(true, 0)
|
||||
topo.SetTopologyId("cluster-a")
|
||||
|
||||
topo.CheckReportedTopologyId("cluster-b")
|
||||
gen := topo.growthFrozenState.Load()
|
||||
if gen == 0 {
|
||||
t.Fatal("mismatching topology id should freeze growth")
|
||||
}
|
||||
topo.CheckReportedTopologyId("cluster-c")
|
||||
if topo.growthFrozenState.Load() != gen {
|
||||
t.Fatal("a repeat mismatch must not re-stamp the armed freeze")
|
||||
}
|
||||
}
|
||||
|
||||
// Snapshots persist the flag and the generation together, so the pair must
|
||||
// come from one read: a lifted freeze reports (false, 0) and an armed one
|
||||
// reports its generation.
|
||||
func TestGrowthFreezeStateConsistentPair(t *testing.T) {
|
||||
topo := NewTopology("topo", nil, 128, 3, false)
|
||||
topo.SetGrowthFreezeConfig(true, 0)
|
||||
|
||||
topo.SetGrowthFrozen(true, time.Now().UnixNano(), 0)
|
||||
armed := topo.growthFrozenState.Load()
|
||||
if frozen, at := topo.GrowthFreezeState(); !frozen || at != armed {
|
||||
t.Fatalf("armed freeze should report (true, %d), got (%v, %d)", armed, frozen, at)
|
||||
}
|
||||
topo.ResumeVolumeGrowth()
|
||||
if frozen, at := topo.GrowthFreezeState(); frozen || at != 0 {
|
||||
t.Fatalf("lifted freeze should report (false, 0), got (%v, %d)", frozen, at)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplicateGrowthUnfreezeSkipsRefreeze(t *testing.T) {
|
||||
applied := make(chan raft.Command, 1)
|
||||
topo := NewTopology("topo", nil, 128, 3, false)
|
||||
topo.RaftServer = appliedRaftServer{applied: applied}
|
||||
|
||||
topo.SetGrowthFrozen(true, 1000, 0)
|
||||
topo.SetGrowthFrozen(true, 2000, 0)
|
||||
topo.replicateGrowthUnfreeze(1000)
|
||||
select {
|
||||
case <-applied:
|
||||
t.Fatal("a stale unfreeze must not lift a freeze that landed after it")
|
||||
default:
|
||||
}
|
||||
if topo.growthFrozenState.Load() != 2000 {
|
||||
t.Fatal("the newer freeze must stay armed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplicateGrowthUnfreezeRearmsOnFailure(t *testing.T) {
|
||||
topo := NewTopology("topo", nil, 128, 3, false)
|
||||
topo.RaftServer = failingRaftServer{}
|
||||
|
||||
topo.SetGrowthFrozen(true, 1000, 0)
|
||||
topo.growthFrozenState.CompareAndSwap(1000, 0) // the local lift
|
||||
topo.replicateGrowthUnfreeze(1000)
|
||||
if topo.growthFrozenState.Load() != 1000 {
|
||||
t.Fatal("a failed replication must re-arm the freeze so a later read retries the lift")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplicateGrowthUnfreezeDoesNotRearmAfterResume(t *testing.T) {
|
||||
topo := NewTopology("topo", nil, 128, 3, false)
|
||||
|
||||
topo.SetGrowthFrozen(true, 1000, 0)
|
||||
topo.SetGrowthFrozen(true, 2000, 0)
|
||||
topo.ResumeVolumeGrowth()
|
||||
topo.replicateGrowthUnfreeze(1000)
|
||||
if topo.growthFrozenState.Load() != 0 {
|
||||
t.Fatal("a superseded lift must not re-arm the freeze an operator resumed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckReportedTopologyIdBeforeTopologyIdKnown(t *testing.T) {
|
||||
topo := NewTopology("topo", nil, 128, 3, false)
|
||||
topo.SetGrowthFreezeConfig(true, 0)
|
||||
@@ -118,17 +258,57 @@ func TestGrowthFreezeCommandApply(t *testing.T) {
|
||||
topo.SetGrowthFreezeConfig(true, 0)
|
||||
|
||||
frozen := true
|
||||
topo.SetGrowthFrozen(frozen)
|
||||
topo.SetGrowthFrozen(frozen, 0, 0)
|
||||
if !topo.IsGrowthFrozen() {
|
||||
t.Fatal("replicated freeze should set the flag")
|
||||
}
|
||||
unfrozen := false
|
||||
topo.SetGrowthFrozen(unfrozen)
|
||||
topo.SetGrowthFrozen(false, 0, 0)
|
||||
if topo.IsGrowthFrozen() {
|
||||
t.Fatal("replicated unfreeze should clear the flag")
|
||||
}
|
||||
}
|
||||
|
||||
// Freeze entries written before the instant field existed replay as their
|
||||
// own stable generation: a lift bound to it still applies, a lift bound to a
|
||||
// different generation cannot clear it, and the recovery window reads as
|
||||
// already elapsed — a replay-time stamp would shadow every committed
|
||||
// unfreeze and re-apply the freeze on each restart.
|
||||
func TestLegacyFreezeReplayStaysAgeless(t *testing.T) {
|
||||
topo := NewTopology("topo", nil, 128, 3, false)
|
||||
topo.SetGrowthFreezeConfig(true, time.Minute)
|
||||
|
||||
topo.SetGrowthFrozen(true, 0, 0)
|
||||
topo.SetGrowthFrozen(false, 0, 1234)
|
||||
if topo.growthFrozenState.Load() != -1 {
|
||||
t.Fatal("a lift bound to a different generation must not clear the ageless freeze")
|
||||
}
|
||||
topo.SetGrowthFrozen(false, 0, -1)
|
||||
if topo.growthFrozenState.Load() != 0 {
|
||||
t.Fatal("a lift bound to the ageless generation should clear it")
|
||||
}
|
||||
|
||||
topo.SetGrowthFrozen(true, 0, 0)
|
||||
if topo.IsGrowthFrozen() {
|
||||
t.Fatal("an ageless freeze should read its recovery window as elapsed")
|
||||
}
|
||||
}
|
||||
|
||||
// A lift bound to a freeze instant must not erase a newer freeze, so a stale
|
||||
// unfreeze cannot overwrite a freeze that committed first but applied later.
|
||||
func TestGrowthUnfreezeBoundToFreezeInstant(t *testing.T) {
|
||||
topo := NewTopology("topo", nil, 128, 3, false)
|
||||
|
||||
topo.SetGrowthFrozen(true, 1000, 0)
|
||||
topo.SetGrowthFrozen(false, 0, 999)
|
||||
if !topo.IsGrowthFrozen() {
|
||||
t.Fatal("a lift for an older freeze instant must not clear a newer freeze")
|
||||
}
|
||||
topo.SetGrowthFrozen(false, 0, 1000)
|
||||
if topo.IsGrowthFrozen() {
|
||||
t.Fatal("a lift bound to the current freeze instant should clear it")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAllocationDelay(t *testing.T) {
|
||||
topo := NewTopology("topo", nil, 128, 3, false)
|
||||
|
||||
|
||||
+163
-45
@@ -92,8 +92,14 @@ type Topology struct {
|
||||
lastPersistedTopologyId atomic.Value
|
||||
lastPersistedGrowthFrozen atomic.Bool
|
||||
|
||||
growthFrozen atomic.Bool
|
||||
growthFrozenAt atomic.Int64
|
||||
// growthFrozenState packs the freeze flag and its generation into one
|
||||
// value: 0 = unfrozen, -1 = frozen without a recorded instant (entries
|
||||
// written before the field existed), >0 = frozen at that instant. Every
|
||||
// transition is a single atomic swap or compare-and-swap, so a lift can
|
||||
// only clear the exact generation it was bound to — a stale unfreeze
|
||||
// cannot erase a newer freeze, a resume cannot lift a generation it did
|
||||
// not observe, and stamps need not be monotonic across masters.
|
||||
growthFrozenState atomic.Int64
|
||||
metaReadOnly atomic.Bool
|
||||
freezeOnMismatch atomic.Bool
|
||||
stateLossRecoveryWait atomic.Int64
|
||||
@@ -122,6 +128,11 @@ type Topology struct {
|
||||
|
||||
reportedTopologyIdsLock sync.Mutex
|
||||
pendingReportedTopologyIds map[string]struct{}
|
||||
// growthFreezeApplyLock serializes replicated freeze writes so the log
|
||||
// order of concurrent mismatch reports cannot land an older freeze
|
||||
// instant after a newer one. Unfreezes skip it: the state machine
|
||||
// applies a lift only against the freeze generation it targeted.
|
||||
growthFreezeApplyLock sync.Mutex
|
||||
|
||||
// quarantinedVolumeIds are (node, vid) pairs force-read-only because the
|
||||
// vid collided across layouts. Kept separate from the reported flag so
|
||||
@@ -1591,15 +1602,22 @@ func (t *Topology) CheckReportedTopologyId(reported string) {
|
||||
return
|
||||
}
|
||||
t.reportedTopologyIdsLock.Lock()
|
||||
defer t.reportedTopologyIdsLock.Unlock()
|
||||
t.checkReportedTopologyIdLocked(reported)
|
||||
freeze := t.checkReportedTopologyIdLocked(reported)
|
||||
t.reportedTopologyIdsLock.Unlock()
|
||||
if freeze {
|
||||
// The decision landed under the lock; the raft wait must not, or a
|
||||
// slow apply would stall every later report — including one that
|
||||
// needs to trigger the next freeze.
|
||||
t.applyGrowthFreeze()
|
||||
}
|
||||
}
|
||||
|
||||
// checkReportedTopologyIdLocked runs the mismatch check inside
|
||||
// reportedTopologyIdsLock, the same critical section the init gate drains
|
||||
// under: a report received before the gate opens is either stashed or frozen
|
||||
// before the gate can release.
|
||||
func (t *Topology) checkReportedTopologyIdLocked(reported string) {
|
||||
// before the gate can release. It returns whether the freeze must be
|
||||
// replicated, which the caller does after releasing the lock.
|
||||
func (t *Topology) checkReportedTopologyIdLocked(reported string) (freeze bool) {
|
||||
own := t.GetTopologyId()
|
||||
if own == "" {
|
||||
// The local topology id is still being minted. Keep the report so a
|
||||
@@ -1609,82 +1627,175 @@ func (t *Topology) checkReportedTopologyIdLocked(reported string) {
|
||||
t.pendingReportedTopologyIds = make(map[string]struct{})
|
||||
}
|
||||
t.pendingReportedTopologyIds[reported] = struct{}{}
|
||||
return
|
||||
return false
|
||||
}
|
||||
if reported == own {
|
||||
return
|
||||
return false
|
||||
}
|
||||
t.growthFrozenAt.Store(time.Now().UnixNano())
|
||||
if t.growthFrozen.CompareAndSwap(false, true) {
|
||||
// Only the first mismatch stamps a generation: a re-stamp never reaches
|
||||
// the log, so bound lifts would target a generation followers never saw.
|
||||
if t.growthFrozenState.CompareAndSwap(0, 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)
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (t *Topology) IsGrowthFrozen() bool {
|
||||
if !t.growthFrozen.Load() {
|
||||
return false
|
||||
frozen, _ := t.GrowthFreezeState()
|
||||
return frozen
|
||||
}
|
||||
|
||||
// GrowthFreezeState returns the freeze flag and its generation from one
|
||||
// load, so snapshots persisting both cannot split the pair across a lift.
|
||||
func (t *Topology) GrowthFreezeState() (frozen bool, frozenAt int64) {
|
||||
state := t.growthFrozenState.Load()
|
||||
if state == 0 {
|
||||
return false, 0
|
||||
}
|
||||
if wait := t.stateLossRecoveryWait.Load(); wait > 0 {
|
||||
if time.Duration(time.Now().UnixNano()-t.growthFrozenAt.Load()) > time.Duration(wait) {
|
||||
if t.growthFrozen.CompareAndSwap(true, false) {
|
||||
// A legacy generation carries no instant; its window counts as
|
||||
// already elapsed.
|
||||
if state < 0 || time.Duration(time.Now().UnixNano()-state) > time.Duration(wait) {
|
||||
if t.growthFrozenState.CompareAndSwap(state, 0) {
|
||||
stats.MasterVolumeGrowthFrozen.Set(0)
|
||||
glog.V(0).Infof("volume growth unfrozen after %v", time.Duration(wait))
|
||||
go t.replicateGrowthUnfreeze(state)
|
||||
return false, 0
|
||||
}
|
||||
return false
|
||||
state = t.growthFrozenState.Load()
|
||||
}
|
||||
}
|
||||
return true
|
||||
return state != 0, state
|
||||
}
|
||||
|
||||
// replicateGrowthUnfreeze pushes the recovery-wait lift through raft like
|
||||
// ResumeVolumeGrowth, so a restart does not replay the freeze. The command
|
||||
// carries the generation it lifts: a newer freeze — even one committed but
|
||||
// not yet applied when this node won an election — makes the apply a no-op
|
||||
// instead of being erased. If the write never lands, the generation is
|
||||
// re-armed so a later IsGrowthFrozen retries the lift.
|
||||
func (t *Topology) replicateGrowthUnfreeze(frozenAt int64) {
|
||||
for attempt := 0; attempt < 3; attempt++ {
|
||||
if attempt > 0 {
|
||||
time.Sleep(time.Duration(attempt) * time.Second)
|
||||
}
|
||||
// The caller's lift already cleared the flag, so unfrozen is the
|
||||
// state to expect; anything else armed is a newer generation this
|
||||
// lift must not touch — and falling through to the re-arm would
|
||||
// undo a resume that followed it.
|
||||
if s := t.growthFrozenState.Load(); s != 0 && s != frozenAt {
|
||||
return
|
||||
}
|
||||
if err := t.applyGrowthUnfreeze(frozenAt); err == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
if frozenAt != 0 && t.growthFrozenState.CompareAndSwap(0, frozenAt) {
|
||||
stats.MasterVolumeGrowthFrozen.Set(1)
|
||||
glog.Warningf("volume growth unfreeze not replicated; freeze re-armed for retry")
|
||||
}
|
||||
}
|
||||
|
||||
func (t *Topology) ResumeVolumeGrowth() bool {
|
||||
if !t.growthFrozen.CompareAndSwap(true, false) {
|
||||
return false
|
||||
for {
|
||||
// The compare binds the resume to the generation it observed: a
|
||||
// freeze landing between the load and the swap just moves to the
|
||||
// next iteration, which lifts the generation actually armed.
|
||||
frozenAt := t.growthFrozenState.Load()
|
||||
if frozenAt == 0 {
|
||||
return false
|
||||
}
|
||||
if t.growthFrozenState.CompareAndSwap(frozenAt, 0) {
|
||||
stats.MasterVolumeGrowthFrozen.Set(0)
|
||||
glog.V(0).Infof("volume growth resumed by operator")
|
||||
t.applyGrowthUnfreeze(frozenAt)
|
||||
return true
|
||||
}
|
||||
}
|
||||
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) {
|
||||
// SetGrowthFrozen applies a replicated freeze transition on every master. A
|
||||
// positive frozenAt pins the instant so all nodes share one recovery
|
||||
// window; a non-positive one — entries written before the field existed —
|
||||
// becomes the ageless generation, a stable identity a bound lift can still
|
||||
// match and the recovery check reads as already elapsed. An unfreeze bound
|
||||
// to unfreezeFor applies only while that exact generation is armed, so a
|
||||
// stale lift cannot erase a newer freeze; zero lifts unconditionally, for
|
||||
// entries written before the binding existed.
|
||||
func (t *Topology) SetGrowthFrozen(frozen bool, frozenAt int64, unfreezeFor int64) {
|
||||
if frozen {
|
||||
t.growthFrozenAt.Store(time.Now().UnixNano())
|
||||
if t.growthFrozen.CompareAndSwap(false, true) {
|
||||
gen := frozenAt
|
||||
if gen == 0 {
|
||||
gen = -1
|
||||
}
|
||||
if t.growthFrozenState.Swap(gen) == 0 {
|
||||
stats.MasterVolumeGrowthFrozen.Set(1)
|
||||
glog.V(0).Infof("volume growth freeze applied from raft")
|
||||
}
|
||||
return
|
||||
}
|
||||
if t.growthFrozen.CompareAndSwap(true, false) {
|
||||
if unfreezeFor == 0 {
|
||||
if t.growthFrozenState.Swap(0) != 0 {
|
||||
stats.MasterVolumeGrowthFrozen.Set(0)
|
||||
}
|
||||
return
|
||||
}
|
||||
if t.growthFrozenState.CompareAndSwap(unfreezeFor, 0) {
|
||||
stats.MasterVolumeGrowthFrozen.Set(0)
|
||||
}
|
||||
}
|
||||
|
||||
// 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) {
|
||||
// applyGrowthFreeze replicates the current freeze flag and instant so
|
||||
// followers keep both across a leader change.
|
||||
func (t *Topology) applyGrowthFreeze() error {
|
||||
t.growthFreezeApplyLock.Lock()
|
||||
defer t.growthFreezeApplyLock.Unlock()
|
||||
cmd := NewMaxVolumeIdCommand(t.GetMaxVolumeId(), t.GetTopologyId())
|
||||
frozen := true
|
||||
cmd.GrowthFrozen = &frozen
|
||||
cmd.GrowthFrozenAt = t.growthFrozenState.Load()
|
||||
return t.applyGrowthFreezeCommand(cmd)
|
||||
}
|
||||
|
||||
// applyGrowthUnfreeze replicates a lift bound to the freeze instant the
|
||||
// caller observed.
|
||||
func (t *Topology) applyGrowthUnfreeze(frozenAt int64) error {
|
||||
cmd := NewMaxVolumeIdCommand(t.GetMaxVolumeId(), t.GetTopologyId())
|
||||
frozen := false
|
||||
cmd.GrowthFrozen = &frozen
|
||||
cmd.GrowthUnfreezeFor = frozenAt
|
||||
return t.applyGrowthFreezeCommand(cmd)
|
||||
}
|
||||
|
||||
// applyGrowthFreezeCommand sends the transition through raft. With no raft
|
||||
// server (single-master test topologies) it applies locally.
|
||||
func (t *Topology) applyGrowthFreezeCommand(cmd *MaxVolumeIdCommand) error {
|
||||
frozen := cmd.GrowthFrozen != nil && *cmd.GrowthFrozen
|
||||
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)
|
||||
return err
|
||||
}
|
||||
} else if t.HashicorpRaft != nil {
|
||||
if b, err := json.Marshal(cmd); err != nil {
|
||||
glog.Warningf("marshal growth freeze %v: %v", frozen, err)
|
||||
} else if future := t.HashicorpRaft.Apply(b, time.Second); future.Error() != nil {
|
||||
glog.Warningf("apply growth freeze %v: %v", frozen, future.Error())
|
||||
}
|
||||
} else {
|
||||
t.SetGrowthFrozen(frozen)
|
||||
return nil
|
||||
}
|
||||
if t.HashicorpRaft != nil {
|
||||
b, err := json.Marshal(cmd)
|
||||
if err != nil {
|
||||
glog.Warningf("marshal growth freeze %v: %v", frozen, err)
|
||||
return err
|
||||
}
|
||||
if future := t.HashicorpRaft.Apply(b, time.Second); future.Error() != nil {
|
||||
glog.Warningf("apply growth freeze %v: %v", frozen, future.Error())
|
||||
return future.Error()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
t.SetGrowthFrozen(*cmd.GrowthFrozen, cmd.GrowthFrozenAt, cmd.GrowthUnfreezeFor)
|
||||
return nil
|
||||
}
|
||||
|
||||
// MarkFreshMint records that this boot minted the TopologyId rather than
|
||||
@@ -1713,13 +1824,17 @@ func (t *Topology) SetTopologyInitPending(pending bool) {
|
||||
// a mismatching report checked here, or still in flight under the same
|
||||
// lock, lands its freeze before any allocation can observe the open gate.
|
||||
t.reportedTopologyIdsLock.Lock()
|
||||
defer t.reportedTopologyIdsLock.Unlock()
|
||||
reported := t.pendingReportedTopologyIds
|
||||
t.pendingReportedTopologyIds = nil
|
||||
freeze := false
|
||||
for id := range reported {
|
||||
t.checkReportedTopologyIdLocked(id)
|
||||
freeze = t.checkReportedTopologyIdLocked(id) || freeze
|
||||
}
|
||||
t.topologyInitPending.Store(false)
|
||||
t.reportedTopologyIdsLock.Unlock()
|
||||
if freeze {
|
||||
t.applyGrowthFreeze()
|
||||
}
|
||||
}
|
||||
|
||||
// ReleaseAllocationOnFirstReport opts the topology into lifting the post-mint
|
||||
@@ -1879,7 +1994,7 @@ func (t *Topology) SetMaxVolumeIdFile(file string) {
|
||||
}
|
||||
if cmd.GrowthFrozen != nil {
|
||||
t.lastPersistedGrowthFrozen.Store(*cmd.GrowthFrozen)
|
||||
t.SetGrowthFrozen(*cmd.GrowthFrozen)
|
||||
t.SetGrowthFrozen(*cmd.GrowthFrozen, cmd.GrowthFrozenAt, 0)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1888,7 +2003,7 @@ func (t *Topology) SetMaxVolumeIdFile(file string) {
|
||||
// recovery, so follower mirrors stay current too.
|
||||
func (t *Topology) PersistMaxVolumeId(vid needle.VolumeId) error {
|
||||
topoId := t.GetTopologyId()
|
||||
frozen := t.IsGrowthFrozen()
|
||||
frozen, frozenAt := t.GrowthFreezeState()
|
||||
mirrored := func() bool {
|
||||
return t.lastPersistedMaxVolumeId.Load() >= uint32(vid) && t.lastPersistedTopologyId.Load() == topoId && t.lastPersistedGrowthFrozen.Load() == frozen
|
||||
}
|
||||
@@ -1908,6 +2023,9 @@ func (t *Topology) PersistMaxVolumeId(vid needle.VolumeId) error {
|
||||
}
|
||||
cmd := NewMaxVolumeIdCommand(vid, topoId)
|
||||
cmd.GrowthFrozen = &frozen
|
||||
if frozen {
|
||||
cmd.GrowthFrozenAt = frozenAt
|
||||
}
|
||||
b, err := json.Marshal(cmd)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
Reference in new issue
Block a user