From c49f04ded67fe4679af917f5cdfde3f7ee0601fe Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Thu, 29 May 2025 16:52:10 -0300 Subject: [PATCH] SIENTIAPDE-1083: add prometheus metrics to opc_manager. --- ingestor/managers/ingestor_manager.py | 3 +- ingestor/managers/opc_manager.py | 53 ++- ingestor/metrics.py | 20 - tests/unit/managers/test_ingestor_manager.py | 4 +- tests/unit/managers/test_opc_manager.py | 433 ++++++++++++++----- tests/unit/test_metrics.py | 44 -- 6 files changed, 354 insertions(+), 203 deletions(-) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 433fcef..85d5602 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -29,6 +29,7 @@ class IngestorManager(): self.opc_servers = {} self.notification_handler = notification_handler + self.pod_id = pod_id def initialize_opc_from_config(self, server_config: dict, data_manager: DataManager, logger: Logger) -> OpcManager | None: @@ -56,7 +57,7 @@ class IngestorManager(): manager = OpcManager( server_config['name'], server_config['url'], data_manager, logger, server_config['server_uri'], - self.notification_handler, server_config.get('cert_path'), + self.notification_handler, self.pod_id, server_config.get('cert_path'), server_config.get('private_key_path'), server_config.get('server_cert_path') ) diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index e218956..35d4a8b 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -7,11 +7,12 @@ from asyncua.sync import Client from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.handlers import NotificationHandler from ingestor.managers.data_manager import DataManager +import ingestor.metrics as metrics class OpcManager(): def __init__(self, name: str, url: str, data_manager: DataManager, - logger: Logger, server_uri: str, notification_handler: NotificationHandler, + logger: Logger, server_uri: str, notification_handler: NotificationHandler, pod_id: str, cert_path: str = None, private_key_path: str = None, server_cert_path: str = None): self.url = url self.name = name @@ -26,8 +27,10 @@ class OpcManager(): self.nodes = {} self.subscriptions = {} self.data_manager = data_manager - self.notification_handler = notification_handler + self.pod_id = pod_id + 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) def __str__(self): return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \ @@ -82,11 +85,20 @@ class OpcManager(): Exception: If the connection to the OPC server fails. """ - self.client = Client(self.url) - if self.cert_path: - self.set_security() - self.logger.info('Starting connection...') - self.client.connect() + metrics.OPC_CONNECTIONS_TOTAL.labels(pod_id=self.pod_id, server_name=self.name).inc() + try: + self.client = Client(self.url) + if self.cert_path: + self.set_security() + self.logger.info(f'Starting connection to {self.name}...') + self.client.connect() + metrics.OPC_CONNECTION_STATUS.labels(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}") + raise def create_subscription(self, name: str, period: int = 500): """ @@ -105,11 +117,14 @@ class OpcManager(): if not self.client: raise ValueError("Client not connected. Call connect first.") - - p = period if period != None else 500 - self.subscriptions[name] = self.client.create_subscription( - p, self) - self.logger.info('Subscription created.') + try: + p = period if period is not None else 500 + self.subscriptions[name] = self.client.create_subscription(p, 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).inc() + except Exception as e: + self.logger.error(f"Failed to create subscription {name} on {self.name}: {e}") + raise def subscribe(self, subscription: str, nodes: dict, collect_period: int): """ @@ -129,14 +144,13 @@ class OpcManager(): """ if not self.subscriptions.get(subscription): - raise ValueError( - "Subscription not created. Call create_subscription first.") + raise ValueError("Subscription not created. Call create_subscription first.") - self.logger.info(f"Subscribing to {subscription}...") + self.logger.info(f"Subscribing to {subscription} on {self.name}...") self.logger.info(f"Subscribing to nodes: {nodes}") - self.addr_nodes = [self.client.get_node( - n) for n in nodes if n not in self.nodes] + self.addr_nodes = [self.client.get_node(n) for n in nodes if n not in self.nodes] self.nodes.update(nodes) + metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(len(self.nodes)) self.collect_period = collect_period for node, config in self.nodes.items(): @@ -202,6 +216,8 @@ class OpcManager(): 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) self.logger.warning("Disconnected from OPC UA server.") def datachange_notification(self, node, _val, data): @@ -236,6 +252,7 @@ class OpcManager(): self.nodes[tag]['cycle_rule']['cycle_count'] = 0 self.non_receive_count = 0 + metrics.OPC_CYCLES_WITHOUT_DATA.labels(pod_id=self.pod_id, server_name=self.name).set(0) data = { 'tag': tag, @@ -278,6 +295,7 @@ class OpcManager(): """ self.non_receive_count += 1 + metrics.OPC_CYCLES_WITHOUT_DATA.labels(pod_id=self.pod_id, server_name=self.name).set(self.non_receive_count) if self.non_receive_count >= 5: self.notification_handler.build_and_send_notification( notification_id=f'OPC_LISTENNING_STOPPED__{self.name}', @@ -287,6 +305,7 @@ class OpcManager(): level=NotificationLevel.ERROR ) if self.non_receive_count >= 15: + metrics.OPC_RECONNECTIONS_TOTAL.labels(pod_id=self.pod_id, server_name=self.name).inc() self.notification_handler.build_and_send_notification( notification_id=f'OPC_CONNECTION_RETRY__{self.name}', message=f'Retrying to connect to server {self.name}', diff --git a/ingestor/metrics.py b/ingestor/metrics.py index bb0976d..b2ddc42 100644 --- a/ingestor/metrics.py +++ b/ingestor/metrics.py @@ -1,11 +1,7 @@ from prometheus_client import Counter, Gauge, Histogram -# It's useful to have a common set of labels, like the pod_id. -# We'll add 'pod_id' to many metrics to distinguish instances. POD_ID_LABEL = ["pod_id"] SERVER_LABELS = ["pod_id", "server_name", "server_url"] -SLOT_LABELS = ["pod_id", "slot_id"] -TAG_LABELS = ["pod_id", "server_name", "tag_id", "tag_name"] KAFKA_LABELS = ["pod_id", "topic"] REDIS_LABELS = ["pod_id", "operation"] NOTIFICATION_LABELS = ["pod_id", "level", "block"] @@ -37,17 +33,14 @@ APP_UP = Gauge( ACTIVE_INGESTORS = Gauge( "ingestor_active_total", "Number of active ingestors reported by Redis", - # No pod_id here, as it's a global view from Redis ) SLOTS_TOTAL = Gauge( "ingestor_slots_total", "Total number of slots configured in Redis", - # No pod_id here, as it's a global view from Redis ) LEASES_TOTAL = Gauge( "ingestor_leases_total", "Total number of leases (allocated slots) in Redis", - # No pod_id here, as it's a global view from Redis ) SLOTS_MANAGED = Gauge( "ingestor_slots_managed_current", @@ -101,19 +94,6 @@ OPC_TAGS_SUBSCRIBED = Gauge( "Current number of OPC tags subscribed on a server", ["pod_id", "server_name"], ) -OPC_DATACHANGE_NOTIFICATIONS = Counter( - "opc_datachange_notifications_total", - "Total data change notifications received (tag readings)", - TAG_LABELS, -) -OPC_TAG_LAST_VALUE = Gauge( - "opc_tag_last_value", "The last value read from an OPC tag", TAG_LABELS -) -OPC_TAG_LAST_READ_TIMESTAMP = Gauge( - "opc_tag_last_read_timestamp_seconds", - "Timestamp of the last value read from an OPC tag", - TAG_LABELS, -) OPC_CYCLES_WITHOUT_DATA = Gauge( "opc_cycles_without_data", "Current number of cycles without receiving data from a server", diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index fa3f7be..f1bcbf3 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -60,7 +60,8 @@ def test_initialize_opc_from_config(opc_manager, ingestor_manager): 'server_uri': 'http://opcua-server.simulator', 'cert_path': '/path/to/cert', 'private_key_path': '/path/to/private_key', - 'server_cert_path': '/path/to/server_cert' + 'server_cert_path': '/path/to/server_cert', + 'pod_id': 'test_pod' } opc_manager.return_value = MagicMock() @@ -70,6 +71,7 @@ def test_initialize_opc_from_config(opc_manager, ingestor_manager): opc_manager.assert_called_once_with( server_config['name'], server_config['url'], ingestor_manager.data_manager, ingestor_manager.logger, server_config['server_uri'], ingestor_manager.notification_handler, + server_config['pod_id'], server_config['cert_path'], server_config['private_key_path'], server_config['server_cert_path'] ) diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index a57ef46..90c25b9 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/tests/unit/managers/test_opc_manager.py @@ -1,6 +1,7 @@ import json from datetime import datetime from unittest.mock import MagicMock, patch +from prometheus_client import Gauge from pytest import fixture from asyncua.crypto.security_policies import SecurityPolicyBasic256 @@ -9,57 +10,65 @@ from ingestor.managers.opc_manager import OpcManager from sientia_do.notifications.models import NotificationLevel tags = { - 'ns=3;i=1001': { - 'aggregation_function': 'LTS', - 'frequency': 1000, - 'max_value': 100, - 'min_value': 0, - 'tag_name': 'Counter' + "ns=3;i=1001": { + "aggregation_function": "LTS", + "frequency": 1000, + "max_value": 100, + "min_value": 0, + "tag_name": "Counter", }, - 'ns=3;i=1003': { - 'aggregation_function': 'AVG', - 'frequency': 1000, - 'max_value': 100, - 'min_value': 0, - 'tag_name': 'Random' + "ns=3;i=1003": { + "aggregation_function": "AVG", + "frequency": 1000, + "max_value": 100, + "min_value": 0, + "tag_name": "Random", + }, + "ns=3;i=1004": { + "aggregation_function": "MDN", + "frequency": 1000, + "max_value": 100, + "min_value": 0, + "tag_name": "Sawtooth", }, - 'ns=3;i=1004': { - 'aggregation_function': 'MDN', - 'frequency': 1000, - 'max_value': 100, - 'min_value': 0, - 'tag_name': 'Sawtooth' - } } @fixture def raw_opc_manager(): return OpcManager( - 'TestConnector', 'opc.tcp://localhost:4840', MagicMock(), - MagicMock(), 'opc.tcp://localhost:4840', MagicMock() + "TestConnector", + "opc.tcp://localhost:4840", + MagicMock(), + MagicMock(), + "opc.tcp://localhost:4840", + MagicMock(), + "localhost", ) @fixture def opc_manager(raw_opc_manager): raw_opc_manager.client = MagicMock() - raw_opc_manager.cert_path = 'cert.pem' - raw_opc_manager.private_key_path = 'private_key.pem' - raw_opc_manager.server_cert_path = 'server_cert.pem' + raw_opc_manager.cert_path = "cert.pem" + raw_opc_manager.private_key_path = "private_key.pem" + raw_opc_manager.server_cert_path = "server_cert.pem" return raw_opc_manager @fixture def opc_manager_subscribed(opc_manager): - opc_manager.subscriptions['sub1'] = MagicMock() + opc_manager.subscriptions["sub1"] = MagicMock() return opc_manager def test___str__(opc_manager): - assert str(opc_manager) == 'OpcManager(name=TestConnector, url=opc.tcp://localhost:4840, server_uri=opc.tcp://localhost:4840)\nnodes={}, subscriptions={}' + assert ( + str(opc_manager) + == "OpcManager(name=TestConnector, url=opc.tcp://localhost:4840, server_uri=opc.tcp://localhost:4840)\nnodes={}, subscriptions={}" + ) def test_set_security_success(opc_manager): @@ -71,7 +80,7 @@ def test_set_security_success(opc_manager): SecurityPolicyBasic256, certificate=opc_manager.cert_path, private_key=opc_manager.private_key_path, - server_certificate=opc_manager.server_cert_path + server_certificate=opc_manager.server_cert_path, ) assert opc_manager.client.secure_channel_timeout == 10000000 @@ -85,16 +94,19 @@ def test_set_security_no_cert(opc_manager): try: opc_manager.set_security() except ValueError as e: - assert str( - e) == "Certificate and private key paths must be provided for secure connection." + assert ( + str(e) + == "Certificate and private key paths must be provided for secure connection." + ) else: assert False, "ValueError not raised" assert opc_manager.client.set_security.call_count == 0 -@patch('ingestor.managers.opc_manager.Client') -def test_connect_no_security(client, raw_opc_manager): +@patch("ingestor.managers.opc_manager.metrics") +@patch("ingestor.managers.opc_manager.Client") +def test_connect_no_security(client, mock_metrics, raw_opc_manager): raw_opc_manager.set_security = MagicMock() raw_opc_manager.connect() @@ -102,13 +114,25 @@ def test_connect_no_security(client, raw_opc_manager): client.assert_called_once_with(raw_opc_manager.url) 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( + pod_id=raw_opc_manager.pod_id, + server_name=raw_opc_manager.name + ) + mock_metrics.OPC_CONNECTIONS_TOTAL.labels.return_value.inc.assert_called_once() + mock_metrics.OPC_CONNECTION_STATUS.labels.assert_called_once_with( + pod_id=raw_opc_manager.pod_id, + server_name=raw_opc_manager.name, + server_url=raw_opc_manager.url + ) + mock_metrics.OPC_CONNECTION_STATUS.labels.return_value.set.assert_called_once_with(1) + mock_metrics.OPC_CONNECTIONS_FAILED.labels.assert_not_called() -@patch('ingestor.managers.opc_manager.Client') +@patch("ingestor.managers.opc_manager.Client") def test_connect_with_security(client, raw_opc_manager): - raw_opc_manager.cert_path = 'cert.pem' - raw_opc_manager.private_key_path = 'private_key.pem' - raw_opc_manager.server_cert_path = 'server_cert.pem' + raw_opc_manager.cert_path = "cert.pem" + raw_opc_manager.private_key_path = "private_key.pem" + raw_opc_manager.server_cert_path = "server_cert.pem" raw_opc_manager.set_security = MagicMock() raw_opc_manager.connect() @@ -118,9 +142,48 @@ def test_connect_with_security(client, raw_opc_manager): raw_opc_manager.set_security.assert_called_once() +@patch("ingestor.managers.opc_manager.Client") +@patch("ingestor.managers.opc_manager.metrics") +def test_connect_exception_handling_and_metrics( + mock_metrics_module, mock_opc_client_class, raw_opc_manager +): + mock_client_instance = mock_opc_client_class.return_value + simulated_error_message = "Erro de conexão simulado" + mock_client_instance.connect.side_effect = Exception(simulated_error_message) + + opc_manager_instance = raw_opc_manager + opc_manager_instance.cert_path = None + + with pytest.raises(Exception, match=simulated_error_message): + 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}" + ) + + def test_create_subscription_no_client(raw_opc_manager): try: - raw_opc_manager.create_subscription('sub1') + raw_opc_manager.create_subscription("sub1") except ValueError as e: assert str(e) == "Client not connected. Call connect first." else: @@ -128,55 +191,95 @@ def test_create_subscription_no_client(raw_opc_manager): def test_create_subscription_success_has_period(opc_manager): - opc_manager.create_subscription('sub1', 1000) + opc_manager.create_subscription("sub1", 1000) - opc_manager.client.create_subscription.assert_called_once_with( - 1000, opc_manager) - assert opc_manager.subscriptions['sub1'] is not None + opc_manager.client.create_subscription.assert_called_once_with(1000, opc_manager) + assert opc_manager.subscriptions["sub1"] is not None def test_create_subscription_success_no_period(opc_manager): - opc_manager.create_subscription('sub1', None) + opc_manager.create_subscription("sub1", None) - opc_manager.client.create_subscription.assert_called_once_with( - 500, opc_manager) - assert opc_manager.subscriptions['sub1'] is not None + opc_manager.client.create_subscription.assert_called_once_with(500, opc_manager) + assert opc_manager.subscriptions["sub1"] is not None -def test_subscribe_no_subscription(opc_manager): +@patch("ingestor.managers.opc_manager.metrics") +def test_create_subscription_with_metrics(metrics, opc_manager): + opc_manager.create_subscription("sub1", 1000) + + metrics.OPC_SUBSCRIPTIONS_CREATED.labels.assert_called_once_with( + pod_id=opc_manager.pod_id, server_name=opc_manager.name, slot_name="sub1" + ) + metrics.OPC_SUBSCRIPTIONS_CREATED.labels.return_value.inc.assert_called_once() + + +@patch("ingestor.managers.opc_manager.metrics") +def test_create_subscription_exception_during_client_call( + mock_metrics_module, raw_opc_manager +): + opc_manager_instance = raw_opc_manager + opc_manager_instance.client = MagicMock() + + 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): + opc_manager_instance.create_subscription( + subscription_name, period=simulated_period + ) + + opc_manager_instance.client.create_subscription.assert_called_once_with( + simulated_period, opc_manager_instance + ) + + opc_manager_instance.logger.error.assert_called_once_with( + f"Failed to create subscription {subscription_name} on {opc_manager_instance.name}: {simulated_error_message}" + ) + + mock_metrics_module.OPC_SUBSCRIPTIONS_CREATED.labels.assert_not_called() + + +@patch("ingestor.managers.opc_manager.metrics") +def test_subscribe_no_subscription(metrics, opc_manager): try: - opc_manager.subscribe('sub1', tags, 1000) + opc_manager.subscribe("sub1", tags, 1000) except ValueError as e: - assert str( - e) == "Subscription not created. Call create_subscription first." + assert str(e) == "Subscription not created. Call create_subscription first." else: assert False, "ValueError not raised" + metrics.OPC_TAGS_SUBSCRIBED.labels.assert_not_called() def test_subscribe_success(opc_manager_subscribed): - opc_manager_subscribed.nodes = { - 'ns=3;i=1001': 'data' - } + opc_manager_subscribed.nodes = {"ns=3;i=1001": "data"} - opc_manager_subscribed.subscribe('sub1', tags, 1000) + opc_manager_subscribed.subscribe("sub1", tags, 1000) assert opc_manager_subscribed.nodes == tags assert opc_manager_subscribed.addr_nodes == [ - opc_manager_subscribed.client.get_node(n) for n in tags if n != 'ns=3;i=1001'] + opc_manager_subscribed.client.get_node(n) for n in tags if n != "ns=3;i=1001" + ] def test_unsubscribe_no_subscription(opc_manager): - opc_manager.unsubscribe('sub1') + opc_manager.unsubscribe("sub1") opc_manager.logger.warning.assert_called_once_with( - "Subscription 'sub1' not found. Cannot unsubscribe.") - assert opc_manager.subscriptions.get('sub1') is None + "Subscription 'sub1' not found. Cannot unsubscribe." + ) + assert opc_manager.subscriptions.get("sub1") is None def test_unsubscribe_success(opc_manager_subscribed): - opc_manager_subscribed.unsubscribe('sub1') + opc_manager_subscribed.unsubscribe("sub1") - opc_manager_subscribed.subscriptions.get('sub1') is None + opc_manager_subscribed.subscriptions.get("sub1") is None def test_disconnect_success(opc_manager_subscribed): @@ -184,85 +287,123 @@ def test_disconnect_success(opc_manager_subscribed): opc_manager_subscribed.disconnect() - opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once() + opc_manager_subscribed.subscriptions["sub1"].delete.assert_called_once() assert opc_manager_subscribed.client is None def test_disconnect_error_unsubscribe(opc_manager_subscribed): opc_manager_subscribed.client = MagicMock() - opc_manager_subscribed.subscriptions['sub1'] = MagicMock( + opc_manager_subscribed.subscriptions["sub1"] = MagicMock( delete=MagicMock(side_effect=Exception("Test error")) ) opc_manager_subscribed.disconnect() - opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once() + 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 clean up subscription: Test error") + "Failed to clean up subscription: Test error" + ) def test_disconnect_error(opc_manager_subscribed): opc_manager_subscribed.client = MagicMock() opc_manager_subscribed.client.disconnect = MagicMock( - side_effect=Exception("Test error")) + side_effect=Exception("Test error") + ) opc_manager_subscribed.disconnect() - opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once() + 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") + "Failed to disconnect from OPC UA server: Test error" + ) -def test_datachange_notification(opc_manager_subscribed): +@patch('ingestor.managers.opc_manager.metrics') +def test_disconnect_metrics_on_successful_path(mock_metrics_module, raw_opc_manager): + mock_metrics_module.OPC_CONNECTION_STATUS.reset_mock() + mock_metrics_module.OPC_TAGS_SUBSCRIBED.reset_mock() + + raw_opc_manager.client = MagicMock() + mock_sub1 = MagicMock() + mock_sub2 = MagicMock() + raw_opc_manager.subscriptions = {"sub1": mock_sub1, "sub2": mock_sub2} + + raw_opc_manager.disconnect() + + mock_metrics_module.OPC_CONNECTION_STATUS.labels.assert_called_once_with( + pod_id=raw_opc_manager.pod_id, + server_name=raw_opc_manager.name, + server_url=raw_opc_manager.url + ) + mock_metrics_module.OPC_CONNECTION_STATUS.labels.return_value.set.assert_called_once_with(0) + + mock_metrics_module.OPC_TAGS_SUBSCRIBED.labels.assert_called_once_with( + pod_id=raw_opc_manager.pod_id, + server_name=raw_opc_manager.name + ) + mock_metrics_module.OPC_TAGS_SUBSCRIBED.labels.return_value.set.assert_called_once_with(0) + + +@patch('ingestor.managers.opc_manager.metrics') +def test_datachange_notification(metrics, opc_manager_subscribed): data = MagicMock( monitored_item=MagicMock( Value=MagicMock( Value=MagicMock(Value=42), SourceTimestamp=datetime.strptime( - '2021-01-01T00:00:00', '%Y-%m-%dT%H:%M:%S') - ))) + "2021-01-01T00:00:00", "%Y-%m-%dT%H:%M:%S" + ), + ) + ) + ) opc_manager_subscribed.nodes = { - 'ns=3;i=1001': { - 'tag_name': 'Counter', - 'cycle_rule': { - 'cycle_increment': 1.0, - 'cycle_count': 2 - }, - 'topics': ['topic1', 'topic2'] + "ns=3;i=1001": { + "tag_name": "Counter", + "cycle_rule": {"cycle_increment": 1.0, "cycle_count": 2}, + "topics": ["topic1", "topic2"], } } - opc_manager_subscribed.datachange_notification( - 'ns=3;i=1001', None, data) + metrics.OPC_CYCLES_WITHOUT_DATA.reset_mock() + + opc_manager_subscribed.datachange_notification("ns=3;i=1001", None, data) opc_manager_subscribed.data_manager.publish.assert_any_call( - 'topic1', { - 'tag': 'ns=3;i=1001', - 'name': 'Counter', - 'timestamp': '2021-01-01 00:00:00', - 'value': 42 - }) + "topic1", + { + "tag": "ns=3;i=1001", + "name": "Counter", + "timestamp": "2021-01-01 00:00:00", + "value": 42, + }, + ) opc_manager_subscribed.data_manager.publish.assert_any_call( - 'topic2', { - 'tag': 'ns=3;i=1001', - 'name': 'Counter', - 'timestamp': '2021-01-01 00:00:00', - 'value': 42 - }) - assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 0 + "topic2", + { + "tag": "ns=3;i=1001", + "name": "Counter", + "timestamp": "2021-01-01 00:00:00", + "value": 42, + }, + ) + assert opc_manager_subscribed.nodes["ns=3;i=1001"]["cycle_rule"]["cycle_count"] == 0 + + metrics.OPC_CYCLES_WITHOUT_DATA.labels.assert_called_once_with( + pod_id=opc_manager_subscribed.pod_id, + server_name=opc_manager_subscribed.name + ) + metrics.OPC_CYCLES_WITHOUT_DATA.labels.return_value.set.assert_called_once_with(0) def test_check_cycles_no_notification(opc_manager): # Setup: node with cycle_count just below threshold opc_manager.nodes = { - 'ns=3;i=1001': { - 'tag_name': 'Counter', - 'cycle_rule': { - 'cycle_increment': 1.0, - 'cycle_count': 3.0 - } + "ns=3;i=1001": { + "tag_name": "Counter", + "cycle_rule": {"cycle_increment": 1.0, "cycle_count": 3.0}, } } opc_manager.notification_handler.build_and_send_notification = MagicMock() @@ -270,20 +411,18 @@ def test_check_cycles_no_notification(opc_manager): opc_manager.check_cycles() # After one increment, cycle_count = 4.0, still below threshold - assert opc_manager.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == pytest.approx( - 4.0) + assert opc_manager.nodes["ns=3;i=1001"]["cycle_rule"][ + "cycle_count" + ] == pytest.approx(4.0) opc_manager.notification_handler.build_and_send_notification.assert_not_called() def test_check_cycles_triggers_notification(opc_manager): # Setup: node with cycle_count just below threshold, increment will cross threshold opc_manager.nodes = { - 'ns=3;i=1001': { - 'tag_name': 'Counter', - 'cycle_rule': { - 'cycle_increment': 2.5, - 'cycle_count': 3.0 - } + "ns=3;i=1001": { + "tag_name": "Counter", + "cycle_rule": {"cycle_increment": 2.5, "cycle_count": 3.0}, } } opc_manager.notification_handler.build_and_send_notification = MagicMock() @@ -291,19 +430,23 @@ def test_check_cycles_triggers_notification(opc_manager): opc_manager.check_cycles() # After increment, cycle_count = 5.5, should trigger notification - assert opc_manager.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == pytest.approx( - 5.5) + assert opc_manager.nodes["ns=3;i=1001"]["cycle_rule"][ + "cycle_count" + ] == pytest.approx(5.5) opc_manager.notification_handler.build_and_send_notification.assert_called_once_with( - notification_id='TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED', - message='5.5 cycles without receive from ns=3;i=1001:Counter', + notification_id="TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED", + message="5.5 cycles without receive from ns=3;i=1001:Counter", block="opc_manager", - level=NotificationLevel.WARNING + level=NotificationLevel.WARNING, ) -def test_check_opc_listenning_no_notification(opc_manager): +@patch('ingestor.managers.opc_manager.metrics') +def test_check_opc_listenning_no_notification(metrics, opc_manager): opc_manager.non_receive_count = 3 opc_manager.notification_handler.build_and_send_notification = MagicMock() + metrics.OPC_CYCLES_WITHOUT_DATA.reset_mock() + metrics.OPC_RECONNECTIONS_TOTAL.reset_mock() result = opc_manager.check_opc_listenning() @@ -311,6 +454,13 @@ def test_check_opc_listenning_no_notification(opc_manager): opc_manager.notification_handler.build_and_send_notification.assert_not_called() assert result is False + metrics.OPC_CYCLES_WITHOUT_DATA.labels.assert_called_once_with( + pod_id=opc_manager.pod_id, + server_name=opc_manager.name + ) + metrics.OPC_CYCLES_WITHOUT_DATA.labels.return_value.set.assert_called_once_with(opc_manager.non_receive_count) + metrics.OPC_RECONNECTIONS_TOTAL.labels.assert_not_called() + def test_check_opc_listenning_warning_notification(opc_manager): opc_manager.non_receive_count = 4 @@ -320,17 +470,20 @@ def test_check_opc_listenning_warning_notification(opc_manager): assert opc_manager.non_receive_count == 5 opc_manager.notification_handler.build_and_send_notification.assert_called_once_with( - notification_id=f'OPC_LISTENNING_STOPPED__{opc_manager.name}', - message=f'5 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}', + notification_id=f"OPC_LISTENNING_STOPPED__{opc_manager.name}", + message=f"5 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}", block="opc_manager", - level=NotificationLevel.ERROR + level=NotificationLevel.ERROR, ) assert result is False -def test_check_opc_listenning_error_notification_and_retry(opc_manager): +@patch('ingestor.managers.opc_manager.metrics') +def test_check_opc_listenning_error_notification_and_retry(metrics, opc_manager): opc_manager.non_receive_count = 14 opc_manager.notification_handler.build_and_send_notification = MagicMock() + metrics.OPC_CYCLES_WITHOUT_DATA.reset_mock() + metrics.OPC_RECONNECTIONS_TOTAL.reset_mock() result = opc_manager.check_opc_listenning() @@ -340,16 +493,56 @@ def test_check_opc_listenning_error_notification_and_retry(opc_manager): calls = opc_manager.notification_handler.build_and_send_notification.call_args_list # First call: 5 cycles warning assert calls[0].kwargs == dict( - notification_id=f'OPC_LISTENNING_STOPPED__{opc_manager.name}', - message=f'15 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}', + notification_id=f"OPC_LISTENNING_STOPPED__{opc_manager.name}", + message=f"15 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}", block="opc_manager", - level=NotificationLevel.ERROR + level=NotificationLevel.ERROR, ) # Second call: 15 cycles retry assert calls[1].kwargs == dict( - notification_id=f'OPC_CONNECTION_RETRY__{opc_manager.name}', - message=f'Retrying to connect to server {opc_manager.name}', + notification_id=f"OPC_CONNECTION_RETRY__{opc_manager.name}", + message=f"Retrying to connect to server {opc_manager.name}", block="opc_manager", - level=NotificationLevel.ERROR + level=NotificationLevel.ERROR, ) assert result is True + + metrics.OPC_CYCLES_WITHOUT_DATA.labels.assert_called_once_with( + pod_id=opc_manager.pod_id, + server_name=opc_manager.name + ) + metrics.OPC_CYCLES_WITHOUT_DATA.labels.return_value.set.assert_called_once_with(opc_manager.non_receive_count) + + metrics.OPC_RECONNECTIONS_TOTAL.labels.assert_called_once_with( + pod_id=opc_manager.pod_id, + server_name=opc_manager.name + ) + metrics.OPC_RECONNECTIONS_TOTAL.labels.return_value.inc.assert_called_once() + + +@patch("ingestor.managers.opc_manager.metrics") +def test_init_metrics_calls_correct_metric_methods(mock_metrics): + opc_manager = OpcManager( + name="TestInitConnector", + url="opc.tcp://init.test:4840", + data_manager=MagicMock(), + logger=MagicMock(), + server_uri="opc.tcp://init.test:4840/uri", + notification_handler=MagicMock(), + pod_id="init_pod_localhost", + ) + + mock_metrics.OPC_CONNECTION_STATUS.labels.assert_called_once_with( + pod_id=opc_manager.pod_id, + server_name=opc_manager.name, + server_url=opc_manager.url, + ) + + mock_metrics.OPC_CONNECTION_STATUS.labels.return_value.set.assert_called_once_with(0) + + mock_metrics.OPC_TAGS_SUBSCRIBED.labels.assert_called_once_with( + pod_id=opc_manager.pod_id, + server_name=opc_manager.name + ) + + mock_metrics.OPC_TAGS_SUBSCRIBED.labels.return_value.set.assert_called_once_with(0) diff --git a/tests/unit/test_metrics.py b/tests/unit/test_metrics.py index 8d9ad7c..8750d42 100644 --- a/tests/unit/test_metrics.py +++ b/tests/unit/test_metrics.py @@ -165,50 +165,6 @@ def test_opc_tags_subscribed(): assert set(metrics.OPC_TAGS_SUBSCRIBED._labelnames) == {"pod_id", "server_name"} -def test_opc_datachange_notifications(): - """Verify the definition of OPC_DATACHANGE_NOTIFICATIONS.""" - assert metrics.OPC_DATACHANGE_NOTIFICATIONS is not None - assert isinstance(metrics.OPC_DATACHANGE_NOTIFICATIONS, Counter) - assert ( - metrics.OPC_DATACHANGE_NOTIFICATIONS._name == "opc_datachange_notifications" - ) # REMOVED _total - assert set(metrics.OPC_DATACHANGE_NOTIFICATIONS._labelnames) == { - "pod_id", - "server_name", - "tag_id", - "tag_name", - } - - -def test_opc_tag_last_value(): - """Verify the definition of OPC_TAG_LAST_VALUE.""" - assert metrics.OPC_TAG_LAST_VALUE is not None - assert isinstance(metrics.OPC_TAG_LAST_VALUE, Gauge) - assert metrics.OPC_TAG_LAST_VALUE._name == "opc_tag_last_value" - assert set(metrics.OPC_TAG_LAST_VALUE._labelnames) == { - "pod_id", - "server_name", - "tag_id", - "tag_name", - } - - -def test_opc_tag_last_read_timestamp(): - """Verify the definition of OPC_TAG_LAST_READ_TIMESTAMP.""" - assert metrics.OPC_TAG_LAST_READ_TIMESTAMP is not None - assert isinstance(metrics.OPC_TAG_LAST_READ_TIMESTAMP, Gauge) - assert ( - metrics.OPC_TAG_LAST_READ_TIMESTAMP._name - == "opc_tag_last_read_timestamp_seconds" - ) - assert set(metrics.OPC_TAG_LAST_READ_TIMESTAMP._labelnames) == { - "pod_id", - "server_name", - "tag_id", - "tag_name", - } - - def test_opc_cycles_without_data(): """Verify the definition of OPC_CYCLES_WITHOUT_DATA.""" assert metrics.OPC_CYCLES_WITHOUT_DATA is not None