mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-04 11:07:35 +01:00
Chris Lu
30f0364256
* 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.
407 lines
14 KiB
Go
407 lines
14 KiB
Go
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")
|
|
}
|