diff --git a/weed/admin/plugin/orphaned_execution.go b/weed/admin/plugin/orphaned_execution.go new file mode 100644 index 0000000000..04b01dba98 --- /dev/null +++ b/weed/admin/plugin/orphaned_execution.go @@ -0,0 +1,90 @@ +package plugin + +import ( + "fmt" + "strings" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" +) + +// orphanedExecutionMisses is how many itemized heartbeats in a row from a +// newer stream of the worker must leave a dispatched job out before the +// admin fails it. The second rides out a heartbeat the worker built while +// it was still picking the job back up. +const orphanedExecutionMisses = 2 + +// pendingExecutionState is an execution request waiting for its JobCompleted. +type pendingExecutionState struct { + completed chan *plugin_pb.JobCompleted + jobID string + jobType string + // session is the stream the request went out on. + session *streamSession + misses int + failed bool + // completionSent is set once a completion is handed to the dispatcher, + // which may already have taken it from the buffer. + completionSent bool +} + +// failExecutionsMissingFromHeartbeat fails the executions sent on an earlier +// stream of the heartbeat's worker that the worker no longer lists. A worker +// that restarted comes back under the same id without the old process's +// jobs, and nothing else ends such a dispatch before its execution timeout, +// so its slot stays reserved all that time. Jobs sent on the heartbeat's own +// stream are left alone, since a worker counts a job before it lists it, and +// so is everything while the heartbeat counts slots it does not itemize, +// since a missing id then proves nothing. +func (r *Plugin) failExecutionsMissingFromHeartbeat(session *streamSession, heartbeat *plugin_pb.WorkerHeartbeat) { + if session == nil || heartbeat == nil { + return + } + itemized := 0 + listed := make(map[string]struct{}, len(heartbeat.GetRunningWork())) + for _, work := range heartbeat.GetRunningWork() { + if work.GetKind() == plugin_pb.WorkKind_WORK_KIND_EXECUTION { + listed[strings.TrimSpace(work.GetWorkId())] = struct{}{} + itemized++ + } + } + itemizesAll := int(heartbeat.GetExecutionSlotsUsed()) <= itemized + + var orphaned []activeJobSnapshot + r.pendingExecutionMu.Lock() + for requestID, pending := range r.pendingExecution { + if pending.failed || pending.session == nil || pending.session == session || pending.session.workerID != session.workerID { + continue + } + if _, ok := listed[strings.TrimSpace(pending.jobID)]; ok { + pending.misses = 0 + continue + } + if !itemizesAll { + continue + } + if pending.completionSent { + continue + } + pending.misses++ + if pending.misses < orphanedExecutionMisses { + continue + } + pending.failed = true + orphaned = append(orphaned, activeJobSnapshot{ + jobID: pending.jobID, + jobType: pending.jobType, + workerID: session.workerID, + requestID: requestID, + }) + } + r.pendingExecutionMu.Unlock() + + for _, snap := range orphaned { + reason := fmt.Sprintf("worker %s re-registered without job %s", snap.workerID, snap.jobID) + glog.Warningf("Plugin failing %s request %s: %s", snap.jobType, snap.requestID, reason) + // off the stream's receive loop: the cancel may wait on a full buffer + go r.expireActiveJob(snap, reason, time.Now().UTC()) + } +} diff --git a/weed/admin/plugin/orphaned_execution_test.go b/weed/admin/plugin/orphaned_execution_test.go new file mode 100644 index 0000000000..0098472191 --- /dev/null +++ b/weed/admin/plugin/orphaned_execution_test.go @@ -0,0 +1,389 @@ +package plugin + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" +) + +const orphanTestJobType = "ec_vacuum" + +type orphanTestResult struct { + completed *plugin_pb.JobCompleted + err error +} + +func connectExecutor(p *Plugin, workerID string) *streamSession { + p.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + WorkerInstanceId: workerID + "-instance-" + time.Now().Format("150405.000000"), + Capabilities: []*plugin_pb.JobTypeCapability{{JobType: orphanTestJobType, CanExecute: true, MaxExecutionConcurrency: 2}}, + }) + session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})} + p.putSession(session) + return session +} + +// dispatchToExecutor runs one execution on session's worker and returns the +// request the worker was sent along with the dispatch's eventual result. +func dispatchToExecutor(t *testing.T, p *Plugin, session *streamSession, jobID string) (string, <-chan orphanTestResult) { + t.Helper() + executor, ok := p.registry.Get(session.workerID) + if !ok { + t.Fatalf("worker %s is not registered", session.workerID) + } + resultCh := make(chan orphanTestResult, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Hour) + defer cancel() + completed, err := p.executeJobWithExecutor(ctx, executor, &plugin_pb.JobSpec{JobId: jobID, JobType: orphanTestJobType}, &plugin_pb.ClusterContext{}, 1) + resultCh <- orphanTestResult{completed: completed, err: err} + }() + select { + case msg := <-session.outgoing: + if msg.GetExecuteJobRequest() == nil { + t.Fatalf("expected an execute_job_request, got %+v", msg) + } + return msg.RequestId, resultCh + case <-time.After(2 * time.Second): + t.Fatal("no execute request") + return "", nil + } +} + +func sendExecutionHeartbeat(p *Plugin, session *streamSession, slotsUsed int32, jobIDs ...string) { + heartbeat := &plugin_pb.WorkerHeartbeat{WorkerId: session.workerID, ExecutionSlotsUsed: slotsUsed, ExecutionSlotsTotal: 2} + for _, jobID := range jobIDs { + heartbeat.RunningWork = append(heartbeat.RunningWork, &plugin_pb.RunningWork{ + WorkId: jobID, + Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, + JobType: orphanTestJobType, + State: plugin_pb.JobState_JOB_STATE_RUNNING, + }) + } + p.handleWorkerMessage(session, &plugin_pb.WorkerToAdminMessage{Body: &plugin_pb.WorkerToAdminMessage_Heartbeat{Heartbeat: heartbeat}}) +} + +func expectDispatchStillPending(t *testing.T, resultCh <-chan orphanTestResult) { + t.Helper() + select { + case got := <-resultCh: + t.Fatalf("dispatch ended early: completed=%+v err=%v", got.completed, got.err) + case <-time.After(100 * time.Millisecond): + } +} + +// The worker process restarted after the job was dispatched: it comes back +// under the same id on a new stream and its heartbeats no longer list the +// job. The dispatch fails instead of holding its slot until the execution +// timeout. +func TestExecutionFailsWhenWorkerRestartsWithoutJob(t *testing.T) { + t.Parallel() + p, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer p.Shutdown() + + const jobID = "job-lost-in-restart" + requestID, resultCh := dispatchToExecutor(t, p, connectExecutor(p, "worker-a"), jobID) + restarted := connectExecutor(p, "worker-a") + + sendExecutionHeartbeat(p, restarted, 0) + expectDispatchStillPending(t, resultCh) + sendExecutionHeartbeat(p, restarted, 0) + + select { + case got := <-resultCh: + if got.err == nil || got.completed == nil || got.completed.Success { + t.Fatalf("expected a failed completion, got completed=%+v err=%v", got.completed, got.err) + } + if want := "worker worker-a re-registered without job " + jobID; !strings.Contains(got.err.Error(), want) { + t.Fatalf("error %q does not say %q", got.err, want) + } + case <-time.After(5 * time.Second): + t.Fatal("dispatch was not failed after its worker restarted without the job") + } + + tracked, ok := p.GetTrackedJob(jobID) + if !ok || tracked == nil { + t.Fatalf("job %s is not tracked", jobID) + } + if tracked.State != StateFailed || tracked.RequestID != requestID { + t.Fatalf("expected request %s failed, got state=%s request=%s", requestID, tracked.State, tracked.RequestID) + } +} + +// A worker that kept the job running across a reconnect lists it from the +// new stream; the dispatch keeps waiting and takes the completion from there. +func TestExecutionKeptWhenWorkerReconnectsWithJob(t *testing.T) { + t.Parallel() + p, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer p.Shutdown() + + const jobID = "job-kept-across-reconnect" + requestID, resultCh := dispatchToExecutor(t, p, connectExecutor(p, "worker-a"), jobID) + reconnected := connectExecutor(p, "worker-a") + + for i := 0; i < orphanedExecutionMisses+1; i++ { + sendExecutionHeartbeat(p, reconnected, 1, jobID) + } + expectDispatchStillPending(t, resultCh) + + p.handleWorkerMessage(reconnected, &plugin_pb.WorkerToAdminMessage{Body: &plugin_pb.WorkerToAdminMessage_JobCompleted{JobCompleted: &plugin_pb.JobCompleted{ + RequestId: requestID, + JobId: jobID, + JobType: orphanTestJobType, + Success: true, + }}}) + select { + case got := <-resultCh: + if got.err != nil || got.completed == nil || !got.completed.Success { + t.Fatalf("expected success, got completed=%+v err=%v", got.completed, got.err) + } + case <-time.After(5 * time.Second): + t.Fatal("completion from the reconnected stream did not reach the dispatch") + } +} + +// Work ids are compared trimmed, so a padded job id still matches its listing. +func TestExecutionKeptWhenReconnectedWorkerListsPaddedJobID(t *testing.T) { + t.Parallel() + p, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer p.Shutdown() + + const jobID = " job-padded " + _, resultCh := dispatchToExecutor(t, p, connectExecutor(p, "worker-a"), jobID) + reconnected := connectExecutor(p, "worker-a") + + for i := 0; i < orphanedExecutionMisses+1; i++ { + sendExecutionHeartbeat(p, reconnected, 1, jobID) + } + expectDispatchStillPending(t, resultCh) +} + +// The worker counts a job before it lists it, and heartbeats can lag; the +// stream the job went out on is never judged by what it has not listed yet. +func TestExecutionKeptWhileDispatchSessionHasNotListedJob(t *testing.T) { + t.Parallel() + p, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer p.Shutdown() + + session := connectExecutor(p, "worker-a") + _, resultCh := dispatchToExecutor(t, p, session, "job-just-dispatched") + + for i := 0; i < orphanedExecutionMisses+1; i++ { + sendExecutionHeartbeat(p, session, 0) + sendExecutionHeartbeat(p, session, 1) + } + expectDispatchStillPending(t, resultCh) +} + +// A heartbeat that counts busy slots without itemizing them (the Rust worker +// reports only the count) cannot show a job is gone. +func TestExecutionKeptWhenHeartbeatDoesNotItemize(t *testing.T) { + t.Parallel() + p, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer p.Shutdown() + + _, resultCh := dispatchToExecutor(t, p, connectExecutor(p, "worker-a"), "job-count-only") + reconnected := connectExecutor(p, "worker-a") + + for i := 0; i < orphanedExecutionMisses+1; i++ { + sendExecutionHeartbeat(p, reconnected, 1) + sendExecutionHeartbeat(p, reconnected, 2, "some-other-job") + } + expectDispatchStillPending(t, resultCh) +} + +// A listing between two omissions shows the worker still runs the job, even +// when that heartbeat also counts a slot it does not itemize. +func TestExecutionKeptWhenReplacementStreamListsJobBetweenOmissions(t *testing.T) { + t.Parallel() + for name, listing := range map[string]int32{"itemized": 1, "partly itemized": 2} { + t.Run(name, func(t *testing.T) { + p, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer p.Shutdown() + + const jobID = "job-listed-between-omissions" + _, resultCh := dispatchToExecutor(t, p, connectExecutor(p, "worker-a"), jobID) + reconnected := connectExecutor(p, "worker-a") + + sendExecutionHeartbeat(p, reconnected, 0) + sendExecutionHeartbeat(p, reconnected, listing, jobID) + sendExecutionHeartbeat(p, reconnected, 0) + expectDispatchStillPending(t, resultCh) + }) + } +} + +func TestExecutionNotTrackedWhenWorkerNotConnected(t *testing.T) { + t.Parallel() + p, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer p.Shutdown() + + p.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: "worker-a", + Capabilities: []*plugin_pb.JobTypeCapability{{JobType: orphanTestJobType, CanExecute: true, MaxExecutionConcurrency: 2}}, + }) + executor, ok := p.registry.Get("worker-a") + if !ok { + t.Fatal("worker-a is not registered") + } + const jobID = "job-without-session" + if _, err := p.executeJobWithExecutor(context.Background(), executor, &plugin_pb.JobSpec{JobId: jobID, JobType: orphanTestJobType}, &plugin_pb.ClusterContext{}, 1); err == nil { + t.Fatal("expected an error for a worker without a stream") + } + if tracked, ok := p.GetTrackedJob(jobID); ok && tracked != nil { + t.Fatalf("job %s tracked as %s for a dispatch that never went out", jobID, tracked.State) + } +} + +// The captured stream closed before the request went out, as when the worker +// reconnects mid-dispatch: no report will ever close the attempt. +func TestExecutionFailedWhenSendToSessionFails(t *testing.T) { + t.Parallel() + p, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer p.Shutdown() + + p.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: "worker-a", + Capabilities: []*plugin_pb.JobTypeCapability{{JobType: orphanTestJobType, CanExecute: true, MaxExecutionConcurrency: 2}}, + }) + closed := &streamSession{workerID: "worker-a", outgoing: make(chan *plugin_pb.AdminToWorkerMessage), done: make(chan struct{})} + closed.close() + p.putSession(closed) + executor, ok := p.registry.Get("worker-a") + if !ok { + t.Fatal("worker-a is not registered") + } + const jobID = "job-send-failed" + if _, err := p.executeJobWithExecutor(context.Background(), executor, &plugin_pb.JobSpec{JobId: jobID, JobType: orphanTestJobType}, &plugin_pb.ClusterContext{}, 1); err == nil { + t.Fatal("expected an error for a closed stream") + } + tracked, ok := p.GetTrackedJob(jobID) + if !ok || tracked == nil || tracked.State != StateFailed { + t.Fatalf("expected job %s failed, got %+v", jobID, tracked) + } +} + +// Shutdown unregisters every pending dispatch before a blocked send returns +// its error; the attempt is still recorded as failed. +func TestExecutionFailedWhenSendFailsAfterShutdownUnregisters(t *testing.T) { + t.Parallel() + p, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer p.Shutdown() + + p.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: "worker-a", + Capabilities: []*plugin_pb.JobTypeCapability{{JobType: orphanTestJobType, CanExecute: true, MaxExecutionConcurrency: 2}}, + }) + blocked := &streamSession{workerID: "worker-a", outgoing: make(chan *plugin_pb.AdminToWorkerMessage), done: make(chan struct{})} + p.putSession(blocked) + executor, ok := p.registry.Get("worker-a") + if !ok { + t.Fatal("worker-a is not registered") + } + const jobID = "job-send-after-shutdown" + errCh := make(chan error, 1) + go func() { + _, err := p.executeJobWithExecutor(context.Background(), executor, &plugin_pb.JobSpec{JobId: jobID, JobType: orphanTestJobType}, &plugin_pb.ClusterContext{}, 1) + errCh <- err + }() + + deadline := time.Now().Add(2 * time.Second) + for unregistered := false; !unregistered; { + if time.Now().After(deadline) { + t.Fatal("dispatch never registered") + } + p.pendingExecutionMu.Lock() + for requestID, pending := range p.pendingExecution { + if pending.jobID == jobID { + close(pending.completed) + delete(p.pendingExecution, requestID) + unregistered = true + } + } + p.pendingExecutionMu.Unlock() + time.Sleep(time.Millisecond) + } + blocked.close() + + select { + case err := <-errCh: + if err == nil { + t.Fatal("expected an error for a closed stream") + } + case <-time.After(2 * time.Second): + t.Fatal("dispatch did not return") + } + tracked, ok := p.GetTrackedJob(jobID) + if !ok || tracked == nil || tracked.State != StateFailed { + t.Fatalf("expected job %s failed, got %+v", jobID, tracked) + } +} + +// The dispatcher can take the completion from the buffer before it +// unregisters the attempt; heartbeats in that gap must not count misses. +func TestExecutionNotOrphanedAfterDispatcherTookCompletion(t *testing.T) { + t.Parallel() + p, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer p.Shutdown() + + const jobID = "job-completion-taken" + const requestID = "exec-completion-taken" + original := connectExecutor(p, "worker-a") + pending := &pendingExecutionState{ + completed: make(chan *plugin_pb.JobCompleted, 1), + jobID: jobID, + jobType: orphanTestJobType, + session: original, + } + p.pendingExecutionMu.Lock() + p.pendingExecution[requestID] = pending + p.pendingExecutionMu.Unlock() + + p.safeSendJobCompleted(requestID, &plugin_pb.JobCompleted{RequestId: requestID, JobId: jobID, JobType: orphanTestJobType, Success: true}) + <-pending.completed + + reconnected := connectExecutor(p, "worker-a") + for i := 0; i < orphanedExecutionMisses+1; i++ { + sendExecutionHeartbeat(p, reconnected, 0) + } + p.pendingExecutionMu.Lock() + failed := pending.failed + p.pendingExecutionMu.Unlock() + if failed { + t.Fatal("attempt whose completion was already delivered was failed as orphaned") + } +} diff --git a/weed/admin/plugin/plugin.go b/weed/admin/plugin/plugin.go index 13db50ab62..b25eced67d 100644 --- a/weed/admin/plugin/plugin.go +++ b/weed/admin/plugin/plugin.go @@ -106,7 +106,7 @@ type Plugin struct { pendingDetection map[string]*pendingDetectionState pendingExecutionMu sync.Mutex - pendingExecution map[string]chan *plugin_pb.JobCompleted + pendingExecution map[string]*pendingExecutionState pendingPreviewMu sync.Mutex pendingPreview map[string]chan *plugin_pb.ObjectPreviewResponse @@ -216,7 +216,7 @@ func New(options Options) (*Plugin, error) { sessions: make(map[string]*streamSession), pendingSchema: make(map[string]chan *plugin_pb.ConfigSchemaResponse), pendingDetection: make(map[string]*pendingDetectionState), - pendingExecution: make(map[string]chan *plugin_pb.JobCompleted), + pendingExecution: make(map[string]*pendingExecutionState), pendingPreview: make(map[string]chan *plugin_pb.ObjectPreviewResponse), observations: NewObservationStore(), nextDetectionAt: make(map[string]time.Time), @@ -289,8 +289,8 @@ func (r *Plugin) Shutdown() { r.pendingDetectionMu.Unlock() r.pendingExecutionMu.Lock() - for requestID, ch := range r.pendingExecution { - close(ch) + for requestID, pending := range r.pendingExecution { + close(pending.completed) delete(r.pendingExecution, requestID) } r.pendingExecutionMu.Unlock() @@ -815,9 +815,24 @@ func (r *Plugin) executeJobWithExecutor( // the single active worker, not 1/N of it. clusterContext = r.decorateClusterContextForJob(clusterContext, job.JobType, adminConfigValues, int(adminRuntime.GetMaxJobsPerDetection())) + // The request goes out on this stream and no other, so a newer stream + // of the worker can tell it never received it. + session, ok := r.getSession(executor.WorkerID) + if !ok { + return nil, fmt.Errorf("worker %s is not connected", executor.WorkerID) + } + + r.trackExecutionStart(requestID, executor.WorkerID, job, attempt) + completedCh := make(chan *plugin_pb.JobCompleted, 1) + pending := &pendingExecutionState{ + completed: completedCh, + jobID: job.JobId, + jobType: job.JobType, + session: session, + } r.pendingExecutionMu.Lock() - r.pendingExecution[requestID] = completedCh + r.pendingExecution[requestID] = pending r.pendingExecutionMu.Unlock() defer func() { r.pendingExecutionMu.Lock() @@ -825,8 +840,6 @@ func (r *Plugin) executeJobWithExecutor( r.pendingExecutionMu.Unlock() }() - r.trackExecutionStart(requestID, executor.WorkerID, job, attempt) - message := &plugin_pb.AdminToWorkerMessage{ RequestId: requestID, SentAt: timestamppb.Now(), @@ -843,7 +856,23 @@ func (r *Plugin) executeJobWithExecutor( }, } - if err := r.sendToWorker(executor.WorkerID, message); err != nil { + if err := r.sendToSession(session, message); err != nil { + // The worker never got the request, so no report will close the + // attempt unless an expiry already claimed it. Unregister first: + // nobody reads completedCh any more. Shutdown may have done so. + r.pendingExecutionMu.Lock() + delete(r.pendingExecution, requestID) + claimed := pending.failed + r.pendingExecutionMu.Unlock() + if !claimed { + r.handleJobCompleted(&plugin_pb.JobCompleted{ + JobId: job.JobId, + JobType: job.JobType, + RequestId: requestID, + ErrorMessage: err.Error(), + CompletedAt: timestamppb.Now(), + }) + } return nil, err } @@ -1157,6 +1186,7 @@ func (r *Plugin) handleWorkerMessage(session *streamSession, message *plugin_pb. return } r.resendUndeliveredCancels(workerID, body.Heartbeat) + r.failExecutionsMissingFromHeartbeat(session, body.Heartbeat) case *plugin_pb.WorkerToAdminMessage_ConfigSchemaResponse: r.handleConfigSchemaResponse(body.ConfigSchemaResponse) case *plugin_pb.WorkerToAdminMessage_DetectionProposals: @@ -1501,8 +1531,12 @@ func (r *Plugin) handleJobCompleted(completed *plugin_pb.JobCompleted) { } func (r *Plugin) safeSendJobCompleted(requestID string, completed *plugin_pb.JobCompleted) { + var ch chan *plugin_pb.JobCompleted r.pendingExecutionMu.Lock() - ch := r.pendingExecution[requestID] + if pending := r.pendingExecution[requestID]; pending != nil { + ch = pending.completed + pending.completionSent = true + } r.pendingExecutionMu.Unlock() safeSendCh(ch, completed, r.shutdownCh) }