diff --git a/netbox/core/models/data.py b/netbox/core/models/data.py index 1de20e194..6e8b4c703 100644 --- a/netbox/core/models/data.py +++ b/netbox/core/models/data.py @@ -6,7 +6,6 @@ from urllib.parse import urlparse import yaml from django.conf import settings -from django.contrib.contenttypes.fields import GenericForeignKey from django.core.exceptions import ValidationError from django.core.validators import RegexValidator from django.db import models @@ -18,6 +17,7 @@ from netbox.constants import CENSOR_TOKEN, CENSOR_TOKEN_CHANGED from netbox.models import PrimaryModel from netbox.models.features import JobsMixin from netbox.registry import registry +from utilities.fields import RestrictedGenericForeignKey from utilities.querysets import RestrictedQuerySet from ..choices import * @@ -393,7 +393,7 @@ class AutoSyncRecord(models.Model): related_name='+' ) object_id = models.PositiveBigIntegerField() - object = GenericForeignKey( + object = RestrictedGenericForeignKey( ct_field='object_type', fk_field='object_id' ) diff --git a/netbox/core/signals.py b/netbox/core/signals.py index cfd10c988..28fa6cae6 100644 --- a/netbox/core/signals.py +++ b/netbox/core/signals.py @@ -4,15 +4,19 @@ from threading import local from django.contrib.contenttypes.models import ContentType from django.core.exceptions import ObjectDoesNotExist, ValidationError from django.core.signals import request_finished +from django.db import transaction from django.db.models import CASCADE, RESTRICT from django.db.models.fields.reverse_related import ManyToManyRel, ManyToOneRel from django.db.models.signals import m2m_changed, post_migrate, post_save, pre_delete from django.dispatch import Signal, receiver from django.utils.translation import gettext_lazy as _ +from django.utils.translation import ngettext from django_prometheus.models import model_deletes, model_inserts, model_updates +from rq.timeouts import JobTimeoutException from core.choices import JobStatusChoices, ObjectChangeActionChoices from core.events import * +from core.exceptions import SyncError from core.models import ObjectType from extras.events import enqueue_event from extras.models import Tag @@ -25,6 +29,8 @@ from utilities.exceptions import AbortRequest from .models import ConfigRevision, DataSource, ObjectChange +logger = logging.getLogger('netbox.core.signals') + __all__ = ( 'clear_events', 'job_end', @@ -299,6 +305,10 @@ def enqueue_sync_job(instance, created, **kwargs): job.delete() +# Keeps the aggregated error readable when a whole source fails at once +_AUTO_SYNC_DETAIL_LIMIT = 10 + + @receiver(post_sync) def auto_sync(instance, **kwargs): """ @@ -306,8 +316,56 @@ def auto_sync(instance, **kwargs): """ from .models import AutoSyncRecord - for autosync in AutoSyncRecord.objects.filter(datafile__source=instance).prefetch_related('object'): - autosync.object.sync(save=True) + failure_count = 0 + details = [] + first_error = None + + records = AutoSyncRecord.objects.filter(datafile__source=instance).order_by('pk') + for autosync in records.select_related('object_type').prefetch_related('object'): + # The object may be unresolvable or mid-failure, so identify by keys + target = f'{autosync.object_type.app_label}.{autosync.object_type.model} ID {autosync.object_id}' + if autosync.object_type.model_class() is None: + # Not an orphaned row, so leave it for remove_stale_contenttypes + logger.warning(f"Skipping AutoSyncRecord for uninstalled model {target}") + continue + try: + # The try must stay outside this, so the savepoint is rolled back before the handler runs + with transaction.atomic(): + obj = autosync.object + if obj is None: + # The prefetch resolves through the default manager, so recheck with the base manager + try: + obj = autosync.object_type.get_object_for_this_type(pk=autosync.object_id) + except ObjectDoesNotExist: + logger.warning(f"Deleting stale AutoSyncRecord for {target}") + autosync.delete() + continue + obj.sync(save=True) + except JobTimeoutException: + # rq arms one alarm per job, so a timeout is not an ordinary per-object failure + raise + except Exception as e: + failure_count += 1 + if first_error is None: + first_error = e + # Not capped, unlike the raised message below + logger.error(f"Error auto-syncing {target}: {e}", exc_info=True) + if len(details) < _AUTO_SYNC_DETAIL_LIMIT: + details.append(f'- {target}: {type(e).__name__}: {e}') + + if first_error is not None: + summary = ngettext( + 'Automatic synchronization failed for {count} object:', + 'Automatic synchronization failed for {count} objects:', + failure_count, + ).format(count=failure_count) + if omitted := failure_count - len(details): + details.append(ngettext( + 'and {count} more failure.', + 'and {count} more failures.', + omitted, + ).format(count=omitted)) + raise SyncError('\n'.join([summary, *details])) from first_error @receiver(post_save, sender=ConfigRevision) diff --git a/netbox/core/tests/test_signals.py b/netbox/core/tests/test_signals.py index d691e72e2..5f47bde44 100644 --- a/netbox/core/tests/test_signals.py +++ b/netbox/core/tests/test_signals.py @@ -5,16 +5,19 @@ from unittest.mock import MagicMock, Mock, patch from django.contrib.contenttypes.models import ContentType from django.core.exceptions import ValidationError from django.core.signals import request_finished -from django.db import transaction +from django.db import IntegrityError, transaction from django.test import RequestFactory, SimpleTestCase, TestCase, override_settings +from django.utils import timezone +from rq.timeouts import JobTimeoutException from core import signals from core.choices import DataSourceStatusChoices, JobStatusChoices, ObjectChangeActionChoices -from core.models import ConfigRevision, DataSource, ObjectChange, ObjectType +from core.exceptions import SyncError +from core.models import AutoSyncRecord, ConfigRevision, DataFile, DataSource, ObjectChange, ObjectType from core.signals import _signals_received, clear_events, post_sync from dcim.choices import InterfaceTypeChoices from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site, SiteGroup -from extras.models import Tag +from extras.models import ConfigContext, Tag from extras.validators import CustomValidator from netbox.context import events_queue from netbox.context_managers import event_tracking @@ -348,28 +351,223 @@ class AutoSyncSignalTestCase(TestCase): DataSource when post_sync fires. """ - def test_post_sync_resyncs_dependent_records(self): - ds = DataSource.objects.create( + @classmethod + def setUpTestData(cls): + cls.datasource = DataSource.objects.create( name='DS 1', type='local', source_url='/tmp/ds1', status=DataSourceStatusChoices.COMPLETED, ) - record_a = SimpleNamespace(object=SimpleNamespace(synced=False)) + cls.object_type = ObjectType.objects.get_for_model(ConfigContext) + + def make_record(self, object_id, obj): + """Stand in for an AutoSyncRecord, carrying the attributes the receiver reads.""" + return SimpleNamespace(object=obj, object_type=self.object_type, object_id=object_id) + + def patch_records(self, autosync_model, records): + """Point the patched manager's queryset chain at the given stand-in records.""" + queryset = autosync_model.objects.filter.return_value.order_by.return_value.select_related.return_value + queryset.prefetch_related.return_value = records + + def test_post_sync_resyncs_dependent_records(self): + record_a = self.make_record(1, SimpleNamespace(synced=False)) record_a.object.sync = lambda save: setattr(record_a.object, 'synced', save) - record_b = SimpleNamespace(object=SimpleNamespace(synced=False)) + record_b = self.make_record(2, SimpleNamespace(synced=False)) record_b.object.sync = lambda save: setattr(record_b.object, 'synced', save) with patch('core.models.AutoSyncRecord') as autosync_model: - autosync_model.objects.filter.return_value.prefetch_related.return_value = [ - record_a, - record_b, - ] - post_sync.send(sender=ds.__class__, instance=ds) + self.patch_records(autosync_model, [record_a, record_b]) + post_sync.send(sender=DataSource, instance=self.datasource) self.assertTrue(record_a.object.synced) self.assertTrue(record_b.object.synced) + def test_post_sync_continues_after_failed_record(self): + """A failing record leaves the connection usable for the records after it.""" + record_a = self.make_record(1, MagicMock()) + record_b = self.make_record(2, MagicMock()) + + def create_duplicate_datasource(save): + # Violates the unique constraint on DataSource.name + DataSource.objects.create(name=self.datasource.name, type='local', source_url='/tmp/duplicate') + + def create_datasource(save): + DataSource.objects.create(name='DS 2', type='local', source_url='/tmp/ds2') + + record_a.object.sync.side_effect = create_duplicate_datasource + record_b.object.sync.side_effect = create_datasource + + with patch('core.models.AutoSyncRecord') as autosync_model: + self.patch_records(autosync_model, [record_a, record_b]) + with self.assertLogs('netbox.core.signals', 'ERROR'), self.assertRaises(SyncError) as cm: + post_sync.send(sender=DataSource, instance=self.datasource) + + record_a.object.sync.assert_called_once_with(save=True) + record_b.object.sync.assert_called_once_with(save=True) + self.assertIsInstance(cm.exception.__cause__, IntegrityError) + self.assertIn('Automatic synchronization failed for 1 object:', str(cm.exception)) + self.assertIn('- extras.configcontext ID 1: IntegrityError:', str(cm.exception)) + self.assertTrue(DataSource.objects.filter(name='DS 2').exists()) + + def test_post_sync_reports_every_failure(self): + """Each failed object is named in the aggregated error.""" + record_a = self.make_record(1, MagicMock()) + record_b = self.make_record(2, MagicMock()) + record_a.object.sync.side_effect = ValueError('First failure') + record_b.object.sync.side_effect = RuntimeError('Second failure') + + with patch('core.models.AutoSyncRecord') as autosync_model: + self.patch_records(autosync_model, [record_a, record_b]) + with self.assertLogs('netbox.core.signals', 'ERROR'), self.assertRaises(SyncError) as cm: + post_sync.send(sender=DataSource, instance=self.datasource) + + record_a.object.sync.assert_called_once_with(save=True) + record_b.object.sync.assert_called_once_with(save=True) + self.assertEqual( + str(cm.exception), + 'Automatic synchronization failed for 2 objects:\n' + '- extras.configcontext ID 1: ValueError: First failure\n' + '- extras.configcontext ID 2: RuntimeError: Second failure', + ) + self.assertIsInstance(cm.exception.__cause__, ValueError) + + def test_post_sync_deletes_dangling_record(self): + """A record whose generic relation no longer resolves is removed instead of failing the sync.""" + dangling = MagicMock(object=None, object_type=self.object_type, object_id=7) + record = self.make_record(1, MagicMock()) + + with patch('core.models.AutoSyncRecord') as autosync_model: + self.patch_records(autosync_model, [dangling, record]) + with self.assertLogs('netbox.core.signals', 'WARNING'): + post_sync.send(sender=DataSource, instance=self.datasource) + + dangling.delete.assert_called_once_with() + record.object.sync.assert_called_once_with(save=True) + + def make_datafile(self, path='dir1/context.yaml'): + """Create a DataFile on the source the receiver filters by.""" + return DataFile.objects.create( + source=self.datasource, + path=path, + last_updated=timezone.now(), + size=1000, + hash='442da078f0111cbdf42f21903724f6597c692535f55bdfbbea758a1ae99ad9e1', + data=b'value: original', + ) + + def test_post_sync_skips_record_for_uninstalled_model(self): + """A record whose model is no longer installed is skipped rather than deleted.""" + # get_for_id() caches the ghost type in the manager, which the rollback does not undo + self.addCleanup(ContentType.objects.clear_cache) + stale_type = ContentType.objects.create(app_label='ghost_plugin', model='ghostmodel') + AutoSyncRecord.objects.create( + datafile=self.make_datafile('dir1/ghost.yaml'), + object_type=stale_type, + object_id=1, + ) + + with self.assertLogs('netbox.core.signals', 'WARNING') as logs: + post_sync.send(sender=DataSource, instance=self.datasource) + + # remove_stale_contenttypes owns this cleanup and cascades to the record + self.assertTrue(AutoSyncRecord.objects.filter(object_type=stale_type).exists()) + self.assertIn('ghost_plugin.ghostmodel ID 1', logs.output[0]) + + def test_post_sync_deletes_real_dangling_record(self): + """The stale record is removed from the database, exercising the unmocked queryset.""" + datafile = self.make_datafile() + AutoSyncRecord.objects.create( + datafile=datafile, + object_type=ObjectType.objects.get_for_model(ConfigContext), + object_id=99999, + ) + + with self.assertLogs('netbox.core.signals', 'WARNING') as logs: + post_sync.send(sender=DataSource, instance=self.datasource) + + self.assertFalse(AutoSyncRecord.objects.filter(datafile=datafile).exists()) + # The mocked tests must not assert a format the real content type never emits + self.assertIn('extras.configcontext ID 99999', logs.output[0]) + + def test_post_sync_syncs_record_hidden_by_default_manager(self): + """A target the default manager excludes is synced through the base manager, not deleted.""" + datafile = self.make_datafile('dir1/hidden.yaml') + context = ConfigContext.objects.create( + name='CC 1', + data={}, + data_source=self.datasource, + data_file=datafile, + data_path=datafile.path, + auto_sync_enabled=True, + ) + hidden = AutoSyncRecord.objects.get(object_type=self.object_type, object_id=context.pk) + missing = AutoSyncRecord.objects.create( + datafile=datafile, + object_type=self.object_type, + object_id=99999, + ) + + # The prefetch reads ConfigContext.objects, so an empty queryset hides a live target + with patch.object(ConfigContext, 'objects', ConfigContext.objects.none()): + with self.assertLogs('netbox.core.signals', 'WARNING') as logs: + post_sync.send(sender=DataSource, instance=self.datasource) + + context.refresh_from_db() + self.assertEqual(context.data, {'value': 'original'}) + self.assertTrue(context.is_synced) + self.assertTrue(AutoSyncRecord.objects.filter(pk=hidden.pk).exists()) + self.assertFalse(AutoSyncRecord.objects.filter(pk=missing.pk).exists()) + self.assertIn('extras.configcontext ID 99999', logs.output[0]) + + def test_post_sync_reports_dangling_record_when_cleanup_is_refused(self): + """A stale record is reported by its target keys when its own cleanup is blocked.""" + dangling = MagicMock(object=None, object_type=self.object_type, object_id=7) + dangling.delete.side_effect = AbortRequest('Deletion is prevented by a protection rule') + + with patch('core.models.AutoSyncRecord') as autosync_model: + self.patch_records(autosync_model, [dangling]) + with self.assertLogs('netbox.core.signals', 'ERROR'), self.assertRaises(SyncError) as cm: + post_sync.send(sender=DataSource, instance=self.datasource) + + self.assertIn('- extras.configcontext ID 7: AbortRequest:', str(cm.exception)) + self.assertNotIn('- None:', str(cm.exception)) + + def test_post_sync_caps_reported_failures(self): + """Only the first N failures are detailed, with the remainder counted.""" + records = [] + for i in range(signals._AUTO_SYNC_DETAIL_LIMIT + 3): + record = self.make_record(i, MagicMock()) + record.object.sync.side_effect = ValueError(f'Failure {i}') + records.append(record) + + with patch('core.models.AutoSyncRecord') as autosync_model: + self.patch_records(autosync_model, records) + with self.assertLogs('netbox.core.signals', 'ERROR') as logs, self.assertRaises(SyncError) as cm: + post_sync.send(sender=DataSource, instance=self.datasource) + + message = str(cm.exception) + self.assertIn(f'failed for {len(records)} objects:', message) + self.assertEqual(message.count('ValueError: Failure'), signals._AUTO_SYNC_DETAIL_LIMIT) + self.assertIn('and 3 more failures.', message) + # Capping the message must not lose an identity, so every failure is still logged + self.assertEqual(len(logs.records), len(records)) + # A capped list is only deterministic if the subset is ordered + autosync_model.objects.filter.return_value.order_by.assert_called_once_with('pk') + + def test_post_sync_propagates_job_timeout(self): + """An rq job timeout escapes the receiver instead of being recorded as a failure.""" + record_a = self.make_record(1, MagicMock()) + record_b = self.make_record(2, MagicMock()) + record_a.object.sync.side_effect = JobTimeoutException('Task exceeded maximum timeout value') + + with patch('core.models.AutoSyncRecord') as autosync_model: + self.patch_records(autosync_model, [record_a, record_b]) + with self.assertRaises(JobTimeoutException): + post_sync.send(sender=DataSource, instance=self.datasource) + + record_b.object.sync.assert_not_called() + class UpdateConfigSignalTestCase(TestCase): """