Merge pull request #21 from Aignosi/SIENTIAPDE-1312-melhorias-e-correcoes-nas-pipelines-de-dados

SIENTIAPDE-1312: Improve OpcManager Connection Handling, Configuration, and Error Logging

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'],

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:
+        except Exception:
-            metrics.OPC_CONNECTION_STATUS.labels(
+            await self.disconnect()
-                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, 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(
-            self.subscriptions[name] = await self.client.create_subscription(p, 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
@@ -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:
+        errors = await self.disconnection_fallback()
-            await self.client.disconnect()
+
-        except Exception as conn_error:
+        if errors:
-            self.logger.error(f'Failed to disconnect from OPC UA server: {conn_error}')
+            self.send_notification(
-        finally:
+                metadata=self.metadata,
-            del self.client
+                notification_id=f'OPC_DISCONNECTION_ERROR_{self.name}',
-            self.client = None
+                message=f'Failed to disconnect from OPC UA server {self.name} after 5 attempts',
-            metrics.OPC_CONNECTION_STATUS.labels(
+                block='opc_manager',
-                pod_id=self.pod_id, server_name=self.name, server_url=self.url
+                level=NotificationLevel.ERROR,
-            ).set(0)
+                attachment_content=json.dumps(errors, indent=4),
-            metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(0)
+            )
+        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.

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',

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(
+    opc_manager_instance.disconnect.assert_called_once()
-        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}'
-    )
 
 
 @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(),

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"