import asyncio from asyncua import Client from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.observability.logger import Logger from sientia_do.observability.metrics_controller import MetricsController from sientia_do.observability.sientia_monitoring import SientiaMonitoring import ingestor.metrics as metrics from ingestor.connection.certs import CertProvider from ingestor.connection.clock import MonotonicClock from ingestor.connection.security import apply_security class OpcConnectionPool(SientiaMonitoring): """Per-server OPC UA client cache (F0 capstone): reuses live connections, evicts idle ones, and reconnects after a failed attempt (never caches it).""" def __init__( self, cert_provider: CertProvider, logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, ): self.cert_provider = cert_provider self.clock = MonotonicClock() self._connections: dict[str, tuple[Client, float]] = {} # Active subscriptions per server. A subscribed connection only receives # callbacks and never calls get(), so without a pin release_idle() would # evict it as idle and silently stop the reads (SIENTIAPDE-1974 G4). self._pins: dict[str, int] = {} # Connection settings each pooled client was opened with (see _check_cfg). self._configs: dict[str, dict] = {} # ponytail: one Lock per server_id, never cleaned up — server set is # small and bounded. Add eviction of stale locks only if it grows. self._locks: dict[str, asyncio.Lock] = {} SientiaMonitoring.__init__( self, logger=logger, notification_handler=notification_handler, metrics_controller=metrics_controller, ) async def get(self, server_id: str, server_cfg: dict) -> Client: """Return the cached client for `server_id`, connecting one if needed. Double-checked under a per-server lock so concurrent callers for the same new server share one connection (no leaked, untracked sessions). `server_id` is normalized to str, so 1 and '1' share the same session.""" server_id = str(server_id) client = self._cached(server_id) if client is not None: self._check_cfg(server_id, server_cfg) return client async with self._locks.setdefault(server_id, asyncio.Lock()): client = self._cached(server_id) # re-check: a racer may have connected if client is not None: self._check_cfg(server_id, server_cfg) return client return await self._open(server_id, server_cfg) def _check_cfg(self, server_id: str, server_cfg: dict) -> None: """Warn when a caller asks for a server with other connection settings than the pooled client was opened with. The client is reused as is: reconnecting on every mismatch would let one malformed write command drop the reads. Config changes from the slot evict the client first (ServerReader.shutdown).""" opened = self._configs.get(server_id, {}) diff = [ k for k in ('url', 'server_uri', 'use_security') if opened.get(k) != server_cfg.get(k) ] if diff: self.logger.warning( f'OPC server {server_id}: reusing the pooled connection, but {diff} differ ' f'from the settings it was opened with' ) def config_for(self, server_id: str) -> dict | None: """Connection settings of the pooled client for `server_id`, or None when the server has no live pooled connection. Lets a writer reuse exactly the settings the read path opened the connection with.""" server_id = str(server_id) if server_id not in self._connections: return None cfg = self._configs.get(server_id) return dict(cfg) if cfg is not None else None def _cached(self, server_id: str) -> Client | None: entry = self._connections.get(server_id) if entry is None: return None client, _ = entry self._connections[server_id] = (client, self.clock.elapsed()) return client async def _open(self, server_id: str, server_cfg: dict) -> Client: try: client = await self._connect(server_cfg) except Exception: if server_cfg.get('use_security'): # The cached certificate may be the cause (replaced in MinIO, # rejected by the server): download it again on the next attempt. self.cert_provider.invalidate() await self.emit_metric( metric_object=metrics.OPC_CONNECTIONS_FAILED, tags={'pod_id': self.pod_id, 'server_name': server_id}, ) raise self._connections[server_id] = (client, self.clock.elapsed()) self._configs[server_id] = dict(server_cfg) await self._record_connected(server_id, server_cfg['url']) return client async def _connect(self, server_cfg: dict) -> Client: client = Client(server_cfg['url'], timeout=10) # Identify the pod on the OPC server, as the legacy OpcManager did: asyncua # sends `name` as ApplicationName and " Session" as the # session name, so each pod's session is recognizable in the server console. client.name = self.pod_id client.description = self.pod_id pod_uri = self.pod_id.replace('-', ':') client.product_uri = pod_uri client.application_uri = pod_uri server_uri = server_cfg.get('server_uri') if server_uri: # Must match the certificate's SAN URI (KEPServer: BadCertificateUriInvalid). client.application_uri = server_uri if server_cfg.get('use_security'): cert, key = await self.cert_provider.fetch() await apply_security(client, server_cfg.get('server_uri'), cert, key, None) await client.connect() return client async def _record_connected(self, server_id: str, url: str): await self.emit_metric( metric_object=metrics.OPC_CONNECTIONS_TOTAL, tags={'pod_id': self.pod_id, 'server_name': server_id}, ) await self.emit_metric( metric_object=metrics.OPC_CONNECTION_STATUS, method='set', value=1, tags={'pod_id': self.pod_id, 'server_name': server_id, 'server_url': url}, ) def pin(self, server_id: str) -> None: """Mark one more active subscription on `server_id` (never evicted as idle).""" server_id = str(server_id) self._pins[server_id] = self._pins.get(server_id, 0) + 1 def unpin(self, server_id: str) -> None: """Drop one active subscription mark from `server_id`.""" server_id = str(server_id) count = self._pins.get(server_id, 0) - 1 if count > 0: self._pins[server_id] = count else: self._pins.pop(server_id, None) async def release_idle(self, max_idle_s: float): """Disconnect and drop clients unused for longer than `max_idle_s`, except the ones pinned by an active subscription.""" now = self.clock.elapsed() idle_ids = [ sid for sid, (_, last_used) in self._connections.items() if now - last_used > max_idle_s and not self._pins.get(sid) ] for server_id in idle_ids: await self._evict(server_id) async def evict(self, server_id: str): """Disconnect `server_id` now (lost server or changed config); the next get() reconnects. No-op when the server has no pooled client.""" server_id = str(server_id) if server_id in self._connections: await self._evict(server_id) async def close_all(self): """Disconnect every pooled client (e.g. on shutdown).""" # list() snapshots the keys: _evict() pops from self._connections, # so iterating the dict directly would raise "changed size during iteration". for server_id in list(self._connections): # NOSONAR await self._evict(server_id) async def _evict(self, server_id: str): client, _ = self._connections.pop(server_id) self._configs.pop(server_id, None) await self._disconnect_with_retries(client, server_id) await self.emit_metric( metric_object=metrics.OPC_DISCONNECTIONS_TOTAL, tags={'pod_id': self.pod_id, 'server_name': server_id}, ) async def _disconnect_with_retries(self, client: Client, server_id: str) -> None: """Tries 5 times to disconnect, with a delay of 100ms x try (mirrors the legacy OpcManager.disconnection_fallback).""" for i in range(5): try: await client.disconnect() return except Exception as e: self.logger.error( f'Failed to disconnect from {server_id}, attempt {i + 1} of 5: {e}' ) await asyncio.sleep(0.1 * i)