vitor-aignosi
ecb055eefb
SIENTIAPDE-1973: Implement per-connector OPC client certificates and remove legacy read engine. This update overhauls certificate management, transitioning from a single platform certificate to individual client certificates defined in the slot configuration. The CertProvider now downloads certificates from MinIO based on certificate_id, applying dynamic security policies (mode, policy) from the slot. The legacy READ_ENGINE=legacy option and OpcManager have been entirely removed, standardizing on the ServerReader with strict slot v2 enforcement. Write command resolution (ServerConfigResolver) now consistently uses connectors joined with opc_certificates. The CERT_BUCKET environment variable is no longer used.
207 lines
9.2 KiB
Python
207 lines
9.2 KiB
Python
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, is_secured
|
|
|
|
|
|
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 (security, certificate) evict the client first:
|
|
the IngestorManager re-initializes the reader (ServerReader.shutdown)."""
|
|
opened = self._configs.get(server_id, {})
|
|
diff = [k for k in ('url', 'security', 'certificate') 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 is_secured(server_cfg):
|
|
# 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(server_cfg['certificate']['certificate_id'])
|
|
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
|
|
# Security only with a client certificate (SIENTIAPDE-1974 D5); apply_security
|
|
# sets the certificate's application URI, which must match its SAN URI.
|
|
if is_secured(server_cfg):
|
|
certificate = server_cfg['certificate']
|
|
cert, key = await self.cert_provider.fetch(certificate)
|
|
await apply_security(client, server_cfg['security'], certificate, cert, key)
|
|
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)
|