mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-07 04:27:49 +01:00
admin: re-send the master vacuum disable every poll so it survives leader changes (#12430)
* admin: re-send the master vacuum disable every poll so it survives leader changes
The admin disabled the master's scheduled vacuum only when a vacuum plugin
worker connected or disconnected. The master keeps that flag in memory and it
is not replicated through raft, so after a leader change or a leader restart
the new leader ran its own vacuum next to the plugin worker. When both picked
the same volume the master's commit failed ("compact swap aborted: missing
.cpd/.cpx") and it dropped that replica, leaving the volume under-replicated.
Send the desired state on every poll instead, and log on the master only
when the flag actually changes.
* admin: shorten the vacuum sync comments
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---------
Co-authored-by: Kevin <kevin@leetify.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
3 files changed
+81
-33
No files matched your search
@@ -380,29 +380,29 @@ func (m *masterVacuumToggler) enableVacuum() error {
|
||||
})
|
||||
}
|
||||
|
||||
// syncVacuumState performs a single sync step: checks if a vacuum-capable worker
|
||||
// is present and calls disable/enable accordingly. Returns the updated state
|
||||
// and whether the call failed (for log dedup on retries).
|
||||
// syncVacuumState tells the master to disable its automatic vacuum while a
|
||||
// vacuum-capable worker is present and to re-enable it otherwise. It re-sends
|
||||
// on every tick because the master keeps the flag in memory only, so a leader
|
||||
// change or master restart drops it. Only transitions are logged. Returns the
|
||||
// state the master was last set to and whether this call failed.
|
||||
func syncVacuumState(hasWorker bool, previouslyActive bool, toggler vacuumToggler, retrying bool) (active bool, failed bool) {
|
||||
if hasWorker == previouslyActive {
|
||||
return previouslyActive, false
|
||||
}
|
||||
changed := hasWorker != previouslyActive
|
||||
if hasWorker {
|
||||
if !retrying {
|
||||
if changed && !retrying {
|
||||
glog.V(0).Infof("Vacuum plugin worker connected, disabling master automatic vacuum")
|
||||
}
|
||||
if err := toggler.disableVacuum(); err != nil {
|
||||
glog.Warningf("Failed to disable vacuum on master: %v", err)
|
||||
return false, true // retry next tick
|
||||
return previouslyActive, true // retry next tick
|
||||
}
|
||||
return true, false
|
||||
}
|
||||
if !retrying {
|
||||
if changed && !retrying {
|
||||
glog.V(0).Infof("Vacuum plugin worker disconnected, re-enabling master automatic vacuum")
|
||||
}
|
||||
if err := toggler.enableVacuum(); err != nil {
|
||||
glog.Warningf("Failed to enable vacuum on master: %v", err)
|
||||
return true, true // retry next tick
|
||||
return previouslyActive, true // retry next tick
|
||||
}
|
||||
return false, false
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package dash
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -56,36 +57,22 @@ func TestVacuumMonitorStateTransitions(t *testing.T) {
|
||||
type fakeToggler struct {
|
||||
disableCalls int
|
||||
enableCalls int
|
||||
err error
|
||||
}
|
||||
|
||||
func (f *fakeToggler) disableVacuum() error {
|
||||
f.disableCalls++
|
||||
return nil
|
||||
return f.err
|
||||
}
|
||||
|
||||
func (f *fakeToggler) enableVacuum() error {
|
||||
f.enableCalls++
|
||||
return nil
|
||||
return f.err
|
||||
}
|
||||
|
||||
func TestSyncVacuumState(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
t.Run("no change when state matches", func(t *testing.T) {
|
||||
tog := &fakeToggler{}
|
||||
result, failed := syncVacuumState(false, false, tog, false)
|
||||
if result != false || failed {
|
||||
t.Error("expected false, no failure")
|
||||
}
|
||||
result, failed = syncVacuumState(true, true, tog, false)
|
||||
if result != true || failed {
|
||||
t.Error("expected true, no failure")
|
||||
}
|
||||
if tog.disableCalls != 0 || tog.enableCalls != 0 {
|
||||
t.Errorf("expected no calls, got disable=%d enable=%d", tog.disableCalls, tog.enableCalls)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("worker connects triggers disable", func(t *testing.T) {
|
||||
tog := &fakeToggler{}
|
||||
result, failed := syncVacuumState(true, false, tog, false)
|
||||
@@ -113,4 +100,61 @@ func TestSyncVacuumState(t *testing.T) {
|
||||
t.Errorf("expected 0 disable calls, got %d", tog.disableCalls)
|
||||
}
|
||||
})
|
||||
|
||||
// The master keeps the flag in memory only, so a new leader or a restarted
|
||||
// master has lost it. The disable has to be re-sent while the worker stays
|
||||
// connected, not only when it first connects.
|
||||
t.Run("disable is re-sent every tick while the worker stays connected", func(t *testing.T) {
|
||||
tog := &fakeToggler{}
|
||||
active := false
|
||||
for i := 0; i < 3; i++ {
|
||||
var failed bool
|
||||
active, failed = syncVacuumState(true, active, tog, false)
|
||||
if !active || failed {
|
||||
t.Fatalf("tick %d: expected active, no failure", i)
|
||||
}
|
||||
}
|
||||
if tog.disableCalls != 3 {
|
||||
t.Errorf("expected 3 disable calls, got %d", tog.disableCalls)
|
||||
}
|
||||
if tog.enableCalls != 0 {
|
||||
t.Errorf("expected 0 enable calls, got %d", tog.enableCalls)
|
||||
}
|
||||
})
|
||||
|
||||
// Same for the enable side: a master that still holds an old plugin disable
|
||||
// and becomes leader would otherwise never vacuum again.
|
||||
t.Run("enable is re-sent every tick while no worker is connected", func(t *testing.T) {
|
||||
tog := &fakeToggler{}
|
||||
active := false
|
||||
for i := 0; i < 3; i++ {
|
||||
var failed bool
|
||||
active, failed = syncVacuumState(false, active, tog, false)
|
||||
if active || failed {
|
||||
t.Fatalf("tick %d: expected inactive, no failure", i)
|
||||
}
|
||||
}
|
||||
if tog.enableCalls != 3 {
|
||||
t.Errorf("expected 3 enable calls, got %d", tog.enableCalls)
|
||||
}
|
||||
if tog.disableCalls != 0 {
|
||||
t.Errorf("expected 0 disable calls, got %d", tog.disableCalls)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("failed call keeps the previous state and reports failure", func(t *testing.T) {
|
||||
tog := &fakeToggler{err: errors.New("master unavailable")}
|
||||
result, failed := syncVacuumState(true, false, tog, false)
|
||||
if result != false || !failed {
|
||||
t.Error("expected previous state false and a failure")
|
||||
}
|
||||
result, failed = syncVacuumState(true, true, tog, false)
|
||||
if result != true || !failed {
|
||||
t.Error("expected previous state true and a failure")
|
||||
}
|
||||
result, failed = syncVacuumState(false, true, tog, false)
|
||||
if result != true || !failed {
|
||||
t.Error("expected previous state true and a failure")
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -1543,17 +1543,21 @@ func (t *Topology) EnableVacuum() {
|
||||
}
|
||||
|
||||
// DisableVacuumByPlugin is called by the admin server's vacuum monitor
|
||||
// when a vacuum plugin worker connects. Only sets the plugin flag.
|
||||
// while a vacuum plugin worker is connected. Only sets the plugin flag.
|
||||
// The monitor repeats the call on every poll, so only a change is logged.
|
||||
func (t *Topology) DisableVacuumByPlugin() {
|
||||
glog.V(0).Infof("DisableVacuum (by plugin worker)")
|
||||
t.vacuumDisabledByPlugin.Store(true)
|
||||
if !t.vacuumDisabledByPlugin.Swap(true) {
|
||||
glog.V(0).Infof("DisableVacuum (by plugin worker)")
|
||||
}
|
||||
}
|
||||
|
||||
// EnableVacuumByPlugin is called by the admin server's vacuum monitor
|
||||
// when a vacuum plugin worker disconnects. Only clears the plugin flag.
|
||||
// while no vacuum plugin worker is connected. Only clears the plugin flag.
|
||||
// The monitor repeats the call on every poll, so only a change is logged.
|
||||
func (t *Topology) EnableVacuumByPlugin() {
|
||||
glog.V(0).Infof("EnableVacuum (by plugin worker)")
|
||||
t.vacuumDisabledByPlugin.Store(false)
|
||||
if t.vacuumDisabledByPlugin.Swap(false) {
|
||||
glog.V(0).Infof("EnableVacuum (by plugin worker)")
|
||||
}
|
||||
}
|
||||
|
||||
// IsVacuumDisabledByPlugin returns whether the plugin monitor has disabled vacuum.
|
||||
|
||||
Reference in new issue
Block a user