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..0ad077d 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -1,4 +1,6 @@ +import asyncio import json +import traceback from pathlib import Path from asyncua import Client @@ -59,6 +61,7 @@ class OpcManager(BaseActivity): self, name: str, url: str, + subscription_period_ms: int, data_manager: DataManager, logger: Logger, server_uri: str, @@ -74,6 +77,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 @@ -187,9 +191,15 @@ 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 + self.client.application_name = self.pod_id + pod_uri = self.pod_id.replace('-', ':') + 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}...') @@ -198,15 +208,12 @@ 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: - 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}') + except Exception: + await self.disconnect() + 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 +237,9 @@ 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 @@ -315,6 +323,32 @@ 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 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 + async def disconnect(self): """ Disconnects from the OPC UA server. @@ -345,19 +379,27 @@ class OpcManager(BaseActivity): except Exception as sub_error: self.logger.error(f'Failed to clean up subscription: {sub_error}') - 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 = await 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. 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 0c4e625..f440e43 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 @@ -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(), @@ -141,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( @@ -168,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() @@ -183,32 +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 @@ -223,24 +207,28 @@ 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' @@ -257,16 +245,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( @@ -318,6 +305,29 @@ 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() @@ -365,9 +375,6 @@ 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') @@ -572,6 +579,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(), diff --git a/values.yaml b/values.yaml index fe51efb..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: @@ -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"