diff --git a/CLAUDE.md b/CLAUDE.md index 7330de9..44c07c9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -40,7 +40,9 @@ app.py (asyncio loop, signals, Prometheus server) -> Ingestor (ingestor.py) — lifecycle: prepare_ingestor() -> loop() -> shutdown() -> IngestorManager (managers/ingestor_manager.py) — central coordinator -> ResourceManager (managers/resource_manager.py) — Redis slot leases, heartbeats, active-ingestor registry - -> OpcManager (managers/opc_manager.py), one instance per OPC server — connection, subscriptions, datachange callbacks + -> ServerReader (managers/server_reader.py), one per OPC server — SubscriptionEngine over the shared OpcConnectionPool + (READ_ENGINE=legacy: OpcManager (managers/opc_manager.py), the rc3 reader kept as plan B until the end of SIENTIAPDE-1974) + -> OpcConnectionPool (connection/pool.py) + CertProvider (connection/certs.py) — pooled connections, cert from MinIO -> DataManager (managers/data_manager.py) — MongoDB persistence + optional Kafka publish ``` @@ -56,7 +58,7 @@ app.py (asyncio loop, signals, Prometheus server) Any change to slot/lease counts should be traced through `manage_leases`/`manage_no_slots` — that's where the rebalancing math lives, not in `ResourceManager` itself (which just does Redis primitives). -**OpcManager** owns one OPC UA `Client` connection (asyncua), handles security policy setup (cert/key-based, `SecurityPolicyBasic256`), subscriptions per tag group, and `datachange_notification` callbacks that push into `data_queue` for `DataManager` to consume. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles`). +**ServerReader** (SIENTIAPDE-1974, default `READ_ENGINE=v2`; the legacy `OpcManager` stays as plan B with `READ_ENGINE=legacy` — slot v1 only, certificate files from the Gitea clone — until it is unused and the tests pass the same without it) reads one OPC server through a `SubscriptionEngine`: the connection comes from the shared `OpcConnectionPool` (also used by NATS writes and gRPC), which fetches the platform certificate from MinIO via `CertProvider` and uses `SecurityPolicyBasic256Sha256` — no certificate files in the repo/image. Slots are parsed by `io/slots.py` (`parse_slot`), which accepts slot v2 (`names[]`) and converts the legacy v1 (`topics`). `datachange_notification` fans each value out to every `names` entry via `DataSink`. Subscribed connections are pinned in the pool so the idle sweep never drops them. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles` / `check_opc_listenning`); a lost server is shut down (pool connection evicted) and re-initialized on the next loop. **Metrics** (`ingestor/metrics.py`): all Prometheus metrics defined here, always labeled by `pod_id`. Emitted via `SientiaMonitoring.emit_metric(...)` (from `sientia_do`), not the prometheus client directly — new metrics should follow that pattern for consistency with logging/notifications. diff --git a/README.md b/README.md index a240e71..f062786 100644 --- a/README.md +++ b/README.md @@ -123,9 +123,9 @@ health probe, NetworkPolicy, cert cache) and the `# OPC Gateway` block in # Ensure services are accessible on localhost with appropriate ports ``` -6. **Generate an OPC UA client certificate** (only if connecting to a secured OPC UA server) +6. **Register the OPC UA client certificate** (only if connecting to a secured OPC UA server) - See [Certificate Generation](#-certificate-generation) below. + See [OPC Client Certificate](#-opc-client-certificate) below. ## Usage @@ -193,37 +193,50 @@ pytest --cov=ingestor --cov-report=html The feeder creates sample OPC tag configurations in Redis that the ingestor can discover and manage. -## 🔐 Certificate Generation +## 🔐 OPC Client Certificate -OPC UA servers that require secure connections (e.g. KEPServer with `Basic256` policy) validate the ingestor via an X.509 client certificate. Use `scripts/generate-and-push-opc-cert.sh` to generate that certificate and push it to the Gitea repo the pod clones on boot. +OPC UA servers that require secure connections validate the gateway with a single, +shared platform X.509 client certificate. Every OPC connection — subscription reads, +NATS writes and gRPC reads — uses it with `Basic256Sha256` (SIENTIAPDE-1974). -```bash -GITEA_PASSWORD='' ./scripts/generate-and-push-opc-cert.sh -``` +The certificate is **never** read from the repository or the image. On connect, the +`CertProvider` (`ingestor/connection/certs.py`): +1. reads the single document of the Mongo collection `platform_certificates`: + `{"client_cert": {"minio_object_name": "..."}, "private_key": {"minio_object_name": "..."}}`; +2. downloads both objects from the MinIO bucket `CERT_BUCKET` (default + `streamlit-connectors`) into `CERT_CACHE_DIR` (default `/tmp/opc-certs`). -Environment variables (all optional except `GITEA_PASSWORD`): +Requirements for a server to accept the connection: +- the server's slot config has `use_security: true` (a slot v1 server with a + `cert_path` is treated the same way; the path itself is ignored); +- the certificate's SAN URI equals the server's `server_uri` in the slot — it is sent + as the client `application_uri` (KEPServer rejects a mismatch with + `BadCertificateUriInvalid`); +- the server exposes a `Basic256Sha256` endpoint and trusts the certificate (e.g. + KEPServer: *OPC UA Configuration Manager → Trusted Clients → Trust*, then restart + its runtime). -| Variable | Description | Default | -|----------|-------------|---------| -| `GITEA_URL` | Gitea in-cluster URL | `https://git.sientia.ai` | -| `GITEA_USER` | Gitea user | `gitea_admin` | -| `GITEA_PASSWORD` | Gitea password (**required**) | — | -| `GITEA_REPO` | Repo the ingestor pod clones | `gitea_admin/sientia-dataops-opc-ingestor` | -| `BRANCH` | Branch to push the cert to | `main` | -| `APP_URI` | SAN URI of the cert — **must match** the `uri` field of the corresponding document in the `OPC_servers` Mongo collection, since the ingestor uses it as `application_uri` (KEPServer rejects with `BadCertificateUriInvalid` on mismatch) | `urn:sientia:opc-ingestor` | -| `CN` | Certificate common name | `sientia-opc-ingestor` | -| `DAYS` | Certificate validity (days) | `3650` | -| `FORCE` | Set to `1` to overwrite an existing cert already pushed to the repo | unset | +MinIO is only contacted when a certificate is first needed, so a MinIO outage only +affects secured servers, which are retried on the next main-loop iteration. -What the script does: -1. Generates a self-signed RSA-2048 X.509 cert + key with `openssl`, with the SAN URI, `keyUsage`, and `extendedKeyUsage` extensions OPC UA / KEPServer require. -2. Clones the ingestor's Gitea repo, copies the cert/key (`.pem`) and a `.der` copy (for manual import) into `certs/`, commits, and pushes. -3. Prints the paths the pod will see after cloning (`/app/certs/opc-ingestor-cert.pem`, `/app/certs/opc-ingestor-key.pem`) and the next manual steps. +### Plan B: legacy read engine (`READ_ENGINE=legacy`) -After pushing, you still need to: -- Set `cert_path` / `private_key_path` (and `uri`) on the server's document in the `OPC_servers` collection so `OpcManager` picks them up (see [OPC Server Configuration](#opc-server-configuration)). -- Restart the ingestor deployment so it re-clones the repo: `kubectl -n sientia rollout restart deployment/opc-ingestor`. -- On first connection, the server rejects the new cert — trust it in the OPC UA server's console (e.g. KEPServer: *OPC UA Configuration Manager → Trusted Clients → Trust*) and restart its runtime. +Until the end of SIENTIAPDE-1974 the legacy reader is kept for validation and as a +fallback. `READ_ENGINE` selects the read path (default `v2`; an invalid value falls +back to `v2` with a warning): + +| `READ_ENGINE` | Reader | Slot | Certificate | +|---|---|---|---| +| `v2` (default) | `ServerReader` + `SubscriptionEngine` over the shared pool | v1 or v2 | MinIO + Mongo (`CertProvider`), `Basic256Sha256` | +| `legacy` | the rc3 `OpcManager`, unchanged, with its own connection | **v1 only** (a v2 slot is rejected and notified) | files from the repository the pod clones (`cert_path` / `private_key_path` in the slot), `Basic256` | + +Writes and gRPC always use the pool (MinIO certificate), whatever the read engine. +The legacy engine needs the certificate files in the Gitea branch the pod clones; +`scripts/generate-and-push-opc-cert.sh` generates them and pushes them to +`certs/` (`GITEA_PASSWORD='' BRANCH= ./scripts/generate-and-push-opc-cert.sh`). +Each version deploy creates a new Gitea branch without `certs/`, so they have to be +copied again for the legacy engine. Remove the legacy engine (OpcManager, its tests +and the script) once it is unused and the test suite passes the same without it. ## 🧪 Testing @@ -306,7 +319,12 @@ gateway-specific environment variables. ### OPC Server Configuration -OPC servers are configured through Redis with the following structure. `cert_path` / `private_key_path` are only required for secured connections (see [Certificate Generation](#-certificate-generation)): +OPC servers are configured by the Orchestrator through Redis, one key per slot +(`slot:opc_tags:`). The gateway reads the **slot v2** structure (contract C1): +one entry per address, with one `names` item per collection / tag name / model, so +the same address can feed several models under different names. For secured +connections set `use_security: true` (the certificate comes from MinIO, see +[OPC Client Certificate](#-opc-client-certificate)): ```json { @@ -315,30 +333,30 @@ OPC servers are configured through Redis with the following structure. `cert_pat "name": "server1", "server_id": "1", "url": "opc.tcp://localhost:4841", - "server_uri": "http://opcua-server.simulator", - "cert_path": "/app/certs/opc-ingestor-cert.pem", - "private_key_path": "/app/certs/opc-ingestor-key.pem", + "server_uri": "urn:sientia:opc-ingestor", + "use_security": true, + "subscription_period_ms": 7500, "tags": { "ns=2;i=2": { - "aggr_func": "avg", - "data_range": [ - -100, - 100 - ], - "frequency": "15000", - "server_id": "1", - "tag_address": "ns=2;i=2", - "tag_name": "Counter", - "topics": [ - "raw_scouter-opcua-orchestrated-pipeline", - "raw_scouter-basic-sum-model" - ] + "frequency": "15000", + "active_read": {"enabled": false}, + "names": [ + {"collection": "raw_scouter-opcua-orchestrated-pipeline", "tag_name": "Counter", "model_name": "m1", "model_id": "1"}, + {"collection": "raw_scouter-basic-sum-model", "tag_name": "Contador", "model_name": "m2", "model_id": "2"} + ] + } } } } } ``` +The legacy **slot v1** (each tag with `tag_name` + `topics`, the server with +`cert_path`) is still accepted and converted on read (`ingestor/io/slots.py`): one +`names` item per topic, and `use_security` is true when `cert_path` is set (the path +itself is ignored). This lets the gateway and the Orchestrator be deployed in any +order. Documents written from a v1 slot have no `model_id`, exactly as before. + ## 🔧 Development ### Project Structure @@ -348,13 +366,14 @@ sientia-dataops-opc-ingestor/ │ ├── managers/ # Component managers │ │ ├── data_manager.py # Data persistence and export │ │ ├── ingestor_manager.py # Main coordination -│ │ ├── opc_manager.py # OPC UA server management -│ │ └── resource_manager.py # Resource coordination +│ │ ├── opc_manager.py # Legacy reader, plan B (READ_ENGINE=legacy) +│ │ ├── resource_manager.py # Resource coordination +│ │ └── server_reader.py # OPC UA server reads (SubscriptionEngine + pool) │ ├── app.py # Main application entry point │ ├── ingestor.py # Core ingestor logic │ └── metrics.py # Prometheus metrics definitions ├── simulator/ # OPC simulation and testing tools -├── scripts/ # Ops scripts (e.g. OPC UA cert generation) +├── scripts/ # Ops scripts (e2e seed; legacy OPC cert generation) ├── tests/ # Test suite └── requirements.txt # Python dependencies ``` diff --git a/ingestor/app.py b/ingestor/app.py index e18ab41..54c78b6 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -58,16 +58,17 @@ async def main(): # NOSONAR ingestor.logger.info('Ingestor prepared. Starting main loop.') # Gateway write/active-read/gRPC wiring (Task 18). Only built if prepare_ingestor - # got far enough to have a data_manager (needed by CertProvider/DataSink) -- - # skipped entirely if prepare failed before that point. Construction can hit - # the network (MinIO bucket check, Postgres engine): a failure there must not - # kill the read path, so the app continues without the gateway runtime. + # got far enough to have an ingestor_manager (its data_manager and connection + # pool are shared) -- skipped entirely if prepare failed before that point. The + # read path (subscriptions) lives in the ingestor_manager and is already running: + # a failure here only disables writes/gRPC, the app continues without the runtime. runtime: GatewayRuntime | None = None if ingestor.ingestor_manager is not None: try: runtime = GatewayRuntime( config=GatewayConfig.from_env(), data_manager=ingestor.ingestor_manager.data_manager, + pool=ingestor.ingestor_manager.pool, logger=ingestor.logger, notification_handler=ingestor.notification_handler, metrics_controller=ingestor.metrics_controller, diff --git a/ingestor/config.py b/ingestor/config.py index dab6676..93d5a9a 100644 --- a/ingestor/config.py +++ b/ingestor/config.py @@ -22,13 +22,25 @@ class GatewayConfig: nats_queue: str = 'gateway-writers' grpc_enabled: bool = False grpc_port: int = 50051 - cert_bucket: str = 'sientia-dataops' + # Legacy frontend bucket the platform already provisions (SIENTIAPDE-1974 D7); + # the design's `sientia-dataops` bucket does not exist in the platform MinIO. + cert_bucket: str = 'streamlit-connectors' cert_cache_dir: str = '/tmp/opc-certs' # noqa: S108 # nosec B108 - default is spec'd, overridable via env write_timestamp: bool = True write_quality: bool = True write_min_pause_s: float = 0.0 idle_conn_max_s: float = 300.0 + @classmethod + def certs_from_env(cls) -> 'GatewayConfig': + """Only the certificate settings — what the read path needs. Plain strings, + no number parsing, so a malformed write/gRPC env var (GRPC_PORT, + WRITE_MIN_PAUSE_S, ...) can only disable the runtime, never the reads.""" + return cls( + cert_bucket=getenv('CERT_BUCKET', cls.cert_bucket), + cert_cache_dir=getenv('CERT_CACHE_DIR', cls.cert_cache_dir), + ) + @classmethod def from_env(cls) -> 'GatewayConfig': return cls( diff --git a/ingestor/connection/certs.py b/ingestor/connection/certs.py index 71dc762..2c8d4a8 100644 --- a/ingestor/connection/certs.py +++ b/ingestor/connection/certs.py @@ -1,12 +1,29 @@ +import asyncio +import os +from os import getenv from pathlib import Path +from sientia_do.repository.minio_repository import MinioRepository + +from ingestor.config import _env_bool + COLLECTION_NAME = 'platform_certificates' +class CertificateNotConfiguredError(RuntimeError): + """No usable platform OPC client certificate is registered (Mongo + MinIO).""" + + class CertProvider: """Downloads the platform's client cert + private key (Mongo metadata, MinIO bytes) into a writable cache dir — pod root FS is read-only (§12), - so `cache_dir` must be an emptyDir/tmpfs mount.""" + so `cache_dir` must be an emptyDir/tmpfs mount. + + The certificate is single and shared by every connection (reads, writes, + gRPC), so it is downloaded once and cached; `invalidate()` forces a fresh + download (the pool calls it when a secured connection fails, e.g. after the + certificate was replaced in MinIO). The lock serializes downloads so two + connections never rewrite the cache files at the same time.""" def __init__(self, mongo, minio, cache_dir: str, bucket: str, metadata: dict | None = None): self.mongo = mongo @@ -14,16 +31,84 @@ class CertProvider: self.cache_dir = cache_dir self.bucket = bucket self.metadata = metadata or {} + self._lock = asyncio.Lock() + self._paths: tuple[str, str] | None = None async def fetch(self) -> tuple[str, str]: + async with self._lock: + if self._paths is None: + self._paths = await self._download_all() + return self._paths + + def invalidate(self) -> None: + self._paths = None + + async def _download_all(self) -> tuple[str, str]: doc = await self.mongo.find_one(COLLECTION_NAME, {}, self.metadata) + try: + cert_object = doc['client_cert']['minio_object_name'] + key_object = doc['private_key']['minio_object_name'] + except (KeyError, TypeError) as e: + raise CertificateNotConfiguredError( + f'No OPC client certificate configured: the Mongo collection ' + f"'{COLLECTION_NAME}' needs a document with client_cert.minio_object_name " + f'and private_key.minio_object_name (objects in the MinIO bucket ' + f"'{self.bucket}'). Register the platform certificate before using " + f'secured OPC servers.' + ) from e Path(self.cache_dir).mkdir(parents=True, exist_ok=True) - cert_path = await self._download(doc['client_cert']['minio_object_name']) - key_path = await self._download(doc['private_key']['minio_object_name']) + cert_path = await self._download(cert_object) + key_path = await self._download(key_object) return cert_path, key_path async def _download(self, object_name: str) -> str: data = await self.minio.download_file(object_name, self.bucket, self.metadata) - local_path = str(Path(self.cache_dir) / Path(object_name).name) - Path(local_path).write_bytes(data) - return local_path + local_path = Path(self.cache_dir) / Path(object_name).name + # Atomic replace: a connection loading the previous file never sees it half-written. + tmp_path = local_path.with_name(local_path.name + '.tmp') + tmp_path.write_bytes(data) + os.replace(tmp_path, local_path) + return str(local_path) + + +class LazyMinio: + """Builds the MinIO client on first use. MinioRepository's constructor hits the + network (bucket check), so building it eagerly would let a MinIO outage at pod + start take the whole read path down. Built lazily, only the servers that need a + certificate fail — and they are retried on the next loop.""" + + def __init__(self, factory): + self._factory = factory + self._client = None + + async def download_file(self, object_name: str, bucket: str, metadata: dict): + if self._client is None: + self._client = self._factory() + return await self._client.download_file(object_name, bucket, metadata) + + +def build_cert_provider( + mongo, config, logger, notification_handler, metrics_controller, metadata=None +) -> CertProvider: + """CertProvider shared by the read path (IngestorManager) and the write path + (GatewayRuntime): one certificate source — MinIO + Mongo — for every connection.""" + + def minio_factory() -> MinioRepository: + return MinioRepository( + endpoint=getenv('MINIO_ENDPOINT', 'localhost:9000'), + access_key=getenv('MINIO_ACCESS_KEY', 'minioadmin'), + secret_key=getenv('MINIO_SECRET_KEY', 'minioadmin'), + secure=_env_bool('MINIO_SECURE', True), + bucket=config.cert_bucket, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + return CertProvider( + mongo=mongo, + minio=LazyMinio(minio_factory), + cache_dir=config.cert_cache_dir, + bucket=config.cert_bucket, + metadata=metadata, + ) diff --git a/ingestor/connection/pool.py b/ingestor/connection/pool.py index 73002f5..d95abc3 100644 --- a/ingestor/connection/pool.py +++ b/ingestor/connection/pool.py @@ -26,6 +26,12 @@ class OpcConnectionPool(SientiaMonitoring): 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] = {} @@ -41,16 +47,35 @@ class OpcConnectionPool(SientiaMonitoring): """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).""" + 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: @@ -63,6 +88,10 @@ class OpcConnectionPool(SientiaMonitoring): 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}, @@ -70,13 +99,23 @@ class OpcConnectionPool(SientiaMonitoring): 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() @@ -96,15 +135,39 @@ class OpcConnectionPool(SientiaMonitoring): 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`.""" + """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 + 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, @@ -114,6 +177,7 @@ class OpcConnectionPool(SientiaMonitoring): 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, @@ -122,7 +186,7 @@ class OpcConnectionPool(SientiaMonitoring): 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 - OpcManager.disconnection_fallback).""" + the legacy OpcManager.disconnection_fallback).""" for i in range(5): try: await client.disconnect() diff --git a/ingestor/engines/subscription.py b/ingestor/engines/subscription.py index 745f2a8..db5b8b7 100644 --- a/ingestor/engines/subscription.py +++ b/ingestor/engines/subscription.py @@ -35,6 +35,9 @@ class SubscriptionEngine(SientiaMonitoring): 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__( @@ -57,6 +60,7 @@ class SubscriptionEngine(SientiaMonitoring): 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, @@ -74,6 +78,7 @@ class SubscriptionEngine(SientiaMonitoring): 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) @@ -96,11 +101,35 @@ class SubscriptionEngine(SientiaMonitoring): 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] + sub = self.subscriptions.pop(subscription) self._clients.pop(subscription, None) - self._servers.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 diff --git a/ingestor/gateway_runtime.py b/ingestor/gateway_runtime.py index c23ec22..59a6a88 100644 --- a/ingestor/gateway_runtime.py +++ b/ingestor/gateway_runtime.py @@ -2,30 +2,25 @@ object graph and wires it into the app's lifecycle. Reuses the existing ingestor's logger/notification_handler/metrics_controller and -its already-connected DataManager (so CertProvider shares the existing Mongo -connection instead of opening a second one -- MongoDBRepository.find_one has the -exact (collection_name, filters, metadata) signature CertProvider.fetch calls). +its already-connected DataManager. The OpcConnectionPool (and its CertProvider, +MinIO + Mongo) belongs to the read path in IngestorManager (SIENTIAPDE-1974): +reads start in prepare_ingestor, before this runtime exists, and must keep running +if anything here fails. The runtime only borrows the pool for writes and gRPC. -# ponytail: subscription v2 swap + scheduled_tick wiring happen at 1974 -# integration; when swapping, build ONE SubscriptionEngine per OPC server (its -# non_receive_count is engine-wide -- see plan ledger F1 carry-forward). +# ponytail: scheduled_tick + gRPC slot_lookup wiring land in the next 1974 RC. """ -from os import getenv - 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.repository.minio_repository import MinioRepository from sientia_do.temporal.activities.postgres import Postgres from sientia_do.utils.connectors_config import build_postgres_config from ingestor.adapters.grpc_server import GrpcServer from ingestor.adapters.nats_consumer import NatsWriteConsumer -from ingestor.config import GatewayConfig, _env_bool -from ingestor.connection.certs import CertProvider +from ingestor.config import GatewayConfig from ingestor.connection.pool import OpcConnectionPool from ingestor.engines.active_read import ActiveReadEngine from ingestor.engines.write import WriteEngine @@ -44,6 +39,7 @@ class GatewayRuntime(SientiaMonitoring): self, config: GatewayConfig, data_manager: DataManager, + pool: OpcConnectionPool, logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, @@ -55,26 +51,7 @@ class GatewayRuntime(SientiaMonitoring): self._nats_started = False self._nats_notified = False # notify once per outage, not per 5s retry - minio = MinioRepository( - endpoint=getenv('MINIO_ENDPOINT', 'localhost:9000'), - access_key=getenv('MINIO_ACCESS_KEY', 'minioadmin'), - secret_key=getenv('MINIO_SECRET_KEY', 'minioadmin'), - secure=_env_bool('MINIO_SECURE', True), - bucket=config.cert_bucket, - logger=logger, - notification_handler=notification_handler, - metrics_controller=metrics_controller, - ) - cert_provider = CertProvider( - mongo=data_manager.mongo_repository, - minio=minio, - cache_dir=config.cert_cache_dir, - bucket=config.cert_bucket, - metadata=self.metadata, - ) - self.pool = OpcConnectionPool( - cert_provider, logger, notification_handler, metrics_controller - ) + self.pool = pool # owned by IngestorManager (read path); closed there postgres = Postgres( **build_postgres_config(), @@ -128,7 +105,8 @@ class GatewayRuntime(SientiaMonitoring): if self._nats_started: await self.nats_consumer.stop() self._nats_started = False - await self.pool.close_all() + # The pool is not closed here: IngestorManager.shutdown() owns it and runs + # after this, once its readers have dropped their subscriptions. async def idle_sweep(self) -> None: await self.pool.release_idle(self.config.idle_conn_max_s) diff --git a/ingestor/io/data_sink.py b/ingestor/io/data_sink.py index 633e1df..8555e2c 100644 --- a/ingestor/io/data_sink.py +++ b/ingestor/io/data_sink.py @@ -21,15 +21,16 @@ class DataSink: await self.data_manager.emit_metric(**kwargs) async def publish_reading(self, tag: str, value, timestamp, names: list[dict]) -> None: - """Publish one MongoDB document per name entry and record its tag value metric.""" + """Publish one MongoDB document per name entry and record its tag value metric. + + `model_id` is only written when the name carries one: names converted from a + slot v1 have none, and their documents must stay identical to the legacy ones.""" for name in names: - doc = { - 'tag': tag, - 'name': name['tag_name'], - 'model_id': name['model_id'], - 'timestamp': timestamp, - 'value': value, - } + doc = {'tag': tag, 'name': name['tag_name']} + if 'model_id' in name: + doc['model_id'] = name['model_id'] + doc['timestamp'] = timestamp + doc['value'] = value await self.data_manager.publish(name['collection'], doc) await self.emit_metric( metric_object=metrics.OPC_TAG_VALUE, @@ -38,7 +39,7 @@ class DataSink: tags={ 'pod_id': self.data_manager.pod_id, 'tag_name': name['tag_name'], - 'model_id': name['model_id'], + 'model_id': name.get('model_id', ''), }, ) diff --git a/ingestor/io/slots.py b/ingestor/io/slots.py index 1e720b3..d8ff695 100644 --- a/ingestor/io/slots.py +++ b/ingestor/io/slots.py @@ -1,25 +1,47 @@ -"""Parse the slot v2 structure written by the Orchestrator to Redis (contract C1).""" +"""Parse the slot written by the Orchestrator to Redis into the slot v2 structure (contract C1). + +Both shapes are accepted so the gateway and the Orchestrator can be deployed in +any order (SIENTIAPDE-1974): + +- v2: each tag carries `names` (one entry per collection / tag name / model); +- v1 (legacy): each tag carries `tag_name` + `topics`, and the server carries the + file-based `cert_path`. v1 is converted to v2: one `names` entry per topic, and + `use_security` is true when the server had a certificate path (the certificate + itself now always comes from MinIO via CertProvider, never from that path). +""" SERVER_FIELDS = ('server_id', 'url', 'server_uri', 'use_security', 'subscription_period_ms') +# What a connection/subscription needs (the pool keys clients by server_id itself). +CONNECTION_FIELDS = tuple(f for f in SERVER_FIELDS if f != 'server_id') + + +def _parse_names(address: str, tag: dict) -> list[dict]: + names = tag.get('names') + if names: + return names + topics = tag.get('topics') + if topics and tag.get('tag_name'): + return [{'collection': topic, 'tag_name': tag['tag_name']} for topic in topics] + raise ValueError(f'tag {address!r} has no names') def _parse_tag(address: str, tag: dict) -> dict: - names = tag.get('names') - if not names: - raise ValueError(f'tag {address!r} has no names') return { 'frequency': tag['frequency'], - 'names': names, + 'names': _parse_names(address, tag), 'active_read': tag.get('active_read', {'enabled': False}), } -def _parse_server(server: dict) -> dict: +def _parse_server(name: str, server: dict) -> dict: parsed = {field: server[field] for field in SERVER_FIELDS if field in server} + parsed['name'] = server.get('name', name) + if 'use_security' not in parsed: + parsed['use_security'] = bool(server.get('cert_path')) parsed['tags'] = {addr: _parse_tag(addr, tag) for addr, tag in server['tags'].items()} return parsed def parse_slot(raw: dict) -> dict: - """Normalize a raw slot v2 dict into `{server_name: {..., tags: {addr: tag}}}`.""" - return {name: _parse_server(server) for name, server in raw.items()} + """Normalize a raw slot (v1 or v2) into `{server_name: {..., tags: {addr: tag}}}`.""" + return {name: _parse_server(name, server) for name, server in raw.items()} diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 0034d0a..0be1773 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -1,5 +1,6 @@ import traceback from copy import deepcopy +from os import getenv from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel @@ -8,9 +9,21 @@ from sientia_do.observability.metrics_controller import MetricsController from sientia_do.observability.sientia_monitoring import SientiaMonitoring import ingestor.metrics as metrics +from ingestor.config import GatewayConfig +from ingestor.connection.certs import build_cert_provider +from ingestor.connection.pool import OpcConnectionPool +from ingestor.io.data_sink import DataSink +from ingestor.io.slots import parse_slot from ingestor.managers.data_manager import DataManager from ingestor.managers.opc_manager import OpcManager from ingestor.managers.resource_manager import ResourceManager +from ingestor.managers.server_reader import ServerReader + +# READ_ENGINE (SIENTIAPDE-1974): 'v2' = ServerReader/SubscriptionEngine over the shared +# pool, certificate from MinIO (default). 'legacy' = plan B: the untouched rc3 +# OpcManager, slot v1 only, certificate files from the Gitea clone (cert_path). Kept +# for validation until the end of 1974; remove once unused and the tests pass without it. +READ_ENGINES = ('v2', 'legacy') class IngestorManager(SientiaMonitoring): @@ -47,7 +60,10 @@ class IngestorManager(SientiaMonitoring): Attributes: data_manager (DataManager): Manages data persistence - opc_managers (dict): Dictionary of OPC managers keyed by server name + opc_managers (dict): ServerReader per OPC server, keyed by server name + pool (OpcConnectionPool): OPC connections shared by reads, writes and gRPC; + owned here because reads start in prepare_ingestor, before GatewayRuntime + sink (DataSink): read egress (fan-out to every slot v2 `names` entry) resource_manager (ResourceManager): Manages Redis-based resource coordination number_of_slots (int): Total number of slots configured in the system poll_interval (int): Main loop polling interval @@ -68,6 +84,8 @@ class IngestorManager(SientiaMonitoring): logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, + gateway_config: GatewayConfig | None = None, + read_engine: str | None = None, ): redis_host: str = redis_data['host'] redis_port: int = int(redis_data['port']) @@ -90,6 +108,26 @@ class IngestorManager(SientiaMonitoring): metrics_controller=metrics_controller, ) self.opc_managers: dict = {} + # Read path (SIENTIAPDE-1974): pooled connections with the certificate from + # MinIO. Nothing here touches the network at construction (MinIO is lazy), so + # a MinIO/Postgres/NATS outage can't stop the reads before they start. + config = gateway_config or GatewayConfig.certs_from_env() + self.cert_provider = build_cert_provider( + mongo=self.data_manager.mongo_repository, + config=config, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + metadata=metadata, + ) + self.pool = OpcConnectionPool( + self.cert_provider, logger, notification_handler, metrics_controller + ) + self.sink = DataSink(self.data_manager, None) + self._server_cert_warned: set[str] = set() + self.read_engine = self._resolve_read_engine( + read_engine or getenv('READ_ENGINE') or 'v2', logger + ) self.resource_manager = ResourceManager( host=redis_host, port=redis_port, @@ -109,36 +147,19 @@ class IngestorManager(SientiaMonitoring): self.metadata = metadata - async def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None: - """ - Initializes an OPC Manager instance using the provided server configuration. + @staticmethod + def _resolve_read_engine(value: str, logger: Logger) -> str: + engine = value.strip().lower() + if engine not in READ_ENGINES: + logger.warning(f'Invalid READ_ENGINE {value!r}, using v2 ({"|".join(READ_ENGINES)})') + engine = 'v2' + logger.info(f'OPC read engine: {engine}') + return engine - This method creates and configures an OPC Manager for a specific OPC UA server, - establishing the connection and preparing it for tag subscriptions. - - Args: - server_config (dict): A dictionary containing the OPC server configuration. - Expected keys include: - - 'name' (str): The name of the OPC server. - - 'url' (str): The URL of the OPC server. - - 'server_uri' (str): The URI of the OPC server. - - 'cert_path' (str, optional): Path to the client certificate file. - - 'private_key_path' (str, optional): Path to the private key file. - - 'server_cert_path' (str, optional): Path to the server certificate file. - - Returns: - OpcManager | None: An initialized OpcManager instance if successful, - otherwise None if an error occurs during initialization. - - Raises: - Exception: If OPC manager initialization fails, the error is logged and - a notification is sent, but the method returns None to allow - the system to continue operating with other servers. - """ - - try: - self.logger.info(f'Initializing OpcManager at {server_config["url"]}') - manager = OpcManager( + def _build_reader(self, server_config: dict) -> ServerReader | OpcManager: + if self.read_engine == 'legacy': + # Plan B: the rc3 OpcManager, unchanged (own connection, cert files). + return OpcManager( name=server_config['name'], url=server_config['url'], subscription_period_ms=server_config['subscription_period_ms'], @@ -152,6 +173,41 @@ class IngestorManager(SientiaMonitoring): server_cert_path=server_config.get('server_cert_path'), metrics_controller=self.metrics_controller, ) + return ServerReader( + name=server_config['name'], + server_config=server_config, + pool=self.pool, + sink=self.sink, + logger=self.logger, + notification_handler=self.notification_handler, + metrics_controller=self.metrics_controller, + metadata=self.metadata, + ) + + async def initialize_opc_from_config( + self, server_config: dict + ) -> ServerReader | OpcManager | None: + """ + Initializes the reader of one OPC UA server from its slot configuration. + + With READ_ENGINE=v2 (default) creates a ServerReader and opens its pooled + connection; with `use_security`, the certificate comes from MinIO. With + READ_ENGINE=legacy (plan B) creates the rc3 OpcManager, which connects on its + own with the certificate files of the slot v1 (`cert_path`). + + Args: + server_config (dict): The server configuration of the slot, without the + 'tags' key. v2: 'name', 'server_id', 'url', 'server_uri', + 'use_security', 'subscription_period_ms'. legacy: the slot v1 fields. + + Returns: + ServerReader | OpcManager | None: The connected reader, or None if the + connection fails (logged and notified, so the other servers keep working). + """ + + try: + self.logger.info(f'Initializing OPC reader at {server_config["url"]}') + manager = self._build_reader(server_config) manager.config = server_config await manager.connect() @@ -176,23 +232,33 @@ class IngestorManager(SientiaMonitoring): Gracefully shuts down the IngestorManager and all its components. This method ensures proper cleanup of: - - All OPC manager instances and their connections + - All OPC readers, their subscriptions and the pooled connections - Data manager connections and resources - - Active subscriptions and server connections The shutdown process is performed asynchronously to allow proper cleanup of all managed resources before termination. """ for _server_name, server in self.opc_managers.items(): await server.shutdown() + await self.pool.close_all() self.data_manager.shutdown() - async def remove_server(self, server: str): + async def remove_server(self, server: str, evict: bool = True): """ Removes an OPC server from the ingestor. + + Args: + server (str): The server name. + evict (bool): Also drop the pooled connection (lost server, failed + subscription). False when the server merely left this pod's slots: + the shared connection stays for writes/gRPC until the idle sweep. """ if server in self.opc_managers: - await self.opc_managers[server].shutdown() + if self.read_engine == 'legacy': + # OpcManager owns its connection: shutdown always disconnects it. + await self.opc_managers[server].shutdown() + else: + await self.opc_managers[server].shutdown(evict=evict) del self.opc_managers[server] for slot, _config in self.managed_tags.items(): self.managed_tags[slot].pop(server, None) @@ -258,9 +324,10 @@ class IngestorManager(SientiaMonitoring): for server in servers: if server not in registered_servers: self.logger.warning( - f'Server {server} not found in managed tags. Desconnecting from server.' + f'Server {server} not found in managed tags. Unsubscribing from server.' ) - await self.remove_server(server) + # Slot moved away: keep the shared connection for writes/gRPC. + await self.remove_server(server, evict=False) await self.emit_metric( metric_object=metrics.OPC_MANAGERS_ACTIVE, @@ -419,14 +486,18 @@ class IngestorManager(SientiaMonitoring): Notes: - If a slot is leased but its details cannot be retrieved - (i.e., `get_tag_slot` returns None), that slot is skipped. + (i.e., `get_tag_slot` returns None) or are malformed, that slot is skipped. + - Slot details are normalized to the slot v2 structure (parse_slot). """ acquired = {} for i in range(1, self.number_of_slots + 1): if await self.resource_manager.lease_tag(str(i)): self.logger.info(f'Leased slot {i}') - slots = await self.resource_manager.get_tag_slot(str(i)) + raw = await self.resource_manager.get_tag_slot(str(i)) + if raw is None: + continue + slots = self.parse_slot_config(str(i), raw) if slots is None: continue acquired[str(i)] = slots @@ -518,16 +589,75 @@ class IngestorManager(SientiaMonitoring): removed_slots: list[str] = [] for slot, _slot_config in self.managed_tags.items(): await self.resource_manager.renew_tag_lease(slot) - update = await self.resource_manager.get_tag_slot(slot) - if update is None: + raw = await self.resource_manager.get_tag_slot(slot) + if raw is None: removed_slots.append(slot) continue + update = self.parse_slot_config(slot, raw) + if update is None: + # Malformed config: keep reading with the last good one. + continue self.managed_tags[slot] = update for slot in removed_slots: self.managed_tags.pop(slot, None) + def parse_slot_config(self, slot: str, raw: dict) -> dict | None: + """ + Normalizes the raw slot read from Redis to the slot v2 structure. + + Accepts slot v1 (`topics`) and v2 (`names`), so the gateway and the + Orchestrator can be deployed in any order (SIENTIAPDE-1974). With + READ_ENGINE=legacy the raw slot v1 is kept as is (what OpcManager reads), and + a slot v2 is rejected. + + Returns: + dict | None: The parsed slot, or None when the config is malformed + (logged and notified). + """ + try: + if self.read_engine == 'legacy': + self._validate_slot_v1(raw) + return raw + parsed = parse_slot(raw) + except (AttributeError, KeyError, TypeError, ValueError) as e: + self.send_notification( + metadata=self.metadata, + notification_id=f'OPC_SLOT_CONFIG_ERROR_{slot}', + message=f'Slot {slot} has an invalid configuration: {e!r}', + block='ingestor_manager', + level=NotificationLevel.ERROR, + ) + self.logger.error(f'Slot {slot} has an invalid configuration: {e!r}') + return None + self._warn_server_cert_path(raw) + return parsed + + @staticmethod + def _validate_slot_v1(raw: dict) -> None: + """READ_ENGINE=legacy: OpcManager only understands slot v1 (`tag_name` + + `topics` per tag); a v2 slot (`names`) would fail in its datachange callback.""" + for name, server in raw.items(): + for address, tag in server['tags'].items(): + if 'topics' not in tag or 'tag_name' not in tag or 'frequency' not in tag: + raise ValueError( + f'READ_ENGINE=legacy needs a slot v1 (tag_name + topics); ' + f'server {name!r}, tag {address!r} is not' + ) + + def _warn_server_cert_path(self, raw: dict) -> None: + """Logs once per server when a slot still pins a server certificate + (`server_cert_path`): connections now come from the pool with the platform + certificate from MinIO, and that file-based pin is no longer applied.""" + for name, server in raw.items(): + if server.get('server_cert_path') and name not in self._server_cert_warned: + self._server_cert_warned.add(name) + self.logger.warning( + f'OPC server {name}: server_cert_path is set in the slot but is no ' + f'longer applied (server certificate validation is disabled)' + ) + async def drop_slot_leases(self, ids: list[str]) -> None: """ Releases the leases associated with the specified slot IDs. diff --git a/ingestor/managers/server_reader.py b/ingestor/managers/server_reader.py new file mode 100644 index 0000000..3335d2f --- /dev/null +++ b/ingestor/managers/server_reader.py @@ -0,0 +1,107 @@ +"""Per-server read path driven by the IngestorManager (SIENTIAPDE-1974). + +Replaces the legacy OpcManager behind the same interface the IngestorManager +already uses (connect / create_subscription / subscribe / unsubscribe / +check_cycles / check_opc_listenning / shutdown), so the slot lease loop is +unchanged. The difference is underneath: + +- the connection comes from the shared OpcConnectionPool, so the certificate is + fetched from MinIO by CertProvider (Basic256Sha256) — never from a file shipped + in the Gitea repository; +- readings go through SubscriptionEngine, which fans each value out to every + `names` entry of the slot v2 (one document per collection / tag name / model). +""" + +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler +from sientia_do.observability.logger import Logger +from sientia_do.observability.metrics_controller import MetricsController + +import ingestor.metrics as metrics +from ingestor.connection.pool import OpcConnectionPool +from ingestor.engines.subscription import SubscriptionEngine +from ingestor.io.data_sink import DataSink +from ingestor.io.slots import CONNECTION_FIELDS + + +class ServerReader: + """Reads one OPC server: one SubscriptionEngine (its `non_receive_count` is + engine-wide, hence one per server) with one subscription per leased slot.""" + + def __init__( + self, + name: str, + server_config: dict, + pool: OpcConnectionPool, + sink: DataSink, + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + metadata: dict | None = None, + ): + self.name = name + self.server_id = str(server_config['server_id']) + self.url = server_config['url'] + self.server_cfg = {k: server_config[k] for k in CONNECTION_FIELDS if k in server_config} + self.pool = pool + self.logger = logger + self.config: dict | None = None + self.engine = SubscriptionEngine( + pool, sink, logger, notification_handler, metrics_controller, metadata + ) + + def __str__(self): + return f'ServerReader(name={self.name}, server_id={self.server_id}, url={self.url})' + + @property + def subscriptions(self) -> dict: + return self.engine.subscriptions + + async def connect(self): + """Open (or reuse) the pooled connection; raises if the server is unreachable + or the certificate can't be fetched, like OpcManager.connect did.""" + await self.pool.get(self.server_id, self.server_cfg) + + async def create_subscription(self, slot: str): + await self.engine.create_subscription(self.server_id, self.server_cfg, slot) + + async def subscribe(self, slot: str, nodes: dict, collect_period: int): + await self.engine.subscribe(slot, nodes, collect_period) + + async def unsubscribe(self, slot: str): + await self.engine.unsubscribe(slot) + + async def check_cycles(self): + await self.engine.check_cycles() + + async def check_opc_listenning(self) -> bool: + return await self.engine.check_opc_listenning(self.server_id) + + async def shutdown(self, evict: bool = True): + """Drop every subscription of this server. + + With `evict` (lost server or changed server config) the pooled connection is + also dropped, so the next initialization reconnects. Without it (the slot just + moved to another pod) the connection stays for writes/gRPC: the connection is + shared, and the idle sweep closes it if nobody uses it anymore.""" + # list() snapshots the keys: unsubscribe() pops from engine.subscriptions, so + # iterating the dict directly would raise "changed size during iteration". + for slot in list(self.engine.subscriptions): # NOSONAR + await self.engine.unsubscribe(slot) + if evict: + await self.pool.evict(self.server_id) + await self.engine.emit_metric( + metric_object=metrics.OPC_CONNECTION_STATUS, + method='set', + value=0, + tags={ + 'pod_id': self.engine.pod_id, + 'server_name': self.server_id, + 'server_url': self.url, + }, + ) + await self.engine.emit_metric( + metric_object=metrics.OPC_TAGS_SUBSCRIBED, + method='set', + value=0, + tags={'pod_id': self.engine.pod_id, 'server_name': self.server_id}, + ) diff --git a/tests/unit/connection/test_certs.py b/tests/unit/connection/test_certs.py index 3b66c91..dcfa97c 100644 --- a/tests/unit/connection/test_certs.py +++ b/tests/unit/connection/test_certs.py @@ -26,3 +26,123 @@ async def test_fetch_returns_local_paths(tmp_path): ) assert (tmp_path / 'client_cert_c.pem').read_bytes() == b'cert-bytes' assert (tmp_path / 'private_key_k.pem').read_bytes() == b'key-bytes' + + +@pytest.mark.asyncio +async def test_lazy_minio_builds_client_once_on_first_download(): + from unittest.mock import MagicMock + + from ingestor.connection.certs import LazyMinio + + client = AsyncMock() + client.download_file.return_value = b'bytes' + factory = MagicMock(return_value=client) + minio = LazyMinio(factory) + + factory.assert_not_called() # no network at construction + assert await minio.download_file('a.pem', 'bucket', {}) == b'bytes' + await minio.download_file('b.pem', 'bucket', {}) + + factory.assert_called_once() + assert client.download_file.await_count == 2 + + +def test_build_cert_provider_is_lazy_and_uses_config(monkeypatch): + from unittest.mock import MagicMock + + from ingestor.config import GatewayConfig + from ingestor.connection import certs + + minio_cls = MagicMock() + monkeypatch.setattr(certs, 'MinioRepository', minio_cls) + mongo = MagicMock() + + provider = certs.build_cert_provider( + mongo=mongo, + config=GatewayConfig(cert_bucket='streamlit-connectors', cert_cache_dir='certs-cache'), + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=MagicMock(), + ) + + assert provider.mongo is mongo + assert provider.bucket == 'streamlit-connectors' + assert provider.cache_dir == 'certs-cache' + assert isinstance(provider.minio, certs.LazyMinio) + minio_cls.assert_not_called() # MinIO is only reached when a cert is needed + + +def _provider(tmp_path, doc=None, data=b'pem'): + from unittest.mock import MagicMock + + mongo = MagicMock() + mongo.find_one = AsyncMock( + return_value=doc + if doc is not None + else { + 'client_cert': {'minio_object_name': 'platform_certificates/c.pem'}, + 'private_key': {'minio_object_name': 'platform_certificates/k.pem'}, + } + ) + minio = MagicMock() + minio.download_file = AsyncMock(return_value=data) + return CertProvider(mongo=mongo, minio=minio, cache_dir=str(tmp_path), bucket='b'), minio + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'doc', + [ + None, + {}, + {'client_cert': {'minio_object_name': 'c.pem'}}, + {'client_cert': None, 'private_key': {}}, + ], +) +async def test_fetch_without_platform_certificate_raises_clear_error(tmp_path, doc): + from ingestor.connection.certs import CertificateNotConfiguredError + + provider, minio = _provider(tmp_path) + provider.mongo.find_one.return_value = doc + + with pytest.raises(CertificateNotConfiguredError, match='platform_certificates'): + await provider.fetch() + minio.download_file.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_fetch_is_cached_until_invalidated(tmp_path): + provider, minio = _provider(tmp_path) + + first = await provider.fetch() + second = await provider.fetch() + assert first == second + assert minio.download_file.await_count == 2 # cert + key, once + + provider.invalidate() + await provider.fetch() + assert minio.download_file.await_count == 4 + + +@pytest.mark.asyncio +async def test_concurrent_fetches_download_once(tmp_path): + import asyncio + + provider, minio = _provider(tmp_path) + + results = await asyncio.gather(*(provider.fetch() for _ in range(5))) + + assert len(set(results)) == 1 + assert minio.download_file.await_count == 2 + + +@pytest.mark.asyncio +async def test_download_replaces_file_atomically(tmp_path): + provider, minio = _provider(tmp_path, data=b'new') + (tmp_path / 'c.pem').write_bytes(b'old') + + cert_path, _ = await provider.fetch() + + assert (tmp_path / 'c.pem').read_bytes() == b'new' + assert cert_path == str(tmp_path / 'c.pem') + assert not list(tmp_path.glob('*.tmp')) diff --git a/tests/unit/connection/test_pool.py b/tests/unit/connection/test_pool.py index d477805..d23dc03 100644 --- a/tests/unit/connection/test_pool.py +++ b/tests/unit/connection/test_pool.py @@ -182,3 +182,163 @@ async def test_concurrent_get_same_server_connects_once(): assert c1 is c2 is clients[0] assert ctor.call_count == 1 assert clients[0].connect.await_count == 1 + + +@pytest.mark.asyncio +async def test_release_idle_keeps_pinned_connection(): + """A subscribed connection only gets callbacks (never get()), so it looks idle: + the pin keeps release_idle from dropping it (SIENTIAPDE-1974 G4).""" + pool = make_pool() + fake_client = AsyncMock() + with patch('ingestor.connection.pool.Client', return_value=fake_client): + await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False}) + + pool.pin('srv') + await pool.release_idle(max_idle_s=0) + + assert 'srv' in pool._connections + fake_client.disconnect.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_release_idle_evicts_after_last_unpin(): + pool = make_pool() + fake_client = AsyncMock() + with patch('ingestor.connection.pool.Client', return_value=fake_client): + await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False}) + + pool.pin('srv') + pool.pin('srv') # two slots subscribed on the same server + pool.unpin('srv') + await pool.release_idle(max_idle_s=0) + assert 'srv' in pool._connections # one subscription still active + + pool.unpin('srv') + await pool.release_idle(max_idle_s=0) + assert 'srv' not in pool._connections + fake_client.disconnect.assert_awaited_once() + + +def test_unpin_never_goes_negative(): + pool = make_pool() + + pool.unpin('srv') + pool.pin('srv') + + assert pool._pins == {'srv': 1} + + +@pytest.mark.asyncio +async def test_evict_disconnects_now_and_next_get_reconnects(): + pool = make_pool() + first, second = AsyncMock(), AsyncMock() + with patch('ingestor.connection.pool.Client', side_effect=[first, second]): + await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False}) + await pool.evict('srv') + client = await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False}) + + first.disconnect.assert_awaited_once() + assert client is second + + +@pytest.mark.asyncio +async def test_evict_unknown_server_is_noop(): + pool = make_pool() + + await pool.evict('nope') + + pool.emit_metric.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_get_normalizes_server_id_to_str(): + pool = make_pool() + fake_client = AsyncMock() + with patch('ingestor.connection.pool.Client', return_value=fake_client): + c1 = await pool.get(1, {'url': 'opc.tcp://h:4840', 'use_security': False}) + c2 = await pool.get('1', {'url': 'opc.tcp://h:4840', 'use_security': False}) + + assert c1 is c2 + assert fake_client.connect.await_count == 1 + + +@pytest.mark.asyncio +async def test_get_with_different_cfg_reuses_client_and_warns(): + pool = make_pool() + fake_client = AsyncMock() + with patch('ingestor.connection.pool.Client', return_value=fake_client): + await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False}) + client = await pool.get('srv', {'url': 'opc.tcp://10.0.0.5:4840', 'use_security': False}) + + assert client is fake_client + assert fake_client.connect.await_count == 1 + pool.logger.warning.assert_called_once() + assert 'url' in pool.logger.warning.call_args.args[0] + + +@pytest.mark.asyncio +async def test_secured_connect_failure_invalidates_cert_cache(): + from unittest.mock import MagicMock + + cert_provider = MagicMock() + cert_provider.fetch = AsyncMock(return_value=('cert.pem', 'key.pem')) + pool = make_pool(cert_provider=cert_provider) + fake_client = AsyncMock() + fake_client.connect.side_effect = RuntimeError('BadCertificateUntrusted') + with ( + patch('ingestor.connection.pool.Client', return_value=fake_client), + patch('ingestor.connection.pool.apply_security', AsyncMock()), + ): + with pytest.raises(RuntimeError): + await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': True}) + + cert_provider.invalidate.assert_called_once() + + +@pytest.mark.asyncio +async def test_insecure_connect_failure_keeps_cert_cache(): + from unittest.mock import MagicMock + + cert_provider = MagicMock() + pool = make_pool(cert_provider=cert_provider) + fake_client = AsyncMock() + fake_client.connect.side_effect = RuntimeError('unreachable') + with patch('ingestor.connection.pool.Client', return_value=fake_client): + with pytest.raises(RuntimeError): + await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False}) + + cert_provider.invalidate.assert_not_called() + + +@pytest.mark.asyncio +async def test_connect_identifies_the_pod_on_the_server(): + """Session/application name = pod id, as the legacy OpcManager did, so each pod's + session is recognizable in the KEPServer console.""" + pool = make_pool() + fake_client = MagicMock(connect=AsyncMock()) + with patch('ingestor.connection.pool.Client', return_value=fake_client): + await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False}) + + assert fake_client.name == pool.pod_id + assert fake_client.description == pool.pod_id + assert fake_client.product_uri == pool.pod_id.replace('-', ':') + assert fake_client.application_uri == pool.pod_id.replace('-', ':') + + +@pytest.mark.asyncio +async def test_connect_server_uri_still_wins_for_application_uri(): + pool = make_pool() + fake_client = MagicMock(connect=AsyncMock()) + with patch('ingestor.connection.pool.Client', return_value=fake_client): + await pool.get( + 'srv', + { + 'url': 'opc.tcp://h:4840', + 'use_security': False, + 'server_uri': 'urn:sientia:opc-ingestor', + }, + ) + + # must match the certificate SAN URI (BadCertificateUriInvalid otherwise) + assert fake_client.application_uri == 'urn:sientia:opc-ingestor' + assert fake_client.name == pool.pod_id diff --git a/tests/unit/engines/test_subscription.py b/tests/unit/engines/test_subscription.py index 5d8cdc8..6ab7cb9 100644 --- a/tests/unit/engines/test_subscription.py +++ b/tests/unit/engines/test_subscription.py @@ -201,3 +201,100 @@ async def test_check_opc_listenning_triggers_reconnect_at_15(): assert result is True assert eng.send_notification_async.call_count == 2 + + +@pytest.mark.asyncio +async def test_create_subscription_pins_pool_connection(): + eng = make_engine() + eng.pool = MagicMock(get=AsyncMock(return_value=AsyncMock())) + + await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1') + + eng.pool.pin.assert_called_once_with('1') + + +@pytest.mark.asyncio +async def test_create_subscription_failure_does_not_pin(): + eng = make_engine() + client = AsyncMock() + client.create_subscription.side_effect = RuntimeError('boom') + eng.pool = MagicMock(get=AsyncMock(return_value=client)) + + with pytest.raises(RuntimeError): + await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1') + + eng.pool.pin.assert_not_called() + + +@pytest.mark.asyncio +async def test_unsubscribe_unpins_pool_connection(): + eng = make_engine() + eng.pool = MagicMock(get=AsyncMock(return_value=AsyncMock())) + await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1') + + await eng.unsubscribe('slot1') + + eng.pool.unpin.assert_called_once_with('1') + assert 'slot1' not in eng.subscriptions + + +@pytest.mark.asyncio +async def test_unsubscribe_tolerates_dead_connection(): + """A lost server's subscription can't be deleted on the server side; it must + still be dropped locally (and unpinned) so the slot can be re-subscribed.""" + eng = make_engine() + sub = AsyncMock() + sub.delete.side_effect = ConnectionError('closed') + client = AsyncMock() + client.create_subscription.return_value = sub + eng.pool = MagicMock(get=AsyncMock(return_value=client)) + await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1') + + await eng.unsubscribe('slot1') # must not raise + + assert 'slot1' not in eng.subscriptions + eng.pool.unpin.assert_called_once_with('1') + eng.logger.warning.assert_called_once() + + +async def _subscribed_engine(slots): + """Engine with one subscription per slot: {slot: [addresses]}.""" + eng = make_engine() + eng.pool = MagicMock( + get=AsyncMock( + return_value=MagicMock(create_subscription=AsyncMock(return_value=AsyncMock())) + ) + ) + for slot, addresses in slots.items(): + await eng.create_subscription('1', {'subscription_period_ms': 500}, slot) + nodes = { + a: {'frequency': '1000', 'names': [{'collection': 'c', 'tag_name': a}]} + for a in addresses + } + await eng.subscribe(slot, nodes, 10) + return eng + + +@pytest.mark.asyncio +async def test_unsubscribe_forgets_the_slot_nodes(): + eng = await _subscribed_engine({'s1': ['ns=2;i=1', 'ns=2;i=2'], 's2': ['ns=2;i=3']}) + + await eng.unsubscribe('s1') + + # no more cycle checks / tag counting for addresses nobody reads anymore + assert set(eng.nodes) == {'ns=2;i=3'} + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_TAGS_SUBSCRIBED, + method='set', + value=1, + tags={'pod_id': eng.pod_id, 'server_name': '1'}, + ) + + +@pytest.mark.asyncio +async def test_unsubscribe_keeps_node_shared_with_another_slot(): + eng = await _subscribed_engine({'s1': ['ns=2;i=1', 'ns=2;i=2'], 's2': ['ns=2;i=2']}) + + await eng.unsubscribe('s1') + + assert set(eng.nodes) == {'ns=2;i=2'} diff --git a/tests/unit/io/test_data_sink_publish.py b/tests/unit/io/test_data_sink_publish.py index a4edfee..eb05517 100644 --- a/tests/unit/io/test_data_sink_publish.py +++ b/tests/unit/io/test_data_sink_publish.py @@ -67,3 +67,26 @@ async def test_publish_reading_emits_tag_value_metric_per_name(): value=42, tags={'pod_id': dm.pod_id, 'tag_name': 'Counter', 'model_id': '1'}, ) + + +@pytest.mark.asyncio +async def test_publish_reading_without_model_id_keeps_legacy_doc(): + """Names converted from a slot v1 carry no model_id: the document must be the + same as the legacy OpcManager one (SIENTIAPDE-1974 NF2).""" + sink, dm = make_sink() + names = [{'collection': 'raw_x', 'tag_name': 'Counter'}] + + await sink.publish_reading( + tag='ns=2;i=2', value=42, timestamp='2026-07-13 00:00:00+0000', names=names + ) + + collection, doc = dm.publish.await_args_list[0].args + assert collection == 'raw_x' + assert doc == { + 'tag': 'ns=2;i=2', + 'name': 'Counter', + 'timestamp': '2026-07-13 00:00:00+0000', + 'value': 42, + } + _, kwargs = sink.emit_metric.await_args + assert kwargs['tags']['model_id'] == '' diff --git a/tests/unit/io/test_slots.py b/tests/unit/io/test_slots.py index f168b50..c4234b3 100644 --- a/tests/unit/io/test_slots.py +++ b/tests/unit/io/test_slots.py @@ -62,3 +62,102 @@ def test_parse_slot_rejects_tag_without_names(): with pytest.raises((KeyError, ValueError)): parse_slot(raw) + + +RAW_V1 = { + 'kepserver': { + 'name': 'kepserver', + 'server_id': '1', + 'url': 'opc.tcp://kepserver:49320', + 'server_uri': 'urn:sientia:opc-ingestor', + 'cert_path': '/app/certs/opc-ingestor-cert.pem', + 'private_key_path': '/app/certs/opc-ingestor-key.pem', + 'server_cert_path': None, + 'subscription_period_ms': 15000.0, + 'tags': { + 'ns=2;s=A.TargetSin': { + 'aggr_func': 'lts', + 'frequency': 30000, + 'server_id': '1', + 'tag_address': 'ns=2;s=A.TargetSin', + 'tag_name': 'TargetSin', + 'topics': ['raw_scouter-sin-approx', 'raw_other-model'], + } + }, + } +} + + +def test_parse_slot_converts_v1_topics_to_names(): + """Slot v1 (current Orchestrator) is read as v2: one name per topic, no model_id, + so the documents stay identical to the legacy ones (SIENTIAPDE-1974 G1).""" + server = parse_slot(RAW_V1)['kepserver'] + tag = server['tags']['ns=2;s=A.TargetSin'] + + assert tag['names'] == [ + {'collection': 'raw_scouter-sin-approx', 'tag_name': 'TargetSin'}, + {'collection': 'raw_other-model', 'tag_name': 'TargetSin'}, + ] + assert tag['frequency'] == 30000 + assert tag['active_read'] == {'enabled': False} + + +def test_parse_slot_v1_cert_path_means_security_but_path_is_dropped(): + server = parse_slot(RAW_V1)['kepserver'] + + assert server['use_security'] is True + assert server['name'] == 'kepserver' + assert server['server_id'] == '1' + assert server['subscription_period_ms'] == 15000.0 + # the certificate always comes from MinIO now: no file path survives the parse + assert 'cert_path' not in server + assert 'private_key_path' not in server + + +def test_parse_slot_v1_without_cert_path_is_insecure(): + raw = { + 's': { + 'server_id': '2', + 'url': 'u', + 'cert_path': None, + 'tags': {'ns=2;i=1': {'tag_name': 't', 'frequency': '1000', 'topics': ['raw_c']}}, + } + } + + server = parse_slot(raw)['s'] + + assert server['use_security'] is False + assert server['name'] == 's' # falls back to the slot key + + +def test_parse_slot_v2_explicit_use_security_wins(): + raw = { + 's': { + 'server_id': '1', + 'url': 'u', + 'use_security': False, + 'cert_path': '/legacy/path', + 'tags': { + 'ns=2;i=1': { + 'frequency': '1000', + 'names': [{'collection': 'c', 'tag_name': 't', 'model_id': '1'}], + } + }, + } + } + + assert parse_slot(raw)['s']['use_security'] is False + + +def test_parse_slot_rejects_v1_tag_without_tag_name(): + raw = { + 's': { + 'server_id': '1', + 'url': 'u', + 'tags': {'ns=2;i=1': {'frequency': '1', 'topics': ['c']}}, + } + } + import pytest + + with pytest.raises(ValueError): + parse_slot(raw) diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 6f52c03..df27f66 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -4,6 +4,9 @@ from unittest.mock import AsyncMock, MagicMock, call, patch from pytest import fixture, mark from sientia_do.notifications.models import NotificationLevel +from ingestor.config import GatewayConfig +from ingestor.connection.pool import OpcConnectionPool +from ingestor.io.slots import parse_slot from ingestor.managers.ingestor_manager import IngestorManager metadata = { @@ -16,6 +19,21 @@ metadata = { } +def raw_slot_v1(tag_name='Counter', address='ns=2;i=2'): + """A slot as the current (v1) Orchestrator writes it to Redis.""" + return { + 'server1': { + 'name': 'server1', + 'server_id': '1', + 'url': 'opc.tcp://localhost:4840', + 'server_uri': 'urn:sientia:opc-ingestor', + 'cert_path': '/app/certs/opc-ingestor-cert.pem', + 'subscription_period_ms': 500.0, + 'tags': {address: {'tag_name': tag_name, 'frequency': '1000', 'topics': ['raw_a']}}, + } + } + + @fixture(autouse=True) def _collect_garbage_after_test(): # ponytail: IngestorManager.__del__ calls asyncio.run(self.shutdown()); if a @@ -52,12 +70,12 @@ def ingestor_manager(data_manager_mock, resource_manager_mock): return ingestor -@patch('ingestor.managers.ingestor_manager.OpcManager') +@patch('ingestor.managers.ingestor_manager.ServerReader') @patch('ingestor.managers.ingestor_manager.DataManager') @patch('ingestor.managers.ingestor_manager.ResourceManager') @patch('ingestor.managers.ingestor_manager.NotificationHandler') def test___init__( - notification_handler_mock, resource_manager_mock, data_manager_mock, opc_manager_mock + notification_handler_mock, resource_manager_mock, data_manager_mock, server_reader_mock ): ingestor = IngestorManager( redis_data={'host': 'localhost', 'port': 6379}, @@ -72,7 +90,7 @@ def test___init__( metrics_controller=MagicMock(), ) - opc_manager_mock.assert_not_called() + server_reader_mock.assert_not_called() data_manager_mock.assert_called_once_with( mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', @@ -100,59 +118,86 @@ def test___init__( assert ingestor.data_manager == data_manager_mock.return_value assert ingestor.resource_manager == resource_manager_mock.return_value + # Read path (SIENTIAPDE-1974): pool + read sink built here, nothing connected yet. + assert isinstance(ingestor.pool, OpcConnectionPool) + assert ingestor.pool.cert_provider is ingestor.cert_provider + assert ingestor.cert_provider.mongo is data_manager_mock.return_value.mongo_repository + assert ingestor.sink.data_manager is data_manager_mock.return_value + assert ingestor.sink.postgres is None -@mark.asyncio -@patch('ingestor.managers.ingestor_manager.OpcManager') -async def test_initialize_opc_from_config(opc_manager, ingestor_manager): - server_config = { - 'name': 'server1', - 'url': 'opc.tcp://localhost:4840', - 'subscription_period_ms': 1000, - 'server_uri': 'http://opcua-server.simulator', - 'cert_path': '/path/to/cert', - 'private_key_path': '/path/to/private_key', - 'server_cert_path': '/path/to/server_cert', - 'pod_id': 'test_pod', - } - opc_manager.return_value = MagicMock(connect=AsyncMock()) - result = await ingestor_manager.initialize_opc_from_config(server_config) - - opc_manager.assert_called_once_with( - name=server_config['name'], - url=server_config['url'], - data_manager=ingestor_manager.data_manager, - logger=ingestor_manager.logger, - subscription_period_ms=server_config['subscription_period_ms'], - server_uri=server_config['server_uri'], - notification_handler=ingestor_manager.notification_handler, - cert_path=server_config['cert_path'], - private_key_path=server_config['private_key_path'], - server_cert_path=server_config['server_cert_path'], +@patch('ingestor.managers.ingestor_manager.DataManager') +@patch('ingestor.managers.ingestor_manager.ResourceManager') +def test___init___cert_provider_uses_gateway_config(resource_manager_mock, data_manager_mock): + ingestor = IngestorManager( + redis_data={'host': 'localhost', 'port': 6379}, + lease_ttl=60, + heartbeat_ttl=60, + poll_interval=5, + mongo_connection_string='mongodb://localhost:27017', + mongo_database='sientia', + logger=MagicMock(), + notification_handler=MagicMock(), metadata=metadata['metadata'], - metrics_controller=ingestor_manager.metrics_controller, + metrics_controller=MagicMock(), + gateway_config=GatewayConfig(cert_bucket='my-bucket', cert_cache_dir='certs-cache'), ) - assert result == opc_manager.return_value - result.connect.assert_called_once() + assert ingestor.cert_provider.bucket == 'my-bucket' + assert ingestor.cert_provider.cache_dir == 'certs-cache' @mark.asyncio -@patch('ingestor.managers.ingestor_manager.OpcManager') -@patch('ingestor.managers.ingestor_manager.traceback') -async def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, ingestor_manager): +@patch('ingestor.managers.ingestor_manager.ServerReader') +async def test_initialize_opc_from_config(server_reader, ingestor_manager): server_config = { 'name': 'server1', + 'server_id': '1', 'url': 'opc.tcp://localhost:4840', + 'server_uri': 'urn:sientia:opc-ingestor', + 'use_security': True, + 'subscription_period_ms': 1000, + } + + server_reader.return_value = MagicMock(connect=AsyncMock()) + result = await ingestor_manager.initialize_opc_from_config(server_config) + + server_reader.assert_called_once_with( + name=server_config['name'], + server_config=server_config, + pool=ingestor_manager.pool, + sink=ingestor_manager.sink, + logger=ingestor_manager.logger, + notification_handler=ingestor_manager.notification_handler, + metrics_controller=ingestor_manager.metrics_controller, + metadata=metadata['metadata'], + ) + + assert result == server_reader.return_value + assert result.config == server_config + result.connect.assert_awaited_once() + + +@mark.asyncio +@patch('ingestor.managers.ingestor_manager.ServerReader') +@patch('ingestor.managers.ingestor_manager.traceback') +async def test_initialize_opc_from_config_exception( + traceback_mock, server_reader, ingestor_manager +): + server_config = { + 'name': 'server1', + 'server_id': '1', + 'url': 'opc.tcp://localhost:4840', + 'server_uri': 'urn:sientia:opc-ingestor', + 'use_security': True, 'subscription_period_ms': 1000, - 'server_uri': 'http://opcua-server.simulator', - 'cert_path': '/path/to/cert', - 'private_key_path': '/path/to/private_key', - 'server_cert_path': '/path/to/server_cert', } ingestor_manager.logger.error = MagicMock() - opc_manager.side_effect = Exception('Initialization error') + # e.g. the server is unreachable or the certificate can't be fetched from MinIO + server_reader.return_value = MagicMock( + connect=AsyncMock(side_effect=Exception('Initialization error')) + ) result = await ingestor_manager.initialize_opc_from_config(server_config) @@ -174,11 +219,14 @@ async def test_shutdown(ingestor_manager): ingestor_manager.opc_managers = {'server1': AsyncMock(), 'server2': AsyncMock()} ingestor_manager.data_manager.shutdown = MagicMock() + ingestor_manager.pool = MagicMock(close_all=AsyncMock()) await ingestor_manager.shutdown() ingestor_manager.opc_managers['server1'].shutdown.assert_called_once() ingestor_manager.opc_managers['server2'].shutdown.assert_called_once() + # The pool is owned by the read path: closed after the readers unsubscribe. + ingestor_manager.pool.close_all.assert_awaited_once() ingestor_manager.data_manager.shutdown.assert_called_once() @@ -218,9 +266,8 @@ async def test_remove_server_not_found(ingestor_manager): @mark.asyncio -@patch('ingestor.managers.ingestor_manager.OpcManager') @patch('ingestor.managers.ingestor_manager.metrics') -async def test_update_opc_servers(metrics, opc_manager, ingestor_manager): +async def test_update_opc_servers(metrics, ingestor_manager): manager1 = MagicMock(config={'config': 'config1'}) manager2 = MagicMock(config={'config': 'config2'}) manager3 = MagicMock(config={'config': 'config3'}) @@ -267,7 +314,9 @@ async def test_update_opc_servers(metrics, opc_manager, ingestor_manager): assert ingestor_manager.opc_managers['server2'].config == {'config': 'config2'} assert ingestor_manager.opc_managers['server3'].config == {'config': 'config3'} - ingestor_manager.remove_server.assert_any_call('server4') + # server4 left this pod's slots: unsubscribe only, the connection stays for writes + ingestor_manager.remove_server.assert_any_call('server4', evict=False) + # server5 failed to initialize: removed with the default (evict) ingestor_manager.remove_server.assert_any_call('server5') assert ingestor_manager.remove_server.call_count == 2 @@ -431,25 +480,47 @@ async def test_get_number_of_slots_empty(metrics, ingestor_manager): @mark.asyncio async def test_get_slot_leases_1_success(ingestor_manager): ingestor_manager.resource_manager.lease_tag = AsyncMock(return_value=True) - ingestor_manager.resource_manager.get_tag_slot = AsyncMock(return_value={'tags': ['tag1']}) + ingestor_manager.resource_manager.get_tag_slot = AsyncMock(return_value=raw_slot_v1()) ingestor_manager.number_of_slots = 1 result = await ingestor_manager.get_slot_leases() - assert result == {'1': {'tags': ['tag1']}} + # Stored already normalized to slot v2 (v1 converted by parse_slot). + assert result == {'1': parse_slot(raw_slot_v1())} + assert ingestor_manager.managed_tags == result @mark.asyncio async def test_get_slot_leases_2_success(ingestor_manager): ingestor_manager.resource_manager.lease_tag = AsyncMock(side_effect=[True, True]) ingestor_manager.resource_manager.get_tag_slot = AsyncMock( - side_effect=[{'tags': ['tag1']}, {'tags': ['tag2']}] + side_effect=[raw_slot_v1('Counter'), raw_slot_v1('Ramp', 'ns=2;i=3')] ) ingestor_manager.number_of_slots = 2 result = await ingestor_manager.get_slot_leases(max_slots=2) - assert result == {'1': {'tags': ['tag1']}, '2': {'tags': ['tag2']}} + assert result == { + '1': parse_slot(raw_slot_v1('Counter')), + '2': parse_slot(raw_slot_v1('Ramp', 'ns=2;i=3')), + } + + +@mark.asyncio +async def test_get_slot_leases_skips_malformed_slot(ingestor_manager): + ingestor_manager.resource_manager.lease_tag = AsyncMock(return_value=True) + # tag without names/topics/frequency + ingestor_manager.resource_manager.get_tag_slot = AsyncMock( + return_value={'server1': {'server_id': '1', 'url': 'u', 'tags': {'ns=2;i=2': {}}}} + ) + + ingestor_manager.number_of_slots = 1 + result = await ingestor_manager.get_slot_leases() + + assert result == {} + _, kwargs = ingestor_manager.send_notification.call_args + assert kwargs['notification_id'] == 'OPC_SLOT_CONFIG_ERROR_1' + assert kwargs['level'] == NotificationLevel.ERROR @mark.asyncio @@ -497,20 +568,20 @@ async def test_unsubscribe_slot(ingestor_manager): @mark.asyncio async def test_update_slot_config(ingestor_manager): ingestor_manager.managed_tags = { - 'slot1': {'config': 'old_config'}, - 'slot2': {'config': 'new_config'}, - 'slot3': {'config': 'old_config'}, + 'slot1': parse_slot(raw_slot_v1('Old')), + 'slot2': parse_slot(raw_slot_v1('Same')), + 'slot3': parse_slot(raw_slot_v1('Old')), } ingestor_manager.resource_manager.get_tag_slot = AsyncMock( - side_effect=[{'config': 'updated_config'}, {'config': 'new_config'}, None] + side_effect=[raw_slot_v1('Updated'), raw_slot_v1('Same'), None] ) ingestor_manager.resource_manager.renew_tag_lease = AsyncMock() await ingestor_manager.update_slot_config() - assert ingestor_manager.managed_tags['slot1'] == {'config': 'updated_config'} - assert ingestor_manager.managed_tags['slot2'] == {'config': 'new_config'} + assert ingestor_manager.managed_tags['slot1'] == parse_slot(raw_slot_v1('Updated')) + assert ingestor_manager.managed_tags['slot2'] == parse_slot(raw_slot_v1('Same')) assert 'slot3' not in ingestor_manager.managed_tags ingestor_manager.resource_manager.renew_tag_lease.assert_any_call('slot1') @@ -652,3 +723,202 @@ async def test_subscribe_to_tags(ingestor_manager): assert ingestor_manager.manage_server.call_count == 3 ingestor_manager.managed_tags['slot1'].pop.assert_called_once_with('server3', None) + + +@mark.asyncio +async def test_update_slot_config_keeps_last_good_config_when_malformed(ingestor_manager): + good = parse_slot(raw_slot_v1('Counter')) + ingestor_manager.managed_tags = {'slot1': good} + ingestor_manager.resource_manager.get_tag_slot = AsyncMock( + return_value={'server1': {'server_id': '1', 'url': 'u', 'tags': {'ns=2;i=2': {}}}} + ) + ingestor_manager.resource_manager.renew_tag_lease = AsyncMock() + + await ingestor_manager.update_slot_config() + + # Still reading with the last good config; the bad one is notified, not applied. + assert ingestor_manager.managed_tags == {'slot1': good} + _, kwargs = ingestor_manager.send_notification.call_args + assert kwargs['notification_id'] == 'OPC_SLOT_CONFIG_ERROR_slot1' + + +@mark.asyncio +async def test_remove_server_forwards_evict(ingestor_manager): + reader = AsyncMock() + ingestor_manager.opc_managers = {'server1': reader} + ingestor_manager.managed_tags = {'slot1': {'server1': {}}} + + await ingestor_manager.remove_server('server1', evict=False) + + reader.shutdown.assert_awaited_once_with(evict=False) + assert 'server1' not in ingestor_manager.opc_managers + assert ingestor_manager.managed_tags == {'slot1': {}} + + +@patch('ingestor.managers.ingestor_manager.DataManager') +@patch('ingestor.managers.ingestor_manager.ResourceManager') +def test___init___ignores_malformed_write_env( + resource_manager_mock, data_manager_mock, monkeypatch +): + """A malformed write/gRPC-only env var must not stop the read path (review #1).""" + monkeypatch.setenv('GRPC_PORT', '') + monkeypatch.setenv('WRITE_MIN_PAUSE_S', '0,5') + monkeypatch.setenv('CERT_BUCKET', 'streamlit-connectors') + + ingestor = IngestorManager( + redis_data={'host': 'localhost', 'port': 6379}, + lease_ttl=60, + heartbeat_ttl=60, + poll_interval=5, + mongo_connection_string='mongodb://localhost:27017', + mongo_database='sientia', + logger=MagicMock(), + notification_handler=MagicMock(), + metadata=metadata['metadata'], + metrics_controller=MagicMock(), + ) + + assert ingestor.cert_provider.bucket == 'streamlit-connectors' + + +def test_parse_slot_config_warns_once_about_server_cert_path(ingestor_manager): + raw = raw_slot_v1() + raw['server1']['server_cert_path'] = '/app/certs/kepserver.der' + + ingestor_manager.parse_slot_config('1', raw) + ingestor_manager.parse_slot_config('1', raw) # every loop refreshes the slot + + warnings = [ + c for c in ingestor_manager.logger.warning.call_args_list if 'server_cert_path' in str(c) + ] + assert len(warnings) == 1 + + +# --- READ_ENGINE=legacy: plan B with the untouched rc3 OpcManager ------------------- + + +def make_manager(read_engine=None): + with ( + patch('ingestor.managers.ingestor_manager.DataManager'), + patch('ingestor.managers.ingestor_manager.ResourceManager'), + ): + manager = IngestorManager( + redis_data={'host': 'localhost', 'port': 6379}, + lease_ttl=60, + heartbeat_ttl=60, + poll_interval=5, + mongo_connection_string='mongodb://localhost:27017', + mongo_database='sientia', + logger=MagicMock(), + notification_handler=MagicMock(), + metadata=metadata['metadata'], + metrics_controller=MagicMock(), + read_engine=read_engine, + ) + manager.send_notification = MagicMock() + manager.emit_metric = AsyncMock() + return manager + + +def test_read_engine_defaults_to_v2(monkeypatch): + monkeypatch.delenv('READ_ENGINE', raising=False) + + assert make_manager().read_engine == 'v2' + + +def test_read_engine_from_env(monkeypatch): + monkeypatch.setenv('READ_ENGINE', 'Legacy ') + + assert make_manager().read_engine == 'legacy' + + +def test_invalid_read_engine_falls_back_to_v2(monkeypatch): + monkeypatch.setenv('READ_ENGINE', 'v3') + + manager = make_manager() + + assert manager.read_engine == 'v2' + assert any('Invalid READ_ENGINE' in str(c) for c in manager.logger.warning.call_args_list) + + +@mark.asyncio +@patch('ingestor.managers.ingestor_manager.ServerReader') +@patch('ingestor.managers.ingestor_manager.OpcManager') +async def test_legacy_initialize_builds_rc3_opc_manager(opc_manager, server_reader): + manager = make_manager('legacy') + server_config = { + 'name': 'server1', + 'server_id': '1', + 'url': 'opc.tcp://localhost:4840', + 'subscription_period_ms': 1000, + 'server_uri': 'urn:sientia:opc-ingestor', + 'cert_path': '/app/certs/opc-ingestor-cert.pem', + 'private_key_path': '/app/certs/opc-ingestor-key.pem', + 'server_cert_path': None, + } + opc_manager.return_value = MagicMock(connect=AsyncMock()) + + result = await manager.initialize_opc_from_config(server_config) + + server_reader.assert_not_called() + opc_manager.assert_called_once_with( + name='server1', + url='opc.tcp://localhost:4840', + subscription_period_ms=1000, + data_manager=manager.data_manager, + logger=manager.logger, + server_uri='urn:sientia:opc-ingestor', + notification_handler=manager.notification_handler, + metadata=metadata['metadata'], + cert_path='/app/certs/opc-ingestor-cert.pem', + private_key_path='/app/certs/opc-ingestor-key.pem', + server_cert_path=None, + metrics_controller=manager.metrics_controller, + ) + assert result is opc_manager.return_value + result.connect.assert_awaited_once() + + +def test_legacy_keeps_raw_slot_v1(): + manager = make_manager('legacy') + raw = raw_slot_v1() + + # OpcManager reads tag_name/topics and the file cert paths straight from the slot + assert manager.parse_slot_config('1', raw) is raw + + +def test_legacy_rejects_slot_v2(): + manager = make_manager('legacy') + raw_v2 = { + 'server1': { + 'name': 'server1', + 'server_id': '1', + 'url': 'opc.tcp://localhost:4840', + 'use_security': True, + 'subscription_period_ms': 500, + 'tags': { + 'ns=2;i=2': { + 'frequency': '1000', + 'names': [{'collection': 'raw_a', 'tag_name': 'Counter', 'model_id': '1'}], + } + }, + } + } + + assert manager.parse_slot_config('1', raw_v2) is None + _, kwargs = manager.send_notification.call_args + assert kwargs['notification_id'] == 'OPC_SLOT_CONFIG_ERROR_1' + assert 'READ_ENGINE=legacy' in kwargs['message'] + + +@mark.asyncio +async def test_legacy_remove_server_calls_plain_shutdown(): + manager = make_manager('legacy') + opc_manager = MagicMock(shutdown=AsyncMock()) + manager.opc_managers = {'server1': opc_manager} + manager.managed_tags = {'slot1': {'server1': {}}} + + await manager.remove_server('server1', evict=False) + + opc_manager.shutdown.assert_awaited_once_with() + assert manager.opc_managers == {} diff --git a/tests/unit/managers/test_server_reader.py b/tests/unit/managers/test_server_reader.py new file mode 100644 index 0000000..b34876e --- /dev/null +++ b/tests/unit/managers/test_server_reader.py @@ -0,0 +1,164 @@ +from unittest.mock import AsyncMock, MagicMock, call + +import pytest + +import ingestor.metrics as metrics +from ingestor.managers.server_reader import ServerReader + +SERVER_CONFIG = { + 'name': 'kepserver', + 'server_id': 1, # the slot may carry it as int; the pool keys by str + 'url': 'opc.tcp://kepserver:49320', + 'server_uri': 'urn:sientia:opc-ingestor', + 'use_security': True, + 'subscription_period_ms': 15000.0, +} + + +def make_reader(): + pool = MagicMock(get=AsyncMock(), evict=AsyncMock()) + reader = ServerReader( + name='kepserver', + server_config=SERVER_CONFIG, + pool=pool, + sink=AsyncMock(), + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=AsyncMock(), + metadata={}, + ) + reader.engine = MagicMock( + subscriptions={}, + pod_id='pod', + create_subscription=AsyncMock(), + subscribe=AsyncMock(), + unsubscribe=AsyncMock(), + check_cycles=AsyncMock(), + check_opc_listenning=AsyncMock(return_value=False), + emit_metric=AsyncMock(), + ) + return reader, pool + + +def test_server_cfg_keeps_only_connection_fields(): + reader, _ = make_reader() + + assert reader.server_id == '1' + assert reader.server_cfg == { + 'url': 'opc.tcp://kepserver:49320', + 'server_uri': 'urn:sientia:opc-ingestor', + 'use_security': True, + 'subscription_period_ms': 15000.0, + } + + +@pytest.mark.asyncio +async def test_connect_opens_pooled_connection(): + reader, pool = make_reader() + + await reader.connect() + + # use_security=True: the pool fetches the certificate from MinIO (CertProvider) + pool.get.assert_awaited_once_with('1', reader.server_cfg) + + +@pytest.mark.asyncio +async def test_connect_propagates_failure(): + reader, pool = make_reader() + pool.get.side_effect = ConnectionError('unreachable') + + with pytest.raises(ConnectionError): + await reader.connect() + + +@pytest.mark.asyncio +async def test_subscription_calls_delegate_to_engine_per_slot(): + reader, _ = make_reader() + nodes = {'ns=2;i=2': {'frequency': '1000', 'names': [{'collection': 'raw_a', 'tag_name': 'a'}]}} + + await reader.create_subscription('1') + await reader.subscribe('1', nodes, 10) + await reader.unsubscribe('1') + + reader.engine.create_subscription.assert_awaited_once_with('1', reader.server_cfg, '1') + reader.engine.subscribe.assert_awaited_once_with('1', nodes, 10) + reader.engine.unsubscribe.assert_awaited_once_with('1') + + +@pytest.mark.asyncio +async def test_supervision_delegates_to_engine(): + reader, _ = make_reader() + reader.engine.check_opc_listenning.return_value = True + + await reader.check_cycles() + lost = await reader.check_opc_listenning() + + reader.engine.check_cycles.assert_awaited_once() + reader.engine.check_opc_listenning.assert_awaited_once_with('1') + assert lost is True + + +def test_subscriptions_are_the_engine_ones(): + reader, _ = make_reader() + reader.engine.subscriptions['1'] = MagicMock() + + assert reader.subscriptions is reader.engine.subscriptions + + +@pytest.mark.asyncio +async def test_shutdown_unsubscribes_all_slots_evicts_and_zeroes_metrics(): + reader, pool = make_reader() + reader.engine.subscriptions.update({'1': MagicMock(), '2': MagicMock()}) + + await reader.shutdown() + + reader.engine.unsubscribe.assert_has_awaits([call('1'), call('2')]) + # Dropping the pooled connection makes the next initialization reconnect. + pool.evict.assert_awaited_once_with('1') + reader.engine.emit_metric.assert_any_await( + metric_object=metrics.OPC_CONNECTION_STATUS, + method='set', + value=0, + tags={'pod_id': 'pod', 'server_name': '1', 'server_url': 'opc.tcp://kepserver:49320'}, + ) + reader.engine.emit_metric.assert_any_await( + metric_object=metrics.OPC_TAGS_SUBSCRIBED, + method='set', + value=0, + tags={'pod_id': 'pod', 'server_name': '1'}, + ) + + +@pytest.mark.asyncio +async def test_shutdown_without_evict_keeps_the_shared_connection(): + """Slot moved to another pod: drop the subscriptions, keep the pooled + connection for NATS writes/gRPC (review #7).""" + reader, pool = make_reader() + reader.engine.subscriptions.update({'1': MagicMock()}) + + await reader.shutdown(evict=False) + + reader.engine.unsubscribe.assert_awaited_once_with('1') + pool.evict.assert_not_awaited() + for c in reader.engine.emit_metric.await_args_list: + assert c.kwargs['metric_object'] is not metrics.OPC_CONNECTION_STATUS + + +@pytest.mark.asyncio +async def test_shutdown_unsubscribes_every_slot_while_engine_pops_them(): + """The real unsubscribe() pops from engine.subscriptions: shutdown must iterate a + snapshot (list()), or the loop breaks with "dictionary changed size during + iteration" and the remaining slots stay subscribed (SonarQube S7504 does not apply).""" + reader, pool = make_reader() + subs = {'1': MagicMock(), '2': MagicMock(), '3': MagicMock()} + reader.engine.subscriptions = subs + + async def pop(slot): + subs.pop(slot) + + reader.engine.unsubscribe = AsyncMock(side_effect=pop) + + await reader.shutdown() + + assert subs == {} + assert reader.engine.unsubscribe.await_count == 3 diff --git a/tests/unit/test_app.py b/tests/unit/test_app.py index ab038a3..e392fc8 100644 --- a/tests/unit/test_app.py +++ b/tests/unit/test_app.py @@ -139,6 +139,7 @@ async def test_main_successful_run_one_loop(mock_app_env, capsys): mock_app_env['GatewayRuntime'].assert_called_once_with( config=mock_app_env['GatewayConfig'].from_env.return_value, data_manager=mock_ingestor_instance.ingestor_manager.data_manager, + pool=mock_ingestor_instance.ingestor_manager.pool, logger=mock_ingestor_instance.logger, notification_handler=mock_ingestor_instance.notification_handler, metrics_controller=mock_ingestor_instance.metrics_controller, @@ -350,8 +351,9 @@ async def test_main_prepare_ingestor_failure(mock_app_env, capsys): @pytest.mark.asyncio async def test_main_gateway_runtime_init_failure_keeps_read_path_alive(mock_app_env): - """GatewayRuntime construction hitting the network (MinIO/Postgres) and failing - must not kill the app: the main loop (read path) still runs and shuts down cleanly.""" + """GatewayRuntime construction failing (e.g. Postgres config) must not kill the + app: the read path lives in the ingestor_manager, so the main loop still runs and + shuts down cleanly.""" mock_ingestor_instance = mock_app_env['Ingestor'].return_value mock_exit_signal = mock_app_env['mock_exit_signal'] diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 7aa9743..8343fe7 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -31,7 +31,7 @@ def test_from_env_defaults(): assert cfg.nats_queue == 'gateway-writers' assert cfg.grpc_enabled is False assert cfg.grpc_port == 50051 - assert cfg.cert_bucket == 'sientia-dataops' + assert cfg.cert_bucket == 'streamlit-connectors' assert cfg.cert_cache_dir == '/tmp/opc-certs' # noqa: S108 assert cfg.write_timestamp is True assert cfg.write_quality is True @@ -69,3 +69,19 @@ def test_bool_parsing_edge_cases(monkeypatch, value, expected): cfg = GatewayConfig.from_env() assert cfg.grpc_enabled is expected + + +def test_certs_from_env_ignores_malformed_numeric_vars(monkeypatch): + import pytest + + monkeypatch.setenv('GRPC_PORT', '') + monkeypatch.setenv('IDLE_CONN_MAX_S', 'abc') + monkeypatch.setenv('CERT_BUCKET', 'my-bucket') + monkeypatch.setenv('CERT_CACHE_DIR', 'certs-cache') + + cfg = GatewayConfig.certs_from_env() + + assert cfg.cert_bucket == 'my-bucket' + assert cfg.cert_cache_dir == 'certs-cache' + with pytest.raises(ValueError): + GatewayConfig.from_env() # the full config still fails, only the runtime uses it diff --git a/tests/unit/test_gateway_runtime.py b/tests/unit/test_gateway_runtime.py index c057753..8da8ddb 100644 --- a/tests/unit/test_gateway_runtime.py +++ b/tests/unit/test_gateway_runtime.py @@ -6,9 +6,6 @@ from ingestor import gateway_runtime as gr from ingestor.config import GatewayConfig _PATCHED_CLASSES = ( - 'MinioRepository', - 'CertProvider', - 'OpcConnectionPool', 'Postgres', 'build_postgres_config', 'DataSink', @@ -36,8 +33,10 @@ def mocks(monkeypatch): 'min_connections': 5, 'max_connections': 20, } - patched['OpcConnectionPool'].return_value.close_all = AsyncMock() - patched['OpcConnectionPool'].return_value.release_idle = AsyncMock() + # The pool is owned by IngestorManager (read path) and only borrowed here. + patched['pool'] = MagicMock() + patched['pool'].close_all = AsyncMock() + patched['pool'].release_idle = AsyncMock() patched['NatsWriteConsumer'].return_value.start = AsyncMock() patched['NatsWriteConsumer'].return_value.stop = AsyncMock() patched['GrpcServer'].return_value.start = AsyncMock() @@ -45,10 +44,13 @@ def mocks(monkeypatch): return patched -def make_runtime(**config_overrides): +def make_runtime(pool=None, **config_overrides): + if pool is None: + pool = MagicMock(close_all=AsyncMock(), release_idle=AsyncMock()) return gr.GatewayRuntime( config=GatewayConfig(**config_overrides), data_manager=MagicMock(), + pool=pool, logger=MagicMock(), notification_handler=MagicMock(), metrics_controller=MagicMock(), @@ -57,26 +59,26 @@ def make_runtime(**config_overrides): def test_builds_object_graph_with_config_flags(mocks): runtime = make_runtime( + pool=mocks['pool'], write_timestamp=False, write_quality=False, write_min_pause_s=0.25, - cert_bucket='my-bucket', - cert_cache_dir='certs-cache', grpc_enabled=False, ) - mocks['OpcConnectionPool'].assert_called_once() + # Writes and active reads share the pool built by the read path. + assert runtime.pool is mocks['pool'] mocks['DataSink'].assert_called_once() mocks['WriteEngine'].assert_called_once() - _, kwargs = mocks['WriteEngine'].call_args + args, kwargs = mocks['WriteEngine'].call_args + assert args[0] is mocks['pool'] assert kwargs['write_timestamp'] is False assert kwargs['write_quality'] is False assert kwargs['min_pause_s'] == 0.25 - _, cert_kwargs = mocks['CertProvider'].call_args - assert cert_kwargs['bucket'] == 'my-bucket' - assert cert_kwargs['cache_dir'] == 'certs-cache' + args, _ = mocks['ActiveReadEngine'].call_args + assert args[0] is mocks['pool'] mocks['GrpcServer'].assert_not_called() assert runtime.grpc_server is None @@ -137,20 +139,21 @@ async def test_nats_start_failure_does_not_raise_and_idle_sweep_retries(mocks): @pytest.mark.asyncio -async def test_stop_stops_started_components_and_closes_pool(mocks): - runtime = make_runtime(grpc_enabled=True) +async def test_stop_stops_started_components_but_not_the_pool(mocks): + runtime = make_runtime(pool=mocks['pool'], grpc_enabled=True) await runtime.start() await runtime.stop() mocks['GrpcServer'].return_value.stop.assert_awaited_once() mocks['NatsWriteConsumer'].return_value.stop.assert_awaited_once() - mocks['OpcConnectionPool'].return_value.close_all.assert_awaited_once() + # IngestorManager.shutdown() closes it, after its readers unsubscribe. + mocks['pool'].close_all.assert_not_called() @pytest.mark.asyncio async def test_stop_skips_nats_stop_if_never_started(mocks): - runtime = make_runtime() + runtime = make_runtime(pool=mocks['pool']) runtime.send_notification_async = AsyncMock() nats_mock = mocks['NatsWriteConsumer'].return_value nats_mock.start = AsyncMock(side_effect=Exception('boom')) @@ -159,14 +162,14 @@ async def test_stop_skips_nats_stop_if_never_started(mocks): await runtime.stop() nats_mock.stop.assert_not_called() - mocks['OpcConnectionPool'].return_value.close_all.assert_awaited_once() + mocks['pool'].close_all.assert_not_called() @pytest.mark.asyncio async def test_idle_sweep_releases_idle_connections(mocks): - runtime = make_runtime(idle_conn_max_s=42.0) + runtime = make_runtime(pool=mocks['pool'], idle_conn_max_s=42.0) runtime._nats_started = True # isolate from the retry path await runtime.idle_sweep() - mocks['OpcConnectionPool'].return_value.release_idle.assert_awaited_once_with(42.0) + mocks['pool'].release_idle.assert_awaited_once_with(42.0)