feat(gateway): SubscriptionEngine multi-name fan-out (C1)

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

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