Fixes #22714: recover system jobs stranded in running status by a killed worker
A worker killed abruptly (SIGKILL, OOM, deploy restart) mid-run leaves its Job row at status=running: start() persists that synchronously, but the finally block in handle() that reschedules the successor never runs. Because running is an enqueued state, enqueue_once() treats the dead row as a live schedule at the next worker startup and never re-arms it, so the recurring job silently stops. This is the running-status sibling of the stale-scheduled case fixed in the preceding commit. RQ liveness can't distinguish an orphan from a live job (a killed job's RQ entry can outlive the worker), and the schedule may be recovered with RQ state wiped, so detection is a pure-DB staleness heuristic on the started timestamp. Add reconcile_stale_system_jobs(), invoked by the worker at startup before enqueue_once(). It moves an object-less system job whose started is older than max(interval, STALE_RUNNING_JOB_GRACE_MINUTES=30) to errored, so the subsequent enqueue_once() re-arms a fresh schedule. errored (not failed) records an unexpected fault rather than a self-declared failure, matching how handle() maps an unhandled exception. terminate() has no notification or webhook side effects for an object-less, userless system job. The query keys on interval so a different system job sharing a name isn't reaped. Also guard the preceding commit's is_stale check against a scheduled row with a null scheduled value, which would otherwise raise TypeError at worker startup on a corrupted row (exactly the DB/Redis restore state this change defends against).
This commit is contained in:
Jason Novinger
parent
6ddb208321
commit
78f549728a
|
|
@ -2,6 +2,7 @@ import logging
|
|||
|
||||
from django_rq.management.commands.rqworker import Command as _Command
|
||||
|
||||
from netbox.jobs import reconcile_stale_system_jobs
|
||||
from netbox.registry import registry
|
||||
|
||||
DEFAULT_QUEUES = ('high', 'default', 'low')
|
||||
|
|
@ -21,6 +22,9 @@ class Command(_Command):
|
|||
interval = kwargs['interval']
|
||||
except KeyError:
|
||||
raise TypeError("System job must specify an interval (in minutes).")
|
||||
# Recover jobs stranded by a killed worker before scheduling, so a stale row
|
||||
# isn't mistaken for a live schedule (see #22714).
|
||||
reconcile_stale_system_jobs(job, interval)
|
||||
logger.debug(f"Scheduling system job {job.name} (interval={interval})")
|
||||
job.enqueue_once(**kwargs)
|
||||
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ from utilities.request import apply_request_processors
|
|||
__all__ = (
|
||||
'AsyncViewJob',
|
||||
'JobRunner',
|
||||
'reconcile_stale_system_jobs',
|
||||
'system_job',
|
||||
)
|
||||
|
||||
|
|
@ -29,6 +30,11 @@ __all__ = (
|
|||
# jobs.py lives at <root>/netbox/netbox/jobs.py, so parents[2] is the root.
|
||||
_INSTALL_ROOT = str(Path(__file__).resolve().parents[2]) + os.sep
|
||||
|
||||
# Floor for the grace period before a "running" system job is treated as stranded. The window
|
||||
# is max(interval, this floor), so a short-interval job still gets time to finish a legitimate
|
||||
# run. See reconcile_stale_system_jobs() and issue #22714.
|
||||
STALE_RUNNING_JOB_GRACE_MINUTES = 30
|
||||
|
||||
|
||||
def system_job(interval):
|
||||
"""
|
||||
|
|
@ -46,6 +52,37 @@ def system_job(interval):
|
|||
return _wrapper
|
||||
|
||||
|
||||
@advisory_lock(ADVISORY_LOCK_KEYS['job-schedules'])
|
||||
def reconcile_stale_system_jobs(job_class, interval):
|
||||
"""
|
||||
Fail any object-less system job of this class left stranded in "running" status by a worker
|
||||
that was killed mid-run (issue #22714). Such a row is never reset, and because "running" is
|
||||
an enqueued state, `enqueue_once()` mistakes it for a live schedule and never re-arms it.
|
||||
|
||||
A running job is treated as stranded once its `started` timestamp is older than
|
||||
`max(interval, STALE_RUNNING_JOB_GRACE_MINUTES)`. RQ is not consulted: a killed job's RQ
|
||||
entry can outlive the worker, and the schedule may be recovered with RQ state wiped.
|
||||
"""
|
||||
grace = max(interval, STALE_RUNNING_JOB_GRACE_MINUTES)
|
||||
cutoff = timezone.now() - timedelta(minutes=grace)
|
||||
|
||||
orphaned = Job.objects.filter(
|
||||
name=job_class.name,
|
||||
object_id__isnull=True,
|
||||
interval=interval,
|
||||
status=JobStatusChoices.STATUS_RUNNING,
|
||||
started__lte=cutoff,
|
||||
)
|
||||
for job in orphaned:
|
||||
# STATUS_ERRORED (not FAILED) records an unexpected fault rather than a self-declared
|
||||
# failure, as handle() does for an unhandled exception. For an object-less, userless
|
||||
# system job, terminate() sends no notification and triggers no event rule.
|
||||
job.terminate(
|
||||
status=JobStatusChoices.STATUS_ERRORED,
|
||||
error="Worker terminated before job completed",
|
||||
)
|
||||
|
||||
|
||||
class JobLogHandler(logging.Handler):
|
||||
"""
|
||||
A logging handler which records entries on a Job.
|
||||
|
|
@ -223,7 +260,11 @@ class JobRunner(ABC):
|
|||
# Redis restart) and must be replaced rather than reused, even though its parameters match.
|
||||
# Running/pending jobs are exempt from this check: their `scheduled` timestamp is expected to be
|
||||
# in the past (or unset) once they've started, and that must not be mistaken for staleness.
|
||||
is_stale = job.status == JobStatusChoices.STATUS_SCHEDULED and job.scheduled <= timezone.now()
|
||||
is_stale = (
|
||||
job.status == JobStatusChoices.STATUS_SCHEDULED and
|
||||
job.scheduled and
|
||||
job.scheduled <= timezone.now()
|
||||
)
|
||||
if not is_stale and (not schedule_at or job.scheduled == schedule_at) and (job.interval == interval):
|
||||
return job
|
||||
job.delete()
|
||||
|
|
|
|||
|
|
@ -33,6 +33,13 @@ class TestSystemJobRunner(JobRunner):
|
|||
pass
|
||||
|
||||
|
||||
@system_job(interval=1)
|
||||
class TestShortIntervalSystemJobRunner(JobRunner):
|
||||
|
||||
def run(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
|
||||
class BaseJobRunnerTestCase(RQQueueTestMixin, TestCase):
|
||||
|
||||
@staticmethod
|
||||
|
|
@ -379,3 +386,146 @@ class SystemJobTestCase(BaseJobRunnerTestCase):
|
|||
interval=interval,
|
||||
)
|
||||
self.assertEqual(enqueued.count(), 2)
|
||||
|
||||
|
||||
class ReconcileStaleJobsTestCase(BaseJobRunnerTestCase):
|
||||
"""
|
||||
Test recovery of system jobs stranded in "running" status by a killed worker (#22714).
|
||||
"""
|
||||
|
||||
def test_reconcile_terminates_orphaned_running_job(self):
|
||||
"""A running system job whose `started` is older than the grace window is an
|
||||
orphan (its worker died) and must be moved to `errored`."""
|
||||
orphan = Job.objects.create(
|
||||
name=TestSystemJobRunner.name,
|
||||
status=JobStatusChoices.STATUS_RUNNING,
|
||||
interval=60,
|
||||
scheduled=timezone.now() - timedelta(hours=2),
|
||||
started=timezone.now() - timedelta(hours=2),
|
||||
job_id=uuid.uuid4(),
|
||||
)
|
||||
|
||||
reconcile_stale_system_jobs(TestSystemJobRunner, 60)
|
||||
|
||||
orphan.refresh_from_db()
|
||||
self.assertEqual(orphan.status, JobStatusChoices.STATUS_ERRORED)
|
||||
self.assertIsNotNone(orphan.completed)
|
||||
self.assertEqual(orphan.error, "Worker terminated before job completed")
|
||||
|
||||
def test_reconcile_preserves_recently_started_running_job(self):
|
||||
"""A running system job that started recently (within the grace window) is a
|
||||
legitimately in-flight job and must NOT be reaped."""
|
||||
live = Job.objects.create(
|
||||
name=TestSystemJobRunner.name,
|
||||
status=JobStatusChoices.STATUS_RUNNING,
|
||||
interval=60,
|
||||
scheduled=timezone.now() - timedelta(minutes=1),
|
||||
started=timezone.now(),
|
||||
job_id=uuid.uuid4(),
|
||||
)
|
||||
|
||||
reconcile_stale_system_jobs(TestSystemJobRunner, 60)
|
||||
|
||||
live.refresh_from_db()
|
||||
self.assertEqual(live.status, JobStatusChoices.STATUS_RUNNING)
|
||||
|
||||
def test_reconcile_grace_window_scales_with_interval(self):
|
||||
"""The grace window is max(interval, floor). A job whose interval exceeds the floor
|
||||
gets the longer window: a 60-minute-interval job started 45 minutes ago is past the
|
||||
30-minute floor but still within its own interval, so it must be preserved."""
|
||||
within_interval = Job.objects.create(
|
||||
name=TestSystemJobRunner.name,
|
||||
status=JobStatusChoices.STATUS_RUNNING,
|
||||
interval=60,
|
||||
scheduled=timezone.now() - timedelta(minutes=45),
|
||||
started=timezone.now() - timedelta(minutes=45),
|
||||
job_id=uuid.uuid4(),
|
||||
)
|
||||
|
||||
reconcile_stale_system_jobs(TestSystemJobRunner, 60)
|
||||
|
||||
within_interval.refresh_from_db()
|
||||
self.assertEqual(within_interval.status, JobStatusChoices.STATUS_RUNNING)
|
||||
|
||||
def test_reconcile_grace_floor_protects_short_interval_job(self):
|
||||
"""For a job whose interval is shorter than the floor, the floor governs the window.
|
||||
A 1-minute-interval job started 20 minutes ago is well past its interval but within
|
||||
the 30-minute floor, so it must NOT be reaped."""
|
||||
within_floor = Job.objects.create(
|
||||
name=TestShortIntervalSystemJobRunner.name,
|
||||
status=JobStatusChoices.STATUS_RUNNING,
|
||||
interval=1,
|
||||
scheduled=timezone.now() - timedelta(minutes=20),
|
||||
started=timezone.now() - timedelta(minutes=20),
|
||||
job_id=uuid.uuid4(),
|
||||
)
|
||||
|
||||
reconcile_stale_system_jobs(TestShortIntervalSystemJobRunner, 1)
|
||||
|
||||
within_floor.refresh_from_db()
|
||||
self.assertEqual(within_floor.status, JobStatusChoices.STATUS_RUNNING)
|
||||
|
||||
def test_reconcile_grace_floor_reaps_short_interval_orphan(self):
|
||||
"""A 1-minute-interval job started 40 minutes ago is past the 30-minute floor and is
|
||||
an orphan, so it must be reaped despite its short interval."""
|
||||
orphan = Job.objects.create(
|
||||
name=TestShortIntervalSystemJobRunner.name,
|
||||
status=JobStatusChoices.STATUS_RUNNING,
|
||||
interval=1,
|
||||
scheduled=timezone.now() - timedelta(minutes=40),
|
||||
started=timezone.now() - timedelta(minutes=40),
|
||||
job_id=uuid.uuid4(),
|
||||
)
|
||||
|
||||
reconcile_stale_system_jobs(TestShortIntervalSystemJobRunner, 1)
|
||||
|
||||
orphan.refresh_from_db()
|
||||
self.assertEqual(orphan.status, JobStatusChoices.STATUS_ERRORED)
|
||||
|
||||
def test_reconcile_ignores_instance_bound_job(self):
|
||||
"""The sweep targets object-less system jobs only. An instance-bound job of the
|
||||
same runner class must be left alone even if it looks stale."""
|
||||
instance = DataSource.objects.create(name='test-ds-reconcile', type='local')
|
||||
bound = Job.objects.create(
|
||||
name=TestSystemJobRunner.name,
|
||||
object=instance,
|
||||
status=JobStatusChoices.STATUS_RUNNING,
|
||||
interval=60,
|
||||
scheduled=timezone.now() - timedelta(hours=2),
|
||||
started=timezone.now() - timedelta(hours=2),
|
||||
job_id=uuid.uuid4(),
|
||||
)
|
||||
|
||||
reconcile_stale_system_jobs(TestSystemJobRunner, 60)
|
||||
|
||||
bound.refresh_from_db()
|
||||
self.assertEqual(bound.status, JobStatusChoices.STATUS_RUNNING)
|
||||
|
||||
def test_reconcile_then_enqueue_once_rearms(self):
|
||||
"""After a stranded job is reconciled to `errored`, the startup `enqueue_once()`
|
||||
call must re-arm a fresh scheduled successor."""
|
||||
orphan = Job.objects.create(
|
||||
name=TestSystemJobRunner.name,
|
||||
status=JobStatusChoices.STATUS_RUNNING,
|
||||
interval=60,
|
||||
scheduled=timezone.now() - timedelta(hours=2),
|
||||
started=timezone.now() - timedelta(hours=2),
|
||||
job_id=uuid.uuid4(),
|
||||
)
|
||||
|
||||
# Mirror rqworker startup: reconcile stale jobs, then enqueue_once.
|
||||
reconcile_stale_system_jobs(TestSystemJobRunner, 60)
|
||||
successor = TestSystemJobRunner.enqueue_once(interval=60)
|
||||
|
||||
orphan.refresh_from_db()
|
||||
self.assertEqual(orphan.status, JobStatusChoices.STATUS_ERRORED)
|
||||
self.assertNotEqual(successor.pk, orphan.pk)
|
||||
self.assertIn(successor.status, JobStatusChoices.ENQUEUED_STATE_CHOICES)
|
||||
|
||||
# Exactly one live (enqueued) successor should remain.
|
||||
enqueued = Job.objects.filter(
|
||||
name=TestSystemJobRunner.name,
|
||||
object_id__isnull=True,
|
||||
status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES,
|
||||
)
|
||||
self.assertEqual(enqueued.count(), 1)
|
||||
|
|
|
|||
Loading…
Reference in New Issue