sientia-dataops-opc-ingestor/ingestor/connection/pool.py

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 _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 "<description> Session<n>" 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)