diff --git a/weed/admin/maintenance/maintenance_manager.go b/weed/admin/maintenance/maintenance_manager.go index 90e9a5d29..c251a38c1 100644 --- a/weed/admin/maintenance/maintenance_manager.go +++ b/weed/admin/maintenance/maintenance_manager.go @@ -141,7 +141,9 @@ func (mm *MaintenanceManager) Start() error { return fmt.Errorf("invalid maintenance configuration: %w", err) } + mm.mutex.Lock() mm.running = true + mm.mutex.Unlock() // Start background processes go mm.scanLoop() @@ -187,30 +189,54 @@ func (mm *MaintenanceManager) validateConfig() error { return nil } -// IsRunning returns whether the maintenance manager is currently running +// IsRunning returns whether the maintenance manager is currently running. +// running is guarded by mm.mutex because the background loops read it on every +// iteration while Start and Stop are called from the admin server's goroutines. func (mm *MaintenanceManager) IsRunning() bool { + mm.mutex.RLock() + defer mm.mutex.RUnlock() return mm.running } -// Stop terminates the maintenance manager +// Stop terminates the maintenance manager. It is a no-op when the manager is not +// running, so a second call cannot close the already closed stop channel. func (mm *MaintenanceManager) Stop() { + mm.mutex.Lock() + if !mm.running { + mm.mutex.Unlock() + return + } mm.running = false close(mm.stopChan) + mm.mutex.Unlock() + glog.Infof("Maintenance manager stopped") } // scanLoop periodically scans for maintenance tasks with adaptive timing func (mm *MaintenanceManager) scanLoop() { scanInterval := time.Duration(mm.config.ScanIntervalSeconds) * time.Second - ticker := time.NewTicker(scanInterval) - defer ticker.Stop() - for mm.running { + // activeInterval is the interval the ticker is actually running at right now. It has + // to be tracked separately from the configured scanInterval, because the error backoff + // replaces the ticker with a much shorter one. Comparing the target against the + // configured interval instead never restores the normal cadence: once the errors stop, + // getScanInterval returns scanInterval again, the comparison comes out false, and the + // ticker is left at the backoff delay. A single transient scan failure therefore pinned + // the scanner to one scan per second forever - see issue #10874, where that produced a + // ~658 KB/s log flood and 193k orphaned task files. + activeInterval := scanInterval + ticker := time.NewTicker(activeInterval) + // Wrapped in a closure so the replacement ticker is stopped, not the one that happened + // to be current when the defer was registered. + defer func() { ticker.Stop() }() + + for mm.IsRunning() { select { case <-mm.stopChan: return case <-ticker.C: - glog.V(1).Infof("Performing maintenance scan every %v", scanInterval) + glog.V(1).Infof("Performing maintenance scan every %v", activeInterval) // Use the same synchronization as TriggerScan to prevent concurrent scans if err := mm.triggerScanInternal(false); err != nil { @@ -220,10 +246,13 @@ func (mm *MaintenanceManager) scanLoop() { // Adjust ticker interval based on error state (read error state safely) currentInterval := mm.getScanInterval(scanInterval) - // Reset ticker with new interval if needed - if currentInterval != scanInterval { + // Reset ticker whenever the target differs from what the ticker is running at, + // which covers both entering the backoff and returning to the normal cadence. + if currentInterval != activeInterval { ticker.Stop() ticker = time.NewTicker(currentInterval) + glog.V(1).Infof("Maintenance scan cadence changed from %v to %v", activeInterval, currentInterval) + activeInterval = currentInterval } } } @@ -255,7 +284,7 @@ func (mm *MaintenanceManager) cleanupLoop() { ticker := time.NewTicker(cleanupInterval) defer ticker.Stop() - for mm.running { + for mm.IsRunning() { select { case <-mm.stopChan: return @@ -272,7 +301,7 @@ func (mm *MaintenanceManager) topologyStatusLoop() { ticker := time.NewTicker(statusInterval) defer ticker.Stop() - for mm.running { + for mm.IsRunning() { select { case <-mm.stopChan: return @@ -541,12 +570,15 @@ func (mm *MaintenanceManager) TriggerScan() error { // triggerScanInternal handles both manual and automatic scan triggers func (mm *MaintenanceManager) triggerScanInternal(isManual bool) error { + // running and scanInProgress are checked under one lock so a Stop that lands + // between the two checks cannot leave a scan running after shutdown. + mm.mutex.Lock() if !mm.running { + mm.mutex.Unlock() return fmt.Errorf("maintenance manager is not running") } // Prevent multiple concurrent scans - mm.mutex.Lock() if mm.scanInProgress { mm.mutex.Unlock() if isManual { diff --git a/weed/admin/maintenance/maintenance_scan_cadence_test.go b/weed/admin/maintenance/maintenance_scan_cadence_test.go new file mode 100644 index 000000000..7dc195d78 --- /dev/null +++ b/weed/admin/maintenance/maintenance_scan_cadence_test.go @@ -0,0 +1,168 @@ +package maintenance + +import ( + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" +) + +// scanCadenceClient is a minimal AdminClient whose first call blocks until the test +// releases it. Blocking the first scan is what makes the cadence test deterministic: +// the scan loop evaluates the error state right after triggering a scan, and that scan +// runs in its own goroutine, so without the block the loop could observe the error +// counter either before or after the scan finished. +type scanCadenceClient struct { + mu sync.Mutex + calls int + release chan struct{} +} + +func (c *scanCadenceClient) WithMasterClient(fn func(client master_pb.SeaweedClient) error) error { + c.mu.Lock() + c.calls++ + first := c.calls == 1 + c.mu.Unlock() + + if first { + <-c.release + } + + // Returning nil without invoking fn yields an empty, successful scan. + return nil +} + +func (c *scanCadenceClient) callCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.calls +} + +// TestScanLoopRestoresIntervalAfterBackoff is the regression test for the scan flood in +// https://github.com/seaweedfs/seaweedfs/issues/10874. The loop shortens its ticker to the +// error backoff delay after a failed scan. It used to compare the target interval against +// the *configured* scan interval rather than against the interval the ticker was actually +// running at, so once the errors stopped the comparison came out false and the ticker was +// never restored: one transient failure pinned the scanner to one scan per second for as +// long as the process lived. +// +// The test seeds the error state, lets the loop drop into the 1s backoff, then lets a scan +// succeed and asserts the loop goes back to the configured 3s cadence instead of keeping +// the 1s one. +func TestScanLoopRestoresIntervalAfterBackoff(t *testing.T) { + const baseInterval = 3 * time.Second + + client := &scanCadenceClient{release: make(chan struct{})} + + config := DefaultMaintenanceConfig() + config.ScanIntervalSeconds = int32(baseInterval / time.Second) + manager := NewMaintenanceManager(client, config, nil) + + // Seed a failed scan so the very first cadence decision drops to the backoff delay. + // backoffDelay is what getScanInterval returns while errorCount > 0. + manager.mutex.Lock() + manager.errorCount = 1 + manager.backoffDelay = time.Second + manager.running = true + manager.mutex.Unlock() + + go manager.scanLoop() + defer manager.Stop() + + // First tick at ~3s: the scan blocks in the client, so the loop still sees errorCount == 1 + // and switches the ticker to the 1s backoff. + deadline := time.Now().Add(baseInterval + 2*time.Second) + for client.callCount() == 0 { + if time.Now().After(deadline) { + t.Fatal("scan loop never triggered its first scan") + } + time.Sleep(20 * time.Millisecond) + } + + // Let the blocked scan complete successfully, which clears the error state. + close(client.release) + + // Wait for the error state to clear so the next cadence decision is unambiguous. + deadline = time.Now().Add(2 * time.Second) + for { + errorCount, _, _ := manager.GetErrorState() + if errorCount == 0 { + break + } + if time.Now().After(deadline) { + t.Fatalf("error tracking never reset, errorCount=%d", errorCount) + } + time.Sleep(20 * time.Millisecond) + } + + // The ticker is at 1s now. Wait for the next tick, where the loop must notice the + // recovery and restore the 3s cadence. + time.Sleep(1500 * time.Millisecond) + before := client.callCount() + + // Observe a window that a 1s cadence would fill with scans and a 3s cadence would not. + const window = 5 * time.Second + time.Sleep(window) + scansInWindow := client.callCount() - before + + // 3s cadence: at most 2 scans in 5s. 1s cadence: about 5. + if scansInWindow > 2 { + t.Errorf("scan loop ran %d scans in %v after recovering from a failed scan; "+ + "the ticker was left at the %v backoff instead of returning to the configured %v", + scansInWindow, window, time.Second, baseInterval) + } + if scansInWindow == 0 { + t.Errorf("scan loop ran no scans in %v, expected the %v cadence to fire at least once", window, baseInterval) + } +} + +// TestScanLoopEntersBackoffOnError checks the other half of the cadence logic: a failing +// scan still has to shorten the ticker, so the fix above did not simply pin the loop to +// the configured interval. +func TestScanLoopEntersBackoffOnError(t *testing.T) { + config := DefaultMaintenanceConfig() + config.ScanIntervalSeconds = 600 + manager := NewMaintenanceManager(nil, config, nil) + + baseInterval := time.Duration(config.ScanIntervalSeconds) * time.Second + + if got := manager.getScanInterval(baseInterval); got != baseInterval { + t.Errorf("healthy scan interval = %v, want the configured %v", got, baseInterval) + } + + manager.mutex.Lock() + manager.errorCount = 1 + manager.backoffDelay = time.Second + manager.mutex.Unlock() + + if got := manager.getScanInterval(baseInterval); got != time.Second { + t.Errorf("scan interval while failing = %v, want the %v backoff", got, time.Second) + } + + manager.mutex.Lock() + manager.resetErrorTracking() + manager.mutex.Unlock() + + if got := manager.getScanInterval(baseInterval); got != baseInterval { + t.Errorf("scan interval after recovery = %v, want the configured %v", got, baseInterval) + } +} + +// TestStopIsIdempotent guards the stop channel against a double close, which used to panic +// when StopMaintenanceManager ran twice (for example on a shutdown path that also runs on +// a signal handler). +func TestStopIsIdempotent(t *testing.T) { + manager := NewMaintenanceManager(nil, DefaultMaintenanceConfig(), nil) + + manager.mutex.Lock() + manager.running = true + manager.mutex.Unlock() + + manager.Stop() + manager.Stop() + + if manager.IsRunning() { + t.Error("manager still reports running after Stop") + } +}