193 lines
8.3 KiB
Python
193 lines
8.3 KiB
Python
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
|