gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1083: add prometheus metrics to opc_manager.

Browse Source
This commit is contained in:
Bruno Domingues
2025-05-29 16:52:10 -03:00
parent 03ed5784b1
commit c49f04ded6
6 changed files with 354 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
ingestor/managers/ingestor_manager.py
View File

@@ -29,6 +29,7 @@ class IngestorManager():
self.opc_servers = {} self.opc_servers = {}
self.notification_handler = notification_handler self.notification_handler = notification_handler
self.pod_id = pod_id
def initialize_opc_from_config(self, server_config: dict, def initialize_opc_from_config(self, server_config: dict,
data_manager: DataManager, logger: Logger) -> OpcManager | None: data_manager: DataManager, logger: Logger) -> OpcManager | None:
@@ -56,7 +57,7 @@ class IngestorManager():
manager = OpcManager( manager = OpcManager(
server_config['name'], server_config['url'], server_config['name'], server_config['url'],
data_manager, logger, server_config['server_uri'], 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('private_key_path'),
server_config.get('server_cert_path') server_config.get('server_cert_path')
) )

53
ingestor/managers/opc_manager.py
View File

@@ -7,11 +7,12 @@ from asyncua.sync import Client
from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.models import NotificationLevel
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import NotificationHandler
from ingestor.managers.data_manager import DataManager from ingestor.managers.data_manager import DataManager
import ingestor.metrics as metrics
class OpcManager(): class OpcManager():
def __init__(self, name: str, url: str, data_manager: DataManager, 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): cert_path: str = None, private_key_path: str = None, server_cert_path: str = None):
self.url = url self.url = url
self.name = name self.name = name
@@ -26,8 +27,10 @@ class OpcManager():
self.nodes = {} self.nodes = {}
self.subscriptions = {} self.subscriptions = {}
self.data_manager = data_manager self.data_manager = data_manager
self.notification_handler = notification_handler 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): def __str__(self):
return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \ 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. Exception: If the connection to the OPC server fails.
""" """
self.client = Client(self.url) metrics.OPC_CONNECTIONS_TOTAL.labels(pod_id=self.pod_id, server_name=self.name).inc()
if self.cert_path: try:
self.set_security() self.client = Client(self.url)
self.logger.info('Starting connection...') if self.cert_path:
self.client.connect() 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): def create_subscription(self, name: str, period: int = 500):
""" """
@@ -105,11 +117,14 @@ class OpcManager():
if not self.client: if not self.client:
raise ValueError("Client not connected. Call connect first.") raise ValueError("Client not connected. Call connect first.")
try:
p = period if period != None else 500 p = period if period is not None else 500
self.subscriptions[name] = self.client.create_subscription( self.subscriptions[name] = self.client.create_subscription(p, self)
p, self) self.logger.info(f'Subscription {name} created on {self.name}.')
self.logger.info('Subscription created.') 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): def subscribe(self, subscription: str, nodes: dict, collect_period: int):
""" """
@@ -129,14 +144,13 @@ class OpcManager():
""" """
if not self.subscriptions.get(subscription): if not self.subscriptions.get(subscription):
raise ValueError( raise ValueError("Subscription not created. Call create_subscription first.")
"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.logger.info(f"Subscribing to nodes: {nodes}")
self.addr_nodes = [self.client.get_node( self.addr_nodes = [self.client.get_node(n) for n in nodes if n not in self.nodes]
n) for n in nodes if n not in self.nodes]
self.nodes.update(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 self.collect_period = collect_period
for node, config in self.nodes.items(): for node, config in self.nodes.items():
@@ -202,6 +216,8 @@ class OpcManager():
finally: finally:
del self.client del self.client
self.client = None 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.") self.logger.warning("Disconnected from OPC UA server.")
def datachange_notification(self, node, _val, data): def datachange_notification(self, node, _val, data):
@@ -236,6 +252,7 @@ class OpcManager():
self.nodes[tag]['cycle_rule']['cycle_count'] = 0 self.nodes[tag]['cycle_rule']['cycle_count'] = 0
self.non_receive_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 = { data = {
'tag': tag, 'tag': tag,
@@ -278,6 +295,7 @@ class OpcManager():
""" """
self.non_receive_count += 1 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: if self.non_receive_count >= 5:
self.notification_handler.build_and_send_notification( self.notification_handler.build_and_send_notification(
notification_id=f'OPC_LISTENNING_STOPPED__{self.name}', notification_id=f'OPC_LISTENNING_STOPPED__{self.name}',
@@ -287,6 +305,7 @@ class OpcManager():
level=NotificationLevel.ERROR level=NotificationLevel.ERROR
) )
if self.non_receive_count >= 15: 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( self.notification_handler.build_and_send_notification(
notification_id=f'OPC_CONNECTION_RETRY__{self.name}', notification_id=f'OPC_CONNECTION_RETRY__{self.name}',
message=f'Retrying to connect to server {self.name}', message=f'Retrying to connect to server {self.name}',

20
ingestor/metrics.py
View File

@@ -1,11 +1,7 @@
from prometheus_client import Counter, Gauge, Histogram 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"] POD_ID_LABEL = ["pod_id"]
SERVER_LABELS = ["pod_id", "server_name", "server_url"] 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"] KAFKA_LABELS = ["pod_id", "topic"]
REDIS_LABELS = ["pod_id", "operation"] REDIS_LABELS = ["pod_id", "operation"]
NOTIFICATION_LABELS = ["pod_id", "level", "block"] NOTIFICATION_LABELS = ["pod_id", "level", "block"]
@@ -37,17 +33,14 @@ APP_UP = Gauge(
ACTIVE_INGESTORS = Gauge( ACTIVE_INGESTORS = Gauge(
"ingestor_active_total", "ingestor_active_total",
"Number of active ingestors reported by Redis", "Number of active ingestors reported by Redis",
# No pod_id here, as it's a global view from Redis
) )
SLOTS_TOTAL = Gauge( SLOTS_TOTAL = Gauge(
"ingestor_slots_total", "ingestor_slots_total",
"Total number of slots configured in Redis", "Total number of slots configured in Redis",
# No pod_id here, as it's a global view from Redis
) )
LEASES_TOTAL = Gauge( LEASES_TOTAL = Gauge(
"ingestor_leases_total", "ingestor_leases_total",
"Total number of leases (allocated slots) in Redis", "Total number of leases (allocated slots) in Redis",
# No pod_id here, as it's a global view from Redis
) )
SLOTS_MANAGED = Gauge( SLOTS_MANAGED = Gauge(
"ingestor_slots_managed_current", "ingestor_slots_managed_current",
@@ -101,19 +94,6 @@ OPC_TAGS_SUBSCRIBED = Gauge(
"Current number of OPC tags subscribed on a server", "Current number of OPC tags subscribed on a server",
["pod_id", "server_name"], ["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 = Gauge(
"opc_cycles_without_data", "opc_cycles_without_data",
"Current number of cycles without receiving data from a server", "Current number of cycles without receiving data from a server",

4
tests/unit/managers/test_ingestor_manager.py
View File

@@ -60,7 +60,8 @@ def test_initialize_opc_from_config(opc_manager, ingestor_manager):
'server_uri': 'http://opcua-server.simulator', 'server_uri': 'http://opcua-server.simulator',
'cert_path': '/path/to/cert', 'cert_path': '/path/to/cert',
'private_key_path': '/path/to/private_key', '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() 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( opc_manager.assert_called_once_with(
server_config['name'], server_config['url'], ingestor_manager.data_manager, ingestor_manager.logger, server_config['name'], server_config['url'], ingestor_manager.data_manager, ingestor_manager.logger,
server_config['server_uri'], ingestor_manager.notification_handler, server_config['server_uri'], ingestor_manager.notification_handler,
server_config['pod_id'],
server_config['cert_path'], server_config['private_key_path'], server_config['cert_path'], server_config['private_key_path'],
server_config['server_cert_path'] server_config['server_cert_path']
) )

433
tests/unit/managers/test_opc_manager.py
View File

@@ -1,6 +1,7 @@
import json import json
from datetime import datetime from datetime import datetime
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from prometheus_client import Gauge
from pytest import fixture from pytest import fixture
from asyncua.crypto.security_policies import SecurityPolicyBasic256 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 from sientia_do.notifications.models import NotificationLevel
tags = { tags = {
'ns=3;i=1001': { "ns=3;i=1001": {
'aggregation_function': 'LTS', "aggregation_function": "LTS",
'frequency': 1000, "frequency": 1000,
'max_value': 100, "max_value": 100,
'min_value': 0, "min_value": 0,
'tag_name': 'Counter' "tag_name": "Counter",
}, },
'ns=3;i=1003': { "ns=3;i=1003": {
'aggregation_function': 'AVG', "aggregation_function": "AVG",
'frequency': 1000, "frequency": 1000,
'max_value': 100, "max_value": 100,
'min_value': 0, "min_value": 0,
'tag_name': 'Random' "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 @fixture
def raw_opc_manager(): def raw_opc_manager():
return OpcManager( return OpcManager(
'TestConnector', 'opc.tcp://localhost:4840', MagicMock(), "TestConnector",
MagicMock(), 'opc.tcp://localhost:4840', MagicMock() "opc.tcp://localhost:4840",
MagicMock(),
MagicMock(),
"opc.tcp://localhost:4840",
MagicMock(),
"localhost",
) )
@fixture @fixture
def opc_manager(raw_opc_manager): def opc_manager(raw_opc_manager):
raw_opc_manager.client = MagicMock() raw_opc_manager.client = MagicMock()
raw_opc_manager.cert_path = 'cert.pem' raw_opc_manager.cert_path = "cert.pem"
raw_opc_manager.private_key_path = 'private_key.pem' raw_opc_manager.private_key_path = "private_key.pem"
raw_opc_manager.server_cert_path = 'server_cert.pem' raw_opc_manager.server_cert_path = "server_cert.pem"
return raw_opc_manager return raw_opc_manager
@fixture @fixture
def opc_manager_subscribed(opc_manager): def opc_manager_subscribed(opc_manager):
opc_manager.subscriptions['sub1'] = MagicMock() opc_manager.subscriptions["sub1"] = MagicMock()
return opc_manager return opc_manager
def test___str__(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): def test_set_security_success(opc_manager):
@@ -71,7 +80,7 @@ def test_set_security_success(opc_manager):
SecurityPolicyBasic256, SecurityPolicyBasic256,
certificate=opc_manager.cert_path, certificate=opc_manager.cert_path,
private_key=opc_manager.private_key_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 assert opc_manager.client.secure_channel_timeout == 10000000
@@ -85,16 +94,19 @@ def test_set_security_no_cert(opc_manager):
try: try:
opc_manager.set_security() opc_manager.set_security()
except ValueError as e: except ValueError as e:
assert str( assert (
e) == "Certificate and private key paths must be provided for secure connection." str(e)
== "Certificate and private key paths must be provided for secure connection."
)
else: else:
assert False, "ValueError not raised" assert False, "ValueError not raised"
assert opc_manager.client.set_security.call_count == 0 assert opc_manager.client.set_security.call_count == 0
@patch('ingestor.managers.opc_manager.Client') @patch("ingestor.managers.opc_manager.metrics")
def test_connect_no_security(client, raw_opc_manager): @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.set_security = MagicMock()
raw_opc_manager.connect() 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) client.assert_called_once_with(raw_opc_manager.url)
raw_opc_manager.client.connect.assert_called_once() raw_opc_manager.client.connect.assert_called_once()
raw_opc_manager.set_security.assert_not_called() 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): def test_connect_with_security(client, raw_opc_manager):
raw_opc_manager.cert_path = 'cert.pem' raw_opc_manager.cert_path = "cert.pem"
raw_opc_manager.private_key_path = 'private_key.pem' raw_opc_manager.private_key_path = "private_key.pem"
raw_opc_manager.server_cert_path = 'server_cert.pem' raw_opc_manager.server_cert_path = "server_cert.pem"
raw_opc_manager.set_security = MagicMock() raw_opc_manager.set_security = MagicMock()
raw_opc_manager.connect() 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() 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): def test_create_subscription_no_client(raw_opc_manager):
try: try:
raw_opc_manager.create_subscription('sub1') raw_opc_manager.create_subscription("sub1")
except ValueError as e: except ValueError as e:
assert str(e) == "Client not connected. Call connect first." assert str(e) == "Client not connected. Call connect first."
else: else:
@@ -128,55 +191,95 @@ def test_create_subscription_no_client(raw_opc_manager):
def test_create_subscription_success_has_period(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( opc_manager.client.create_subscription.assert_called_once_with(1000, opc_manager)
1000, opc_manager) assert opc_manager.subscriptions["sub1"] is not None
assert opc_manager.subscriptions['sub1'] is not None
def test_create_subscription_success_no_period(opc_manager): 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( opc_manager.client.create_subscription.assert_called_once_with(500, opc_manager)
500, opc_manager) assert opc_manager.subscriptions["sub1"] is not None
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: try:
opc_manager.subscribe('sub1', tags, 1000) opc_manager.subscribe("sub1", tags, 1000)
except ValueError as e: except ValueError as e:
assert str( assert str(e) == "Subscription not created. Call create_subscription first."
e) == "Subscription not created. Call create_subscription first."
else: else:
assert False, "ValueError not raised" assert False, "ValueError not raised"
metrics.OPC_TAGS_SUBSCRIBED.labels.assert_not_called()
def test_subscribe_success(opc_manager_subscribed): def test_subscribe_success(opc_manager_subscribed):
opc_manager_subscribed.nodes = { opc_manager_subscribed.nodes = {"ns=3;i=1001": "data"}
'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.nodes == tags
assert opc_manager_subscribed.addr_nodes == [ 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): def test_unsubscribe_no_subscription(opc_manager):
opc_manager.unsubscribe('sub1') opc_manager.unsubscribe("sub1")
opc_manager.logger.warning.assert_called_once_with( opc_manager.logger.warning.assert_called_once_with(
"Subscription 'sub1' not found. Cannot unsubscribe.") "Subscription 'sub1' not found. Cannot unsubscribe."
assert opc_manager.subscriptions.get('sub1') is None )
assert opc_manager.subscriptions.get("sub1") is None
def test_unsubscribe_success(opc_manager_subscribed): 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): 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.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 assert opc_manager_subscribed.client is None
def test_disconnect_error_unsubscribe(opc_manager_subscribed): def test_disconnect_error_unsubscribe(opc_manager_subscribed):
opc_manager_subscribed.client = MagicMock() opc_manager_subscribed.client = MagicMock()
opc_manager_subscribed.subscriptions['sub1'] = MagicMock( opc_manager_subscribed.subscriptions["sub1"] = MagicMock(
delete=MagicMock(side_effect=Exception("Test error")) delete=MagicMock(side_effect=Exception("Test error"))
) )
opc_manager_subscribed.disconnect() 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.client = None
opc_manager_subscribed.logger.error.assert_called_once_with( 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): def test_disconnect_error(opc_manager_subscribed):
opc_manager_subscribed.client = MagicMock() opc_manager_subscribed.client = MagicMock()
opc_manager_subscribed.client.disconnect = MagicMock( opc_manager_subscribed.client.disconnect = MagicMock(
side_effect=Exception("Test error")) side_effect=Exception("Test error")
)
opc_manager_subscribed.disconnect() 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.client = None
opc_manager_subscribed.logger.error.assert_called_once_with( 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( data = MagicMock(
monitored_item=MagicMock( monitored_item=MagicMock(
Value=MagicMock( Value=MagicMock(
Value=MagicMock(Value=42), Value=MagicMock(Value=42),
SourceTimestamp=datetime.strptime( 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 = { opc_manager_subscribed.nodes = {
'ns=3;i=1001': { "ns=3;i=1001": {
'tag_name': 'Counter', "tag_name": "Counter",
'cycle_rule': { "cycle_rule": {"cycle_increment": 1.0, "cycle_count": 2},
'cycle_increment': 1.0, "topics": ["topic1", "topic2"],
'cycle_count': 2
},
'topics': ['topic1', 'topic2']
} }
} }
opc_manager_subscribed.datachange_notification( metrics.OPC_CYCLES_WITHOUT_DATA.reset_mock()
'ns=3;i=1001', None, data)
opc_manager_subscribed.datachange_notification("ns=3;i=1001", None, data)
opc_manager_subscribed.data_manager.publish.assert_any_call( opc_manager_subscribed.data_manager.publish.assert_any_call(
'topic1', { "topic1",
'tag': 'ns=3;i=1001', {
'name': 'Counter', "tag": "ns=3;i=1001",
'timestamp': '2021-01-01 00:00:00', "name": "Counter",
'value': 42 "timestamp": "2021-01-01 00:00:00",
}) "value": 42,
},
)
opc_manager_subscribed.data_manager.publish.assert_any_call( opc_manager_subscribed.data_manager.publish.assert_any_call(
'topic2', { "topic2",
'tag': 'ns=3;i=1001', {
'name': 'Counter', "tag": "ns=3;i=1001",
'timestamp': '2021-01-01 00:00:00', "name": "Counter",
'value': 42 "timestamp": "2021-01-01 00:00:00",
}) "value": 42,
assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 0 },
)
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): def test_check_cycles_no_notification(opc_manager):
# Setup: node with cycle_count just below threshold # Setup: node with cycle_count just below threshold
opc_manager.nodes = { opc_manager.nodes = {
'ns=3;i=1001': { "ns=3;i=1001": {
'tag_name': 'Counter', "tag_name": "Counter",
'cycle_rule': { "cycle_rule": {"cycle_increment": 1.0, "cycle_count": 3.0},
'cycle_increment': 1.0,
'cycle_count': 3.0
}
} }
} }
opc_manager.notification_handler.build_and_send_notification = MagicMock() 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() opc_manager.check_cycles()
# After one increment, cycle_count = 4.0, still below threshold # After one increment, cycle_count = 4.0, still below threshold
assert opc_manager.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == pytest.approx( assert opc_manager.nodes["ns=3;i=1001"]["cycle_rule"][
4.0) "cycle_count"
] == pytest.approx(4.0)
opc_manager.notification_handler.build_and_send_notification.assert_not_called() opc_manager.notification_handler.build_and_send_notification.assert_not_called()
def test_check_cycles_triggers_notification(opc_manager): def test_check_cycles_triggers_notification(opc_manager):
# Setup: node with cycle_count just below threshold, increment will cross threshold # Setup: node with cycle_count just below threshold, increment will cross threshold
opc_manager.nodes = { opc_manager.nodes = {
'ns=3;i=1001': { "ns=3;i=1001": {
'tag_name': 'Counter', "tag_name": "Counter",
'cycle_rule': { "cycle_rule": {"cycle_increment": 2.5, "cycle_count": 3.0},
'cycle_increment': 2.5,
'cycle_count': 3.0
}
} }
} }
opc_manager.notification_handler.build_and_send_notification = MagicMock() 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() opc_manager.check_cycles()
# After increment, cycle_count = 5.5, should trigger notification # After increment, cycle_count = 5.5, should trigger notification
assert opc_manager.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == pytest.approx( assert opc_manager.nodes["ns=3;i=1001"]["cycle_rule"][
5.5) "cycle_count"
] == pytest.approx(5.5)
opc_manager.notification_handler.build_and_send_notification.assert_called_once_with( opc_manager.notification_handler.build_and_send_notification.assert_called_once_with(
notification_id='TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED', notification_id="TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED",
message='5.5 cycles without receive from ns=3;i=1001:Counter', message="5.5 cycles without receive from ns=3;i=1001:Counter",
block="opc_manager", 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.non_receive_count = 3
opc_manager.notification_handler.build_and_send_notification = MagicMock() 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() 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() opc_manager.notification_handler.build_and_send_notification.assert_not_called()
assert result is False 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): def test_check_opc_listenning_warning_notification(opc_manager):
opc_manager.non_receive_count = 4 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 assert opc_manager.non_receive_count == 5
opc_manager.notification_handler.build_and_send_notification.assert_called_once_with( opc_manager.notification_handler.build_and_send_notification.assert_called_once_with(
notification_id=f'OPC_LISTENNING_STOPPED__{opc_manager.name}', 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)}', message=f"5 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}",
block="opc_manager", block="opc_manager",
level=NotificationLevel.ERROR level=NotificationLevel.ERROR,
) )
assert result is False 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.non_receive_count = 14
opc_manager.notification_handler.build_and_send_notification = MagicMock() 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() 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 calls = opc_manager.notification_handler.build_and_send_notification.call_args_list
# First call: 5 cycles warning # First call: 5 cycles warning
assert calls[0].kwargs == dict( assert calls[0].kwargs == dict(
notification_id=f'OPC_LISTENNING_STOPPED__{opc_manager.name}', 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)}', message=f"15 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}",
block="opc_manager", block="opc_manager",
level=NotificationLevel.ERROR level=NotificationLevel.ERROR,
) )
# Second call: 15 cycles retry # Second call: 15 cycles retry
assert calls[1].kwargs == dict( assert calls[1].kwargs == dict(
notification_id=f'OPC_CONNECTION_RETRY__{opc_manager.name}', notification_id=f"OPC_CONNECTION_RETRY__{opc_manager.name}",
message=f'Retrying to connect to server {opc_manager.name}', message=f"Retrying to connect to server {opc_manager.name}",
block="opc_manager", block="opc_manager",
level=NotificationLevel.ERROR level=NotificationLevel.ERROR,
) )
assert result is True 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)

44
tests/unit/test_metrics.py
View File

@@ -165,50 +165,6 @@ def test_opc_tags_subscribed():
assert set(metrics.OPC_TAGS_SUBSCRIBED._labelnames) == {"pod_id", "server_name"} 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(): def test_opc_cycles_without_data():
"""Verify the definition of OPC_CYCLES_WITHOUT_DATA.""" """Verify the definition of OPC_CYCLES_WITHOUT_DATA."""
assert metrics.OPC_CYCLES_WITHOUT_DATA is not None assert metrics.OPC_CYCLES_WITHOUT_DATA is not None