sientia-dataops-opc-ingestor/ingestor/engines/subscription.py

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 = {}
        # Addresses each subscription monitors, so unsubscribe can forget them
        # (otherwise check_cycles keeps alarming on tags nobody reads anymore).
        self._sub_nodes: dict[str, set[str]] = {}
        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.pool.pin(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._sub_nodes.setdefault(subscription, set()).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
        sub = self.subscriptions.pop(subscription)
        self._clients.pop(subscription, None)
        server_id = self._servers.pop(subscription, None)
        if server_id is not None:
            self.pool.unpin(server_id)
        self._forget_nodes(subscription)
        try:
            await sub.delete()
        except Exception as e:
            # The connection may already be gone (lost server); the subscription is
            # dropped from the books either way so the slot can be re-subscribed.
            self.logger.warning(f'Failed to delete subscription {subscription}: {e}')
        self.logger.info(f'Unsubscribed from {subscription}.')
        if server_id is not None:
            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 _forget_nodes(self, subscription: str) -> None:
        """Drop the subscription's addresses, except those another subscription
        of this engine still monitors."""
        nodes = self._sub_nodes.pop(subscription, set())
        still_monitored = set().union(*self._sub_nodes.values())
        for node in nodes:
            if node not in still_monitored:
                self.nodes.pop(node, None)

    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