From 6d2a0e46be95eb95ee4d97dbaeed0fefbce0fa35 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Oct 2025 11:46:02 -0300 Subject: [PATCH 1/4] Refactor OPC server integrity checks to be asynchronous Updated the Ingestor and IngestorManager classes to use async methods for checking OPC server integrity and shutting down OPC managers. Adjusted related tests to ensure proper mocking of asynchronous behavior. --- ingestor/ingestor.py | 2 +- ingestor/managers/ingestor_manager.py | 7 ++++--- ingestor/managers/opc_manager.py | 3 +++ tests/unit/managers/test_ingestor_manager.py | 7 ++++--- tests/unit/test_ingestor.py | 8 ++++++++ 5 files changed, 20 insertions(+), 7 deletions(-) diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 662f539..5f27759 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -414,7 +414,7 @@ class Ingestor: # Check OPC cycles self.logger.info('Checking OPC servers integrity...') - self.ingestor_manager.check_opc_servers_integrity() + await self.ingestor_manager.check_opc_servers_integrity() self.logger.info('Updating managed tags...') await self.update_ingestor_manager(current_managed_tags) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index bf964c0..f82116c 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -228,7 +228,7 @@ class IngestorManager(BaseActivity): elif server_instance.config != server_config: self.logger.warning(f'Reinitializing OPC manager for server {server}') - server_instance.disconnect() + await server_instance.shutdown() del self.opc_managers[server] server_instance = await self.initialize_opc_from_config(server_config) else: @@ -254,12 +254,12 @@ class IngestorManager(BaseActivity): self.logger.warning( f'Server {server} not found in managed tags. Desconnecting from server.' ) - await self.opc_managers[server].disconnect() + await self.opc_managers[server].shutdown() self.opc_managers.pop(server, None) metrics.OPC_MANAGERS_ACTIVE.labels(pod_id=self.pod_id).set(len(self.opc_managers)) - def check_opc_servers_integrity(self): + async def check_opc_servers_integrity(self): """ Checks the integrity of the OPC servers and updates the OPC servers if necessary. @@ -282,6 +282,7 @@ class IngestorManager(BaseActivity): self.logger.warning(f'OPC server {server} is lost. Server will be disconnected.') for slot, _config in self.managed_tags.items(): + await self.managed_tags[slot][server].shutdown() self.managed_tags[slot].pop(server, None) metrics.OPC_MANAGERS_ACTIVE.labels(pod_id=self.pod_id).set(len(self.opc_managers)) diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index 0ad077d..d1232dd 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -107,6 +107,9 @@ class OpcManager(BaseActivity): f'nodes={self.nodes}, subscriptions={self.subscriptions}' ) + def __del__(self): + asyncio.run(self.shutdown()) + async def shutdown(self): """ Comprehensive cleanup method for graceful shutdown. diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index d6dc973..b218e4e 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -184,7 +184,7 @@ async def test_update_opc_servers(metrics, opc_manager, ingestor_manager): 'slot2': {'server3': {'config': 'config3'}, 'server1': {'config': 'config1'}}, } - mock = MagicMock(config={'config': 'old_config2'}) + mock = AsyncMock(config={'config': 'old_config2'}) ingestor_manager.opc_managers['server3'] = AsyncMock(config={'config': 'config3'}) ingestor_manager.opc_managers['server2'] = mock ingestor_manager.opc_managers['server4'] = AsyncMock() @@ -472,8 +472,9 @@ async def test_subscribe_to_tags(ingestor_manager): ingestor_manager.managed_tags['slot1'].pop.assert_called_once_with('server3', None) +@mark.asyncio @patch('ingestor.managers.ingestor_manager.metrics') -def test_check_opc_servers_integrity_all_healthy(metrics, ingestor_manager): +async def test_check_opc_servers_integrity_all_healthy(metrics, ingestor_manager): # Setup mock OPC managers opc_manager1 = MagicMock() opc_manager1.check_cycles.return_value = None @@ -491,7 +492,7 @@ def test_check_opc_servers_integrity_all_healthy(metrics, ingestor_manager): ingestor_manager.initialize_opc_from_config = MagicMock() # Call the method - ingestor_manager.check_opc_servers_integrity() + await ingestor_manager.check_opc_servers_integrity() # Verify that check_cycles and check_opc_listenning were called for each server opc_manager1.check_cycles.assert_called_once() diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index b636f2b..c98fbf3 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -238,6 +238,7 @@ async def test_loop(ingestor_manager_started): ingestor_manager_started.manage_no_slots = MagicMock() ingestor_manager_started.manage_leases = AsyncMock() ingestor_manager_started.update_ingestor_manager = AsyncMock() + ingestor_manager_started.ingestor_manager.check_opc_servers_integrity = AsyncMock() ingestor_manager_started.ingestor_manager.managed_tags = { 'slot1': 'server1', 'slot2': 'server2', @@ -262,12 +263,16 @@ async def test_loop(ingestor_manager_started): ingestor_manager_started.manage_leases.assert_called_once_with(4, 3, 2) ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once() + ingestor_manager_started.ingestor_manager.check_opc_servers_integrity.assert_called_once() + ingestor_manager_started.update_ingestor_manager.assert_called_once() + @mark.asyncio async def test_loop_no_managed(ingestor_manager_started): ingestor_manager_started.manage_no_slots = MagicMock() ingestor_manager_started.manage_leases = AsyncMock() ingestor_manager_started.update_ingestor_manager = AsyncMock() + ingestor_manager_started.ingestor_manager.check_opc_servers_integrity = AsyncMock() ingestor_manager_started.ingestor_manager.managed_tags = {} ingestor_manager_started.ingestor_manager.get_active_ingestors = MagicMock( return_value=['ingestor1', 'ingestor2'] @@ -288,6 +293,9 @@ async def test_loop_no_managed(ingestor_manager_started): ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once() ingestor_manager_started.logger.info.assert_any_call('No slots acquired in this loop') + ingestor_manager_started.ingestor_manager.check_opc_servers_integrity.assert_called_once() + ingestor_manager_started.update_ingestor_manager.assert_called_once() + @mark.asyncio async def test_loop_no_ingestor_manager(ingestor_manager_started): From eea6ccb48f27d6b7f2592a2daca47827a2e80491 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Oct 2025 11:52:00 -0300 Subject: [PATCH 2/4] Refactor OPC manager shutdown logic for improved reliability Updated the IngestorManager class to ensure that OPC managers are properly shut down and removed from the management list when servers are lost or removed. This change enhances the integrity checks for OPC server management by ensuring asynchronous handling of shutdown operations. --- ingestor/managers/ingestor_manager.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index f82116c..bc4e094 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -243,10 +243,12 @@ class IngestorManager(BaseActivity): f'Failed to initialize OPC manager for server {server}, ' f'removing server from managed tags.' ) - _a = [ + + for slot, _value in current_managed_tags.items(): + if server in self.opc_managers: + await self.opc_managers[server].shutdown() + del self.opc_managers[server] self.managed_tags[slot].pop(server, None) - for slot, _value in current_managed_tags.items() - ] servers = list(self.opc_managers.keys()) for server in servers: @@ -282,7 +284,9 @@ class IngestorManager(BaseActivity): self.logger.warning(f'OPC server {server} is lost. Server will be disconnected.') for slot, _config in self.managed_tags.items(): - await self.managed_tags[slot][server].shutdown() + if server in self.opc_managers: + await self.opc_managers[server].shutdown() + del self.opc_managers[server] self.managed_tags[slot].pop(server, None) metrics.OPC_MANAGERS_ACTIVE.labels(pod_id=self.pod_id).set(len(self.opc_managers)) @@ -605,4 +609,7 @@ class IngestorManager(BaseActivity): to_remove.append([slot, server]) for slot, server in to_remove: + if server in self.opc_managers: + await self.opc_managers[server].shutdown() + del self.opc_managers[server] self.managed_tags[slot].pop(server, None) From f415207d5bbe884a7d56477cb2a4a064c7e0cbfe Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Oct 2025 11:55:55 -0300 Subject: [PATCH 3/4] Update OPC server update method to include NOSONAR comment for code quality tools --- ingestor/managers/ingestor_manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index bc4e094..a84aab3 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -189,7 +189,7 @@ class IngestorManager(BaseActivity): def __del__(self): asyncio.run(self.shutdown()) - async def update_opc_servers(self): + async def update_opc_servers(self): # NOSONAR """ Updates the OPC (OLE for Process Control) server connections managed by the ingestor. This method ensures that the OPC servers defined in `self.managed_tags` are properly From 83e43edf95f240a50d873f026d43a31475115349 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Oct 2025 11:58:00 -0300 Subject: [PATCH 4/4] Fix formatting of NOSONAR comment in update_opc_servers method for consistency --- ingestor/managers/ingestor_manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index a84aab3..f1258b3 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -189,7 +189,7 @@ class IngestorManager(BaseActivity): def __del__(self): asyncio.run(self.shutdown()) - async def update_opc_servers(self): # NOSONAR + async def update_opc_servers(self): # NOSONAR """ Updates the OPC (OLE for Process Control) server connections managed by the ingestor. This method ensures that the OPC servers defined in `self.managed_tags` are properly