Compare commits

...

2 Commits

Author SHA1 Message Date
Jeremy Stretch
c40640af81 Omit the system filepath north of the installation root 2026-03-04 13:47:54 -05:00
Jeremy Stretch
3c6596de8f Closes #20916: Record a stack trace in the job log for unhandled exceptions 2026-03-04 13:39:08 -05:00
3 changed files with 30 additions and 0 deletions

View File

@@ -1,4 +1,6 @@
import django_tables2 as tables
from django.utils.html import conditional_escape
from django.utils.safestring import mark_safe
from django.utils.translation import gettext_lazy as _
from core.constants import JOB_LOG_ENTRY_LEVELS
@@ -82,3 +84,9 @@ class JobLogEntryTable(BaseTable):
class Meta(BaseTable.Meta):
empty_text = _('No log entries')
fields = ('timestamp', 'level', 'message')
def render_message(self, record, value):
if record.get('level') == 'error' and '\n' in value:
value = conditional_escape(value)
return mark_safe(f'<pre class="p-0">{value}</pre>')
return value

View File

@@ -1,6 +1,9 @@
import logging
import os
import traceback
from abc import ABC, abstractmethod
from datetime import timedelta
from pathlib import Path
from django.core.exceptions import ImproperlyConfigured
from django.utils import timezone
@@ -21,6 +24,11 @@ __all__ = (
'system_job',
)
# The installation root, e.g. "/opt/netbox/". Used to strip absolute path
# prefixes from traceback file paths before recording them in the job log.
# 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
def system_job(interval):
"""
@@ -107,6 +115,13 @@ class JobRunner(ABC):
job.terminate(status=JobStatusChoices.STATUS_FAILED)
except Exception as e:
tb_str = traceback.format_exc().replace(_INSTALL_ROOT, '')
tb_record = logging.makeLogRecord({
'levelno': logging.ERROR,
'levelname': 'ERROR',
'msg': tb_str,
})
job.log(tb_record)
job.terminate(status=JobStatusChoices.STATUS_ERRORED, error=repr(e))
if type(e) is JobTimeoutException:
logger.error(e)

View File

@@ -10,6 +10,7 @@ from core.models import DataSource, Job
from utilities.testing import disable_warnings
from ..jobs import *
from ..jobs import _INSTALL_ROOT
class TestJobRunner(JobRunner):
@@ -83,6 +84,12 @@ class JobRunnerTest(JobRunnerTestCase):
self.assertEqual(job.status, JobStatusChoices.STATUS_ERRORED)
self.assertEqual(job.error, repr(ErroredJobRunner.EXP))
self.assertEqual(len(job.log_entries), 1)
self.assertEqual(job.log_entries[0]['level'], 'error')
tb_message = job.log_entries[0]['message']
self.assertIn('Traceback', tb_message)
self.assertIn('Test error', tb_message)
self.assertNotIn(_INSTALL_ROOT, tb_message)
class EnqueueTest(JobRunnerTestCase):