From 30f0364256c8841bc2015ca2b26e169d46c69483 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Nov 2026 12:28:39 -0800 Subject: [PATCH] admin: fail an in-flight detection when its worker disconnects (#12376) * admin: keep a reconnected plugin worker registered when its old stream unwinds cleanupSession removed whatever session was registered under the worker ID, so a worker that reconnected while its previous stream was still unwinding lost the new session and its registry entry to the old stream's deferred cleanup. Tie cleanup to the stream that is ending and publish the new stream before the registry upsert so a lingering predecessor sees itself superseded. * admin: fail an in-flight detection when its worker disconnects RunDetectionWithReport waited only on the caller's context and the DetectionComplete channel, so a detection whose worker stream ended mid-request could only finish by timeout. On a customer cluster a planned restart of ec-vacuum-1 during an ec_vacuum detection left the default lane in 'detecting' with every execution slot idle until the 3600s detection timeout, blocking every job type in that lane. Bind the request to the stream it was sent on and fail the detection as soon as that stream closes, whether by disconnect or by a reconnect that replaced the session. A liveness tick also fails it once the registry no longer considers the worker available (heartbeat stale), sending a cancel in that case because the stream may still be open. The detector lease is dropped so the next pick starts fresh, and a 'disconnected' detector activity is recorded. WorkerDisconnectForTest now goes through the real session cleanup. * admin: retry a scheduled detection once when its detector drops When a detection fails because its worker disconnected, pick a detector again and retry once within the same iteration while the job type's runtime budget allows, instead of giving the whole cycle up. Any other failure, or a failed retry, ends the run as before and the lane moves on; the next cycle re-detects anyway. * admin: register a plugin worker's hello and stream as one step WorkerStream published the stream under sessionsMu but ran the registry upsert after releasing it, so two concurrent streams for one worker ID could publish their hellos in the opposite order from their sessions: the registry then described the older hello's capabilities while only the newer stream was sendable, and detector picks routed on them. registerSession upserts the hello and installs the stream under one sessionsMu critical section, and a hello or heartbeat arriving on a stream that is no longer the registered one is dropped instead of rewriting the live stream's entry. * admin: steer a detection retry off the stream that dropped The retry picked while the lost worker could still be in the registry: cleanupSession closed done before taking the registry entry down, and a stream closed by a reconnect keeps its entry on purpose. The lost worker ranks first on free slots, so the single retry went back to it and failed on send while a healthy worker sat idle. The disconnect error now carries the stream it came from and the retry picks around it; pickDetector only returns workers whose current stream is open, excluding registry entries whose stream is closed; and cleanupSession removes the registry entry before closing done, so a waiter woken by the closure never sees the dead worker as a candidate. --- weed/admin/plugin/plugin.go | 299 ++++++++++--- .../plugin_detection_disconnect_test.go | 406 ++++++++++++++++++ weed/admin/plugin/plugin_scheduler.go | 36 +- weed/admin/plugin/plugin_scheduler_test.go | 7 +- weed/admin/plugin/preview_test.go | 6 +- weed/admin/plugin/registry.go | 22 +- 6 files changed, 692 insertions(+), 84 deletions(-) create mode 100644 weed/admin/plugin/plugin_detection_disconnect_test.go diff --git a/weed/admin/plugin/plugin.go b/weed/admin/plugin/plugin.go index 9082cdc4b5..346631fc4f 100644 --- a/weed/admin/plugin/plugin.go +++ b/weed/admin/plugin/plugin.go @@ -27,6 +27,10 @@ const ( defaultReconnectDelay = 5 defaultPendingSchemaBuffer = 1 adminScriptJobType = "admin_script" + // How often an in-flight detection re-checks that its worker is still + // registered and heartbeating; a session that went stale fails the + // detection the same way a closed stream does. + defaultDetectionLivenessInterval = 30 * time.Second ) type Options struct { @@ -59,6 +63,8 @@ type Plugin struct { outgoingBuffer int sendTimeout time.Duration + detectionLivenessInterval time.Duration + schedulerTick time.Duration clusterContextProvider func(context.Context) (*plugin_pb.ClusterContext, error) configDefaultsProvider func(config *plugin_pb.PersistedJobTypeConfig) *plugin_pb.PersistedJobTypeConfig @@ -140,6 +146,23 @@ type pendingDetectionState struct { workerID string } +// detectionWorkerDisconnectedError reports a detection whose worker session +// ended (stream closed, replaced by a reconnect, or heartbeat went stale) +// before it answered. +type detectionWorkerDisconnectedError struct { + WorkerID string + Reason string + session *streamSession +} + +func (e *detectionWorkerDisconnectedError) Error() string { + msg := fmt.Sprintf("detection worker %s disconnected", e.WorkerID) + if e.Reason != "" { + msg += ": " + e.Reason + } + return msg +} + // DetectionReport captures one detection run including request metadata. type DetectionReport struct { RequestID string @@ -178,6 +201,7 @@ func New(options Options) (*Plugin, error) { registry: NewRegistry(), outgoingBuffer: bufferSize, sendTimeout: sendTimeout, + detectionLivenessInterval: defaultDetectionLivenessInterval, schedulerTick: schedulerTick, clusterContextProvider: options.ClusterContextProvider, configDefaultsProvider: options.ConfigDefaultsProvider, @@ -290,15 +314,13 @@ func (r *Plugin) WorkerStream(stream plugin_pb.PluginControlService_WorkerStream } workerID := hello.WorkerId - r.registry.UpsertFromHello(hello) - session := &streamSession{ workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, r.outgoingBuffer), done: make(chan struct{}), } - r.putSession(session) - defer r.cleanupSession(workerID) + r.registerSession(session, hello) + defer r.cleanupSession(session) glog.V(0).Infof("Plugin worker connected: %s (%s)", workerID, hello.Address) @@ -309,7 +331,7 @@ func (r *Plugin) WorkerStream(stream plugin_pb.PluginControlService_WorkerStream sendErrCh <- r.sendLoop(stream.Context(), stream, session) }() - if err := r.sendAdminHello(workerID); err != nil { + if err := r.sendAdminHello(session); err != nil { glog.Warningf("failed to send plugin admin hello to %s: %v", workerID, err) } go r.prefetchDescriptorsFromHello(hello) @@ -324,7 +346,7 @@ func (r *Plugin) WorkerStream(stream plugin_pb.PluginControlService_WorkerStream recvErrCh <- recvErr return } - r.handleWorkerMessage(workerID, message) + r.handleWorkerMessage(session, message) } }() @@ -462,11 +484,23 @@ func (r *Plugin) RunDetectionWithReport( jobType string, clusterContext *plugin_pb.ClusterContext, maxResults int32, +) (*DetectionReport, error) { + return r.runDetectionWithReport(ctx, jobType, clusterContext, maxResults, nil) +} + +// runDetectionWithReport is RunDetectionWithReport with a stream the pick +// must avoid: the one whose closure a retry is recovering from. +func (r *Plugin) runDetectionWithReport( + ctx context.Context, + jobType string, + clusterContext *plugin_pb.ClusterContext, + maxResults int32, + avoid *streamSession, ) (*DetectionReport, error) { releaseGate := r.acquireDetectionExecutionGate(jobType, false) defer releaseGate() - detector, err := r.pickDetector(jobType) + detector, err := r.pickDetector(jobType, avoid) if err != nil { return nil, err } @@ -530,7 +564,16 @@ func (r *Plugin) RunDetectionWithReport( }, } - if err := r.sendToWorker(detector.WorkerID, message); err != nil { + // The request is bound to the stream it was sent on: if that stream ends + // (disconnect or a reconnect with a new session) the answer can never + // arrive, so the wait below fails instead of running out the timeout. + session, connected := r.getSession(detector.WorkerID) + if !connected { + err = fmt.Errorf("worker %s is not connected", detector.WorkerID) + } else { + err = r.sendToSession(session, message) + } + if err != nil { r.clearDetectorLease(jobType, detector.WorkerID) r.appendActivity(JobActivity{ JobType: jobType, @@ -544,49 +587,89 @@ func (r *Plugin) RunDetectionWithReport( return nil, err } - select { - case <-ctx.Done(): - r.sendCancel(detector.WorkerID, requestID, plugin_pb.WorkKind_WORK_KIND_DETECTION, ctx.Err()) - r.appendActivity(JobActivity{ - JobType: jobType, - RequestID: requestID, - WorkerID: detector.WorkerID, - Source: "detector", - Stage: "canceled", - Message: "detection canceled", - OccurredAt: timeToPtr(time.Now().UTC()), - }) + emptyReport := func() *DetectionReport { return &DetectionReport{ RequestID: requestID, JobType: jobType, WorkerID: detector.WorkerID, - }, ctx.Err() - case complete, ok := <-state.complete: - if !ok { - return &DetectionReport{ + } + } + + livenessInterval := r.detectionLivenessInterval + if livenessInterval <= 0 { + livenessInterval = defaultDetectionLivenessInterval + } + liveness := time.NewTicker(livenessInterval) + defer liveness.Stop() + + for { + select { + case <-ctx.Done(): + r.sendCancel(detector.WorkerID, requestID, plugin_pb.WorkKind_WORK_KIND_DETECTION, ctx.Err()) + r.appendActivity(JobActivity{ + JobType: jobType, + RequestID: requestID, + WorkerID: detector.WorkerID, + Source: "detector", + Stage: "canceled", + Message: "detection canceled", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + return emptyReport(), ctx.Err() + case <-session.done: + return emptyReport(), r.failDetectionWorkerLost(jobType, requestID, session, "", false) + case <-liveness.C: + if r.isWorkerAvailable(detector.WorkerID) { + continue + } + // The stream may still be open, so the worker is told to stop. + return emptyReport(), r.failDetectionWorkerLost(jobType, requestID, session, "heartbeat stale", true) + case complete, ok := <-state.complete: + if !ok { + return emptyReport(), fmt.Errorf("detection request %s interrupted", requestID) + } + proposals := cloneJobProposals(state.proposals) + report := &DetectionReport{ RequestID: requestID, JobType: jobType, WorkerID: detector.WorkerID, - }, fmt.Errorf("detection request %s interrupted", requestID) + Proposals: proposals, + Complete: complete, + } + if complete == nil { + return report, fmt.Errorf("detection request %s returned no completion state", requestID) + } + if !complete.Success { + return report, fmt.Errorf("detection failed for %s: %s", jobType, complete.ErrorMessage) + } + return report, nil } - proposals := cloneJobProposals(state.proposals) - report := &DetectionReport{ - RequestID: requestID, - JobType: jobType, - WorkerID: detector.WorkerID, - Proposals: proposals, - Complete: complete, - } - if complete == nil { - return report, fmt.Errorf("detection request %s returned no completion state", requestID) - } - if !complete.Success { - return report, fmt.Errorf("detection failed for %s: %s", jobType, complete.ErrorMessage) - } - return report, nil } } +// failDetectionWorkerLost closes out a detection whose worker session ended +// before it answered. The detector lease is dropped so the next pick starts +// fresh; the worker is only told to cancel when its stream may still be open. +func (r *Plugin) failDetectionWorkerLost(jobType, requestID string, session *streamSession, reason string, cancelOnWorker bool) error { + workerID := session.workerID + err := &detectionWorkerDisconnectedError{WorkerID: workerID, Reason: reason, session: session} + r.clearDetectorLease(jobType, workerID) + if cancelOnWorker { + r.sendCancel(workerID, requestID, plugin_pb.WorkKind_WORK_KIND_DETECTION, err) + } + glog.Warningf("Plugin detection %s for %s failed: %v", requestID, jobType, err) + r.appendActivity(JobActivity{ + JobType: jobType, + RequestID: requestID, + WorkerID: workerID, + Source: "detector", + Stage: "disconnected", + Message: err.Error(), + OccurredAt: timeToPtr(time.Now().UTC()), + }) + return err +} + // RunDetection requests one detector worker to produce job proposals for a job type. func (r *Plugin) RunDetection( ctx context.Context, @@ -834,17 +917,20 @@ func (r *Plugin) FilterProposalsWithActiveJobs(jobType string, proposals []*plug } func (r *Plugin) PickDetectorWorker(jobType string) (*WorkerSession, error) { - return r.pickDetector(jobType) + return r.pickDetector(jobType, nil) } func (r *Plugin) PickExecutorWorker(jobType string) (*WorkerSession, error) { return r.registry.PickExecutor(jobType) } -func (r *Plugin) pickDetector(jobType string) (*WorkerSession, error) { +// pickDetector picks a detector that can actually be sent to: registered, +// with an open stream, and not the stream in avoid. The registry alone is +// not enough because a stream can close before its entry is removed. +func (r *Plugin) pickDetector(jobType string, avoid *streamSession) (*WorkerSession, error) { leasedWorkerID := r.getDetectorLease(jobType) if leasedWorkerID != "" { - if worker, ok := r.registry.Get(leasedWorkerID); ok { + if worker, ok := r.registry.Get(leasedWorkerID); ok && r.sessionIsOpen(leasedWorkerID, avoid) { if capability := worker.Capabilities[jobType]; capability != nil && capability.CanDetect { return worker, nil } @@ -852,13 +938,31 @@ func (r *Plugin) pickDetector(jobType string) (*WorkerSession, error) { r.clearDetectorLease(jobType, leasedWorkerID) } - detector, err := r.registry.PickDetector(jobType) - if err != nil { - return nil, err + var excluded []string + for { + detector, err := r.registry.PickDetector(jobType, excluded...) + if err != nil { + return nil, err + } + if r.sessionIsOpen(detector.WorkerID, avoid) { + r.setDetectorLease(jobType, detector.WorkerID) + return detector, nil + } + excluded = append(excluded, detector.WorkerID) } +} - r.setDetectorLease(jobType, detector.WorkerID) - return detector, nil +func (r *Plugin) sessionIsOpen(workerID string, avoid *streamSession) bool { + session, ok := r.getSession(workerID) + if !ok || session == avoid { + return false + } + select { + case <-session.done: + return false + default: + return true + } } func (r *Plugin) getDetectorLease(jobType string) string { @@ -920,7 +1024,7 @@ func (r *Plugin) sendCancel(workerID, targetID string, kind plugin_pb.WorkKind, } } -func (r *Plugin) sendAdminHello(workerID string) error { +func (r *Plugin) sendAdminHello(session *streamSession) error { msg := &plugin_pb.AdminToWorkerMessage{ RequestId: "", SentAt: timestamppb.Now(), @@ -933,7 +1037,7 @@ func (r *Plugin) sendAdminHello(workerID string) error { }, }, } - return r.sendToWorker(workerID, msg) + return r.sendToSession(session, msg) } func (r *Plugin) sendLoop( @@ -960,37 +1064,50 @@ func (r *Plugin) sendLoop( } } -func (r *Plugin) sendToWorker(workerID string, message *plugin_pb.AdminToWorkerMessage) error { +func (r *Plugin) getSession(workerID string) (*streamSession, bool) { r.sessionsMu.RLock() session, ok := r.sessions[workerID] r.sessionsMu.RUnlock() + return session, ok +} + +func (r *Plugin) sendToWorker(workerID string, message *plugin_pb.AdminToWorkerMessage) error { + session, ok := r.getSession(workerID) if !ok { return fmt.Errorf("worker %s is not connected", workerID) } + return r.sendToSession(session, message) +} +func (r *Plugin) sendToSession(session *streamSession, message *plugin_pb.AdminToWorkerMessage) error { select { case <-r.shutdownCh: return fmt.Errorf("plugin is shutting down") case <-session.done: - return fmt.Errorf("worker %s session is closed", workerID) + return fmt.Errorf("worker %s session is closed", session.workerID) case session.outgoing <- message: return nil case <-time.After(r.sendTimeout): - return fmt.Errorf("timed out sending message to worker %s", workerID) + return fmt.Errorf("timed out sending message to worker %s", session.workerID) } } -func (r *Plugin) handleWorkerMessage(workerID string, message *plugin_pb.WorkerToAdminMessage) { +func (r *Plugin) handleWorkerMessage(session *streamSession, message *plugin_pb.WorkerToAdminMessage) { if message == nil { return } + workerID := session.workerID switch body := message.Body.(type) { case *plugin_pb.WorkerToAdminMessage_Hello: - r.registry.UpsertFromHello(body.Hello) + if !r.updateRegistryFromSession(session, func() { r.registry.UpsertFromHello(body.Hello) }) { + return + } r.wakeScheduler() case *plugin_pb.WorkerToAdminMessage_Heartbeat: - r.registry.UpdateHeartbeat(workerID, body.Heartbeat) + if !r.updateRegistryFromSession(session, func() { r.registry.UpdateHeartbeat(workerID, body.Heartbeat) }) { + return + } case *plugin_pb.WorkerToAdminMessage_ConfigSchemaResponse: r.handleConfigSchemaResponse(body.ConfigSchemaResponse) case *plugin_pb.WorkerToAdminMessage_DetectionProposals: @@ -1335,27 +1452,73 @@ func (r *Plugin) safeSendJobCompleted(requestID string, completed *plugin_pb.Job safeSendCh(ch, completed, r.shutdownCh) } +// The registry entry for a worker always describes the stream that is +// currently sendable for it: every registry write for a worker happens under +// sessionsMu, together with or checked against r.sessions[workerID]. + func (r *Plugin) putSession(session *streamSession) { r.sessionsMu.Lock() defer r.sessionsMu.Unlock() - - if old, exists := r.sessions[session.workerID]; exists { - old.close() - } - r.sessions[session.workerID] = session + r.publishSessionLocked(session) } -func (r *Plugin) cleanupSession(workerID string) { - r.registry.Remove(workerID) +// registerSession publishes a stream and its hello as one step so a second +// stream for the same worker cannot leave the registry describing the older +// hello while only the newer stream is sendable. +func (r *Plugin) registerSession(session *streamSession, hello *plugin_pb.WorkerHello) { + r.sessionsMu.Lock() + defer r.sessionsMu.Unlock() + r.registry.UpsertFromHello(hello) + r.publishSessionLocked(session) +} + +// publishSessionLocked installs the stream and closes the one it replaces. +// Anything waiting on the old stream's done wakes to a registry and session +// map that already describe the replacement. +func (r *Plugin) publishSessionLocked(session *streamSession) { + old, exists := r.sessions[session.workerID] + r.sessions[session.workerID] = session + if exists && old != session { + old.close() + } +} + +// updateRegistryFromSession applies a registry write for a stream's worker +// only while that stream is the registered one; a superseded stream's late +// hello or heartbeat must not describe the live stream. +func (r *Plugin) updateRegistryFromSession(session *streamSession, update func()) bool { + r.sessionsMu.RLock() + defer r.sessionsMu.RUnlock() + if current, ok := r.sessions[session.workerID]; !ok || current != session { + return false + } + update() + return true +} + +// cleanupSession ends one stream. Only the stream currently registered for +// the worker takes the registry entry down with it; a stream that a reconnect +// already replaced must not evict its successor. The registry entry goes +// before done closes so nothing woken by the closure can still pick it. +func (r *Plugin) cleanupSession(session *streamSession) { + workerID := session.workerID r.sessionsMu.Lock() - session, exists := r.sessions[workerID] - if exists { + current, exists := r.sessions[workerID] + superseded := exists && current != session + if exists && !superseded { delete(r.sessions, workerID) - session.close() } + if !superseded { + r.registry.Remove(workerID) + } + session.close() r.sessionsMu.Unlock() + if superseded { + glog.V(1).Infof("Plugin worker stream closed: %s (replaced by a newer session)", workerID) + return + } glog.V(0).Infof("Plugin worker disconnected: %s", workerID) } @@ -1494,5 +1657,9 @@ func (r *Plugin) WorkerConnectForTest(hello *plugin_pb.WorkerHello) { // WorkerDisconnectForTest simulates a worker disconnecting (test helper). func (r *Plugin) WorkerDisconnectForTest(workerID string) { + if session, ok := r.getSession(workerID); ok { + r.cleanupSession(session) + return + } r.registry.Remove(workerID) } diff --git a/weed/admin/plugin/plugin_detection_disconnect_test.go b/weed/admin/plugin/plugin_detection_disconnect_test.go new file mode 100644 index 0000000000..3a0eef85f3 --- /dev/null +++ b/weed/admin/plugin/plugin_detection_disconnect_test.go @@ -0,0 +1,406 @@ +package plugin + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/stretchr/testify/require" +) + +func newDetectorSession(p *Plugin, workerID, jobType string) *streamSession { + p.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + WorkerInstanceId: workerID + "-instance-" + time.Now().Format("150405.000000"), + Capabilities: []*plugin_pb.JobTypeCapability{{JobType: jobType, CanDetect: true, MaxDetectionConcurrency: 1}}, + }) + session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})} + p.putSession(session) + return session +} + +func awaitDetectionRequest(t *testing.T, session *streamSession) *plugin_pb.AdminToWorkerMessage { + t.Helper() + select { + case request := <-session.outgoing: + require.NotNil(t, request.GetRunDetectionRequest(), "expected run_detection_request, got %+v", request) + return request + case <-time.After(2 * time.Second): + t.Fatal("no detection request") + return nil + } +} + +func findActivity(activities []JobActivity, source, stage string) *JobActivity { + for i := range activities { + if activities[i].Source == source && activities[i].Stage == stage { + return &activities[i] + } + } + return nil +} + +func (r *Plugin) detectionInFlightForTest(jobType string) bool { + r.schedulerMu.Lock() + defer r.schedulerMu.Unlock() + return r.detectionInFlight[jobType] +} + +func (r *Plugin) pendingDetectionCountForTest() int { + r.pendingDetectionMu.Lock() + defer r.pendingDetectionMu.Unlock() + return len(r.pendingDetection) +} + +func TestRunDetectionFailsWhenDetectorDisconnects(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + + const jobType = "vacuum" + session := newDetectorSession(p, "worker-a", jobType) + + ctx, cancel := context.WithTimeout(context.Background(), time.Hour) + defer cancel() + type result struct { + report *DetectionReport + err error + } + resultCh := make(chan result, 1) + go func() { + report, runErr := p.RunDetectionWithReport(ctx, jobType, &plugin_pb.ClusterContext{}, 10) + resultCh <- result{report: report, err: runErr} + }() + request := awaitDetectionRequest(t, session) + + p.WorkerDisconnectForTest("worker-a") + + var got result + select { + case got = <-resultCh: + case <-time.After(5 * time.Second): + t.Fatal("detection did not fail after its worker disconnected") + } + require.Error(t, got.err) + var lost *detectionWorkerDisconnectedError + require.True(t, errors.As(got.err, &lost), "unexpected error type: %v", got.err) + require.Equal(t, "detection worker worker-a disconnected", got.err.Error()) + require.NotNil(t, got.report) + require.Equal(t, request.RequestId, got.report.RequestID) + require.Equal(t, 0, p.pendingDetectionCountForTest()) + require.Equal(t, "", p.getDetectorLease(jobType), "detector lease must not pin the lost worker") + + activity := findActivity(p.ListActivities(jobType, 50), "detector", "disconnected") + require.NotNil(t, activity, "expected a disconnected detector activity") + require.Equal(t, request.RequestId, activity.RequestID) + require.Equal(t, "worker-a", activity.WorkerID) + + // A replacement worker can serve the next detection right away. + next := newDetectorSession(p, "worker-b", jobType) + go func() { + _, _ = p.RunDetectionWithReport(ctx, jobType, &plugin_pb.ClusterContext{}, 10) + }() + awaitDetectionRequest(t, next) +} + +func TestRunDetectionFailsWhenDetectorReconnectsWithNewSession(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + + const jobType = "vacuum" + session := newDetectorSession(p, "worker-a", jobType) + + ctx, cancel := context.WithTimeout(context.Background(), time.Hour) + defer cancel() + errCh := make(chan error, 1) + go func() { + _, runErr := p.RunDetectionWithReport(ctx, jobType, &plugin_pb.ClusterContext{}, 10) + errCh <- runErr + }() + awaitDetectionRequest(t, session) + + // The worker process restarted: same worker ID, new instance, new stream. + replacement := newDetectorSession(p, "worker-a", jobType) + + select { + case runErr := <-errCh: + var lost *detectionWorkerDisconnectedError + require.True(t, errors.As(runErr, &lost), "unexpected error: %v", runErr) + case <-time.After(5 * time.Second): + t.Fatal("detection did not fail after its worker reconnected with a new session") + } + require.Len(t, replacement.outgoing, 0, "the replacement session must not receive a cancel for the old request") + + // The old stream unwinding afterwards must not evict the new session. + p.cleanupSession(session) + current, ok := p.getSession("worker-a") + require.True(t, ok) + require.Same(t, replacement, current) + _, registered := p.registry.Get("worker-a") + require.True(t, registered) +} + +func TestRunDetectionSurvivesOtherWorkerDisconnect(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + + const jobType = "vacuum" + sessionA := newDetectorSession(p, "worker-a", jobType) + sessionB := newDetectorSession(p, "worker-b", jobType) + + ctx, cancel := context.WithTimeout(context.Background(), time.Hour) + defer cancel() + errCh := make(chan error, 1) + go func() { + _, runErr := p.RunDetectionWithReport(ctx, jobType, &plugin_pb.ClusterContext{}, 10) + errCh <- runErr + }() + + var request *plugin_pb.AdminToWorkerMessage + detector, other := "", "" + select { + case request = <-sessionA.outgoing: + detector, other = "worker-a", "worker-b" + case request = <-sessionB.outgoing: + detector, other = "worker-b", "worker-a" + case <-time.After(2 * time.Second): + t.Fatal("no detection request") + } + require.NotNil(t, request.GetRunDetectionRequest()) + + p.WorkerDisconnectForTest(other) + + select { + case runErr := <-errCh: + t.Fatalf("detection on %s ended after %s disconnected: %v", detector, other, runErr) + case <-time.After(300 * time.Millisecond): + } + require.Equal(t, 1, p.pendingDetectionCountForTest()) + + p.handleDetectionComplete(detector, &plugin_pb.DetectionComplete{RequestId: request.RequestId, JobType: jobType, Success: true}) + select { + case runErr := <-errCh: + require.NoError(t, runErr) + case <-time.After(2 * time.Second): + t.Fatal("detection did not complete") + } +} + +func TestRunDetectionFailsWhenDetectorHeartbeatGoesStale(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + p.registry.staleAfter = 50 * time.Millisecond + p.detectionLivenessInterval = 20 * time.Millisecond + + const jobType = "vacuum" + session := newDetectorSession(p, "worker-a", jobType) + + ctx, cancel := context.WithTimeout(context.Background(), time.Hour) + defer cancel() + errCh := make(chan error, 1) + go func() { + _, runErr := p.RunDetectionWithReport(ctx, jobType, &plugin_pb.ClusterContext{}, 10) + errCh <- runErr + }() + request := awaitDetectionRequest(t, session) + + select { + case runErr := <-errCh: + require.Error(t, runErr) + require.Equal(t, "detection worker worker-a disconnected: heartbeat stale", runErr.Error()) + case <-time.After(5 * time.Second): + t.Fatal("detection did not fail after its worker went stale") + } + + // The stream may still be open, so the worker is asked to stop the request. + select { + case message := <-session.outgoing: + cancelReq := message.GetCancelRequest() + require.NotNil(t, cancelReq, "expected cancel_request, got %+v", message) + require.Equal(t, request.RequestId, cancelReq.TargetId) + require.Equal(t, plugin_pb.WorkKind_WORK_KIND_DETECTION, cancelReq.TargetKind) + case <-time.After(2 * time.Second): + t.Fatal("no cancel request after stale worker") + } +} + +func TestScheduledDetectionFailsFastOnWorkerDisconnect(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + p.clusterContextProvider = func(context.Context) (*plugin_pb.ClusterContext, error) { + return &plugin_pb.ClusterContext{}, nil + } + + const jobType = "vacuum" + session := newDetectorSession(p, "worker-a", jobType) + + done := make(chan struct{}) + go func() { + p.runJobTypeIteration(jobType, schedulerPolicy{DetectionTimeout: time.Hour, JobTypeMaxRuntime: time.Hour}) + close(done) + }() + awaitDetectionRequest(t, session) + require.True(t, p.detectionInFlightForTest(jobType)) + + p.WorkerDisconnectForTest("worker-a") + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("scheduled iteration kept waiting after its detector disconnected") + } + require.False(t, p.detectionInFlightForTest(jobType)) + require.Equal(t, 0, p.pendingDetectionCountForTest()) + require.Equal(t, "error", p.snapshotSchedulerRun(jobType).lastRunStatus) + require.False(t, p.snapshotSchedulerDetection(jobType).lastErrorAt.IsZero()) + + activities := p.ListActivities(jobType, 50) + require.NotNil(t, findActivity(activities, "detector", "disconnected")) + retrying := findActivity(activities, "admin_scheduler", "retrying") + require.NotNil(t, retrying, "the scheduler should have tried a replacement pick") + require.Contains(t, retrying.Message, "detection worker worker-a disconnected") + failed := findActivity(activities, "admin_scheduler", "failed") + require.NotNil(t, failed, "with no other detector the run must end as failed") + + // The next cycle can detect again. + next := newDetectorSession(p, "worker-b", jobType) + go func() { + p.runJobTypeIteration(jobType, schedulerPolicy{DetectionTimeout: time.Hour, JobTypeMaxRuntime: time.Hour}) + }() + awaitDetectionRequest(t, next) +} + +func TestScheduledDetectionRetriesOnAnotherWorkerAfterDisconnect(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + p.clusterContextProvider = func(context.Context) (*plugin_pb.ClusterContext, error) { + return &plugin_pb.ClusterContext{}, nil + } + + const jobType = "vacuum" + sessions := map[string]*streamSession{ + "worker-a": newDetectorSession(p, "worker-a", jobType), + "worker-b": newDetectorSession(p, "worker-b", jobType), + } + + done := make(chan struct{}) + go func() { + p.runJobTypeIteration(jobType, schedulerPolicy{DetectionTimeout: time.Hour, JobTypeMaxRuntime: time.Hour}) + close(done) + }() + + first, other := "", "" + select { + case <-sessions["worker-a"].outgoing: + first, other = "worker-a", "worker-b" + case <-sessions["worker-b"].outgoing: + first, other = "worker-b", "worker-a" + case <-time.After(2 * time.Second): + t.Fatal("no detection request") + } + + p.WorkerDisconnectForTest(first) + + retry := awaitDetectionRequest(t, sessions[other]) + p.handleDetectionComplete(other, &plugin_pb.DetectionComplete{RequestId: retry.RequestId, JobType: jobType, Success: true}) + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("scheduled iteration did not finish after the retry completed") + } + require.False(t, p.detectionInFlightForTest(jobType)) + require.Equal(t, "success", p.snapshotSchedulerRun(jobType).lastRunStatus) + require.False(t, p.snapshotSchedulerDetection(jobType).lastDetectedAt.IsZero()) + require.NotNil(t, findActivity(p.ListActivities(jobType, 50), "admin_scheduler", "retrying")) +} + +func TestRegisterSessionKeepsRegistryOnLiveStream(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + + helloA := &plugin_pb.WorkerHello{WorkerId: "worker-a", WorkerInstanceId: "a-1", Capabilities: []*plugin_pb.JobTypeCapability{{JobType: "ec_vacuum", CanDetect: true}}} + helloB := &plugin_pb.WorkerHello{WorkerId: "worker-a", WorkerInstanceId: "a-2", Capabilities: []*plugin_pb.JobTypeCapability{{JobType: "s3_lifecycle", CanDetect: true}}} + streamA := &streamSession{workerID: "worker-a", outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 1), done: make(chan struct{})} + streamB := &streamSession{workerID: "worker-a", outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 1), done: make(chan struct{})} + + p.registerSession(streamA, helloA) + p.registerSession(streamB, helloB) + select { + case <-streamA.done: + default: + t.Fatal("the replaced stream must be closed") + } + + // Stream A's hello and heartbeat arriving after it was replaced must not + // describe the live stream B, and neither must A's own cleanup. + p.handleWorkerMessage(streamA, &plugin_pb.WorkerToAdminMessage{Body: &plugin_pb.WorkerToAdminMessage_Hello{Hello: helloA}}) + p.handleWorkerMessage(streamA, &plugin_pb.WorkerToAdminMessage{Body: &plugin_pb.WorkerToAdminMessage_Heartbeat{Heartbeat: &plugin_pb.WorkerHeartbeat{DetectionSlotsTotal: 9}}}) + p.cleanupSession(streamA) + + worker, ok := p.registry.Get("worker-a") + require.True(t, ok) + require.Equal(t, "a-2", worker.WorkerInstance) + require.Nil(t, worker.Capabilities["ec_vacuum"]) + require.NotNil(t, worker.Capabilities["s3_lifecycle"]) + require.Nil(t, worker.Heartbeat, "a superseded stream's heartbeat must be dropped") + current, ok := p.getSession("worker-a") + require.True(t, ok) + require.Same(t, streamB, current) + _, err = p.pickDetector("ec_vacuum", nil) + require.Error(t, err, "the capability only stream A advertised must be gone") + + p.handleWorkerMessage(streamB, &plugin_pb.WorkerToAdminMessage{Body: &plugin_pb.WorkerToAdminMessage_Heartbeat{Heartbeat: &plugin_pb.WorkerHeartbeat{DetectionSlotsTotal: 2}}}) + worker, ok = p.registry.Get("worker-a") + require.True(t, ok) + require.NotNil(t, worker.Heartbeat) + require.EqualValues(t, 2, worker.Heartbeat.DetectionSlotsTotal) +} + +func TestScheduledDetectionRetryAvoidsClosedDetectorStream(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + p.clusterContextProvider = func(context.Context) (*plugin_pb.ClusterContext, error) { + return &plugin_pb.ClusterContext{}, nil + } + + const jobType = "vacuum" + sessionA := newDetectorSession(p, "worker-a", jobType) + sessionB := newDetectorSession(p, "worker-b", jobType) + // worker-a ranks first on free detection slots, for the first pick and + // for any retry that still sees it in the registry. + p.registry.UpdateHeartbeat("worker-a", &plugin_pb.WorkerHeartbeat{DetectionSlotsTotal: 4}) + p.registry.UpdateHeartbeat("worker-b", &plugin_pb.WorkerHeartbeat{DetectionSlotsTotal: 1}) + + done := make(chan struct{}) + go func() { + p.runJobTypeIteration(jobType, schedulerPolicy{DetectionTimeout: time.Hour, JobTypeMaxRuntime: time.Hour}) + close(done) + }() + awaitDetectionRequest(t, sessionA) + + // Close the stream while its registry entry is still present: the + // ordering where the retry picks before the disconnect is fully cleaned up. + sessionA.close() + + retry := awaitDetectionRequest(t, sessionB) + require.Equal(t, "worker-b", p.getDetectorLease(jobType)) + p.handleDetectionComplete("worker-b", &plugin_pb.DetectionComplete{RequestId: retry.RequestId, JobType: jobType, Success: true}) + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("scheduled iteration did not finish after the retry completed") + } + require.Equal(t, "success", p.snapshotSchedulerRun(jobType).lastRunStatus) + require.Len(t, sessionA.outgoing, 0, "nothing may be sent on the closed stream") +} diff --git a/weed/admin/plugin/plugin_scheduler.go b/weed/admin/plugin/plugin_scheduler.go index 759581e193..af62a5d788 100644 --- a/weed/admin/plugin/plugin_scheduler.go +++ b/weed/admin/plugin/plugin_scheduler.go @@ -322,7 +322,6 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo } defer releaseDetectionLock() - detectionTimeout := policy.DetectionTimeout remaining := time.Until(start.Add(maxRuntime)) if remaining <= 0 { r.appendActivity(JobActivity{ @@ -335,16 +334,35 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo r.recordSchedulerRunComplete(jobType, "timeout") return false } - if detectionTimeout <= 0 { - detectionTimeout = defaultScheduledDetectionTimeout - } - if detectionTimeout > remaining { - detectionTimeout = remaining + detect := func(avoid *streamSession) (*DetectionReport, error) { + detectionTimeout := policy.DetectionTimeout + if detectionTimeout <= 0 { + detectionTimeout = defaultScheduledDetectionTimeout + } + if remaining := time.Until(start.Add(maxRuntime)); detectionTimeout > remaining { + detectionTimeout = remaining + } + detectCtx, cancelDetect := context.WithTimeout(jobCtx, detectionTimeout) + defer cancelDetect() + return r.runDetectionWithReport(detectCtx, jobType, clusterContext, policy.MaxResults, avoid) } - detectCtx, cancelDetect := context.WithTimeout(jobCtx, detectionTimeout) - report, err := r.RunDetectionWithReport(detectCtx, jobType, clusterContext, policy.MaxResults) - cancelDetect() + report, err := detect(nil) + // A detector that dropped mid-request gets one replacement pick, steered + // off the stream that dropped, in this iteration while the runtime budget + // allows; anything else waits for the next cycle, which re-detects anyway. + var lost *detectionWorkerDisconnectedError + if errors.As(err, &lost) && time.Until(start.Add(maxRuntime)) > 0 { + r.recordSchedulerDetectionError(jobType, err) + r.appendActivity(JobActivity{ + JobType: jobType, + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduled detection retrying on another worker: %v", err), + Stage: "retrying", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + report, err = detect(lost.session) + } releaseDetectionLock() if err != nil { r.recordSchedulerDetectionError(jobType, err) diff --git a/weed/admin/plugin/plugin_scheduler_test.go b/weed/admin/plugin/plugin_scheduler_test.go index b7aa3757ca..8b7a384f23 100644 --- a/weed/admin/plugin/plugin_scheduler_test.go +++ b/weed/admin/plugin/plugin_scheduler_test.go @@ -562,9 +562,11 @@ func TestPickDetectorPrefersLeasedWorker(t *testing.T) { }, }) + pluginSvc.putSession(&streamSession{workerID: "worker-a", outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 1), done: make(chan struct{})}) + pluginSvc.putSession(&streamSession{workerID: "worker-b", outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 1), done: make(chan struct{})}) pluginSvc.setDetectorLease("vacuum", "worker-b") - detector, err := pluginSvc.pickDetector("vacuum") + detector, err := pluginSvc.pickDetector("vacuum", nil) if err != nil { t.Fatalf("pickDetector: %v", err) } @@ -588,9 +590,10 @@ func TestPickDetectorReassignsWhenLeaseIsStale(t *testing.T) { {JobType: "vacuum", CanDetect: true}, }, }) + pluginSvc.putSession(&streamSession{workerID: "worker-a", outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 1), done: make(chan struct{})}) pluginSvc.setDetectorLease("vacuum", "worker-stale") - detector, err := pluginSvc.pickDetector("vacuum") + detector, err := pluginSvc.pickDetector("vacuum", nil) if err != nil { t.Fatalf("pickDetector: %v", err) } diff --git a/weed/admin/plugin/preview_test.go b/weed/admin/plugin/preview_test.go index a5f4345802..85ead7a666 100644 --- a/weed/admin/plugin/preview_test.go +++ b/weed/admin/plugin/preview_test.go @@ -69,7 +69,7 @@ func TestRequestObjectPreviewReturnsTheWorkersRows(t *testing.T) { t.Fatalf("request lost its parameters: %+v", asked) } - pluginSvc.handleWorkerMessage(workerID, &plugin_pb.WorkerToAdminMessage{ + pluginSvc.handleWorkerMessage(session, &plugin_pb.WorkerToAdminMessage{ WorkerId: workerID, Body: &plugin_pb.WorkerToAdminMessage_ObjectPreviewResponse{ ObjectPreviewResponse: &plugin_pb.ObjectPreviewResponse{ @@ -114,7 +114,7 @@ func TestRequestObjectPreviewSurfacesTheWorkersError(t *testing.T) { }() request := <-session.outgoing - pluginSvc.handleWorkerMessage(workerID, &plugin_pb.WorkerToAdminMessage{ + pluginSvc.handleWorkerMessage(session, &plugin_pb.WorkerToAdminMessage{ WorkerId: workerID, Body: &plugin_pb.WorkerToAdminMessage_ObjectPreviewResponse{ ObjectPreviewResponse: &plugin_pb.ObjectPreviewResponse{ @@ -187,7 +187,7 @@ func TestObjectPreviewResponseRacesShutdown(t *testing.T) { }() go func() { defer wg.Done() - pluginSvc.handleWorkerMessage(workerID, &plugin_pb.WorkerToAdminMessage{ + pluginSvc.handleWorkerMessage(session, &plugin_pb.WorkerToAdminMessage{ WorkerId: workerID, Body: &plugin_pb.WorkerToAdminMessage_ObjectPreviewResponse{ ObjectPreviewResponse: &plugin_pb.ObjectPreviewResponse{ diff --git a/weed/admin/plugin/registry.go b/weed/admin/plugin/registry.go index 6e27d30411..4fa141fb06 100644 --- a/weed/admin/plugin/registry.go +++ b/weed/admin/plugin/registry.go @@ -272,9 +272,10 @@ func (r *Registry) PickSchemaProvider(jobType string) (*WorkerSession, error) { return cloneWorkerSession(candidates[0]), nil } -// PickDetector picks one detector worker for a job type. -func (r *Registry) PickDetector(jobType string) (*WorkerSession, error) { - return r.pickByKind(jobType, true) +// PickDetector picks one detector worker for a job type, skipping any +// worker IDs in excluded. +func (r *Registry) PickDetector(jobType string, excluded ...string) (*WorkerSession, error) { + return r.pickByKind(jobType, true, excluded) } // PickExecutor picks one executor worker for a job type. @@ -322,11 +323,24 @@ func (r *Registry) ListExecutors(jobType string, preferredHosts ...string) ([]*W return out, nil } -func (r *Registry) pickByKind(jobType string, detect bool) (*WorkerSession, error) { +func (r *Registry) pickByKind(jobType string, detect bool, excluded []string) (*WorkerSession, error) { r.mu.Lock() defer r.mu.Unlock() candidates := r.collectByKindLocked(jobType, detect, time.Now()) + if len(excluded) > 0 { + skip := make(map[string]struct{}, len(excluded)) + for _, workerID := range excluded { + skip[workerID] = struct{}{} + } + kept := candidates[:0] + for _, candidate := range candidates { + if _, found := skip[candidate.WorkerID]; !found { + kept = append(kept, candidate) + } + } + candidates = kept + } if len(candidates) == 0 { kind := "executor"