diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 2244d61..e78dd0c 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -268,12 +268,14 @@ class Ingestor(SientiaMonitoring): overleases = list(self.ingestor_manager.managed_tags.keys())[1:] - await self.ingestor_manager.drop_slot_leases(overleases) - + # Unsubscribe before giving the leases back: another pod may take a released + # slot right away, and must not find this pod still reading it for lease in overleases: await self.ingestor_manager.unsubscribe_slot(lease) self.ingestor_manager.managed_tags.pop(lease) + await self.ingestor_manager.drop_slot_leases(overleases) + await self.emit_metric( metric_object=metrics.SLOTS_MANAGED, method='set', diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index caffaa9..5520ee8 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -549,17 +549,46 @@ class IngestorManager(SientiaMonitoring): also left the subscription open on the OPC server. Unsubscribing a slot that no reader holds is a no-op. + A reader that fails to unsubscribe (e.g. the legacy OpcManager, whose delete() + raises when the server session is gone) is notified and logged, and the slot is + dropped from its books, as the v2 engine already does, so it can be subscribed + again. The other readers are still unsubscribed and the main loop goes on. + Args: slot (str): The name of the slot to unsubscribe. Side Effects: - Removes the slot's subscription from every OPC reader that has one - Cleans up subscription resources on the OPC servers + - Sends OPC_UNSUBSCRIBE_ERROR_: when a reader fails """ - for reader in self.opc_managers.values(): - if slot in reader.subscriptions: + # list() snapshots the readers: the loop awaits, so a reader added or removed + # meanwhile must not break the iteration. + for server, reader in list(self.opc_managers.items()): # NOSONAR + if slot not in reader.subscriptions: + continue + try: await reader.unsubscribe(slot) + except Exception as e: + trace = traceback.format_exc() + await self.send_notification_async( + metadata=self.metadata, + notification_id=f'OPC_UNSUBSCRIBE_ERROR_{slot}:{server}', + message=f'Failed to unsubscribe slot {slot} from {server}: {e}', + block='opc_manager', + level=NotificationLevel.ERROR, + attachment_content=trace, + ) + self.logger.error(trace) + reader.subscriptions.pop(slot, None) + + async def _unsubscribe_removed_slots(self, slots: list[str]) -> None: + """Unsubscribe and forget slots whose configuration is gone from Redis.""" + for slot in slots: + self.logger.warning(f'Slot {slot} is no longer configured; unsubscribing it') + await self.unsubscribe_slot(slot) + self.managed_tags.pop(slot, None) async def update_slot_config(self): """ @@ -569,7 +598,7 @@ class IngestorManager(SientiaMonitoring): This method performs the following steps: 1. Renews the lease for each managed slot using the resource manager. 2. Fetches the latest configuration for each slot. - 3. Logs and removes slots whose configurations are no longer available. + 3. Logs, unsubscribes and removes slots whose configurations are no longer available. 4. Updates the configuration of slots if changes are detected. 5. Unsubscribes and re-subscribes to slots with updated configurations. 6. Removes slots from the managed tags if they are no longer valid. @@ -602,8 +631,9 @@ class IngestorManager(SientiaMonitoring): continue self.managed_tags[slot] = update - for slot in removed_slots: - self.managed_tags.pop(slot, None) + # Unsubscribe right away: leaving it to update_ingestor_manager's snapshot diff + # would leak the subscription if anything in between failed + await self._unsubscribe_removed_slots(removed_slots) def parse_slot_config(self, slot: str, raw: dict) -> dict | None: """ diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py new file mode 100644 index 0000000..4119e99 --- /dev/null +++ b/tests/unit/conftest.py @@ -0,0 +1,20 @@ +from unittest.mock import AsyncMock, MagicMock + +from pytest import fixture + + +@fixture +def reader_double(): + """Factory of OPC reader doubles holding subscriptions for the given slots. + + Like the real readers (ServerReader, OpcManager), `subscriptions` is keyed by slot + and `unsubscribe` drops the slot from it. + """ + + def make(*slots): + reader = MagicMock() + reader.subscriptions = {slot: MagicMock() for slot in slots} + reader.unsubscribe = AsyncMock(side_effect=lambda slot: reader.subscriptions.pop(slot)) + return reader + + return make diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index f39b2ef..4c14b29 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -1,5 +1,5 @@ import gc -from unittest.mock import AsyncMock, MagicMock, call, patch +from unittest.mock import ANY, AsyncMock, MagicMock, call, patch from pytest import fixture, mark from sientia_do.notifications.models import NotificationLevel @@ -8,6 +8,8 @@ from ingestor.config import GatewayConfig from ingestor.connection.pool import OpcConnectionPool from ingestor.io.slots import parse_slot from ingestor.managers.ingestor_manager import IngestorManager +from ingestor.managers.opc_manager import OpcManager +from ingestor.managers.server_reader import ServerReader metadata = { 'metadata': { @@ -547,24 +549,16 @@ async def test_get_slot_leases_1_failure(ingestor_manager): assert result == {} -def _reader(*slots): - """An OPC reader double whose unsubscribe drops the slot, like the real readers.""" - reader = MagicMock() - reader.subscriptions = dict.fromkeys(slots, MagicMock()) - reader.unsubscribe = AsyncMock(side_effect=lambda slot: reader.subscriptions.pop(slot)) - return reader - - @mark.asyncio -async def test_unsubscribe_slot(ingestor_manager): +async def test_unsubscribe_slot(ingestor_manager, reader_double): ingestor_manager.managed_tags = { 'slot1': {'server1': {'tags': 'config1'}, 'server2': {'tags': 'config2'}}, 'slot2': {'server3': {'tags': 'config3'}, 'server1': {'tags': 'config1'}}, } ingestor_manager.opc_managers = { - 'server1': _reader('slot1', 'slot2'), - 'server2': _reader('slot1'), - 'server3': _reader('slot2'), + 'server1': reader_double('slot1', 'slot2'), + 'server2': reader_double('slot1'), + 'server3': reader_double('slot2'), } await ingestor_manager.unsubscribe_slot('slot1') @@ -575,12 +569,12 @@ async def test_unsubscribe_slot(ingestor_manager): @mark.asyncio -async def test_unsubscribe_slot_not_in_managed_tags(ingestor_manager): +async def test_unsubscribe_slot_not_in_managed_tags(ingestor_manager, reader_double): """The slot is already gone from managed_tags: the readers still holding it are unsubscribed""" ingestor_manager.managed_tags = {'slot1': {'server1': {'tags': 'config1'}}} ingestor_manager.opc_managers = { - 'server1': _reader('slot1', 'slot3'), - 'server2': _reader('slot3'), + 'server1': reader_double('slot1', 'slot3'), + 'server2': reader_double('slot3'), } await ingestor_manager.unsubscribe_slot('slot3') @@ -590,10 +584,10 @@ async def test_unsubscribe_slot_not_in_managed_tags(ingestor_manager): @mark.asyncio -async def test_unsubscribe_slot_twice_is_a_no_op(ingestor_manager): +async def test_unsubscribe_slot_twice_is_a_no_op(ingestor_manager, reader_double): """A second unsubscribe of the same slot (lease already dropped) does nothing and never raises""" ingestor_manager.managed_tags = {} - ingestor_manager.opc_managers = {'server1': _reader('slot2')} + ingestor_manager.opc_managers = {'server1': reader_double('slot2')} await ingestor_manager.unsubscribe_slot('slot2') await ingestor_manager.unsubscribe_slot('slot2') @@ -601,6 +595,95 @@ async def test_unsubscribe_slot_twice_is_a_no_op(ingestor_manager): ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot2') +@mark.asyncio +async def test_unsubscribe_slot_reader_failure_does_not_raise(ingestor_manager, reader_double): + """A reader that fails to unsubscribe is notified and forgotten; the others still go""" + broken = reader_double('slot1') + broken.unsubscribe = AsyncMock(side_effect=ConnectionError('session closed')) + healthy = reader_double('slot1') + ingestor_manager.opc_managers = {'server1': broken, 'server2': healthy} + + await ingestor_manager.unsubscribe_slot('slot1') + + healthy.unsubscribe.assert_called_once_with('slot1') + assert 'slot1' not in broken.subscriptions + ingestor_manager.send_notification_async.assert_called_once_with( + metadata=ingestor_manager.metadata, + notification_id='OPC_UNSUBSCRIBE_ERROR_slot1:server1', + message='Failed to unsubscribe slot slot1 from server1: session closed', + block='opc_manager', + level=NotificationLevel.ERROR, + attachment_content=ANY, + ) + + +@mark.asyncio +async def test_unsubscribe_slot_with_real_readers(ingestor_manager): + """Contract: both real readers expose `subscriptions` keyed by slot, and + unsubscribe_slot closes the slot's subscription on each of them""" + pool = MagicMock(get=AsyncMock(), evict=AsyncMock()) + server_reader = ServerReader( + name='kepserver', + server_config={'server_id': '1', 'url': 'opc.tcp://kepserver:49320'}, + pool=pool, + sink=AsyncMock(), + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=AsyncMock(), + metadata={}, + ) + server_reader.engine.emit_metric = AsyncMock() + v2_subscription = AsyncMock() + server_reader.engine.subscriptions['slot1'] = v2_subscription + server_reader.engine._servers['slot1'] = '1' + + with patch('ingestor.managers.opc_manager.metrics'): + legacy_reader = OpcManager( + name='legacy', + url='opc.tcp://legacy:4840', + data_manager=AsyncMock(), + subscription_period_ms=1000, + logger=MagicMock(), + server_uri='opc.tcp://legacy:4840', + notification_handler=MagicMock(), + metadata={}, + metrics_controller=AsyncMock(), + ) + legacy_subscription = AsyncMock() + legacy_reader.subscriptions['slot1'] = legacy_subscription + + ingestor_manager.opc_managers = {'kepserver': server_reader, 'legacy': legacy_reader} + + await ingestor_manager.unsubscribe_slot('slot1') + + v2_subscription.delete.assert_awaited_once() + legacy_subscription.delete.assert_awaited_once() + assert 'slot1' not in server_reader.subscriptions + assert 'slot1' not in legacy_reader.subscriptions + pool.unpin.assert_called_once_with('1') + + +@mark.asyncio +async def test_update_slot_config_unsubscribes_removed_slot(ingestor_manager, reader_double): + """A slot deleted from Redis is unsubscribed right away, not left to the main loop""" + ingestor_manager.managed_tags = { + 'slot1': parse_slot(raw_slot_v1('Same')), + 'slot2': parse_slot(raw_slot_v1('Same')), + } + reader = reader_double('slot1', 'slot2') + ingestor_manager.opc_managers = {'test_server': reader} + ingestor_manager.resource_manager.get_tag_slot = AsyncMock( + side_effect=lambda slot: raw_slot_v1('Same') if slot == 'slot1' else None + ) + ingestor_manager.resource_manager.renew_tag_lease = AsyncMock() + + await ingestor_manager.update_slot_config() + + reader.unsubscribe.assert_called_once_with('slot2') + assert list(ingestor_manager.managed_tags) == ['slot1'] + assert list(reader.subscriptions) == ['slot1'] + + @mark.asyncio async def test_update_slot_config(ingestor_manager): ingestor_manager.managed_tags = { diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index bd2e03c..de8361f 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -231,6 +231,10 @@ async def test_manage_leases_no_available_slots_extra_sltos(ingestor_manager_sta ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_called_once_with( ['tag2', 'tag3'] ) + # The extra slots are unsubscribed before their leases are given back + calls = [c[0] for c in ingestor_manager_started.ingestor_manager.mock_calls] + assert calls.index('drop_slot_leases') > calls.index('unsubscribe_slot') + assert list(ingestor_manager_started.ingestor_manager.managed_tags) == ['tag1'] @mark.asyncio @@ -345,19 +349,12 @@ def _manager_with_real_unsubscribe(managed_tags, readers): return manager -def _reader(*slots): - reader = MagicMock() - reader.subscriptions = dict.fromkeys(slots, MagicMock()) - reader.unsubscribe = AsyncMock(side_effect=lambda slot: reader.subscriptions.pop(slot)) - return reader - - @mark.asyncio -async def test_dropped_extra_lease_does_not_stop_the_loop(ingestor_manager_started): +async def test_dropped_extra_lease_does_not_stop_the_loop(ingestor_manager_started, reader_double): """Regression: manage_leases drops slot 2, then update_ingestor_manager sees slot 2 in the snapshot only and unsubscribes it again. That used to raise KeyError and stop the main loop (container restart).""" - reader = _reader('1', '2') + reader = reader_double('1', '2') ingestor_manager_started.ingestor_manager = _manager_with_real_unsubscribe( {'1': {'kepserver': {}}, '2': {'kepserver': {}}}, {'kepserver': reader} ) @@ -372,11 +369,11 @@ async def test_dropped_extra_lease_does_not_stop_the_loop(ingestor_manager_start @mark.asyncio -async def test_slot_removed_from_redis_is_unsubscribed(ingestor_manager_started): +async def test_slot_removed_from_redis_is_unsubscribed(ingestor_manager_started, reader_double): """Regression: update_slot_config drops a slot deleted by the orchestrator from managed_tags without unsubscribing it; update_ingestor_manager must close its subscription instead of raising KeyError.""" - reader = _reader('1', '3') + reader = reader_double('1', '3') ingestor_manager_started.ingestor_manager = _manager_with_real_unsubscribe( {'1': {'kepserver': {}}}, {'kepserver': reader} )