From 50578fdba89a5ed77480043232d69f2f75703501 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 6 Oct 2026 13:27:18 -0300 Subject: [PATCH] SIENTIAPDE-1974: feat(read): slot v2 reads via SubscriptionEngine, OPC cert from MinIO MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Swaps the legacy OpcManager read path for the gateway's SubscriptionEngine over the shared OpcConnectionPool. The OPC client certificate now always comes from MinIO + Mongo (CertProvider, Basic256Sha256) — no more certs/ shipped in the Gitea repository, so a new version branch no longer breaks the reads. - io/slots.py: parse_slot accepts slot v2 (names[]) and converts the legacy v1 (tag_name + topics; cert_path => use_security, path dropped), so gateway and Orchestrator can be deployed in any order. - managers/server_reader.py: new per-server reader with the OpcManager interface the IngestorManager already drives (connect, create_subscription, subscribe, unsubscribe, check_cycles, check_opc_listenning, shutdown). - IngestorManager owns the read path: lazy-MinIO CertProvider, the pool and a read DataSink, built without touching the network. Slots from Redis are normalized on lease and refresh; a malformed slot is notified and the last good config is kept. - GatewayRuntime borrows the pool (writes/gRPC) instead of building it, and no longer closes it on stop: a Postgres/NATS failure can't stop the reads. - pool: pin/unpin per active subscription so release_idle never evicts a subscribed connection; public evict() for lost servers / changed config. - SubscriptionEngine pins on subscribe, unpins on unsubscribe, and drops a subscription locally even when the server-side delete fails. - DataSink: model_id only written when the name has one, keeping v1 documents identical to the legacy ones. - config: CERT_BUCKET default is the legacy streamlit-connectors bucket (D7). - Removes OpcManager, its tests and scripts/generate-and-push-opc-cert.sh; README/CLAUDE.md document the MinIO certificate and the slot v2 format. Tests: 218 unit tests pass; ruff check/format clean. Local end-to-end against an asyncua server: v1 slot with two topics -> one doc per collection, legacy doc shape, idle sweep keeps the subscription, lost-server shutdown and re-initialization work. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 5 +- README.md | 97 ++- ingestor/app.py | 9 +- ingestor/config.py | 4 +- ingestor/connection/certs.py | 48 ++ ingestor/connection/pool.py | 31 +- ingestor/engines/subscription.py | 14 +- ingestor/gateway_runtime.py | 42 +- ingestor/io/data_sink.py | 19 +- ingestor/io/slots.py | 36 +- ingestor/managers/ingestor_manager.py | 117 ++-- ingestor/managers/opc_manager.py | 588 ----------------- ingestor/managers/server_reader.py | 101 +++ scripts/generate-and-push-opc-cert.sh | 95 --- tests/unit/connection/test_certs.py | 44 ++ tests/unit/connection/test_pool.py | 66 ++ tests/unit/engines/test_subscription.py | 54 ++ tests/unit/io/test_data_sink_publish.py | 23 + tests/unit/io/test_slots.py | 99 +++ tests/unit/managers/test_ingestor_manager.py | 192 ++++-- tests/unit/managers/test_opc_manager.py | 627 ------------------- tests/unit/managers/test_server_reader.py | 129 ++++ tests/unit/test_app.py | 6 +- tests/unit/test_config.py | 2 +- tests/unit/test_gateway_runtime.py | 43 +- 25 files changed, 957 insertions(+), 1534 deletions(-) delete mode 100644 ingestor/managers/opc_manager.py create mode 100644 ingestor/managers/server_reader.py delete mode 100755 scripts/generate-and-push-opc-cert.sh delete mode 100644 tests/unit/managers/test_opc_manager.py create mode 100644 tests/unit/managers/test_server_reader.py diff --git a/CLAUDE.md b/CLAUDE.md index 7330de9..0a9caef 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -40,7 +40,8 @@ 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 + -> 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 +57,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, replaced the legacy `OpcManager`) 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..56e5711 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,31 @@ 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 | - -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. - -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. +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. ## 🧪 Testing @@ -306,7 +300,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 +314,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 +347,13 @@ 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 +│ │ ├── 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 (e.g. e2e seed) ├── 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..87492c0 100644 --- a/ingestor/config.py +++ b/ingestor/config.py @@ -22,7 +22,9 @@ 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 diff --git a/ingestor/connection/certs.py b/ingestor/connection/certs.py index 71dc762..8c57b5b 100644 --- a/ingestor/connection/certs.py +++ b/ingestor/connection/certs.py @@ -1,5 +1,10 @@ +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' @@ -27,3 +32,46 @@ class CertProvider: local_path = str(Path(self.cache_dir) / Path(object_name).name) Path(local_path).write_bytes(data) return 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..1f66a5c 100644 --- a/ingestor/connection/pool.py +++ b/ingestor/connection/pool.py @@ -26,6 +26,10 @@ 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] = {} # 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] = {} @@ -96,15 +100,36 @@ 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).""" + 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`.""" + 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.""" + 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, @@ -122,7 +147,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..16ebabe 100644 --- a/ingestor/engines/subscription.py +++ b/ingestor/engines/subscription.py @@ -57,6 +57,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, @@ -96,10 +97,17 @@ 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) + 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}.') async def datachange_notification(self, node, _val, data): 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..d9e0180 100644 --- a/ingestor/io/slots.py +++ b/ingestor/io/slots.py @@ -1,25 +1,45 @@ -"""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') -def _parse_tag(address: str, tag: dict) -> dict: +def _parse_names(address: str, tag: dict) -> list[dict]: names = tag.get('names') - if not names: - raise ValueError(f'tag {address!r} has no 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: 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..c7e0504 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -8,9 +8,14 @@ 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 class IngestorManager(SientiaMonitoring): @@ -47,7 +52,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 +76,7 @@ class IngestorManager(SientiaMonitoring): logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, + gateway_config: GatewayConfig | None = None, ): redis_host: str = redis_data['host'] redis_port: int = int(redis_data['port']) @@ -90,6 +99,22 @@ 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.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.resource_manager = ResourceManager( host=redis_host, port=redis_port, @@ -109,48 +134,34 @@ class IngestorManager(SientiaMonitoring): self.metadata = metadata - async def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None: + async def initialize_opc_from_config(self, server_config: dict) -> ServerReader | None: """ - Initializes an OPC Manager instance using the provided server configuration. + Initializes the reader of one OPC UA server from its slot v2 configuration. - This method creates and configures an OPC Manager for a specific OPC UA server, - establishing the connection and preparing it for tag subscriptions. + Creates a ServerReader and opens its pooled connection, so the server is ready + for tag subscriptions. With `use_security`, the certificate comes from MinIO. 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. + server_config (dict): The server configuration of a parsed slot (parse_slot), + without the 'tags' key. Expected keys: 'name', 'server_id', 'url', + 'server_uri', 'use_security', 'subscription_period_ms'. 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. + ServerReader | 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 OpcManager at {server_config["url"]}') - manager = OpcManager( + self.logger.info(f'Initializing OPC reader at {server_config["url"]}') + manager = ServerReader( name=server_config['name'], - url=server_config['url'], - subscription_period_ms=server_config['subscription_period_ms'], - data_manager=self.data_manager, + server_config=server_config, + pool=self.pool, + sink=self.sink, logger=self.logger, - server_uri=server_config['server_uri'], notification_handler=self.notification_handler, - metadata=self.metadata, - cert_path=server_config.get('cert_path'), - private_key_path=server_config.get('private_key_path'), - server_cert_path=server_config.get('server_cert_path'), metrics_controller=self.metrics_controller, + metadata=self.metadata, ) manager.config = server_config @@ -176,15 +187,15 @@ 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): @@ -419,14 +430,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 +533,44 @@ 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). + + Returns: + dict | None: The parsed slot, or None when the config is malformed + (logged and notified). + """ + try: + return 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 + async def drop_slot_leases(self, ids: list[str]) -> None: """ Releases the leases associated with the specified slot IDs. diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py deleted file mode 100644 index c08d5ab..0000000 --- a/ingestor/managers/opc_manager.py +++ /dev/null @@ -1,588 +0,0 @@ -import asyncio -import json -import traceback -from pathlib import Path - -from asyncua import Client -from asyncua.crypto.security_policies import SecurityPolicyBasic256 -from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler -from sientia_do.notifications.models import NotificationLevel -from sientia_do.observability.logger import Logger -from sientia_do.observability.metrics_controller import MetricsController -from sientia_do.observability.sientia_monitoring import SientiaMonitoring -from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, OPC_TIMEZONE - -import ingestor.metrics as metrics -from ingestor.managers.data_manager import DataManager - - -class OpcManager(SientiaMonitoring): - """ - Manages OPC UA server connections and tag subscriptions. - - The OpcManager is responsible for: - - Establishing and maintaining secure connections to OPC UA servers - - Managing tag subscriptions and data collection - - Handling server reconnection and error recovery - - Processing OPC data and forwarding it to the data manager - - Monitoring connection health and performance metrics - - The manager supports both secure and unsecured connections, with optional - certificate-based authentication for enhanced security. - - Args: - name (str): Unique identifier for the OPC server - url (str): OPC UA server endpoint URL - data_manager (DataManager): Manager for data persistence and export - logger (Logger): Logger instance for application logging - server_uri (str): OPC UA server application URI - notification_handler (NotificationHandler): Handler for sending notifications - metadata (dict): Application metadata for notifications and tracking - cert_path (str, optional): Path to client certificate file for secure connections - private_key_path (str, optional): Path to client private key file - server_cert_path (str, optional): Path to server certificate file for validation - - Attributes: - url (str): OPC UA server endpoint URL - name (str): Unique identifier for the OPC server - server_uri (str): OPC UA server application URI - data_queue (dict): Queue for buffering OPC data before processing - non_receive_count (int): Counter for cycles without data reception - client (Client): OPC UA client instance - cert_path (str): Path to client certificate file - private_key_path (str): Path to client private key file - server_cert_path (str): Path to server certificate file - nodes (dict): Dictionary of OPC node references - subscriptions (dict): Active OPC subscriptions - data_manager (DataManager): Manager for data persistence and export - metadata (dict): Application metadata - """ - - def __init__( - self, - name: str, - url: str, - subscription_period_ms: int, - data_manager: DataManager, - logger: Logger, - server_uri: str, - notification_handler: NotificationHandler, - metrics_controller: MetricsController, - metadata: dict, - cert_path: str | None = None, - private_key_path: str | None = None, - server_cert_path: str | None = None, - ): - self.url = url - self.name = name - self.server_uri = server_uri - self.data_queue: dict = {} - self.non_receive_count = 0 - self.client: Client | None = None - self.subscription_period_ms = subscription_period_ms - self.cert_path = cert_path - self.private_key_path = private_key_path - self.server_cert_path = server_cert_path - self.nodes: dict = {} - self.subscriptions: dict = {} - self.data_manager = data_manager - self.metadata = metadata - - SientiaMonitoring.__init__( - self, - logger=logger, - notification_handler=notification_handler, - metrics_controller=metrics_controller, - ) - - metrics.OPC_CONNECTION_STATUS.labels( - pod_id=self.pod_id, server_name=self.name, server_url=self.url - ).set(0) - metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(0) - - def __str__(self): - """ - String representation of the OPC Manager. - - Returns: - str: Human-readable representation showing server details and current state. - """ - return ( - f'OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n' - f'nodes={self.nodes}, subscriptions={self.subscriptions}' - ) - - async def shutdown(self): - """ - Comprehensive cleanup method for graceful shutdown. - - This method ensures proper cleanup of all OPC UA resources: - - Closes active subscriptions - - Disconnects from the OPC server - - Releases allocated resources - - Should be called before the application terminates to prevent resource leaks - and ensure clean disconnection from OPC servers. - """ - try: - await self.disconnect() - except Exception as e: - self.logger.error(f'Error during cleanup: {e}') - - async def set_security(self): - """ - Configures the security settings for the OPC UA client. - - This method sets up the security policy, certificates, and timeouts - required for establishing a secure connection with the OPC UA server. - It implements Basic256 security policy with certificate-based authentication. - - Raises: - ValueError: If either the certificate path or private key path is not provided. - - Security Settings: - - Security Policy: Basic256 - - Secure Channel Timeout: 10,000,000 ms - - Session Timeout: 10,000,000 ms - - The method configures: - - Client application URI - - Certificate-based authentication - - Server certificate validation (if provided) - - Connection timeouts for stability - """ - - if not all([self.cert_path, self.private_key_path]): - raise ValueError( - 'Certificate and private key paths must be provided for secure connection.' - ) - cert = str(Path(self.cert_path)) if self.cert_path else None - private_key = str(Path(self.private_key_path)) if self.private_key_path else None - server_cert = str(Path(self.server_cert_path)) if self.server_cert_path else None - assert cert is not None and private_key is not None - - if self.client: - self.client.application_uri = self.server_uri - self.logger.info('Setting security...') - await self.client.set_security( - SecurityPolicyBasic256, - certificate=cert, - private_key=private_key, - server_certificate=server_cert, - ) - self.client.secure_channel_timeout = 10000000 - self.client.session_timeout = 10000000 - - async def connect(self): - """ - Establishes a connection to the OPC server. - - This method initializes the OPC client using the provided URL and - sets up security if a certificate path is specified. It then - attempts to connect to the server and logs the connection status. - - The connection process includes: - 1. Client initialization with server URL - 2. Security configuration (if certificates are provided) - 3. Connection establishment - 4. Metrics recording for monitoring - - Raises: - Exception: If the connection to the OPC server fails. - - Metrics: - - OPC_CONNECTIONS_TOTAL: Incremented on connection attempt - - OPC_CONNECTION_STATUS: Set to 1 on successful connection - """ - - await self.emit_metric( - metric_object=metrics.OPC_CONNECTIONS_TOTAL, - tags={ - 'pod_id': self.pod_id, - 'server_name': self.name, - }, - ) - try: - self.client = Client(self.url, timeout=10, watchdog_intervall=3600000) - assert self.client is not None # Informa ao mypy que client não é None - - self.client.name = self.pod_id - self.client.description = self.pod_id - pod_uri = self.pod_id.replace('-', ':') - self.client.application_uri = pod_uri - self.client.product_uri = pod_uri - - if self.cert_path: - await self.set_security() - self.logger.info(f'Starting connection to {self.name}...') - await self.client.connect() - await self.emit_metric( - metric_object=metrics.OPC_CONNECTION_STATUS, - method='set', - value=1, - tags={ - 'pod_id': self.pod_id, - 'server_name': self.name, - 'server_url': self.url, - }, - ) - self.logger.info(f'Connection to {self.name} successful.') - except Exception: - await self.disconnect() - - raise - - async def create_subscription(self, name: str): - """ - Creates a subscription with the specified monitoring period. - - This method establishes a subscription to monitor data changes or events - from the OPC UA server. If the client is not connected, an exception is raised. - - Args: - name (str): The name identifier for the subscription - period (int, optional): The monitoring period in milliseconds. Defaults to 500 ms. - - Raises: - ValueError: If the client is not connected. - - Side Effects: - - Sets the `self.period` attribute to the specified or default period. - - Creates a subscription and assigns it to `self.subscriptions[name]`. - - Logs the creation of the subscription. - - Increments subscription creation metrics. - """ - - if not self.client: - raise ValueError('Client not connected. Call connect first.') - try: - self.subscriptions[name] = await self.client.create_subscription( - self.subscription_period_ms, self - ) - self.logger.info(f'Subscription {name} created on {self.name}.') - await self.emit_metric( - metric_object=metrics.OPC_SUBSCRIPTIONS_CREATED, - method='inc', - value=1, - tags={ - 'pod_id': self.pod_id, - 'server_name': self.name, - 'slot_name': name, - }, - ) - except Exception as e: - self.logger.error(f'Failed to create subscription {name} on {self.name}: {e}') - raise - - async def subscribe(self, subscription: str, nodes: dict, collect_period: int): - """ - Subscribes to a set of OPC UA nodes for data change notifications. - - This method adds the specified nodes to the subscription and configures - their data collection rules based on the provided collection period and - node-specific frequency. - - Args: - subscription (str): The name of the subscription to use - nodes (dict): A dictionary where keys are node identifiers and values are - configurations for each node. Each configuration must include a 'frequency' - key indicating the frequency of data collection in Hz. - collect_period (int): The data collection period in seconds. - - Raises: - ValueError: If the subscription has not been created by calling - `create_subscription` prior to this method. - - Side Effects: - - Updates internal node tracking and cycle rules - - Establishes data change monitoring for specified nodes - - Updates metrics for subscribed tags count - """ - - if not self.subscriptions.get(subscription): - raise ValueError('Subscription not created. Call create_subscription first.') - - self.logger.info(f'Subscribing to {subscription} on {self.name}...') - self.logger.info(f'Subscribing to nodes: {nodes}') - assert self.client is not None # Informa ao mypy que client não é None - addr_nodes = [self.client.get_node(n) for n in nodes] - self.logger.debug(f'Addr nodes: {addr_nodes}') - self.nodes.update(nodes) - self.logger.debug(f'Nodes: {self.nodes}') - - self.collect_period = collect_period - - for node, config in self.nodes.items(): - self.nodes[node]['cycle_rule'] = { - 'cycle_increment': collect_period * 1000 / float(config['frequency']), - 'cycle_count': 0, - } - - await self.subscriptions[subscription].subscribe_data_change(addr_nodes) - - await self.emit_metric( - metric_object=metrics.OPC_TAGS_SUBSCRIBED, - method='set', - value=len(self.nodes), - tags={ - 'pod_id': self.pod_id, - 'server_name': self.name, - }, - ) - - async def unsubscribe(self, subscription: str): - """ - Unsubscribes from a given subscription. - - This method removes the specified subscription and cleans up associated - resources. It handles cases where the subscription doesn't exist gracefully. - - Args: - subscription (str): The name of the subscription to unsubscribe from. - - Logs: - - A warning if the specified subscription does not exist. - - An info message upon successful unsubscription. - - Behavior: - - If the subscription exists, it is deleted and removed from the - subscriptions dictionary. - - If the subscription does not exist, no action is taken. - """ - - if not self.subscriptions.get(subscription): - self.logger.warning(f"Subscription '{subscription}' not found. Cannot unsubscribe.") - return - await self.subscriptions[subscription].delete() - del self.subscriptions[subscription] - self.logger.info(f'Unsubscribed from {subscription}.') - - async def disconnection_fallback(self) -> list: - """ - Tries 5 times to disconnect from the OPC UA server, with a delay of 100ms x try. - """ - - assert self.client is not None - error_stack = [] - for i in range(5): - try: - self.logger.info(f'Disconnecting from OPC UA server, attempt {i + 1} of 5') - await self.client.disconnect() - return [] - except Exception as e: - self.logger.error( - f'Failed to disconnect from OPC UA serve in attempt {i + 1} of 5: {e}' - ) - error_stack.append( - { - 'attempt': i + 1, - 'error': str(e), - 'traceback': traceback.format_exc(), - } - ) - await asyncio.sleep(0.1 * i) - return error_stack - - async def disconnect(self): - """ - Disconnects from the OPC UA server. - - This method handles the disconnection process by deleting all subscriptions - and disconnecting the client from the OPC UA server. It logs the disconnection - process and handles any exceptions that may occur during cleanup. - - Raises: - Exception: If an error occurs while deleting the subscription or disconnecting - from the OPC UA server, it logs the error details. - - Side Effects: - - Deletes all active subscriptions - - Disconnects the OPC client - - Updates connection status metrics - - Clears internal client reference - """ - - self.logger.warning('Disconnecting from OPC server') - if self.client is None: - self.logger.warning('Client already disconnected.') - return - try: - for sub in self.subscriptions: - await self.subscriptions[sub].delete() - self.logger.warning('Deleted all subscriptions.') - except Exception as sub_error: - self.logger.error(f'Failed to clean up subscription: {sub_error}') - - errors = await self.disconnection_fallback() - - if errors: - await self.send_notification_async( - metadata=self.metadata, - notification_id=f'OPC_DISCONNECTION_ERROR_{self.name}', - message=f'Failed to disconnect from OPC UA server {self.name} after 5 attempts', - block='opc_manager', - level=NotificationLevel.ERROR, - attachment_content=json.dumps(errors, indent=4), - ) - else: - self.logger.warning('Disconnected from OPC UA server.') - - del self.client - self.client = None - await self.emit_metric( - metric_object=metrics.OPC_CONNECTION_STATUS, - method='set', - value=0, - tags={ - 'pod_id': self.pod_id, - 'server_name': self.name, - 'server_url': self.url, - }, - ) - await self.emit_metric( - metric_object=metrics.OPC_TAGS_SUBSCRIBED, - method='set', - value=0, - tags={ - 'pod_id': self.pod_id, - 'server_name': self.name, - }, - ) - - async def datachange_notification(self, node, _val, data): - """ - Handles data change notifications for monitored OPC UA nodes. - - This method is triggered when a monitored node's value changes. It processes - the notification, updates internal state, and publishes the data to the - appropriate topics. - - Args: - node (NodeId): The OPC UA node that triggered the data change notification. - _val (Any): The new value of the node (unused in this implementation). - data (DataChangeNotification): The data change notification object containing - details about the change. - - Behavior: - - Extracts the value and source timestamp from the monitored item. - - Resets the cycle count for the node's cycle rule. - - Resets the non-receive count. - - Constructs a data dictionary containing the tag, tag name, timestamp, and value. - - Publishes the data to all topics associated with the node. - """ - - # get data value - monitored_item = data.monitored_item - value = monitored_item.Value.Value.Value - # source_timestamp - source_timestamp = monitored_item.Value.SourceTimestamp.replace(tzinfo=OPC_TIMEZONE) - tag = str(node) - - self.logger.debug( - f'Data change notification received for tag:' - f'{tag} after {self.nodes[tag]["cycle_rule"]["cycle_count"]} cycles' - ) - - data = { - 'tag': tag, - 'name': self.nodes[str(node)]['tag_name'], - 'timestamp': source_timestamp.strftime(DATETIME_FORMAT_WITH_TZ), - 'value': value, - } - - for topic in self.nodes[tag]['topics']: - await self.data_manager.publish(topic, data) - - self.nodes[tag]['cycle_rule']['cycle_count'] = 0 - self.non_receive_count = 0 - await self.emit_metric( - metric_object=metrics.OPC_CYCLES_WITHOUT_DATA, - method='set', - value=0, - tags={ - 'pod_id': self.pod_id, - 'server_name': self.name, - }, - ) - - async def check_cycles(self): - """ - Checks the cycle counts for all monitored nodes and sends - notifications if thresholds are exceeded. - - This method iterates through all monitored nodes and updates their cycle counts based on - configured increments. If a node's cycle count exceeds a threshold (5 cycles), it triggers - a warning notification. - - Side Effects: - - Updates cycle counts for all monitored nodes - - Sends warning notifications for nodes exceeding cycle thresholds - """ - for node, config in self.nodes.items(): - self.nodes[node]['cycle_rule']['cycle_count'] += config['cycle_rule']['cycle_increment'] - if self.nodes[node]['cycle_rule']['cycle_count'] >= 5: - name = config['tag_name'] - cycles = self.nodes[node]['cycle_rule']['cycle_count'] - await self.send_notification_async( - metadata=self.metadata, - notification_id=f'TAG_{node}:{name}_LISTENNING_STOPPED', - message=f'{cycles} cycles without receive from {node}:{name}', - block='opc_manager', - level=NotificationLevel.WARNING, - ) - - async def check_opc_listenning(self) -> bool: - """ - Checks the OPC connection and triggers notifications if the connection is lost. - - This method monitors the data reception health by tracking cycles without - data. It sends notifications at different thresholds and can trigger - reconnection attempts. - - Returns: - bool: True if the connection is lost and reconnection should be attempted, - False otherwise. - - Side Effects: - - Increments non-receive count - - Updates metrics for cycles without data - - Sends warning notifications at 5 cycles - - Sends error notifications and triggers reconnection at 15 cycles - """ - - self.non_receive_count += 1 - await self.emit_metric( - metric_object=metrics.OPC_CYCLES_WITHOUT_DATA, - method='set', - value=self.non_receive_count, - tags={ - 'pod_id': self.pod_id, - 'server_name': self.name, - }, - ) - if self.non_receive_count >= 5: - await self.send_notification_async( - metadata=self.metadata, - notification_id=f'OPC_LISTENNING_STOPPED__{self.name}', - message=f'{self.non_receive_count} cycles without ' - f'receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}', - block='opc_manager', - level=NotificationLevel.ERROR, - ) - if self.non_receive_count >= 15: - await self.emit_metric( - metric_object=metrics.OPC_RECONNECTIONS_TOTAL, - tags={ - 'pod_id': self.pod_id, - 'server_name': self.name, - }, - ) - await self.send_notification_async( - metadata=self.metadata, - notification_id=f'OPC_CONNECTION_RETRY__{self.name}', - message=f'Retrying to connect to server {self.name}', - block='opc_manager', - level=NotificationLevel.ERROR, - ) - return True - return False diff --git a/ingestor/managers/server_reader.py b/ingestor/managers/server_reader.py new file mode 100644 index 0000000..7bb892c --- /dev/null +++ b/ingestor/managers/server_reader.py @@ -0,0 +1,101 @@ +"""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 + +_CONNECTION_FIELDS = ('url', 'server_uri', 'use_security', 'subscription_period_ms') + + +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): + """Drop every subscription and the pooled connection, so the next + initialization reconnects (lost server or changed server config).""" + for slot in list(self.engine.subscriptions): + await self.engine.unsubscribe(slot) + 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/scripts/generate-and-push-opc-cert.sh b/scripts/generate-and-push-opc-cert.sh deleted file mode 100755 index 713965e..0000000 --- a/scripts/generate-and-push-opc-cert.sh +++ /dev/null @@ -1,95 +0,0 @@ -#!/usr/bin/env bash -# Gera o certificado X.509 de cliente OPC UA do ingestor e sobe pro Gitea -# in-cluster (repo que o pod clona no boot — nada muda no values/deployment, -# só rollout restart depois do push). -# -# Uso: -# GITEA_PASSWORD='' ./scripts/generate-and-push-opc-cert.sh -# -# Vars (defaults pro cluster dev/gcp): -# GITEA_URL default https://git.sientia.ai -# GITEA_USER default gitea_admin -# GITEA_PASSWORD obrigatória -# GITEA_REPO default gitea_admin/sientia-dataops-opc-ingestor -# BRANCH default main -# APP_URI default urn:sientia:opc-ingestor (SAN URI do cert — TEM que -# ser igual ao campo `uri` do documento na collection -# OPC_servers: o ingestor usa esse valor como application_uri -# e o KEPServer rejeita com BadCertificateUriInvalid se divergir) -# DAYS default 3650 -# FORCE=1 sobrescreve cert já existente no repo -set -euo pipefail - -GITEA_URL=${GITEA_URL:-https://git.sientia.ai} -GITEA_USER=${GITEA_USER:-gitea_admin} -GITEA_PASSWORD=${GITEA_PASSWORD:?export GITEA_PASSWORD com a senha do Gitea} -GITEA_REPO=${GITEA_REPO:-gitea_admin/sientia-dataops-opc-ingestor} -BRANCH=${BRANCH:-main} -APP_URI=${APP_URI:-urn:sientia:opc-ingestor} -CN=${CN:-sientia-opc-ingestor} -DAYS=${DAYS:-3650} - -CERT_DIR_IN_REPO="certs" -CERT_NAME="opc-ingestor-cert.pem" -KEY_NAME="opc-ingestor-key.pem" -DER_NAME="opc-ingestor-cert.der" - -workdir=$(mktemp -d) -trap 'rm -rf "$workdir"' EXIT - -echo ">> Gerando certificado (CN=${CN}, URI=${APP_URI}, ${DAYS} dias)..." -# Extensões exigidas por OPC UA (KEPServer valida keyUsage/EKU e o SAN URI). -openssl req -x509 -newkey rsa:2048 -sha256 -nodes -days "$DAYS" \ - -keyout "$workdir/$KEY_NAME" -out "$workdir/$CERT_NAME" \ - -subj "/CN=${CN}/O=Aignosi/OU=Sientia" \ - -addext "subjectAltName=URI:${APP_URI},DNS:${CN}" \ - -addext "keyUsage=critical,digitalSignature,nonRepudiation,keyEncipherment,dataEncipherment" \ - -addext "extendedKeyUsage=critical,clientAuth,serverAuth" \ - -addext "basicConstraints=critical,CA:FALSE" - -# Cópia DER — só referência p/ import manual no KEPServer (o trust normal é -# aceitar o cert rejeitado na 1ª conexão). -openssl x509 -in "$workdir/$CERT_NAME" -outform der -out "$workdir/$DER_NAME" - -echo ">> Clonando ${GITEA_REPO}@${BRANCH} do Gitea..." -encoded_pass=$(python3 - <&1 | sed "s#${encoded_pass}#***#g" - -if [[ -f "$workdir/repo/$CERT_DIR_IN_REPO/$CERT_NAME" && "${FORCE:-0}" != "1" ]]; then - echo "ERRO: $CERT_DIR_IN_REPO/$CERT_NAME já existe no repo. Re-rodar com FORCE=1 sobrescreve" >&2 - echo " (o cert antigo deixa de valer — retrust no KEPServer necessário)." >&2 - exit 1 -fi - -mkdir -p "$workdir/repo/$CERT_DIR_IN_REPO" -cp "$workdir/$CERT_NAME" "$workdir/$KEY_NAME" "$workdir/$DER_NAME" "$workdir/repo/$CERT_DIR_IN_REPO/" - -cd "$workdir/repo" -git add "$CERT_DIR_IN_REPO" -git -c user.name="opc-cert-script" -c user.email="platform@aignosi.com.br" \ - commit -m "Add OPC UA client certificate for KEPServer connection (URI ${APP_URI})" -git push origin "$BRANCH" 2>&1 | sed "s#${encoded_pass}#***#g" - -fingerprint=$(openssl x509 -in "$CERT_DIR_IN_REPO/$CERT_NAME" -noout -fingerprint -sha1) - -cat < ${CERT_DIR_IN_REPO}/{${CERT_NAME},${KEY_NAME},${DER_NAME}} -${fingerprint} - -Paths dentro do pod (o entrypoint clona o repo em /app): - cert_path: /app/${CERT_DIR_IN_REPO}/${CERT_NAME} - private_key_path: /app/${CERT_DIR_IN_REPO}/${KEY_NAME} - -Próximos passos: - 1. Documento na collection OPC_servers com "uri": "${APP_URI}" (ver README/PR). - 2. kubectl -n sientia rollout restart deployment/opc-ingestor # re-clona o repo - 3. Pod tenta conectar -> no KEPServer: OPC UA Configuration Manager -> - Trusted Clients -> cert '${CN}' rejeitado -> Trust -> restart runtime. -EOF diff --git a/tests/unit/connection/test_certs.py b/tests/unit/connection/test_certs.py index 3b66c91..db98ce7 100644 --- a/tests/unit/connection/test_certs.py +++ b/tests/unit/connection/test_certs.py @@ -26,3 +26,47 @@ 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 diff --git a/tests/unit/connection/test_pool.py b/tests/unit/connection/test_pool.py index d477805..4b1a21b 100644 --- a/tests/unit/connection/test_pool.py +++ b/tests/unit/connection/test_pool.py @@ -182,3 +182,69 @@ 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() diff --git a/tests/unit/engines/test_subscription.py b/tests/unit/engines/test_subscription.py index 5d8cdc8..6a07a64 100644 --- a/tests/unit/engines/test_subscription.py +++ b/tests/unit/engines/test_subscription.py @@ -201,3 +201,57 @@ 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() 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..de6580c 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'}) @@ -431,25 +478,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 +566,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 +721,20 @@ 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' diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py deleted file mode 100644 index 407c166..0000000 --- a/tests/unit/managers/test_opc_manager.py +++ /dev/null @@ -1,627 +0,0 @@ -import json -from datetime import datetime -from unittest.mock import ANY, AsyncMock, MagicMock, call, patch - -import pytest -from asyncua.crypto.security_policies import SecurityPolicyBasic256 -from pytest import fixture, mark -from sientia_do.notifications.models import NotificationLevel - -from ingestor.managers.opc_manager import OpcManager - -tags = { - 'ns=3;i=1001': { - 'aggregation_function': 'LTS', - 'frequency': 1000, - 'max_value': 100, - 'min_value': 0, - 'tag_name': 'Counter', - }, - 'ns=3;i=1003': { - 'aggregation_function': 'AVG', - 'frequency': 1000, - 'max_value': 100, - 'min_value': 0, - 'tag_name': 'Random', - }, - 'ns=3;i=1004': { - 'aggregation_function': 'MDN', - 'frequency': 1000, - 'max_value': 100, - 'min_value': 0, - 'tag_name': 'Sawtooth', - }, -} - -metadata = { - 'metadata': { - 'model_id': 'test_model', - 'model_name': 'test_model', - 'workflow_name': 'test_workflow', - 'schema_name': 'test_schedule', - }, -} - - -@fixture -@patch('ingestor.managers.opc_manager.metrics') -def raw_opc_manager(mock_metrics): - opc_manager = OpcManager( - name='TestConnector', - url='opc.tcp://localhost:4840', - data_manager=AsyncMock(), - subscription_period_ms=1000, - logger=MagicMock(), - server_uri='opc.tcp://localhost:4840', - notification_handler=MagicMock(), - metadata=metadata['metadata'], - metrics_controller=AsyncMock(), - ) - - opc_manager.emit_metric = AsyncMock() - opc_manager.send_notification_async = AsyncMock() - opc_manager.send_notification = MagicMock() - - return opc_manager - - -@fixture -def opc_manager(raw_opc_manager): - raw_opc_manager.client = AsyncMock() - raw_opc_manager.cert_path = 'cert.pem' - raw_opc_manager.private_key_path = 'private_key.pem' - raw_opc_manager.server_cert_path = 'server_cert.pem' - - return raw_opc_manager - - -@fixture -def opc_manager_subscribed(opc_manager): - opc_manager.subscriptions['sub1'] = AsyncMock() - - return opc_manager - - -def test___str__(opc_manager): - assert ( - str(opc_manager) - == 'OpcManager(name=TestConnector, url=opc.tcp://localhost:4840, server_uri=opc.tcp://localhost:4840)\nnodes={}, subscriptions={}' - ) - - -@mark.asyncio -async def test_shutdown_success(opc_manager): - opc_manager.disconnect = AsyncMock() - - await opc_manager.shutdown() - - opc_manager.disconnect.assert_called_once() - - -@mark.asyncio -async def test_shutdown_error(opc_manager): - opc_manager.disconnect = AsyncMock(side_effect=Exception('Test error')) - - await opc_manager.shutdown() - - opc_manager.logger.error.assert_called_once_with('Error during cleanup: Test error') - - -@mark.asyncio -async def test_set_security_success(opc_manager): - await opc_manager.set_security() - - assert opc_manager.client.application_uri == opc_manager.server_uri - - opc_manager.client.set_security.assert_called_once_with( - SecurityPolicyBasic256, - certificate=opc_manager.cert_path, - private_key=opc_manager.private_key_path, - server_certificate=opc_manager.server_cert_path, - ) - - assert opc_manager.client.secure_channel_timeout == 10000000 - assert opc_manager.client.session_timeout == 10000000 - - -@mark.asyncio -async def test_set_security_no_cert(opc_manager): - opc_manager.cert_path = None - opc_manager.private_key_path = None - - try: - await opc_manager.set_security() - except ValueError as e: - assert str(e) == 'Certificate and private key paths must be provided for secure connection.' - else: - raise AssertionError('ValueError not raised') - - assert opc_manager.client.set_security.call_count == 0 - - -@mark.asyncio -@patch('ingestor.managers.opc_manager.metrics') -@patch('ingestor.managers.opc_manager.Client') -async def test_connect_no_security(client, mock_metrics, raw_opc_manager): - raw_opc_manager.set_security = AsyncMock() - client.return_value = AsyncMock() - - await raw_opc_manager.connect() - - client.assert_called_once_with(raw_opc_manager.url, timeout=10, watchdog_intervall=3600000) - raw_opc_manager.client.connect.assert_called_once() - raw_opc_manager.set_security.assert_not_called() - raw_opc_manager.emit_metric.assert_has_calls( - [ - call( - metric_object=mock_metrics.OPC_CONNECTIONS_TOTAL, - tags={ - 'pod_id': raw_opc_manager.pod_id, - 'server_name': raw_opc_manager.name, - }, - ), - call( - metric_object=mock_metrics.OPC_CONNECTION_STATUS, - method='set', - value=1, - tags={ - 'pod_id': raw_opc_manager.pod_id, - 'server_name': raw_opc_manager.name, - 'server_url': raw_opc_manager.url, - }, - ), - ], - any_order=True, - ) - - -@mark.asyncio -@patch('ingestor.managers.opc_manager.Client') -async def test_connect_with_security(client, raw_opc_manager): - raw_opc_manager.cert_path = 'cert.pem' - raw_opc_manager.private_key_path = 'private_key.pem' - raw_opc_manager.server_cert_path = 'server_cert.pem' - raw_opc_manager.set_security = AsyncMock() - client.return_value = AsyncMock() - - await raw_opc_manager.connect() - - client.assert_called_once_with(raw_opc_manager.url, timeout=10, watchdog_intervall=3600000) - raw_opc_manager.client.connect.assert_called_once() - raw_opc_manager.set_security.assert_called_once() - - -@mark.asyncio -@patch('ingestor.managers.opc_manager.Client') -@patch('ingestor.managers.opc_manager.metrics') -async def test_connect_exception_handling_and_metrics( - mock_metrics_module, mock_opc_client_class, raw_opc_manager -): - mock_client_instance = mock_opc_client_class.return_value - simulated_error_message = 'Erro de conexão simulado' - mock_client_instance.connect.side_effect = Exception(simulated_error_message) - - raw_opc_manager.disconnect = AsyncMock() - - opc_manager_instance = raw_opc_manager - opc_manager_instance.cert_path = None - - with pytest.raises(Exception, match=simulated_error_message): - await opc_manager_instance.connect() - - opc_manager_instance.disconnect.assert_called_once() - - -@mark.asyncio -async def test_create_subscription_no_client(raw_opc_manager): - try: - await raw_opc_manager.create_subscription('sub1') - except ValueError as e: - assert str(e) == 'Client not connected. Call connect first.' - else: - raise AssertionError('ValueError not raised') - - -@mark.asyncio -async def test_create_subscription_success_has_period(opc_manager): - await opc_manager.create_subscription('sub1') - - opc_manager.client.create_subscription.assert_called_once_with( - opc_manager.subscription_period_ms, opc_manager - ) - assert opc_manager.subscriptions['sub1'] is not None - - -@mark.asyncio -async def test_create_subscription_success_no_period(opc_manager): - await opc_manager.create_subscription('sub1') - - opc_manager.client.create_subscription.assert_called_once_with( - opc_manager.subscription_period_ms, opc_manager - ) - assert opc_manager.subscriptions['sub1'] is not None - - -@patch('ingestor.managers.opc_manager.metrics') -@mark.asyncio -async def test_create_subscription_with_metrics(metrics, opc_manager): - await opc_manager.create_subscription('sub1') - - opc_manager.emit_metric.assert_called_once_with( - metric_object=metrics.OPC_SUBSCRIPTIONS_CREATED, - method='inc', - value=1, - tags={ - 'pod_id': opc_manager.pod_id, - 'server_name': opc_manager.name, - 'slot_name': 'sub1', - }, - ) - - -@patch('ingestor.managers.opc_manager.metrics') -@mark.asyncio -async def test_create_subscription_exception_during_client_call( - mock_metrics_module, raw_opc_manager -): - opc_manager_instance = raw_opc_manager - opc_manager_instance.client = AsyncMock() - - subscription_name = 'test_sub_client_error' - simulated_error_message = 'Falha ao criar subscrição no cliente OPC' - - opc_manager_instance.client.create_subscription.side_effect = Exception(simulated_error_message) - - with pytest.raises(Exception, match=simulated_error_message): - await opc_manager_instance.create_subscription(subscription_name) - - opc_manager_instance.client.create_subscription.assert_called_once_with( - opc_manager_instance.subscription_period_ms, opc_manager_instance - ) - - opc_manager_instance.logger.error.assert_called_once_with( - f'Failed to create subscription {subscription_name} on {opc_manager_instance.name}: {simulated_error_message}' - ) - - mock_metrics_module.OPC_SUBSCRIPTIONS_CREATED.labels.assert_not_called() - - -@patch('ingestor.managers.opc_manager.metrics') -@mark.asyncio -async def test_subscribe_no_subscription(metrics, opc_manager): - try: - await opc_manager.subscribe('sub1', tags, 1000) - except ValueError as e: - assert str(e) == 'Subscription not created. Call create_subscription first.' - else: - raise AssertionError('ValueError not raised') - metrics.OPC_TAGS_SUBSCRIBED.labels.assert_not_called() - - -@mark.asyncio -async def test_subscribe_success(opc_manager_subscribed): - opc_manager_subscribed.client.get_node = MagicMock() - opc_manager_subscribed.nodes = {'ns=3;i=1001': 'data'} - - await opc_manager_subscribed.subscribe('sub1', tags, 1000) - - assert opc_manager_subscribed.nodes == tags - opc_manager_subscribed.subscriptions['sub1'].subscribe_data_change.assert_called_once_with( - [opc_manager_subscribed.client.get_node(n) for n in tags] - ) - - -@mark.asyncio -async def test_unsubscribe_no_subscription(opc_manager): - await opc_manager.unsubscribe('sub1') - - opc_manager.logger.warning.assert_called_once_with( - "Subscription 'sub1' not found. Cannot unsubscribe." - ) - assert opc_manager.subscriptions.get('sub1') is None - - -@mark.asyncio -async def test_unsubscribe_success(opc_manager_subscribed): - await opc_manager_subscribed.unsubscribe('sub1') - - assert opc_manager_subscribed.subscriptions.get('sub1') is None - - -@mark.asyncio -async def test_disconnection_fallback_success(opc_manager): - opc_manager.client = AsyncMock() - opc_manager.client.disconnect.return_value = True - result = await opc_manager.disconnection_fallback() - assert result == [] - - -@mark.asyncio -async def test_disconnection_fallback_fail(opc_manager): - opc_manager.client = AsyncMock() - opc_manager.client.disconnect.side_effect = Exception('Test error') - result = await opc_manager.disconnection_fallback() - assert result == [ - {'attempt': 1, 'error': 'Test error', 'traceback': ANY}, - {'attempt': 2, 'error': 'Test error', 'traceback': ANY}, - {'attempt': 3, 'error': 'Test error', 'traceback': ANY}, - {'attempt': 4, 'error': 'Test error', 'traceback': ANY}, - {'attempt': 5, 'error': 'Test error', 'traceback': ANY}, - ] - assert opc_manager.client.disconnect.call_count == 5 - - -@mark.asyncio -async def test_disconnect_success(opc_manager_subscribed): - opc_manager_subscribed.client = MagicMock() - - await opc_manager_subscribed.disconnect() - - opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once() - assert opc_manager_subscribed.client is None - - -@mark.asyncio -async def test_disconnect_no_client(opc_manager_subscribed): - opc_manager_subscribed.client = None - assert await opc_manager_subscribed.disconnect() is None - - opc_manager_subscribed.logger.warning.assert_has_calls( - [ - call('Client already disconnected.'), - ] - ) - - -@mark.asyncio -async def test_disconnect_error_unsubscribe(opc_manager_subscribed): - opc_manager_subscribed.client = MagicMock(disconnect=AsyncMock()) - opc_manager_subscribed.subscriptions['sub1'] = MagicMock( - delete=AsyncMock(side_effect=Exception('Test error')) - ) - - await opc_manager_subscribed.disconnect() - - opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once() - opc_manager_subscribed.client = None - opc_manager_subscribed.logger.error.assert_called_once_with( - 'Failed to clean up subscription: Test error' - ) - - -@mark.asyncio -async def test_disconnect_error(opc_manager_subscribed): - opc_manager_subscribed.client = MagicMock() - opc_manager_subscribed.client.disconnect = MagicMock(side_effect=Exception('Test error')) - - await opc_manager_subscribed.disconnect() - - opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once() - opc_manager_subscribed.client = None - - -@patch('ingestor.managers.opc_manager.metrics') -@mark.asyncio -async def test_disconnect_metrics_on_successful_path(mock_metrics_module, raw_opc_manager): - mock_metrics_module.OPC_CONNECTION_STATUS.reset_mock() - mock_metrics_module.OPC_TAGS_SUBSCRIBED.reset_mock() - - raw_opc_manager.client = MagicMock() - mock_sub1 = MagicMock() - mock_sub2 = MagicMock() - raw_opc_manager.subscriptions = {'sub1': mock_sub1, 'sub2': mock_sub2} - - await raw_opc_manager.disconnect() - - raw_opc_manager.emit_metric.assert_has_calls( - [ - call( - metric_object=mock_metrics_module.OPC_CONNECTION_STATUS, - method='set', - value=0, - tags={ - 'pod_id': raw_opc_manager.pod_id, - 'server_name': raw_opc_manager.name, - 'server_url': raw_opc_manager.url, - }, - ), - call( - metric_object=mock_metrics_module.OPC_TAGS_SUBSCRIBED, - method='set', - value=0, - tags={ - 'pod_id': raw_opc_manager.pod_id, - 'server_name': raw_opc_manager.name, - }, - ), - ], - any_order=True, - ) - - -@patch('ingestor.managers.opc_manager.metrics') -@mark.asyncio -async def test_datachange_notification(metrics, opc_manager_subscribed): - data = MagicMock( - monitored_item=MagicMock( - Value=MagicMock( - Value=MagicMock(Value=42), - SourceTimestamp=datetime.strptime('2021-01-01T00:00:00', '%Y-%m-%dT%H:%M:%S'), - ) - ) - ) - opc_manager_subscribed.nodes = { - 'ns=3;i=1001': { - 'tag_name': 'Counter', - 'cycle_rule': {'cycle_increment': 1.0, 'cycle_count': 2}, - 'topics': ['topic1', 'topic2'], - } - } - - await opc_manager_subscribed.datachange_notification('ns=3;i=1001', None, data) - - opc_manager_subscribed.data_manager.publish.assert_any_call( - 'topic1', - { - 'tag': 'ns=3;i=1001', - 'name': 'Counter', - 'timestamp': '2021-01-01 00:00:00-0300', - 'value': 42, - }, - ) - opc_manager_subscribed.data_manager.publish.assert_any_call( - 'topic2', - { - 'tag': 'ns=3;i=1001', - 'name': 'Counter', - 'timestamp': '2021-01-01 00:00:00-0300', - 'value': 42, - }, - ) - assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 0 - - opc_manager_subscribed.emit_metric.assert_called_once_with( - metric_object=metrics.OPC_CYCLES_WITHOUT_DATA, - method='set', - value=0, - tags={ - 'pod_id': opc_manager_subscribed.pod_id, - 'server_name': opc_manager_subscribed.name, - }, - ) - - -@mark.asyncio -async def test_check_cycles_no_notification(opc_manager): - # Setup: node with cycle_count just below threshold - opc_manager.nodes = { - 'ns=3;i=1001': { - 'tag_name': 'Counter', - 'cycle_rule': {'cycle_increment': 1.0, 'cycle_count': 3.0}, - } - } - - await opc_manager.check_cycles() - - # After one increment, cycle_count = 4.0, still below threshold - assert opc_manager.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == pytest.approx(4.0) - opc_manager.send_notification_async.assert_not_called() - - -@mark.asyncio -async def test_check_cycles_triggers_notification(opc_manager): - # Setup: node with cycle_count just below threshold, increment will cross threshold - opc_manager.nodes = { - 'ns=3;i=1001': { - 'tag_name': 'Counter', - 'cycle_rule': {'cycle_increment': 2.5, 'cycle_count': 3.0}, - } - } - opc_manager.notification_handler.build_and_send_notification = MagicMock() - - await opc_manager.check_cycles() - - # After increment, cycle_count = 5.5, should trigger notification - assert opc_manager.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == pytest.approx(5.5) - opc_manager.send_notification_async.assert_called_once_with( - notification_id='TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED', - message='5.5 cycles without receive from ns=3;i=1001:Counter', - block='opc_manager', - level=NotificationLevel.WARNING, - metadata=metadata['metadata'], - ) - - -@patch('ingestor.managers.opc_manager.metrics') -@mark.asyncio -async def test_check_opc_listenning_no_notification(metrics, opc_manager): - opc_manager.non_receive_count = 3 - - result = await opc_manager.check_opc_listenning() - - assert opc_manager.non_receive_count == 4 - opc_manager.send_notification_async.assert_not_called() - assert result is False - - opc_manager.emit_metric.assert_called_once_with( - metric_object=metrics.OPC_CYCLES_WITHOUT_DATA, - method='set', - value=opc_manager.non_receive_count, - tags={ - 'pod_id': opc_manager.pod_id, - 'server_name': opc_manager.name, - }, - ) - - -@mark.asyncio -async def test_check_opc_listenning_warning_notification(opc_manager): - opc_manager.non_receive_count = 4 - - result = await opc_manager.check_opc_listenning() - - assert opc_manager.non_receive_count == 5 - opc_manager.send_notification_async.assert_called_once_with( - notification_id=f'OPC_LISTENNING_STOPPED__{opc_manager.name}', - message=f'5 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}', - block='opc_manager', - level=NotificationLevel.ERROR, - metadata=metadata['metadata'], - ) - assert result is False - - -@patch('ingestor.managers.opc_manager.metrics') -@mark.asyncio -async def test_check_opc_listenning_error_notification_and_retry(metrics, opc_manager): - opc_manager.non_receive_count = 14 - - result = await opc_manager.check_opc_listenning() - - assert opc_manager.non_receive_count == 15 - # Should be called twice: once for 5, once for 15 - assert opc_manager.send_notification_async.call_count == 2 - calls = opc_manager.send_notification_async.call_args_list - # First call: 5 cycles warning - assert calls[0].kwargs == { - 'notification_id': f'OPC_LISTENNING_STOPPED__{opc_manager.name}', - 'message': f'15 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}', - 'block': 'opc_manager', - 'level': NotificationLevel.ERROR, - 'metadata': metadata['metadata'], - } - # Second call: 15 cycles retry - assert calls[1].kwargs == { - 'notification_id': f'OPC_CONNECTION_RETRY__{opc_manager.name}', - 'message': f'Retrying to connect to server {opc_manager.name}', - 'block': 'opc_manager', - 'level': NotificationLevel.ERROR, - 'metadata': metadata['metadata'], - } - assert result is True - - opc_manager.emit_metric.assert_has_calls( - [ - call( - metric_object=metrics.OPC_CYCLES_WITHOUT_DATA, - method='set', - value=opc_manager.non_receive_count, - tags={ - 'pod_id': opc_manager.pod_id, - 'server_name': opc_manager.name, - }, - ), - ] - ) - - opc_manager.emit_metric.assert_has_calls( - [ - call( - metric_object=metrics.OPC_RECONNECTIONS_TOTAL, - tags={ - 'pod_id': opc_manager.pod_id, - 'server_name': opc_manager.name, - }, - ), - ] - ) diff --git a/tests/unit/managers/test_server_reader.py b/tests/unit/managers/test_server_reader.py new file mode 100644 index 0000000..97b9cf6 --- /dev/null +++ b/tests/unit/managers/test_server_reader.py @@ -0,0 +1,129 @@ +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'}, + ) 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..04482a2 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 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)