mirror of
https://github.com/netbox-community/netbox.git
synced 2025-07-21 03:27:21 -06:00
Introduce reusable BackgroundJob framework
A new abstract class can be used to implement job function classes. It handles the necessary logic for starting and stopping jobs, including exception handling and rescheduling of recurring jobs. This commit also includes the migration of data source jobs to the new framework.
This commit is contained in:
parent
388ba3d736
commit
5fab8e4839
@ -1,33 +1,32 @@
|
|||||||
import logging
|
import logging
|
||||||
|
|
||||||
from netbox.search.backends import search_backend
|
from utilities.jobs import BackgroundJob
|
||||||
from .choices import *
|
|
||||||
from .exceptions import SyncError
|
|
||||||
from .models import DataSource
|
|
||||||
from rq.timeouts import JobTimeoutException
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def sync_datasource(job, *args, **kwargs):
|
class SyncDataSourceJob(BackgroundJob):
|
||||||
"""
|
"""
|
||||||
Call sync() on a DataSource.
|
Call sync() on a DataSource.
|
||||||
"""
|
"""
|
||||||
datasource = DataSource.objects.get(pk=job.object_id)
|
|
||||||
|
|
||||||
try:
|
@classmethod
|
||||||
job.start()
|
def run(cls, job, *args, **kwargs):
|
||||||
datasource.sync()
|
from netbox.search.backends import search_backend
|
||||||
|
from .choices import DataSourceStatusChoices
|
||||||
|
from .exceptions import SyncError
|
||||||
|
from .models import DataSource
|
||||||
|
|
||||||
# Update the search cache for DataFiles belonging to this source
|
datasource = DataSource.objects.get(pk=job.object_id)
|
||||||
search_backend.cache(datasource.datafiles.iterator())
|
|
||||||
|
|
||||||
job.terminate()
|
try:
|
||||||
|
datasource.sync()
|
||||||
|
|
||||||
except Exception as e:
|
# Update the search cache for DataFiles belonging to this source
|
||||||
job.terminate(status=JobStatusChoices.STATUS_ERRORED, error=repr(e))
|
search_backend.cache(datasource.datafiles.iterator())
|
||||||
DataSource.objects.filter(pk=datasource.pk).update(status=DataSourceStatusChoices.FAILED)
|
|
||||||
if type(e) in (SyncError, JobTimeoutException):
|
except Exception as e:
|
||||||
logging.error(e)
|
DataSource.objects.filter(pk=datasource.pk).update(status=DataSourceStatusChoices.FAILED)
|
||||||
else:
|
if type(e) is SyncError:
|
||||||
|
logging.error(e)
|
||||||
raise e
|
raise e
|
||||||
|
@ -12,7 +12,6 @@ from django.core.validators import RegexValidator
|
|||||||
from django.db import models
|
from django.db import models
|
||||||
from django.urls import reverse
|
from django.urls import reverse
|
||||||
from django.utils import timezone
|
from django.utils import timezone
|
||||||
from django.utils.module_loading import import_string
|
|
||||||
from django.utils.translation import gettext as _
|
from django.utils.translation import gettext as _
|
||||||
|
|
||||||
from netbox.constants import CENSOR_TOKEN, CENSOR_TOKEN_CHANGED
|
from netbox.constants import CENSOR_TOKEN, CENSOR_TOKEN_CHANGED
|
||||||
@ -162,8 +161,8 @@ class DataSource(JobsMixin, PrimaryModel):
|
|||||||
DataSource.objects.filter(pk=self.pk).update(status=self.status)
|
DataSource.objects.filter(pk=self.pk).update(status=self.status)
|
||||||
|
|
||||||
# Enqueue a sync job
|
# Enqueue a sync job
|
||||||
return Job.enqueue(
|
from ..jobs import SyncDataSourceJob
|
||||||
import_string('core.jobs.sync_datasource'),
|
return SyncDataSourceJob.enqueue(
|
||||||
instance=self,
|
instance=self,
|
||||||
user=request.user
|
user=request.user
|
||||||
)
|
)
|
||||||
|
68
netbox/utilities/jobs.py
Normal file
68
netbox/utilities/jobs.py
Normal file
@ -0,0 +1,68 @@
|
|||||||
|
import logging
|
||||||
|
from abc import ABC, abstractmethod
|
||||||
|
from datetime import timedelta
|
||||||
|
|
||||||
|
from rq.timeouts import JobTimeoutException
|
||||||
|
|
||||||
|
from core.choices import JobStatusChoices
|
||||||
|
from core.models import Job
|
||||||
|
|
||||||
|
|
||||||
|
class BackgroundJob(ABC):
|
||||||
|
"""
|
||||||
|
Background Job helper class.
|
||||||
|
|
||||||
|
This class handles the execution of a background job. It is responsible for maintaining its state, reporting errors,
|
||||||
|
and scheduling recurring jobs.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
@abstractmethod
|
||||||
|
def run(cls, *args, **kwargs) -> None:
|
||||||
|
"""
|
||||||
|
Run the job.
|
||||||
|
|
||||||
|
A `BackgroundJob` class needs to implement this method to execute all commands of the job.
|
||||||
|
"""
|
||||||
|
pass
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def handle(cls, job, *args, **kwargs):
|
||||||
|
"""
|
||||||
|
Handle the execution of a `BackgroundJob`.
|
||||||
|
|
||||||
|
This method is called by the Job Scheduler to handle the execution of all job commands. It will maintain the
|
||||||
|
job's metadata and handle errors. For periodic jobs, a new job is automatically scheduled using its `interval'.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
job.start()
|
||||||
|
cls.run(job, *args, **kwargs)
|
||||||
|
job.terminate()
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
job.terminate(status=JobStatusChoices.STATUS_ERRORED, error=repr(e))
|
||||||
|
if type(e) is JobTimeoutException:
|
||||||
|
logging.error(e)
|
||||||
|
|
||||||
|
# If the executed job is a periodic job, schedule its next execution at the specified interval.
|
||||||
|
finally:
|
||||||
|
if job.interval:
|
||||||
|
new_scheduled_time = (job.scheduled or job.started) + timedelta(minutes=job.interval)
|
||||||
|
cls.enqueue(
|
||||||
|
instance=job.object,
|
||||||
|
name=job.name,
|
||||||
|
user=job.user,
|
||||||
|
schedule_at=new_scheduled_time,
|
||||||
|
interval=job.interval,
|
||||||
|
**kwargs,
|
||||||
|
)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def enqueue(cls, *args, **kwargs):
|
||||||
|
"""
|
||||||
|
Enqueue a new `BackgroundJob`.
|
||||||
|
|
||||||
|
This method is a wrapper of `Job.enqueue` using `handle()` as function callback. See its documentation for
|
||||||
|
parameters.
|
||||||
|
"""
|
||||||
|
return Job.enqueue(cls.handle, *args, **kwargs)
|
Loading…
Reference in New Issue
Block a user