Closes: #128 - Update JobRunner to use SystemJobs

Author: DanSheps

netbox_config_backup/__init__.py

@@ -39,26 +39,7 @@ def ready(self, *args, **kwargs):
         super().ready()
         import sys
         if len(sys.argv) > 1 and 'rqworker' in sys.argv[1]:
-            from netbox import settings
             from netbox_config_backup.jobs.backup import BackupRunner
-            frequency = settings.PLUGINS_CONFIG.get('netbox_config_backup', {}).get('frequency') / 60
-            lastjob = BackupRunner.get_jobs().order_by('pk').last()
-
-            if not lastjob:
-                BackupRunner.enqueue_once(interval=frequency)
-            elif lastjob.status in JobStatusChoices.ENQUEUED_STATE_CHOICES:
-                if lastjob.scheduled and lastjob.scheduled < timezone.now():
-                    BackupRunner.enqueue_once(interval=frequency)
-                elif not lastjob.scheduled:
-                    lastjob.scheduled = timezone.now()
-                    lastjob.clean()
-                    lastjob.save()
-            elif lastjob.status in JobStatusChoices.TERMINAL_STATE_CHOICES:
-                scheduled = lastjob.created + timezone.timedelta(minutes=frequency)
-                if scheduled < timezone.now():
-                    scheduled = None
-                BackupRunner.enqueue_once(interval=frequency, schedule_at=scheduled)
-
 
 
 config = NetboxConfigBackup

netbox_config_backup/backup/processing.py

@@ -2,11 +2,14 @@
 import os
 import time
 import traceback
+from datetime import timedelta
 
+import uuid
 from django.db.models import Q
 from django.utils import timezone
 
 from core.choices import JobStatusChoices
+from netbox import settings
 from netbox.api.exceptions import ServiceUnavailable
 from netbox_config_backup.models import BackupJob, Backup
 from netbox_config_backup.utils.db import close_db
@@ -91,6 +94,21 @@ def run_backup(job_id):
             commit = backup.set_config(configs)
 
             d.close()
+
+            frequency = timedelta(
+                seconds=settings.PLUGINS_CONFIG.get('netbox_config_backup', {}).get('frequency', 3600)
+            )
+            new = BackupJob(
+                runner=None,
+                backup=job.backup,
+                status=JobStatusChoices.STATUS_SCHEDULED,
+                scheduled=timezone.now() + frequency,
+                job_id=uuid.uuid4(),
+                data={},
+            )
+            new.full_clean()
+            new.save()
+
             logger.info(f'{backup}: Backup complete')
             job.status = JobStatusChoices.STATUS_COMPLETED
             job.completed = timezone.now()

netbox_config_backup/jobs/backup.py

@@ -5,12 +5,11 @@
 from datetime import timedelta
 from multiprocessing import Process
 
-from django.db.models import Q
 from django.utils import timezone
-from rq.job import JobStatus
 
-from core.choices import JobStatusChoices
-from netbox.jobs import JobRunner
+from core.choices import JobStatusChoices, JobIntervalChoices
+from netbox import settings
+from netbox.jobs import JobRunner, system_job
 from netbox_config_backup.backup.processing import run_backup
 from netbox_config_backup.choices import StatusChoices
 from netbox_config_backup.models import Backup, BackupJob
@@ -20,46 +19,52 @@
 logger = logging.getLogger(f"netbox_config_backup")
 
 
-class SchedulerRunner(JobRunner):
-    class Meta:
-        name = "The scheduler"
-
-
-
-
+@system_job(interval=JobIntervalChoices.INTERVAL_MINUTELY)
 class BackupRunner(JobRunner):
     processes = {}
 
     class Meta:
         name = 'The Backup Job Runner'
 
-    def clean_stale_jobs(self):
+    @classmethod
+    def fail_job(cls, job: BackupJob, status: str, error: str = ''):
+        job.status = status
+        if not job.data:
+            job.data = {}
+        job.data.update({'error': 'Process terminated'})
+        job.save()
+        job.refresh_from_db()
+
+
+    @classmethod
+    def clean_stale_jobs(cls):
+        results = {
+            'stale': 0,
+            'scheduled': 0
+        }
+
         jobs = BackupJob.objects.order_by('created').filter(
-            status=JobStatusChoices.ENQUEUED_STATE_CHOICES,
-        ).prefetch_related('device')
-        scheduled = jobs.filter(status=JobStatusChoices.STATUS_SCHEDULED)
-        stale = jobs.filter(scheduled__lt=timezone.now() - timedelta(minutes=30))
+            status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES,
+        ).prefetch_related('backup', 'backup__device')
 
+        stale = jobs.filter(scheduled__lt=timezone.now() - timedelta(minutes=30))
         for job in stale:
-            if job.pid:
-                pass
-            job.status = JobStatusChoices.STATUS_ERRORED
-            if not job.data:
-                job.data = {}
-            job.data.update({'error': 'Job hung'})
-            job.save()
-            job.refresh_from_db()
+            results['stale'] += 1
+            cls.fail_job(job, JobStatusChoices.STATUS_FAILED, 'Job hung')
             logger.warning(f'Job {job.backup} appears stuck, deleting')
 
+        scheduled = jobs.filter(status=JobStatusChoices.STATUS_SCHEDULED)
         for job in scheduled:
             if job != scheduled.filter(backup=job.backup).last():
-                job.status = JobStatusChoices.STATUS_FAILED
-                if not job.data:
-                    job.data = {}
-                job.data.update({'error': 'Process terminated'})
-                job.save()
+                results['scheduled'] += 1
+                cls.fail_job(job, JobStatusChoices.STATUS_ERRORED, 'Job missed')
+                logger.warning(f'Job {job.backup} appears to have been missed, deleting')
 
-    def schedule_jobs(self, backup=None, device=None):
+        return results
+
+    @classmethod
+    def schedule_jobs(cls, runner, backup=None, device=None):
+        scheduled_status = 0
         if backup:
             logging.debug(f'Scheduling backup for backup: {backup}')
             backups = Backup.objects.filter(pk=backup.pk, status=StatusChoices.STATUS_ACTIVE, device__isnull=False)
@@ -70,38 +75,47 @@ def schedule_jobs(self, backup=None, device=None):
             logging.debug(f'Scheduling all backups')
             backups = Backup.objects.filter(status=StatusChoices.STATUS_ACTIVE, device__isnull=False)
 
+        frequency = timedelta(seconds=settings.PLUGINS_CONFIG.get('netbox_config_backup', {}).get('frequency', 3600))
+
         for backup in backups:
             if can_backup(backup):
                 logger.debug(f'Queuing device {backup.device} for backup')
-                jobs = BackupJob.objects.filter(backup=backup, status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES)
-                job = jobs.last()
-                if job is not None:
-                    job.runner = self.job
-                    job.status = JobStatusChoices.STATUS_SCHEDULED
-                    job.scheduled = timezone.now()
-                    job.save()
-                else:
+                jobs = BackupJob.objects.filter(backup=backup)
+                if jobs.filter(status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES).count() == 0:
+                    scheduled = timezone.now()
                     job = BackupJob(
-                        runner=self.job,
+                        runner=None,
                         backup=backup,
                         status=JobStatusChoices.STATUS_SCHEDULED,
-                        scheduled=timezone.now(),
+                        scheduled=scheduled,
                         job_id=uuid.uuid4(),
                         data={},
                     )
-                job.full_clean()
-                job.save()
+                    job.full_clean()
+                    job.save()
+                    scheduled_status += 1
             else:
                 jobs = BackupJob.objects.filter(backup=backup, status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES)
                 for job in jobs:
-                    job.status = JobStatusChoices.STATUS_FAILED
-                    if not job.data:
-                        job.data = {}
-                    job.data.update({'error': f'Cannot queue job'})
-                    job.save()
+                    cls.fail_job(job, JobStatusChoices.STATUS_FAILED, f'Cannot queue job')
+
+        return scheduled_status
 
     def run_processes(self):
-        for job in BackupJob.objects.filter(runner=self.job, status=JobStatusChoices.STATUS_SCHEDULED):
+        jobs = BackupJob.objects.filter(
+            runner=None,
+            status=JobStatusChoices.STATUS_SCHEDULED,
+            scheduled__lte=timezone.now()
+        )
+        for job in jobs:
+            job.runner = self.job
+            job.status = JobStatusChoices.STATUS_PENDING
+            job.save()
+
+        self.job.data.update({'status': {'pending': jobs.count()}})
+        self.job.save()
+
+        for job in jobs:
             try:
                 process = self.fork_process(job)
                 process.join(1)
@@ -128,6 +142,9 @@ def fork_process(self, job):
     def handle_processes(self):
         close_db()
         for pk in list(self.processes.keys()):
+            terminated = self.job.data.get('status', {}).get('terminated', 0)
+            completed = self.job.data.get('status', {}).get('completed', 0)
+
             process = self.processes.get(pk, {}).get('process')
             job_pk = self.processes.get(pk, {}).get('job')
             backup = self.processes.get(pk, {}).get('backup')
@@ -137,16 +154,32 @@ def handle_processes(self):
                 del self.processes[pk]
                 job = BackupJob.objects.filter(pk=job_pk).first()
                 if job and job.status != JobStatusChoices.STATUS_COMPLETED:
+                    self.job.data.update({'status': {'terminated': terminated}})
                     job.status = JobStatusChoices.STATUS_ERRORED
                     if not job.data:
                         job.data = {}
                     job.data.update({'error': 'Process terminated'})
                     job.save()
+                else:
+                    self.job.data.update({'status': {'completed': completed}})
+        self.job.save()
+        self.job.refresh_from_db()
 
     def run(self, backup=None, device=None, *args, **kwargs):
+
+        if not self.job.data:
+            self.job.data = {}
+            self.job.save()
+
         try:
-            self.clean_stale_jobs()
-            self.schedule_jobs(backup=backup, device=device)
+            status = self.clean_stale_jobs()
+            self.job.data.update({'status': status})
+
+            status = self.schedule_jobs(runner=self.job, backup=backup, device=device)
+            self.job.data.update({'status': {'scheduled': status}})
+
+            self.job.save()
+
             self.run_processes()
             while(True):
                 self.handle_processes()
@@ -156,3 +189,4 @@ def run(self, backup=None, device=None, *args, **kwargs):
         except Exception as e:
             logger.warning(f'{traceback.format_exc()}')
             logger.error(f'{e}')
+            raise e

netbox_config_backup/querysets/__init__.py

@@ -19,9 +19,7 @@ def default_annotate(self):
             next_attempt=models.Subquery(
                 BackupJob.objects.filter(
                     backup=models.OuterRef('id'),
-                    status__in=[
-                        JobStatusChoices.STATUS_PENDING, JobStatusChoices.STATUS_RUNNING
-                    ]
+                    status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES
                 ).order_by('-scheduled').values('scheduled')[:1]
             ),
             last_change=models.Subquery(

pyproject.toml

@@ -17,7 +17,7 @@ description = "A NetBox Switch Configuration Backup Plugin"
 readme = "README.md"
 requires-python = ">=3.10"
 keywords = ["netbox-plugin", ]
-version = "2.1.7"
+version = "2.1.8"
 license = {file = "LICENSE"}
 classifiers = [
     "Programming Language :: Python :: 3",