diff --git a/weed/admin/dash/admin_lock_manager.go b/weed/admin/dash/admin_lock_manager.go index 41ec0ed1b..5be4c421f 100644 --- a/weed/admin/dash/admin_lock_manager.go +++ b/weed/admin/dash/admin_lock_manager.go @@ -14,16 +14,38 @@ const ( adminLockClientName = "admin-plugin" ) +const ( + // A competing shell client polls the master once a second, while a local + // re-acquire would win immediately. Leave the lock free for slightly more + // than one poll interval so a waiting operator gets it first. + defaultAdminLockYieldWindow = 2 * time.Second + // After the lock has been held continuously for this long (overlapping + // reference-counted holds can keep it held indefinitely), new Acquire + // calls wait for a full release so shell clients are not starved. + defaultAdminLockFairnessWindow = time.Minute +) + +// adminLocker is the subset of exclusive_locks.ExclusiveLocker the manager uses. +type adminLocker interface { + RequestLock(clientName string) + ReleaseLock() + SetMessage(message string) +} + // AdminLockManager coordinates exclusive admin locks with reference counting. // It is safe for concurrent use. type AdminLockManager struct { - locker *exclusive_locks.ExclusiveLocker - clientName string + locker adminLocker + clientName string + yieldWindow time.Duration + fairnessWindow time.Duration mu sync.Mutex cond *sync.Cond acquiring bool + releasing bool holdCount int + heldSince time.Time lastAcquiredAt time.Time lastReleasedAt time.Time @@ -40,13 +62,19 @@ func NewAdminLockManager(masterClient *wdclient.MasterClient, clientName string) clientName = adminLockClientName } manager := &AdminLockManager{ - locker: exclusive_locks.NewExclusiveLocker(masterClient, adminLockName), - clientName: clientName, + locker: exclusive_locks.NewExclusiveLocker(masterClient, adminLockName), + clientName: clientName, + yieldWindow: defaultAdminLockYieldWindow, + fairnessWindow: defaultAdminLockFairnessWindow, } manager.cond = sync.NewCond(&manager.mu) return manager } +// Acquire takes the shared cluster lock, reference counted so overlapping +// admin-side holders piggyback on one lease. It must not be called while the +// caller's own call stack already holds the lock: once the fairness window +// makes new acquires wait for a full release, a nested Acquire deadlocks. func (m *AdminLockManager) Acquire(reason string) (func(), error) { if m == nil || m.locker == nil { return func() {}, nil @@ -57,18 +85,28 @@ func (m *AdminLockManager) Acquire(reason string) (func(), error) { m.locker.SetMessage(reason) m.currentReason = reason } - for m.acquiring { + for m.acquiring || m.releasing || (m.holdCount > 0 && time.Since(m.heldSince) > m.fairnessWindow) { m.cond.Wait() } if m.holdCount == 0 { m.acquiring = true m.waitingSince = time.Now().UTC() m.waitingReason = reason + yield := time.Duration(0) + if !m.lastReleasedAt.IsZero() { + if since := time.Since(m.lastReleasedAt); since < m.yieldWindow { + yield = m.yieldWindow - since + } + } m.mu.Unlock() + if yield > 0 { + time.Sleep(yield) + } m.locker.RequestLock(m.clientName) m.mu.Lock() m.acquiring = false m.holdCount = 1 + m.heldSince = time.Now() m.lastAcquiredAt = time.Now().UTC() m.waitingSince = time.Time{} m.waitingReason = "" @@ -96,14 +134,22 @@ func (m *AdminLockManager) Release() { } m.holdCount-- shouldRelease := m.holdCount == 0 + if shouldRelease { + // Block new acquires until ReleaseLock completes: a RequestLock + // issued while the locker still considers itself locked would no-op + // and leave a hold with no live lease. + m.releasing = true + } m.mu.Unlock() if shouldRelease { + m.locker.ReleaseLock() m.mu.Lock() + m.releasing = false m.lastReleasedAt = time.Now().UTC() m.currentReason = "" + m.cond.Broadcast() m.mu.Unlock() - m.locker.ReleaseLock() } } diff --git a/weed/admin/dash/admin_lock_manager_test.go b/weed/admin/dash/admin_lock_manager_test.go new file mode 100644 index 000000000..4f3b26289 --- /dev/null +++ b/weed/admin/dash/admin_lock_manager_test.go @@ -0,0 +1,156 @@ +package dash + +import ( + "sync" + "testing" + "time" +) + +type fakeAdminLocker struct { + releaseDelay time.Duration + + mu sync.Mutex + requests []time.Time + releases []time.Time // recorded when ReleaseLock returns +} + +func (f *fakeAdminLocker) RequestLock(clientName string) { + f.mu.Lock() + f.requests = append(f.requests, time.Now()) + f.mu.Unlock() +} + +func (f *fakeAdminLocker) ReleaseLock() { + if f.releaseDelay > 0 { + time.Sleep(f.releaseDelay) + } + f.mu.Lock() + f.releases = append(f.releases, time.Now()) + f.mu.Unlock() +} + +func (f *fakeAdminLocker) SetMessage(message string) {} + +func (f *fakeAdminLocker) counts() (int, int) { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.requests), len(f.releases) +} + +func newTestLockManager(locker adminLocker, yieldWindow, fairnessWindow time.Duration) *AdminLockManager { + m := &AdminLockManager{ + locker: locker, + clientName: "test", + yieldWindow: yieldWindow, + fairnessWindow: fairnessWindow, + } + m.cond = sync.NewCond(&m.mu) + return m +} + +func TestAdminLockManagerRefCountsOverlappingHolds(t *testing.T) { + t.Parallel() + + locker := &fakeAdminLocker{} + m := newTestLockManager(locker, defaultAdminLockYieldWindow, defaultAdminLockFairnessWindow) + + release1, _ := m.Acquire("a") + release2, _ := m.Acquire("b") + + if requests, releases := locker.counts(); requests != 1 || releases != 0 { + t.Fatalf("expected one lease request while overlapping, got requests=%d releases=%d", requests, releases) + } + + release1() + if _, releases := locker.counts(); releases != 0 { + t.Fatalf("lock released while still held by another caller") + } + release2() + if _, releases := locker.counts(); releases != 1 { + t.Fatalf("expected one release after last holder, got %d", releases) + } +} + +func TestAdminLockManagerYieldsBeforeReacquire(t *testing.T) { + t.Parallel() + + locker := &fakeAdminLocker{} + m := newTestLockManager(locker, 100*time.Millisecond, defaultAdminLockFairnessWindow) + + release, _ := m.Acquire("first") + release() + release, _ = m.Acquire("second") + release() + + locker.mu.Lock() + gap := locker.requests[1].Sub(locker.releases[0]) + locker.mu.Unlock() + if gap < 90*time.Millisecond { + t.Fatalf("re-acquire did not yield to competing clients: gap=%v", gap) + } +} + +func TestAdminLockManagerFairnessWindowForcesFullRelease(t *testing.T) { + t.Parallel() + + locker := &fakeAdminLocker{} + m := newTestLockManager(locker, 20*time.Millisecond, 50*time.Millisecond) + + release1, _ := m.Acquire("long-hold") + m.mu.Lock() + m.heldSince = time.Now().Add(-time.Second) + m.mu.Unlock() + + acquired := make(chan func(), 1) + go func() { + release2, _ := m.Acquire("late-joiner") + acquired <- release2 + }() + + select { + case <-acquired: + t.Fatalf("late acquire joined a hold older than the fairness window") + case <-time.After(50 * time.Millisecond): + } + + release1() + + select { + case release2 := <-acquired: + release2() + case <-time.After(time.Second): + t.Fatalf("late acquire did not proceed after full release") + } + + if requests, releases := locker.counts(); requests != 2 || releases != 2 { + t.Fatalf("expected a full release/re-acquire cycle, got requests=%d releases=%d", requests, releases) + } +} + +func TestAdminLockManagerAcquireWaitsForReleaseInFlight(t *testing.T) { + t.Parallel() + + locker := &fakeAdminLocker{releaseDelay: 50 * time.Millisecond} + m := newTestLockManager(locker, time.Millisecond, defaultAdminLockFairnessWindow) + + release1, _ := m.Acquire("first") + releaseDone := make(chan struct{}) + go func() { + release1() + close(releaseDone) + }() + // Let the release enter the slow ReleaseLock before re-acquiring. + time.Sleep(10 * time.Millisecond) + release2, _ := m.Acquire("second") + <-releaseDone + release2() + + locker.mu.Lock() + defer locker.mu.Unlock() + if len(locker.requests) != 2 || len(locker.releases) != 2 { + t.Fatalf("unexpected lease traffic: requests=%d releases=%d", len(locker.requests), len(locker.releases)) + } + if locker.requests[1].Before(locker.releases[0]) { + t.Fatalf("RequestLock ran while ReleaseLock was still in flight") + } +} diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index 1bf10fec2..4adbc2178 100644 --- a/weed/admin/dash/admin_server.go +++ b/weed/admin/dash/admin_server.go @@ -1459,8 +1459,8 @@ func (s *AdminServer) RunPluginDetectionWithReport( // DispatchPluginProposals dispatches a batch of proposals using the same // capacity-aware dispatch logic as the scheduler loop (executor reservation with -// backoff, per-job retry on transient errors). The plugin lock must already be -// held by the caller. +// backoff, per-job retry on transient errors). The dispatch takes the cluster +// admin lock around each job itself; callers must not hold it. func (s *AdminServer) DispatchPluginProposals( ctx context.Context, jobType string, diff --git a/weed/admin/dash/plugin_api.go b/weed/admin/dash/plugin_api.go index 7f3700306..e21a4e1d1 100644 --- a/weed/admin/dash/plugin_api.go +++ b/weed/admin/dash/plugin_api.go @@ -595,20 +595,15 @@ func (s *AdminServer) TriggerPluginDetectionAPI(w http.ResponseWriter, r *http.R } // RunPluginJobTypeAPI runs full workflow for one job type: detect then dispatch detected jobs. +// RunPluginDetection and the dispatch path take the cluster admin lock +// themselves (around detection and around each job), so no lock is held +// across the whole workflow here. func (s *AdminServer) RunPluginJobTypeAPI(w http.ResponseWriter, r *http.Request) { jobType := strings.TrimSpace(mux.Vars(r)["jobType"]) if jobType == "" { writeJSONError(w, http.StatusBadRequest, "jobType is required") return } - releaseLock, err := s.acquirePluginLock(fmt.Sprintf("plugin detect+execute %s", jobType)) - if err != nil { - writeJSONError(w, http.StatusInternalServerError, err.Error()) - return - } - if releaseLock != nil { - defer releaseLock() - } var req struct { ClusterContext json.RawMessage `json:"cluster_context"` diff --git a/weed/admin/plugin/plugin.go b/weed/admin/plugin/plugin.go index d50d7d2a3..8f1944d8c 100644 --- a/weed/admin/plugin/plugin.go +++ b/weed/admin/plugin/plugin.go @@ -424,7 +424,13 @@ func (r *Plugin) acquireAdminLock(reason string) (func(), error) { if r == nil || r.lockManager == nil { return func() {}, nil } - return r.lockManager.Acquire(reason) + release, err := r.lockManager.Acquire(reason) + if release == nil { + // Lock manager implementations may return a nil release when + // disabled; callers always invoke the release function. + release = func() {} + } + return release, err } // RunDetectionWithReport requests one detector worker and returns proposals with request metadata. diff --git a/weed/admin/plugin/plugin_scheduler.go b/weed/admin/plugin/plugin_scheduler.go index 5db4ddf11..294bfd6f1 100644 --- a/weed/admin/plugin/plugin_scheduler.go +++ b/weed/admin/plugin/plugin_scheduler.go @@ -98,9 +98,11 @@ func (r *Plugin) laneSchedulerLoop(ls *schedulerLaneState) { // runLaneSchedulerIteration runs one scheduling pass for a single lane, // processing only the job types assigned to that lane. // -// For lanes that require a lock (e.g. LaneDefault), all job types are -// processed sequentially under one admin lock because their volume -// management operations share global state. +// For lanes that require a lock (e.g. LaneDefault), job types are processed +// sequentially because their volume management operations share global +// state, and the cluster admin lock is taken around each detection and each +// dispatched job rather than the whole pass, so manual shell operations +// sharing the same lock can interleave between jobs. // // For lanes that do not require a lock (e.g. LaneIceberg, LaneLifecycle), // each job type runs independently in its own goroutine so they do not @@ -122,7 +124,7 @@ func (r *Plugin) runLaneSchedulerIteration(ls *schedulerLaneState) bool { } if LaneRequiresLock(ls.lane) { - return r.runLaneSchedulerIterationLocked(ls, jobTypes) + return r.runLaneSchedulerIterationSequential(ls, jobTypes) } return r.runLaneSchedulerIterationConcurrent(ls, jobTypes) } @@ -162,21 +164,13 @@ func (r *Plugin) collectDueJobTypes(ls *schedulerLaneState, jobTypes []string) ( return active, due } -// runLaneSchedulerIterationLocked processes job types sequentially under a -// single admin lock. Used by the default lane where volume management -// operations must be serialised. -func (r *Plugin) runLaneSchedulerIterationLocked(ls *schedulerLaneState, jobTypes []string) bool { - r.setLaneLoopState(ls, "", "waiting_for_lock") - lockName := fmt.Sprintf("plugin scheduler:%s", ls.lane) - releaseLock, err := r.acquireAdminLock(lockName) - if err != nil { - glog.Warningf("Plugin scheduler [%s] failed to acquire lock: %v", ls.lane, err) - r.setLaneLoopState(ls, "", "idle") - return false - } - if releaseLock != nil { - defer releaseLock() - } +// runLaneSchedulerIterationSequential processes job types one at a time. +// Used by the default lane where volume management operations must be +// serialised. The cluster admin lock is not held across the pass; +// runJobTypeIteration and dispatchScheduledProposals take it around each +// detection and each job. +func (r *Plugin) runLaneSchedulerIterationSequential(ls *schedulerLaneState, jobTypes []string) bool { + r.setLaneLoopState(ls, "", "busy") active, due := r.collectDueJobTypes(ls, jobTypes) hadJobs := false @@ -288,6 +282,30 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo return false } + // Detection plans against the live topology, so it runs under the shared + // cluster admin lock; the lock is released as soon as detection returns. + releaseDetectionLock := func() {} + if LaneRequiresLock(JobTypeLane(jobType)) { + r.setSchedulerLoopStateForJobType(jobType, "waiting_for_lock") + release, lockErr := r.acquireAdminLock(fmt.Sprintf("plugin scheduler %s detection", jobType)) + if lockErr != nil { + r.recordSchedulerDetectionError(jobType, lockErr) + r.appendActivity(JobActivity{ + JobType: jobType, + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduled detection aborted: %v", lockErr), + Stage: "failed", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + r.recordSchedulerRunComplete(jobType, "error") + return false + } + var releaseOnce sync.Once + releaseDetectionLock = func() { releaseOnce.Do(release) } + r.setSchedulerLoopStateForJobType(jobType, "detecting") + } + defer releaseDetectionLock() + detectionTimeout := policy.DetectionTimeout remaining := time.Until(start.Add(maxRuntime)) if remaining <= 0 { @@ -311,6 +329,7 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo detectCtx, cancelDetect := context.WithTimeout(jobCtx, detectionTimeout) proposals, err := r.RunDetection(detectCtx, jobType, clusterContext, policy.MaxResults) cancelDetect() + releaseDetectionLock() if err != nil { r.recordSchedulerDetectionError(jobType, err) stage := "failed" @@ -375,26 +394,6 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo r.setSchedulerLoopStateForJobType(jobType, "executing") - // Scan proposals for the maximum estimated_runtime_seconds so the - // execution phase gets enough time for large jobs (e.g. vacuum on - // big volumes). If any proposal needs more time than the remaining - // JobTypeMaxRuntime, extend the execution context accordingly. - var maxEstimatedRuntime time.Duration - for _, p := range filtered { - if p.Parameters != nil { - if est, ok := p.Parameters["estimated_runtime_seconds"]; ok { - if v := est.GetInt64Value(); v > 0 { - if d := time.Duration(v) * time.Second; d > maxEstimatedRuntime { - maxEstimatedRuntime = d - } - } - } - } - } - if maxEstimatedRuntime > maxEstimatedRuntimeCap { - maxEstimatedRuntime = maxEstimatedRuntimeCap - } - remaining = time.Until(start.Add(maxRuntime)) if remaining <= 0 { r.appendActivity(JobActivity{ @@ -408,17 +407,6 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo return detected } - // If the longest estimated job exceeds the remaining JobTypeMaxRuntime, - // create a new execution context with enough headroom instead of using - // jobCtx which would cancel too early. - execCtx := jobCtx - execCancel := context.CancelFunc(func() {}) - if maxEstimatedRuntime > 0 && maxEstimatedRuntime > remaining { - execCtx, execCancel = context.WithTimeout(context.Background(), maxEstimatedRuntime) - remaining = maxEstimatedRuntime - } - defer execCancel() - execPolicy := policy if execPolicy.ExecutionTimeout <= 0 { execPolicy.ExecutionTimeout = defaultScheduledExecutionTimeout @@ -427,10 +415,15 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo execPolicy.ExecutionTimeout = remaining } - successCount, errorCount, canceledCount := r.dispatchScheduledProposals(execCtx, jobType, filtered, clusterContext, execPolicy) + // jobCtx bounds how long this run keeps STARTING jobs; a started job + // runs on its own estimated-runtime deadline (see + // executeScheduledJobWithExecutor). Jobs still queued when the window + // closes are canceled and re-proposed by a later detection, so one large + // backlog cannot monopolise the lane for hours. + successCount, errorCount, canceledCount := r.dispatchScheduledProposals(jobCtx, jobType, filtered, clusterContext, execPolicy) status := "success" - if execCtx.Err() != nil { + if jobCtx.Err() != nil { status = "timeout" } else if errorCount > 0 || canceledCount > 0 { status = "error" @@ -818,6 +811,11 @@ func (r *Plugin) dispatchScheduledProposals( ctx = context.Background() } + // Volume management jobs mutate volume placement, so each one runs under + // the shared cluster admin lock. Taking it per job instead of around the + // whole dispatch lets manual shell operations interleave between jobs. + jobsRequireLock := LaneRequiresLock(JobTypeLane(jobType)) + jobQueue := make(chan *plugin_pb.JobSpec, len(proposals)) for index, proposal := range proposals { job := buildScheduledJobSpec(jobType, proposal, index) @@ -900,7 +898,42 @@ func (r *Plugin) dispatchScheduledProposals( break } + var releaseJobLock func() + if jobsRequireLock { + var lockErr error + releaseJobLock, lockErr = r.acquireAdminLock(fmt.Sprintf("plugin scheduler %s job %s", jobType, job.JobId)) + if lockErr != nil { + release() + statsMu.Lock() + errorCount++ + statsMu.Unlock() + r.appendActivity(JobActivity{ + JobID: job.JobId, + JobType: job.JobType, + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduled execution lock failed: %v", lockErr), + Stage: "failed", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + break + } + // The lock may have been held by an operator for a + // while; re-check the execution window. + if ctx.Err() != nil { + releaseJobLock() + release() + r.cancelQueuedJob(job, ctx.Err()) + statsMu.Lock() + canceledCount++ + statsMu.Unlock() + continue jobLoop + } + } + err := r.executeScheduledJobWithExecutor(ctx, executor, job, clusterContext, policy) + if releaseJobLock != nil { + releaseJobLock() + } release() if errors.Is(err, errExecutorAtCapacity) { r.trackExecutionQueued(job) @@ -1164,10 +1197,6 @@ func (r *Plugin) executeScheduledJobWithExecutor( return ctx.Err() } - parent := ctx - if parent == nil { - parent = context.Background() - } // Use the job's estimated runtime if provided and larger than the // default execution timeout. This lets handlers like vacuum scale // the timeout based on volume size so large volumes are not killed. @@ -1185,7 +1214,11 @@ func (r *Plugin) executeScheduledJobWithExecutor( } } } - execCtx, cancel := context.WithTimeout(parent, timeout) + // The attempt runs against its own deadline, detached from the + // dispatch window (ctx): a job started near the end of the window + // may run to completion; the window only stops further jobs from + // starting. + execCtx, cancel := context.WithTimeout(context.Background(), timeout) _, err := r.executeJobWithExecutor(execCtx, executor, job, clusterContext, int32(attempt)) cancel() if err == nil { diff --git a/weed/admin/plugin/plugin_scheduler_test.go b/weed/admin/plugin/plugin_scheduler_test.go index 99b374b77..371f6a0b0 100644 --- a/weed/admin/plugin/plugin_scheduler_test.go +++ b/weed/admin/plugin/plugin_scheduler_test.go @@ -664,6 +664,12 @@ func TestRunLaneSchedulerIterationLockBehavior(t *testing.T) { t.Fatalf("SaveJobTypeConfig: %v", err) } + // Make the job type due immediately so the iteration reaches + // detection; the lock is only taken around due work now. + pluginSvc.schedulerMu.Lock() + pluginSvc.nextDetectionAt[tt.jobType] = time.Now().UTC().Add(-time.Second) + pluginSvc.schedulerMu.Unlock() + ls := pluginSvc.lanes[tt.lane] pluginSvc.runLaneSchedulerIteration(ls) @@ -674,6 +680,61 @@ func TestRunLaneSchedulerIterationLockBehavior(t *testing.T) { } } +func TestDispatchScheduledProposalsLocksPerJob(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + jobType string + wantLocks int + }{ + {"DefaultLaneLocksEachJob", "volume_balance", 3}, + {"LifecycleLaneNeedsNoLock", "s3_lifecycle", 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + lm := &trackingLockManager{} + pluginSvc, err := New(Options{LockManager: lm}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + // The worker has capabilities but no connected session, so each + // job reserves capacity, fails to send, and moves on. + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: "worker-a", + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: tt.jobType, CanExecute: true, MaxExecutionConcurrency: 4}, + }, + }) + + policy := schedulerPolicy{ + ExecutionConcurrency: 1, + PerWorkerConcurrency: 1, + ExecutionTimeout: time.Second, + ExecutorReserveBackoff: time.Millisecond, + } + proposals := []*plugin_pb.JobProposal{ + {ProposalId: "p1", JobType: tt.jobType, DedupeKey: "k1"}, + {ProposalId: "p2", JobType: tt.jobType, DedupeKey: "k2"}, + {ProposalId: "p3", JobType: tt.jobType, DedupeKey: "k3"}, + } + + success, errors, canceled := pluginSvc.dispatchScheduledProposals( + context.Background(), tt.jobType, proposals, &plugin_pb.ClusterContext{}, policy) + if success != 0 || canceled != 0 || errors != 3 { + t.Fatalf("unexpected dispatch counts: success=%d errors=%d canceled=%d", success, errors, canceled) + } + if got := lm.count(); got != tt.wantLocks { + t.Fatalf("lock acquired %d times, want %d", got, tt.wantLocks) + } + }) + } +} + // ---------- lane-scoped prune ---------- func TestPruneSchedulerState_DefaultLaneKeepsForeignLanesAndPrunesOwnStale(t *testing.T) { diff --git a/weed/admin/plugin/scheduler_lane.go b/weed/admin/plugin/scheduler_lane.go index d90494136..abb69145c 100644 --- a/weed/admin/plugin/scheduler_lane.go +++ b/weed/admin/plugin/scheduler_lane.go @@ -6,8 +6,8 @@ import ( ) // SchedulerLane identifies an independent scheduling track. Each lane runs -// its own goroutine, maintains its own detection timing, and acquires its -// own admin lock so that workloads in different lanes never block each other. +// its own goroutine and maintains its own detection timing so that +// workloads in different lanes never block each other. type SchedulerLane string const ( @@ -39,17 +39,19 @@ var laneIdleSleep = map[SchedulerLane]time.Duration{ } // laneRequiresLock maps each lane to whether its job types must be -// serialised under a single admin lock. The default lane needs this -// because volume-management operations share global state. Other -// lanes run each job type independently. +// serialised and run under the shared cluster admin lock. The default +// lane needs this because volume-management operations share global +// state; the lock is taken around each detection and each job so manual +// shell operations can interleave. Other lanes run each job type +// independently without the lock. var laneRequiresLock = map[SchedulerLane]bool{ LaneDefault: true, LaneIceberg: false, LaneLifecycle: false, } -// LaneRequiresLock returns true if the given lane needs a single admin -// lock to serialise its job types. Unknown lanes default to true. +// LaneRequiresLock returns true if the given lane serialises its job types +// and runs them under the cluster admin lock. Unknown lanes default to true. func LaneRequiresLock(lane SchedulerLane) bool { if v, ok := laneRequiresLock[lane]; ok { return v diff --git a/weed/plugin/worker/volume_lookup.go b/weed/plugin/worker/volume_lookup.go new file mode 100644 index 000000000..b910a0bfb --- /dev/null +++ b/weed/plugin/worker/volume_lookup.go @@ -0,0 +1,75 @@ +package pluginworker + +import ( + "context" + "fmt" + "strconv" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "google.golang.org/grpc" +) + +// LookupVolumeLocations asks the masters for the current registered locations +// of a volume, bypassing any client-side caches. Execution handlers use it to +// re-check a proposal against the live topology, since a proposal can go +// stale between detection and execution. Returns the location URLs +// (host:port) the master reports; an empty slice means the volume is not +// registered anywhere. +func LookupVolumeLocations(ctx context.Context, masterAddresses []string, grpcDialOption grpc.DialOption, volumeID uint32) ([]string, error) { + if grpcDialOption == nil { + return nil, fmt.Errorf("grpc dial option is not configured") + } + if len(masterAddresses) == 0 { + return nil, fmt.Errorf("no master addresses provided in cluster context") + } + + vid := strconv.FormatUint(uint64(volumeID), 10) + var lastErr error + for _, address := range masterAddresses { + for _, candidate := range MasterAddressCandidates(address) { + if ctx.Err() != nil { + return nil, ctx.Err() + } + + dialCtx, cancelDial := context.WithTimeout(ctx, 5*time.Second) + conn, err := pb.GrpcDial(dialCtx, candidate, false, grpcDialOption) + cancelDial() + if err != nil { + lastErr = err + continue + } + + client := master_pb.NewSeaweedClient(conn) + callCtx, cancelCall := context.WithTimeout(ctx, 10*time.Second) + resp, callErr := client.LookupVolume(callCtx, &master_pb.LookupVolumeRequest{ + VolumeOrFileIds: []string{vid}, + }) + cancelCall() + _ = conn.Close() + if callErr != nil { + lastErr = callErr + continue + } + + for _, vidLocation := range resp.VolumeIdLocations { + if vidLocation.VolumeOrFileId != vid { + continue + } + locations := make([]string, 0, len(vidLocation.Locations)) + for _, location := range vidLocation.Locations { + locations = append(locations, location.Url) + } + return locations, nil + } + // The master answered but does not know the volume. + return nil, nil + } + } + + if lastErr == nil { + lastErr = fmt.Errorf("no valid master address candidate") + } + return nil, lastErr +} diff --git a/weed/worker/tasks/balance/plugin_handler.go b/weed/worker/tasks/balance/plugin_handler.go index d61220a04..813f770c2 100644 --- a/weed/worker/tasks/balance/plugin_handler.go +++ b/weed/worker/tasks/balance/plugin_handler.go @@ -723,19 +723,23 @@ func (h *VolumeBalanceHandler) executeSingleMove( return err } - if err := task.Execute(execCtx, params); err != nil { + execErr := h.checkMoveStillValid(execCtx, request.GetClusterContext().GetMasterGrpcAddresses(), params.VolumeId, params.Sources[0].Node, params.Targets[0].Node) + if execErr == nil { + execErr = task.Execute(execCtx, params) + } + if execErr != nil { _ = sender.SendProgress(&plugin_pb.JobProgressUpdate{ JobId: request.Job.JobId, JobType: request.Job.JobType, State: plugin_pb.JobState_JOB_STATE_FAILED, ProgressPercent: 100, Stage: "failed", - Message: err.Error(), + Message: execErr.Error(), Activities: []*plugin_pb.ActivityEvent{ - pluginworker.BuildExecutorActivity("failed", err.Error()), + pluginworker.BuildExecutorActivity("failed", execErr.Error()), }, }) - return err + return execErr } sourceNode := params.Sources[0].Node @@ -766,6 +770,41 @@ func (h *VolumeBalanceHandler) executeSingleMove( }) } +// checkMoveStillValid re-checks a planned move against the master's live +// view right before executing it. The admin lock is released between +// detection and execution, so the plan can go stale: the volume may have +// left the source, or the target may have gained a replica that the copy +// would overwrite and the source delete would then reduce to a single copy. +// Without master addresses (older admin) the check is skipped and the move +// relies on the task's own execution-time guards. +func (h *VolumeBalanceHandler) checkMoveStillValid(ctx context.Context, masterAddresses []string, volumeID uint32, sourceNode, targetNode string) error { + if len(masterAddresses) == 0 { + glog.Warningf("volume balance: no master addresses in cluster context, skipping pre-move check for volume %d", volumeID) + return nil + } + locations, err := pluginworker.LookupVolumeLocations(ctx, masterAddresses, h.grpcDialOption, volumeID) + if err != nil { + return fmt.Errorf("pre-move check for volume %d: %w", volumeID, err) + } + return checkMovePreconditions(locations, volumeID, sourceNode, targetNode) +} + +func checkMovePreconditions(locations []string, volumeID uint32, sourceNode, targetNode string) error { + sourceFound := false + for _, location := range locations { + switch strings.TrimSpace(location) { + case targetNode: + return fmt.Errorf("stale move: volume %d already has a replica on target %s", volumeID, targetNode) + case sourceNode: + sourceFound = true + } + } + if !sourceFound { + return fmt.Errorf("stale move: volume %d is no longer on source %s", volumeID, sourceNode) + } + return nil +} + // executeBatchMoves runs multiple volume moves concurrently within a single job. func (h *VolumeBalanceHandler) executeBatchMoves( ctx context.Context, @@ -881,6 +920,7 @@ func (h *VolumeBalanceHandler) executeBatchMoves( sem := make(chan struct{}, maxConcurrent) results := make(chan moveResult, totalMoves) + masterAddresses := request.GetClusterContext().GetMasterGrpcAddresses() for i, move := range moves { sem <- struct{}{} // acquire slot @@ -899,7 +939,10 @@ func (h *VolumeBalanceHandler) executeBatchMoves( }) moveParams := buildMoveTaskParams(m, bp) - err := task.Execute(batchCtx, moveParams) + err := h.checkMoveStillValid(batchCtx, masterAddresses, m.VolumeId, m.SourceNode, m.TargetNode) + if err == nil { + err = task.Execute(batchCtx, moveParams) + } results <- moveResult{ index: idx, volumeID: m.VolumeId, diff --git a/weed/worker/tasks/balance/plugin_handler_test.go b/weed/worker/tasks/balance/plugin_handler_test.go index 97b00a13e..a3c2294b1 100644 --- a/weed/worker/tasks/balance/plugin_handler_test.go +++ b/weed/worker/tasks/balance/plugin_handler_test.go @@ -795,3 +795,30 @@ func (r *recordingDetectionSender) SendActivity(event *plugin_pb.ActivityEvent) } return nil } + +func TestCheckMovePreconditions(t *testing.T) { + tests := []struct { + name string + locations []string + wantErr string + }{ + {"valid move", []string{"10.0.0.1:8080", "10.0.0.3:8080"}, ""}, + {"volume left the source", []string{"10.0.0.3:8080"}, "no longer on source"}, + {"volume gone entirely", nil, "no longer on source"}, + {"target already has a replica", []string{"10.0.0.1:8080", "10.0.0.2:8080"}, "already has a replica on target"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := checkMovePreconditions(tt.locations, 5, "10.0.0.1:8080", "10.0.0.2:8080") + if tt.wantErr == "" { + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + return + } + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("expected error containing %q, got %v", tt.wantErr, err) + } + }) + } +}