From 47b3159e7177d9163dbc29e2e38012308983762b Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 20 Oct 2025 12:41:05 -0300 Subject: [PATCH 1/7] SIENTIAPDE-1312 Enhance OpcManager to accept subscription period from configuration - Updated OpcManager to include subscription_period_ms as a parameter for better configurability. - Refactored create_subscription method to utilize the new subscription period parameter. - Adjusted unit tests to reflect changes in subscription handling and ensure correct functionality. --- ingestor/managers/ingestor_manager.py | 1 + ingestor/managers/opc_manager.py | 26 ++++++++++++++++++++++--- tests/unit/managers/test_opc_manager.py | 14 ++++++++----- 3 files changed, 33 insertions(+), 8 deletions(-) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index ea0b3a1..bf964c0 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -141,6 +141,7 @@ class IngestorManager(BaseActivity): manager = OpcManager( name=server_config['name'], url=server_config['url'], + subscription_period_ms=server_config['subscription_period_ms'], data_manager=self.data_manager, logger=self.logger, server_uri=server_config['server_uri'], diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index bf66ef1..abd81c1 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -59,6 +59,7 @@ class OpcManager(BaseActivity): self, name: str, url: str, + subscription_period_ms: int, data_manager: DataManager, logger: Logger, server_uri: str, @@ -74,6 +75,7 @@ class OpcManager(BaseActivity): self.data_queue: dict = {} self.non_receive_count = 0 self.client: Client | None = None + self.subscription_period_ms = subscription_period_ms self.cert_path = cert_path self.private_key_path = private_key_path self.server_cert_path = server_cert_path @@ -82,6 +84,7 @@ class OpcManager(BaseActivity): self.data_manager = data_manager self.metadata = metadata + BaseActivity.__init__( self, logger=logger, notification_handler=notification_handler, set_error_counter=True ) @@ -189,7 +192,13 @@ class OpcManager(BaseActivity): try: self.client = Client(self.url, watchdog_intervall=3600000) assert self.client is not None # Informa ao mypy que client não é None + self.client.name = self.pod_id + self.client.application_name = self.pod_id + pod_uri = f'urn:sientia-do:pod:{self.pod_id}' + self.client.application_uri = pod_uri + self.client.product_uri = pod_uri + if self.cert_path: await self.set_security() self.logger.info(f'Starting connection to {self.name}...') @@ -199,6 +208,15 @@ class OpcManager(BaseActivity): ).set(1) self.logger.info(f'Connection to {self.name} successful.') except Exception as e: + if self.client: + try: + await self.client.disconnect() + except Exception as internal_e: + self.logger.error(f'And error occurred while creating connection from {self.name}, and an exception occurred while disconnecting: {internal_e}') + + del self.client + self.client = None + metrics.OPC_CONNECTION_STATUS.labels( pod_id=self.pod_id, server_name=self.name, server_url=self.url ).set(0) @@ -206,7 +224,7 @@ class OpcManager(BaseActivity): self.logger.error(f'Failed to connect to {self.name}: {e}') raise - async def create_subscription(self, name: str, period: int = 500): + async def create_subscription(self, name: str): """ Creates a subscription with the specified monitoring period. @@ -230,8 +248,8 @@ class OpcManager(BaseActivity): if not self.client: raise ValueError('Client not connected. Call connect first.') try: - p = period if period is not None else 500 - self.subscriptions[name] = await self.client.create_subscription(p, self) + self.subscriptions[name] = await self.client.create_subscription( + self.subscription_period_ms, self) self.logger.info(f'Subscription {name} created on {self.name}.') metrics.OPC_SUBSCRIPTIONS_CREATED.labels( pod_id=self.pod_id, server_name=self.name, slot_name=name @@ -345,10 +363,12 @@ class OpcManager(BaseActivity): except Exception as sub_error: self.logger.error(f'Failed to clean up subscription: {sub_error}') + # Inserir backoff para disconnect, emitindo métrica e alerta try: await self.client.disconnect() except Exception as conn_error: self.logger.error(f'Failed to disconnect from OPC UA server: {conn_error}') + finally: del self.client self.client = None diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index 0c4e625..d7a2eac 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/tests/unit/managers/test_opc_manager.py @@ -50,6 +50,7 @@ def raw_opc_manager(mock_metrics): name='TestConnector', url='opc.tcp://localhost:4840', data_manager=MagicMock(), + subscription_period_ms=1000, logger=MagicMock(), server_uri='opc.tcp://localhost:4840', notification_handler=MagicMock(), @@ -223,24 +224,26 @@ async def test_create_subscription_no_client(raw_opc_manager): @mark.asyncio async def test_create_subscription_success_has_period(opc_manager): - await opc_manager.create_subscription('sub1', 1000) + await opc_manager.create_subscription('sub1') - opc_manager.client.create_subscription.assert_called_once_with(1000, opc_manager) + opc_manager.client.create_subscription.assert_called_once_with( + opc_manager.subscription_period_ms, opc_manager) assert opc_manager.subscriptions['sub1'] is not None @mark.asyncio async def test_create_subscription_success_no_period(opc_manager): - await opc_manager.create_subscription('sub1', None) + await opc_manager.create_subscription('sub1') - opc_manager.client.create_subscription.assert_called_once_with(500, opc_manager) + opc_manager.client.create_subscription.assert_called_once_with( + opc_manager.subscription_period_ms, opc_manager) assert opc_manager.subscriptions['sub1'] is not None @patch('ingestor.managers.opc_manager.metrics') @mark.asyncio async def test_create_subscription_with_metrics(metrics, opc_manager): - await opc_manager.create_subscription('sub1', 1000) + await opc_manager.create_subscription('sub1') metrics.OPC_SUBSCRIPTIONS_CREATED.labels.assert_called_once_with( pod_id=opc_manager.pod_id, server_name=opc_manager.name, slot_name='sub1' @@ -572,6 +575,7 @@ def test_init_metrics_calls_correct_metric_methods(metrics): name='TestInitConnector', url='opc.tcp://init.test:4840', data_manager=MagicMock(), + subscription_period_ms=1000, logger=MagicMock(), server_uri='opc.tcp://init.test:4840/uri', notification_handler=MagicMock(), From 44a9367663d7e1db024158d1d141bf07cf77d6c1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 20 Oct 2025 12:41:30 -0300 Subject: [PATCH 2/7] SIENTIAPDE-1312 Update GITHUB_BRANCH in values.yaml to reflect changes for SIENTIAPDE-1312, focusing on improvements and corrections in data pipelines. --- values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/values.yaml b/values.yaml index fe51efb..c9a926d 100644 --- a/values.yaml +++ b/values.yaml @@ -139,7 +139,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-opc-ingestor.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1205-alterar-opc-para-assincrono" + value: "SIENTIAPDE-1312-melhorias-e-correcoes-nas-pipelines-de-dados" - name: PYTHON_APP value: "ingestor.app" From 7001b5e9bd5f46a2402c6bfa0f61f3d635769814 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 20 Oct 2025 15:27:05 -0300 Subject: [PATCH 3/7] SIENTIAPDE-1312 SIENTIAPDE-1318 Enhance OpcManager disconnection handling - Introduced a disconnection_fallback method to attempt multiple disconnects from the OPC UA server, improving reliability. - Updated error handling during connection attempts to ensure proper logging and notification of disconnection failures. - Refactored disconnect method to utilize the new fallback mechanism and maintain metrics accurately. --- ingestor/managers/opc_manager.py | 68 +++++++++++++++++++++++--------- 1 file changed, 49 insertions(+), 19 deletions(-) diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index abd81c1..113473f 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -1,5 +1,7 @@ +import asyncio import json from pathlib import Path +import traceback from asyncua import Client from asyncua.crypto.security_policies import SecurityPolicyBasic256 @@ -198,7 +200,7 @@ class OpcManager(BaseActivity): pod_uri = f'urn:sientia-do:pod:{self.pod_id}' self.client.application_uri = pod_uri self.client.product_uri = pod_uri - + if self.cert_path: await self.set_security() self.logger.info(f'Starting connection to {self.name}...') @@ -208,11 +210,10 @@ class OpcManager(BaseActivity): ).set(1) self.logger.info(f'Connection to {self.name} successful.') except Exception as e: - if self.client: - try: - await self.client.disconnect() - except Exception as internal_e: - self.logger.error(f'And error occurred while creating connection from {self.name}, and an exception occurred while disconnecting: {internal_e}') + try: + await self.disconnect() + except Exception as internal_e: + self.logger.error(f'And error occurred while creating connection from {self.name}, and an exception occurred while disconnecting: {internal_e}') del self.client self.client = None @@ -333,6 +334,28 @@ class OpcManager(BaseActivity): del self.subscriptions[subscription] self.logger.info(f'Unsubscribed from {subscription}.') + async def disconnection_fallback(self) -> list: + """ + Tries 5 times to disconnect from the OPC UA server, with a delay of 100ms x try. + """ + + assert self.client is not None + error_stack = [] + for i in range(5): + try: + self.logger.info(f'Disconnecting from OPC UA server, attempt {i+1} of 5') + await self.client.disconnect() + return [] + except Exception as e: + self.logger.error(f'Failed to disconnect from OPC UA server: {e}') + error_stack.append({ + 'attempt': i+1, + 'error': str(e), + 'traceback': traceback.format_exc(), + }) + await asyncio.sleep(0.1 * i) + return error_stack + async def disconnect(self): """ Disconnects from the OPC UA server. @@ -363,21 +386,28 @@ class OpcManager(BaseActivity): except Exception as sub_error: self.logger.error(f'Failed to clean up subscription: {sub_error}') - # Inserir backoff para disconnect, emitindo métrica e alerta - try: - await self.client.disconnect() - except Exception as conn_error: - self.logger.error(f'Failed to disconnect from OPC UA server: {conn_error}') - - finally: - del self.client - self.client = None - metrics.OPC_CONNECTION_STATUS.labels( - pod_id=self.pod_id, server_name=self.name, server_url=self.url - ).set(0) - metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(0) + + errors = self.disconnection_fallback() + + if errors: + self.send_notification( + metadata=self.metadata, + notification_id=f'OPC_DISCONNECTION_ERROR_{self.name}', + message=f'Failed to disconnect from OPC UA server {self.name} after 5 attempts', + block='opc_manager', + level=NotificationLevel.ERROR, + attachment_content=json.dumps(errors, indent=4), + ) + else: self.logger.warning('Disconnected from OPC UA server.') + del self.client + self.client = None + metrics.OPC_CONNECTION_STATUS.labels( + pod_id=self.pod_id, server_name=self.name, server_url=self.url + ).set(0) + metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(0) + async def datachange_notification(self, node, _val, data): """ Handles data change notifications for monitored OPC UA nodes. From 8c552f50a83a38425a312f447b5d5afafc084456 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 20 Oct 2025 15:56:00 -0300 Subject: [PATCH 4/7] SIENTIAPDE-1312 Fix pod URI formatting and enhance disconnection error logging in OpcManager - Updated pod URI construction to replace hyphens with colons for correct formatting. - Improved error logging during disconnection attempts to include the attempt number for better traceability. - Changed disconnection_fallback method call to be awaited for proper asynchronous handling. --- ingestor/managers/opc_manager.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index 113473f..09ee184 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -197,7 +197,7 @@ class OpcManager(BaseActivity): self.client.name = self.pod_id self.client.application_name = self.pod_id - pod_uri = f'urn:sientia-do:pod:{self.pod_id}' + pod_uri = self.pod_id.replace('-', ':') self.client.application_uri = pod_uri self.client.product_uri = pod_uri @@ -347,7 +347,7 @@ class OpcManager(BaseActivity): await self.client.disconnect() return [] except Exception as e: - self.logger.error(f'Failed to disconnect from OPC UA server: {e}') + self.logger.error(f'Failed to disconnect from OPC UA serve in attempt {i+1} of 5: {e}') error_stack.append({ 'attempt': i+1, 'error': str(e), @@ -387,7 +387,7 @@ class OpcManager(BaseActivity): self.logger.error(f'Failed to clean up subscription: {sub_error}') - errors = self.disconnection_fallback() + errors = await self.disconnection_fallback() if errors: self.send_notification( From aaa8d00f56bd3f8fe770678b2b5c9a49b27c1ddc Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 20 Oct 2025 16:26:24 -0300 Subject: [PATCH 5/7] SIENTIAPDE-1312 Update OpcManager to set a timeout for the client connection - Added a timeout parameter to the Client initialization in OpcManager to enhance connection reliability and prevent indefinite blocking during connection attempts. --- ingestor/managers/opc_manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index 09ee184..45bacd1 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -192,7 +192,7 @@ class OpcManager(BaseActivity): metrics.OPC_CONNECTIONS_TOTAL.labels(pod_id=self.pod_id, server_name=self.name).inc() try: - self.client = Client(self.url, watchdog_intervall=3600000) + self.client = Client(self.url, timeout=10, watchdog_intervall=3600000) assert self.client is not None # Informa ao mypy que client não é None self.client.name = self.pod_id From 3eacf27faf7e48ccba268c1835dea79acce05e75 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 20 Oct 2025 17:08:40 -0300 Subject: [PATCH 6/7] SIENTIAPDE-1312 Refactor OpcManager connection error handling and enhance unit tests - Simplified error handling during connection attempts in OpcManager by removing redundant disconnection logic. - Updated unit tests to include subscription_period_ms in server configuration and adjusted connection assertions to include a timeout parameter. - Added new tests for disconnection fallback functionality to ensure robust error handling during disconnect attempts. --- ingestor/managers/opc_manager.py | 13 +--- tests/unit/managers/test_ingestor_manager.py | 3 + tests/unit/managers/test_opc_manager.py | 63 ++++++++++---------- 3 files changed, 36 insertions(+), 43 deletions(-) diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index 45bacd1..db77e35 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -210,19 +210,8 @@ class OpcManager(BaseActivity): ).set(1) self.logger.info(f'Connection to {self.name} successful.') except Exception as e: - try: - await self.disconnect() - except Exception as internal_e: - self.logger.error(f'And error occurred while creating connection from {self.name}, and an exception occurred while disconnecting: {internal_e}') + await self.disconnect() - del self.client - self.client = None - - metrics.OPC_CONNECTION_STATUS.labels( - pod_id=self.pod_id, server_name=self.name, server_url=self.url - ).set(0) - metrics.OPC_CONNECTIONS_FAILED.labels(pod_id=self.pod_id, server_name=self.name).inc() - self.logger.error(f'Failed to connect to {self.name}: {e}') raise async def create_subscription(self, name: str): diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 2625fd3..d6dc973 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -94,6 +94,7 @@ async def test_initialize_opc_from_config(opc_manager, ingestor_manager): server_config = { 'name': 'server1', 'url': 'opc.tcp://localhost:4840', + 'subscription_period_ms': 1000, 'server_uri': 'http://opcua-server.simulator', 'cert_path': '/path/to/cert', 'private_key_path': '/path/to/private_key', @@ -109,6 +110,7 @@ async def test_initialize_opc_from_config(opc_manager, ingestor_manager): url=server_config['url'], data_manager=ingestor_manager.data_manager, logger=ingestor_manager.logger, + subscription_period_ms=server_config['subscription_period_ms'], server_uri=server_config['server_uri'], notification_handler=ingestor_manager.notification_handler, cert_path=server_config['cert_path'], @@ -128,6 +130,7 @@ async def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, server_config = { 'name': 'server1', 'url': 'opc.tcp://localhost:4840', + 'subscription_period_ms': 1000, 'server_uri': 'http://opcua-server.simulator', 'cert_path': '/path/to/cert', 'private_key_path': '/path/to/private_key', diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index d7a2eac..fe72f46 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/tests/unit/managers/test_opc_manager.py @@ -1,6 +1,6 @@ import json from datetime import datetime -from unittest.mock import AsyncMock, MagicMock, call, patch +from unittest.mock import ANY, AsyncMock, MagicMock, call, patch import pytest from asyncua.crypto.security_policies import SecurityPolicyBasic256 @@ -142,7 +142,7 @@ async def test_connect_no_security(client, mock_metrics, raw_opc_manager): await raw_opc_manager.connect() - client.assert_called_once_with(raw_opc_manager.url, watchdog_intervall=3600000) + client.assert_called_once_with(raw_opc_manager.url, timeout=10, watchdog_intervall=3600000) raw_opc_manager.client.connect.assert_called_once() raw_opc_manager.set_security.assert_not_called() mock_metrics.OPC_CONNECTIONS_TOTAL.labels.assert_called_once_with( @@ -169,7 +169,7 @@ async def test_connect_with_security(client, raw_opc_manager): await raw_opc_manager.connect() - client.assert_called_once_with(raw_opc_manager.url, watchdog_intervall=3600000) + client.assert_called_once_with(raw_opc_manager.url, timeout=10, watchdog_intervall=3600000) raw_opc_manager.client.connect.assert_called_once() raw_opc_manager.set_security.assert_called_once() @@ -184,33 +184,15 @@ async def test_connect_exception_handling_and_metrics( simulated_error_message = 'Erro de conexão simulado' mock_client_instance.connect.side_effect = Exception(simulated_error_message) + raw_opc_manager.disconnect = AsyncMock() + opc_manager_instance = raw_opc_manager opc_manager_instance.cert_path = None with pytest.raises(Exception, match=simulated_error_message): await opc_manager_instance.connect() - - mock_metrics_module.OPC_CONNECTIONS_TOTAL.labels.assert_called_once_with( - pod_id=opc_manager_instance.pod_id, server_name=opc_manager_instance.name - ) - mock_metrics_module.OPC_CONNECTIONS_TOTAL.labels.return_value.inc.assert_called_once() - - mock_metrics_module.OPC_CONNECTION_STATUS.labels.assert_called_once_with( - pod_id=opc_manager_instance.pod_id, - server_name=opc_manager_instance.name, - server_url=opc_manager_instance.url, - ) - mock_metrics_module.OPC_CONNECTION_STATUS.labels.return_value.set.assert_called_once_with(0) - - mock_metrics_module.OPC_CONNECTIONS_FAILED.labels.assert_called_once_with( - pod_id=opc_manager_instance.pod_id, server_name=opc_manager_instance.name - ) - mock_metrics_module.OPC_CONNECTIONS_FAILED.labels.return_value.inc.assert_called_once() - - opc_manager_instance.logger.error.assert_called_once_with( - f'Failed to connect to {opc_manager_instance.name}: {simulated_error_message}' - ) - + + opc_manager_instance.disconnect.assert_called_once() @mark.asyncio async def test_create_subscription_no_client(raw_opc_manager): @@ -260,16 +242,15 @@ async def test_create_subscription_exception_during_client_call( opc_manager_instance.client = AsyncMock() subscription_name = 'test_sub_client_error' - simulated_period = 750 simulated_error_message = 'Falha ao criar subscrição no cliente OPC' opc_manager_instance.client.create_subscription.side_effect = Exception(simulated_error_message) with pytest.raises(Exception, match=simulated_error_message): - await opc_manager_instance.create_subscription(subscription_name, period=simulated_period) + await opc_manager_instance.create_subscription(subscription_name) opc_manager_instance.client.create_subscription.assert_called_once_with( - simulated_period, opc_manager_instance + opc_manager_instance.subscription_period_ms, opc_manager_instance ) opc_manager_instance.logger.error.assert_called_once_with( @@ -321,6 +302,28 @@ async def test_unsubscribe_success(opc_manager_subscribed): assert opc_manager_subscribed.subscriptions.get('sub1') is None + +@mark.asyncio +async def test_disconnection_fallback_success(opc_manager): + opc_manager.client = AsyncMock() + opc_manager.client.disconnect.return_value = True + result = await opc_manager.disconnection_fallback() + assert result == [] + +@mark.asyncio +async def test_disconnection_fallback_fail(opc_manager): + opc_manager.client = AsyncMock() + opc_manager.client.disconnect.side_effect = Exception('Test error') + result = await opc_manager.disconnection_fallback() + assert result == [ + {'attempt': 1, 'error': 'Test error', 'traceback': ANY}, + {'attempt': 2, 'error': 'Test error', 'traceback': ANY}, + {'attempt': 3, 'error': 'Test error', 'traceback': ANY}, + {'attempt': 4, 'error': 'Test error', 'traceback': ANY}, + {'attempt': 5, 'error': 'Test error', 'traceback': ANY}, ] + assert opc_manager.client.disconnect.call_count == 5 + + @mark.asyncio async def test_disconnect_success(opc_manager_subscribed): opc_manager_subscribed.client = MagicMock() @@ -368,9 +371,7 @@ async def test_disconnect_error(opc_manager_subscribed): opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once() opc_manager_subscribed.client = None - opc_manager_subscribed.logger.error.assert_called_once_with( - 'Failed to disconnect from OPC UA server: Test error' - ) + @patch('ingestor.managers.opc_manager.metrics') From 95c9ad85f9aa4a363075511585256436c2f65d9c Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 21 Oct 2025 17:03:58 -0300 Subject: [PATCH 7/7] SIENTIAPDE-1312 Update image tag in values.yaml and refactor OpcManager error handling - Updated the image tag in values.yaml from "0.4.5" to "0.4.9" for the latest version. - Refactored error handling in OpcManager to improve logging and maintain code clarity. - Adjusted unit tests for better readability and consistency in assertions. --- ingestor/managers/opc_manager.py | 27 ++++++++++++++----------- tests/unit/managers/test_opc_manager.py | 17 +++++++++------- values.yaml | 2 +- 3 files changed, 26 insertions(+), 20 deletions(-) diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index db77e35..0ad077d 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -1,7 +1,7 @@ import asyncio import json -from pathlib import Path import traceback +from pathlib import Path from asyncua import Client from asyncua.crypto.security_policies import SecurityPolicyBasic256 @@ -86,7 +86,6 @@ class OpcManager(BaseActivity): self.data_manager = data_manager self.metadata = metadata - BaseActivity.__init__( self, logger=logger, notification_handler=notification_handler, set_error_counter=True ) @@ -209,7 +208,7 @@ class OpcManager(BaseActivity): pod_id=self.pod_id, server_name=self.name, server_url=self.url ).set(1) self.logger.info(f'Connection to {self.name} successful.') - except Exception as e: + except Exception: await self.disconnect() raise @@ -239,7 +238,8 @@ class OpcManager(BaseActivity): raise ValueError('Client not connected. Call connect first.') try: self.subscriptions[name] = await self.client.create_subscription( - self.subscription_period_ms, self) + self.subscription_period_ms, self + ) self.logger.info(f'Subscription {name} created on {self.name}.') metrics.OPC_SUBSCRIPTIONS_CREATED.labels( pod_id=self.pod_id, server_name=self.name, slot_name=name @@ -332,16 +332,20 @@ class OpcManager(BaseActivity): error_stack = [] for i in range(5): try: - self.logger.info(f'Disconnecting from OPC UA server, attempt {i+1} of 5') + self.logger.info(f'Disconnecting from OPC UA server, attempt {i + 1} of 5') await self.client.disconnect() return [] except Exception as e: - self.logger.error(f'Failed to disconnect from OPC UA serve in attempt {i+1} of 5: {e}') - error_stack.append({ - 'attempt': i+1, - 'error': str(e), - 'traceback': traceback.format_exc(), - }) + self.logger.error( + f'Failed to disconnect from OPC UA serve in attempt {i + 1} of 5: {e}' + ) + error_stack.append( + { + 'attempt': i + 1, + 'error': str(e), + 'traceback': traceback.format_exc(), + } + ) await asyncio.sleep(0.1 * i) return error_stack @@ -375,7 +379,6 @@ class OpcManager(BaseActivity): except Exception as sub_error: self.logger.error(f'Failed to clean up subscription: {sub_error}') - errors = await self.disconnection_fallback() if errors: diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index fe72f46..f440e43 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/tests/unit/managers/test_opc_manager.py @@ -191,9 +191,10 @@ async def test_connect_exception_handling_and_metrics( with pytest.raises(Exception, match=simulated_error_message): await opc_manager_instance.connect() - + opc_manager_instance.disconnect.assert_called_once() + @mark.asyncio async def test_create_subscription_no_client(raw_opc_manager): try: @@ -209,7 +210,8 @@ async def test_create_subscription_success_has_period(opc_manager): await opc_manager.create_subscription('sub1') opc_manager.client.create_subscription.assert_called_once_with( - opc_manager.subscription_period_ms, opc_manager) + opc_manager.subscription_period_ms, opc_manager + ) assert opc_manager.subscriptions['sub1'] is not None @@ -218,7 +220,8 @@ async def test_create_subscription_success_no_period(opc_manager): await opc_manager.create_subscription('sub1') opc_manager.client.create_subscription.assert_called_once_with( - opc_manager.subscription_period_ms, opc_manager) + opc_manager.subscription_period_ms, opc_manager + ) assert opc_manager.subscriptions['sub1'] is not None @@ -302,7 +305,6 @@ async def test_unsubscribe_success(opc_manager_subscribed): assert opc_manager_subscribed.subscriptions.get('sub1') is None - @mark.asyncio async def test_disconnection_fallback_success(opc_manager): opc_manager.client = AsyncMock() @@ -310,6 +312,7 @@ async def test_disconnection_fallback_success(opc_manager): result = await opc_manager.disconnection_fallback() assert result == [] + @mark.asyncio async def test_disconnection_fallback_fail(opc_manager): opc_manager.client = AsyncMock() @@ -317,10 +320,11 @@ async def test_disconnection_fallback_fail(opc_manager): result = await opc_manager.disconnection_fallback() assert result == [ {'attempt': 1, 'error': 'Test error', 'traceback': ANY}, - {'attempt': 2, 'error': 'Test error', 'traceback': ANY}, + {'attempt': 2, 'error': 'Test error', 'traceback': ANY}, {'attempt': 3, 'error': 'Test error', 'traceback': ANY}, {'attempt': 4, 'error': 'Test error', 'traceback': ANY}, - {'attempt': 5, 'error': 'Test error', 'traceback': ANY}, ] + {'attempt': 5, 'error': 'Test error', 'traceback': ANY}, + ] assert opc_manager.client.disconnect.call_count == 5 @@ -373,7 +377,6 @@ async def test_disconnect_error(opc_manager_subscribed): opc_manager_subscribed.client = None - @patch('ingestor.managers.opc_manager.metrics') @mark.asyncio async def test_disconnect_metrics_on_successful_path(mock_metrics_module, raw_opc_manager): diff --git a/values.yaml b/values.yaml index c9a926d..5d22543 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.4.5" + tag: "0.4.9" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: