fix(core): Isolate auto-sync failures per object
An exception from any single object exited the auto_sync receiver, so every record after it kept stale data. Each object now synchronizes in its own transaction and all failures are reported together, with stale records removed rather than failing the sync. Fixes #23122
This commit is contained in:
Martin Hauser
parent
d2191e0fb3
commit
ce3aa55241
|
|
@ -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'
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
"""
|
||||
|
|
|
|||
Loading…
Reference in New Issue