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 0be1773..4b3d8f8 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -538,26 +538,84 @@ class IngestorManager(SientiaMonitoring): async def unsubscribe_slot(self, slot: str): """ - Unsubscribes a specific slot from all associated OPC servers. + Unsubscribes a specific slot from every OPC reader that holds a subscription for it. - This method removes all subscriptions for a given slot across all - OPC servers that were managing it. It ensures clean cleanup of - resources when slots are released or reconfigured. + The readers' own subscriptions are the source of truth, not managed_tags: the + main loop removes a slot from managed_tags before update_ingestor_manager asks + to unsubscribe it, both when manage_leases drops an extra lease (which already + unsubscribed it) and when update_slot_config finds the slot gone from Redis + (which did not). Looking the slot up in managed_tags raised KeyError in both + cases, which stopped the main loop and restarted the container; the second case + 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 is handled by _unsubscribe_reader: the other + readers are still unsubscribed and the main loop goes on. Args: slot (str): The name of the slot to unsubscribe. - Raises: - KeyError: If the specified slot does not exist in the managed tags. - Side Effects: - - Removes subscriptions from all OPC servers for the specified slot + - Removes the slot's subscription from every OPC reader that has one - Cleans up subscription resources on the OPC servers """ - for server in self.managed_tags[slot].keys(): - if server in self.opc_managers: - await self.opc_managers[server].unsubscribe(slot) + # list() snapshots the readers: the loop awaits, so a reader added or removed + # meanwhile must not break the iteration. SonarQube flags it as an unnecessary + # list() (S7504); the issue is accepted in SonarQube for this reason. + for server, reader in list(self.opc_managers.items()): + if slot in reader.subscriptions: + await self._unsubscribe_reader(server, reader, slot) + + async def _unsubscribe_reader( + self, server: str, reader: ServerReader | OpcManager, slot: str + ) -> None: + """ + Unsubscribe one reader from a slot without letting a failure stop the main loop. + + A failure (e.g. the legacy OpcManager, whose delete() raises when the server + session is gone) increments OPC_SUBSCRIPTION_ERRORS, sends + OPC_UNSUBSCRIBE_ERROR_: and is logged. The slot is dropped from the + reader's books, as the v2 engine already does, so it can be subscribed again. + + Known limitation: if delete() failed while the session is still alive (e.g. a + timeout), the server-side subscription may keep delivering data until that + session ends (reconnection or pod restart). + + Args: + server (str): The reader's key in opc_managers + reader (ServerReader | OpcManager): The OPC reader + slot (str): The slot to unsubscribe + """ + try: + await reader.unsubscribe(slot) + except Exception as e: + await self.emit_metric( + metric_object=metrics.OPC_SUBSCRIPTION_ERRORS, + tags={ + 'pod_id': self.pod_id, + 'server': server, + 'slot': slot, + }, + ) + 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): """ @@ -567,7 +625,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. @@ -600,8 +658,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: """ @@ -764,7 +823,7 @@ class IngestorManager(SientiaMonitoring): self.logger.error(trace) self.logger.warning(f'Removing subscription from server {server} for slot {slot}') - await self.opc_managers[server].unsubscribe(slot) + await self._unsubscribe_reader(server, self.opc_managers[server], slot) return 2 return 0 diff --git a/ingestor/metrics.py b/ingestor/metrics.py index a9ee17f..9b7e50b 100644 --- a/ingestor/metrics.py +++ b/ingestor/metrics.py @@ -93,7 +93,7 @@ OPC_MANAGERS_ACTIVE = Gauge( ) OPC_SUBSCRIPTION_ERRORS = Counter( 'ingestor_opc_subscription_errors_total', - 'Errors when trying to subscribe to OPC tags', + 'Errors when trying to subscribe to or unsubscribe from OPC tags', ['pod_id', 'server', 'slot'], ) 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 df27f66..a91b1a5 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -1,13 +1,16 @@ 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 +import ingestor.metrics as metrics 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': { @@ -548,21 +551,153 @@ async def test_get_slot_leases_1_failure(ingestor_manager): @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': AsyncMock(), - 'server2': AsyncMock(), - 'server3': AsyncMock(), + 'server1': reader_double('slot1', 'slot2'), + 'server2': reader_double('slot1'), + 'server3': reader_double('slot2'), } await ingestor_manager.unsubscribe_slot('slot1') ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot1') ingestor_manager.opc_managers['server2'].unsubscribe.assert_called_once_with('slot1') ingestor_manager.opc_managers['server3'].unsubscribe.assert_not_called() + assert list(ingestor_manager.opc_managers['server1'].subscriptions) == ['slot2'] + + +@mark.asyncio +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_double('slot1', 'slot3'), + 'server2': reader_double('slot3'), + } + + await ingestor_manager.unsubscribe_slot('slot3') + + ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot3') + ingestor_manager.opc_managers['server2'].unsubscribe.assert_called_once_with('slot3') + + +@mark.asyncio +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_double('slot2')} + + await ingestor_manager.unsubscribe_slot('slot2') + await ingestor_manager.unsubscribe_slot('slot2') + + 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.emit_metric.assert_called_once_with( + metric_object=metrics.OPC_SUBSCRIPTION_ERRORS, + tags={'pod_id': ingestor_manager.pod_id, 'server': 'server1', 'slot': 'slot1'}, + ) + 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 keep `subscriptions` keyed by slot, and + unsubscribe_slot closes the slot's subscription on each of them""" + v2_subscription = AsyncMock() + client = MagicMock(create_subscription=AsyncMock(return_value=v2_subscription)) + pool = MagicMock(get=AsyncMock(return_value=client), evict=AsyncMock()) + server_reader = ServerReader( + name='kepserver', + server_config={ + 'server_id': '1', + 'url': 'opc.tcp://kepserver:49320', + 'subscription_period_ms': 1000.0, + }, + pool=pool, + sink=AsyncMock(), + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=AsyncMock(), + metadata={}, + ) + server_reader.engine.emit_metric = AsyncMock() + await server_reader.create_subscription('slot1') + + 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_reader.emit_metric = AsyncMock() + legacy_subscription = AsyncMock() + legacy_reader.client = MagicMock( + create_subscription=AsyncMock(return_value=legacy_subscription) + ) + await legacy_reader.create_subscription('slot1') + + assert 'slot1' in server_reader.subscriptions + assert 'slot1' in legacy_reader.subscriptions + 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.pin.assert_called_once_with('1') + 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 @@ -646,6 +781,30 @@ async def test_manage_server_create_subscription_failure(ingestor_manager): ingestor_manager.opc_managers['server1'].subscribe.assert_not_called() +@mark.asyncio +async def test_manage_server_subscribe_and_unsubscribe_failure(ingestor_manager): + """subscribe() fails and so does the cleanup unsubscribe (legacy reader, session + gone): manage_server reports the failure and nothing escapes to the main loop""" + reader = AsyncMock() + reader.subscriptions = {} + reader.subscribe.side_effect = Exception('Subscription error') + reader.unsubscribe.side_effect = ConnectionError('session closed') + ingestor_manager.opc_managers = {'server1': reader} + server_config = {'tags': {'tag1': 'config1'}} + + result = await ingestor_manager.manage_server('slot1', 'server1', server_config, server_config) + + assert result == 2 + reader.unsubscribe.assert_called_once_with('slot1') + notified = [ + c.kwargs['notification_id'] for c in ingestor_manager.send_notification_async.call_args_list + ] + assert notified == [ + 'OPC_SUBSCRIPTION_ERROR_slot1:server1', + 'OPC_UNSUBSCRIBE_ERROR_slot1:server1', + ] + + @mark.asyncio async def test_manage_server(ingestor_manager): ingestor_manager.opc_managers = {'server1': AsyncMock(), 'server2': AsyncMock()} diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 8bbccc4..5c404b3 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -1,8 +1,10 @@ +from copy import deepcopy from unittest.mock import ANY, AsyncMock, MagicMock, call, patch from pytest import fixture, mark from ingestor.ingestor import Ingestor +from ingestor.managers.ingestor_manager import IngestorManager @patch('ingestor.ingestor.getenv') @@ -229,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 @@ -332,3 +338,52 @@ async def test_update_ingestor_manager_no_ingestor_manager(ingestor_manager_star ingestor_manager_started.ingestor_manager = None await ingestor_manager_started.update_ingestor_manager({}) + + +def _manager_with_real_unsubscribe(managed_tags, readers): + """A mocked IngestorManager whose unsubscribe path (unsubscribe_slot and + _unsubscribe_reader) is the real one.""" + manager = AsyncMock(update_opc_servers=AsyncMock(), subscribe_to_tags=AsyncMock()) + manager.managed_tags = managed_tags + manager.opc_managers = readers + manager.unsubscribe_slot = lambda slot: IngestorManager.unsubscribe_slot(manager, slot) + manager._unsubscribe_reader = lambda server, reader, slot: IngestorManager._unsubscribe_reader( + manager, server, reader, slot + ) + return manager + + +@mark.asyncio +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_double('1', '2') + ingestor_manager_started.ingestor_manager = _manager_with_real_unsubscribe( + {'1': {'kepserver': {}}, '2': {'kepserver': {}}}, {'kepserver': reader} + ) + snapshot = deepcopy(ingestor_manager_started.ingestor_manager.managed_tags) + + await ingestor_manager_started.manage_leases(0, 0, 1) + await ingestor_manager_started.update_ingestor_manager(snapshot) + + reader.unsubscribe.assert_called_once_with('2') + assert list(reader.subscriptions) == ['1'] + assert list(ingestor_manager_started.ingestor_manager.managed_tags) == ['1'] + + +@mark.asyncio +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_double('1', '3') + ingestor_manager_started.ingestor_manager = _manager_with_real_unsubscribe( + {'1': {'kepserver': {}}}, {'kepserver': reader} + ) + snapshot = {'1': {'kepserver': {}}, '3': {'kepserver': {}}} + + await ingestor_manager_started.update_ingestor_manager(snapshot) + + reader.unsubscribe.assert_called_once_with('3') + assert list(reader.subscriptions) == ['1']