diff --git a/ingestor/engines/__init__.py b/ingestor/engines/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestor/engines/subscription.py b/ingestor/engines/subscription.py new file mode 100644 index 0000000..745f2a8 --- /dev/null +++ b/ingestor/engines/subscription.py @@ -0,0 +1,192 @@ +import json + +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler +from sientia_do.notifications.models import NotificationLevel +from sientia_do.observability.logger import Logger +from sientia_do.observability.metrics_controller import MetricsController +from sientia_do.observability.sientia_monitoring import SientiaMonitoring +from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, OPC_TIMEZONE + +import ingestor.metrics as metrics +from ingestor.connection.clock import MonotonicClock +from ingestor.connection.pool import OpcConnectionPool +from ingestor.io.data_sink import DataSink + + +class SubscriptionEngine(SientiaMonitoring): + """Adapts OpcManager's subscription/datachange path (F1 capstone) to the + shared OpcConnectionPool and the multi-name slot v2 structure: on data + change it fans one reading out to every configured name via DataSink, + instead of the old flat per-`topics` publish loop.""" + + def __init__( + self, + pool: OpcConnectionPool, + sink: DataSink, + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + metadata: dict | None = None, + ): + self.pool = pool + self.sink = sink + self.metadata = metadata or {} + self.nodes: dict = {} + self.subscriptions: dict = {} + self._clients: dict = {} + self._servers: dict = {} + self.non_receive_count = 0 + + SientiaMonitoring.__init__( + self, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + async def create_subscription(self, server_id: str, server_cfg: dict, name: str): + """Get a client for `server_id` from the pool and create a subscription + named `name` on it, with the engine itself as the datachange handler.""" + client = await self.pool.get(server_id, server_cfg) + try: + self.subscriptions[name] = await client.create_subscription( + server_cfg['subscription_period_ms'], self + ) + except Exception as e: + self.logger.error(f'Failed to create subscription {name} on {server_id}: {e}') + raise + self._clients[name] = client + self._servers[name] = server_id + self.logger.info(f'Subscription {name} created on {server_id}.') + await self.emit_metric( + metric_object=metrics.OPC_SUBSCRIPTIONS_CREATED, + method='inc', + value=1, + tags={'pod_id': self.pod_id, 'server_name': server_id, 'slot_name': name}, + ) + + async def subscribe(self, subscription: str, nodes: dict, collect_period: int): + """Register `nodes` (slot v2 tags: frequency + names + active_read) under + `subscription` and start monitoring them for data changes.""" + if not self.subscriptions.get(subscription): + raise ValueError('Subscription not created. Call create_subscription first.') + client = self._clients[subscription] + server_id = self._servers[subscription] + addr_nodes = [client.get_node(n) for n in nodes] + self.nodes.update(nodes) + self._init_cycle_rules(nodes, collect_period, server_id) + + await self.subscriptions[subscription].subscribe_data_change(addr_nodes) + await self.emit_metric( + metric_object=metrics.OPC_TAGS_SUBSCRIBED, + method='set', + value=len(self.nodes), + tags={'pod_id': self.pod_id, 'server_name': server_id}, + ) + + def _init_cycle_rules(self, nodes: dict, collect_period: int, server_id: str): + for node in nodes: + self.nodes[node]['server_name'] = server_id + self.nodes[node]['cycle_rule'] = { + 'cycle_increment': collect_period * 1000 / float(self.nodes[node]['frequency']), + 'cycle_count': 0, + } + + async def unsubscribe(self, subscription: str): + if not self.subscriptions.get(subscription): + self.logger.warning(f"Subscription '{subscription}' not found. Cannot unsubscribe.") + return + await self.subscriptions[subscription].delete() + del self.subscriptions[subscription] + self._clients.pop(subscription, None) + self._servers.pop(subscription, None) + self.logger.info(f'Unsubscribed from {subscription}.') + + async def datachange_notification(self, node, _val, data): + """asyncua subscription callback: fan the new value out to every name + configured for this address, then reset cycle/lag bookkeeping.""" + value = data.monitored_item.Value.Value.Value + source_timestamp = data.monitored_item.Value.SourceTimestamp.replace(tzinfo=OPC_TIMEZONE) + tag = str(node) + config = self.nodes[tag] + + await self.sink.publish_reading( + tag=tag, + value=value, + timestamp=source_timestamp.strftime(DATETIME_FORMAT_WITH_TZ), + names=config['names'], + ) + await self._record_read_lag(config, source_timestamp) + await self._reset_cycle_state(config) + + async def _record_read_lag(self, config: dict, source_timestamp): + lag_seconds = MonotonicClock.now_wall() - source_timestamp.timestamp() + await self.emit_metric( + metric_object=metrics.OPC_READ_LAG, + method='set', + value=lag_seconds, + tags={'pod_id': self.pod_id, 'server_name': config.get('server_name', 'unknown')}, + ) + + async def _reset_cycle_state(self, config: dict): + config['cycle_rule']['cycle_count'] = 0 + self.non_receive_count = 0 + await self.emit_metric( + metric_object=metrics.OPC_CYCLES_WITHOUT_DATA, + method='set', + value=0, + tags={'pod_id': self.pod_id, 'server_name': config.get('server_name', 'unknown')}, + ) + + async def check_cycles(self): + """Increment every node's cycle count and warn on nodes stuck >= 5 cycles.""" + for node, config in self.nodes.items(): + config['cycle_rule']['cycle_count'] += config['cycle_rule']['cycle_increment'] + if config['cycle_rule']['cycle_count'] >= 5: + await self._notify_node_stopped(node, config) + + async def _notify_node_stopped(self, node: str, config: dict): + cycles = config['cycle_rule']['cycle_count'] + names = ','.join(n['tag_name'] for n in config['names']) + await self.send_notification_async( + metadata=self.metadata, + notification_id=f'TAG_{node}:{names}_LISTENNING_STOPPED', + message=f'{cycles} cycles without receive from {node}:{names}', + block='subscription_engine', + level=NotificationLevel.WARNING, + ) + + async def check_opc_listenning(self, server_name: str) -> bool: + """Track cycles without any datachange for `server_name`; warn at 5, + error + signal reconnect at 15. `server_name` is explicit (unlike + OpcManager's `self.name`) since one engine can serve several servers.""" + self.non_receive_count += 1 + await self.emit_metric( + metric_object=metrics.OPC_CYCLES_WITHOUT_DATA, + method='set', + value=self.non_receive_count, + tags={'pod_id': self.pod_id, 'server_name': server_name}, + ) + if self.non_receive_count >= 5: + await self.send_notification_async( + metadata=self.metadata, + notification_id=f'OPC_LISTENNING_STOPPED__{server_name}', + message=f'{self.non_receive_count} cycles without receive from OPC ' + f'{server_name}. Tags: {json.dumps(self.nodes, default=str)}', + block='subscription_engine', + level=NotificationLevel.ERROR, + ) + if self.non_receive_count >= 15: + await self.emit_metric( + metric_object=metrics.OPC_RECONNECTIONS_TOTAL, + tags={'pod_id': self.pod_id, 'server_name': server_name}, + ) + await self.send_notification_async( + metadata=self.metadata, + notification_id=f'OPC_CONNECTION_RETRY__{server_name}', + message=f'Retrying to connect to server {server_name}', + block='subscription_engine', + level=NotificationLevel.ERROR, + ) + return True + return False diff --git a/tests/unit/engines/__init__.py b/tests/unit/engines/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/engines/test_subscription.py b/tests/unit/engines/test_subscription.py new file mode 100644 index 0000000..a088657 --- /dev/null +++ b/tests/unit/engines/test_subscription.py @@ -0,0 +1,205 @@ +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest +from sientia_do.temporal.constants import OPC_TIMEZONE + +import ingestor.metrics as metrics +from ingestor.engines.subscription import SubscriptionEngine + + +def make_engine(): + eng = SubscriptionEngine( + pool=AsyncMock(), + sink=AsyncMock(), + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=AsyncMock(), + metadata={}, + ) + eng.emit_metric = AsyncMock() + eng.send_notification_async = AsyncMock() + return eng + + +def make_node(name='ns=2;i=2'): + node = MagicMock() + node.__str__ = lambda self: name + return node + + +def make_data(value=42, timestamp=None): + data = MagicMock() + data.monitored_item.Value.Value.Value = value + if timestamp is not None: + data.monitored_item.Value.SourceTimestamp = timestamp + return data + + +@pytest.mark.asyncio +async def test_datachange_fans_out(): + eng = make_engine() + eng.nodes['ns=2;i=2'] = { + 'names': [{'tag_name': 'a'}, {'tag_name': 'b'}], + 'cycle_rule': {'cycle_increment': 1, 'cycle_count': 3}, + } + node = make_node() + data = make_data(42) + + await eng.datachange_notification(node, 42, data) + + assert eng.sink.publish_reading.await_count == 1 + _, kwargs = eng.sink.publish_reading.call_args + assert len(kwargs['names']) == 2 + assert kwargs['tag'] == 'ns=2;i=2' + assert kwargs['value'] == 42 + + +@pytest.mark.asyncio +async def test_datachange_resets_cycle_count_and_non_receive_count(): + eng = make_engine() + eng.non_receive_count = 7 + eng.nodes['ns=2;i=2'] = { + 'names': [{'tag_name': 'a'}], + 'cycle_rule': {'cycle_increment': 1, 'cycle_count': 3}, + } + + await eng.datachange_notification(make_node(), 42, make_data(42)) + + assert eng.nodes['ns=2;i=2']['cycle_rule']['cycle_count'] == 0 + assert eng.non_receive_count == 0 + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_CYCLES_WITHOUT_DATA, + method='set', + value=0, + tags={'pod_id': eng.pod_id, 'server_name': 'unknown'}, + ) + + +@pytest.mark.asyncio +async def test_datachange_emits_read_lag(): + eng = make_engine() + eng.nodes['ns=2;i=2'] = { + 'names': [{'tag_name': 'a'}], + 'cycle_rule': {'cycle_increment': 1, 'cycle_count': 0}, + 'server_name': 'srv1', + } + source_ts = datetime.now(tz=OPC_TIMEZONE) + + await eng.datachange_notification(make_node(), 42, make_data(42, timestamp=source_ts)) + + lag_call = next( + c + for c in eng.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_READ_LAG + ) + assert lag_call.kwargs['tags'] == {'pod_id': eng.pod_id, 'server_name': 'srv1'} + assert abs(lag_call.kwargs['value']) < 5 + + +@pytest.mark.asyncio +async def test_create_subscription_uses_pool_client(): + eng = make_engine() + client = AsyncMock() + eng.pool.get.return_value = client + + await eng.create_subscription('srv1', {'subscription_period_ms': 500}, 'sub1') + + eng.pool.get.assert_awaited_once_with('srv1', {'subscription_period_ms': 500}) + client.create_subscription.assert_awaited_once_with(500, eng) + assert eng.subscriptions['sub1'] is not None + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_SUBSCRIPTIONS_CREATED, + method='inc', + value=1, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'slot_name': 'sub1'}, + ) + + +@pytest.mark.asyncio +async def test_subscribe_registers_nodes_with_cycle_rule(): + eng = make_engine() + client = AsyncMock() + eng.pool.get.return_value = client + await eng.create_subscription('srv1', {'subscription_period_ms': 500}, 'sub1') + + nodes = { + 'ns=2;i=2': { + 'frequency': 1000, + 'names': [{'tag_name': 'a'}, {'tag_name': 'b'}], + 'active_read': {'enabled': False}, + } + } + + await eng.subscribe('sub1', nodes, collect_period=1) + + assert eng.nodes['ns=2;i=2']['server_name'] == 'srv1' + assert eng.nodes['ns=2;i=2']['cycle_rule']['cycle_count'] == 0 + assert eng.nodes['ns=2;i=2']['cycle_rule']['cycle_increment'] == pytest.approx(1.0) + eng.subscriptions['sub1'].subscribe_data_change.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_subscribe_without_subscription_raises(): + eng = make_engine() + with pytest.raises(ValueError, match='Subscription not created'): + await eng.subscribe('sub1', {}, collect_period=1) + + +@pytest.mark.asyncio +async def test_unsubscribe_missing_logs_warning(): + eng = make_engine() + await eng.unsubscribe('sub1') + eng.logger.warning.assert_called_once_with( + "Subscription 'sub1' not found. Cannot unsubscribe." + ) + + +@pytest.mark.asyncio +async def test_unsubscribe_deletes_subscription(): + eng = make_engine() + client = AsyncMock() + eng.pool.get.return_value = client + await eng.create_subscription('srv1', {'subscription_period_ms': 500}, 'sub1') + sub = eng.subscriptions['sub1'] + + await eng.unsubscribe('sub1') + + sub.delete.assert_awaited_once() + assert 'sub1' not in eng.subscriptions + + +@pytest.mark.asyncio +async def test_check_cycles_triggers_notification_over_threshold(): + eng = make_engine() + eng.nodes['ns=2;i=2'] = { + 'names': [{'tag_name': 'a'}], + 'cycle_rule': {'cycle_increment': 5.5, 'cycle_count': 0}, + } + + await eng.check_cycles() + + eng.send_notification_async.assert_called_once() + + +@pytest.mark.asyncio +async def test_check_opc_listenning_below_threshold(): + eng = make_engine() + eng.non_receive_count = 2 + + result = await eng.check_opc_listenning('srv1') + + assert eng.non_receive_count == 3 + assert result is False + eng.send_notification_async.assert_not_called() + + +@pytest.mark.asyncio +async def test_check_opc_listenning_triggers_reconnect_at_15(): + eng = make_engine() + eng.non_receive_count = 14 + + result = await eng.check_opc_listenning('srv1') + + assert result is True + assert eng.send_notification_async.call_count == 2