From 0ac817a479db58edd61d6b1dc007ea21ae64d56a Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 6 Oct 2026 14:12:58 -0300 Subject: [PATCH] SIENTIAPDE-1974: feat(read): keep the legacy OpcManager as plan B (READ_ENGINE) The rc3 OpcManager is restored unchanged (with its tests and scripts/generate-and-push-opc-cert.sh) for validation and as a fallback until the end of SIENTIAPDE-1974; it is removed only once it is unused and the test suite passes the same without it. - READ_ENGINE=v2 (default): ServerReader/SubscriptionEngine over the shared pool, certificate from MinIO. READ_ENGINE=legacy: the rc3 OpcManager with its own connection, slot v1 only (a v2 slot is rejected and notified) and the certificate files of the Gitea clone. An invalid value falls back to v2 with a warning. Writes/gRPC always use the pool. - pool: OPC clients carry the pod id as application/session name and product URI (as OpcManager did), so each pod's session is recognizable in the server console; application_uri stays the server_uri (certificate SAN). - README/CLAUDE.md document the plan B and how to remove it. Tests: 275 unit tests pass (restored OpcManager tests unchanged); ruff clean. Local e2e: both engines read the same slot v1 from an asyncua server and write identical documents to the same collections. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 3 +- README.md | 22 +- ingestor/connection/pool.py | 9 + ingestor/managers/ingestor_manager.py | 109 +++- ingestor/managers/opc_manager.py | 588 +++++++++++++++++ scripts/generate-and-push-opc-cert.sh | 95 +++ tests/unit/connection/test_pool.py | 34 + tests/unit/managers/test_ingestor_manager.py | 130 ++++ tests/unit/managers/test_opc_manager.py | 627 +++++++++++++++++++ 9 files changed, 1593 insertions(+), 24 deletions(-) create mode 100644 ingestor/managers/opc_manager.py create mode 100755 scripts/generate-and-push-opc-cert.sh create mode 100644 tests/unit/managers/test_opc_manager.py diff --git a/CLAUDE.md b/CLAUDE.md index 0a9caef..44c07c9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -41,6 +41,7 @@ app.py (asyncio loop, signals, Prometheus server) -> IngestorManager (managers/ingestor_manager.py) — central coordinator -> ResourceManager (managers/resource_manager.py) — Redis slot leases, heartbeats, active-ingestor registry -> ServerReader (managers/server_reader.py), one per OPC server — SubscriptionEngine over the shared OpcConnectionPool + (READ_ENGINE=legacy: OpcManager (managers/opc_manager.py), the rc3 reader kept as plan B until the end of SIENTIAPDE-1974) -> OpcConnectionPool (connection/pool.py) + CertProvider (connection/certs.py) — pooled connections, cert from MinIO -> DataManager (managers/data_manager.py) — MongoDB persistence + optional Kafka publish ``` @@ -57,7 +58,7 @@ app.py (asyncio loop, signals, Prometheus server) Any change to slot/lease counts should be traced through `manage_leases`/`manage_no_slots` — that's where the rebalancing math lives, not in `ResourceManager` itself (which just does Redis primitives). -**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. +**ServerReader** (SIENTIAPDE-1974, default `READ_ENGINE=v2`; the legacy `OpcManager` stays as plan B with `READ_ENGINE=legacy` — slot v1 only, certificate files from the Gitea clone — until it is unused and the tests pass the same without it) reads one OPC server through a `SubscriptionEngine`: the connection comes from the shared `OpcConnectionPool` (also used by NATS writes and gRPC), which fetches the platform certificate from MinIO via `CertProvider` and uses `SecurityPolicyBasic256Sha256` — no certificate files in the repo/image. Slots are parsed by `io/slots.py` (`parse_slot`), which accepts slot v2 (`names[]`) and converts the legacy v1 (`topics`). `datachange_notification` fans each value out to every `names` entry via `DataSink`. Subscribed connections are pinned in the pool so the idle sweep never drops them. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles` / `check_opc_listenning`); a lost server is shut down (pool connection evicted) and re-initialized on the next loop. **Metrics** (`ingestor/metrics.py`): all Prometheus metrics defined here, always labeled by `pod_id`. Emitted via `SientiaMonitoring.emit_metric(...)` (from `sientia_do`), not the prometheus client directly — new metrics should follow that pattern for consistency with logging/notifications. diff --git a/README.md b/README.md index 56e5711..f062786 100644 --- a/README.md +++ b/README.md @@ -219,6 +219,25 @@ Requirements for a server to accept the connection: 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. +### Plan B: legacy read engine (`READ_ENGINE=legacy`) + +Until the end of SIENTIAPDE-1974 the legacy reader is kept for validation and as a +fallback. `READ_ENGINE` selects the read path (default `v2`; an invalid value falls +back to `v2` with a warning): + +| `READ_ENGINE` | Reader | Slot | Certificate | +|---|---|---|---| +| `v2` (default) | `ServerReader` + `SubscriptionEngine` over the shared pool | v1 or v2 | MinIO + Mongo (`CertProvider`), `Basic256Sha256` | +| `legacy` | the rc3 `OpcManager`, unchanged, with its own connection | **v1 only** (a v2 slot is rejected and notified) | files from the repository the pod clones (`cert_path` / `private_key_path` in the slot), `Basic256` | + +Writes and gRPC always use the pool (MinIO certificate), whatever the read engine. +The legacy engine needs the certificate files in the Gitea branch the pod clones; +`scripts/generate-and-push-opc-cert.sh` generates them and pushes them to +`certs/` (`GITEA_PASSWORD='' BRANCH= ./scripts/generate-and-push-opc-cert.sh`). +Each version deploy creates a new Gitea branch without `certs/`, so they have to be +copied again for the legacy engine. Remove the legacy engine (OpcManager, its tests +and the script) once it is unused and the test suite passes the same without it. + ## 🧪 Testing ### Unit Tests @@ -347,13 +366,14 @@ sientia-dataops-opc-ingestor/ │ ├── managers/ # Component managers │ │ ├── data_manager.py # Data persistence and export │ │ ├── ingestor_manager.py # Main coordination +│ │ ├── opc_manager.py # Legacy reader, plan B (READ_ENGINE=legacy) │ │ ├── resource_manager.py # Resource coordination │ │ └── server_reader.py # OPC UA server reads (SubscriptionEngine + pool) │ ├── app.py # Main application entry point │ ├── ingestor.py # Core ingestor logic │ └── metrics.py # Prometheus metrics definitions ├── simulator/ # OPC simulation and testing tools -├── scripts/ # Ops scripts (e.g. e2e seed) +├── scripts/ # Ops scripts (e2e seed; legacy OPC cert generation) ├── tests/ # Test suite └── requirements.txt # Python dependencies ``` diff --git a/ingestor/connection/pool.py b/ingestor/connection/pool.py index ea0bde6..d95abc3 100644 --- a/ingestor/connection/pool.py +++ b/ingestor/connection/pool.py @@ -105,8 +105,17 @@ class OpcConnectionPool(SientiaMonitoring): async def _connect(self, server_cfg: dict) -> Client: client = Client(server_cfg['url'], timeout=10) + # Identify the pod on the OPC server, as the legacy OpcManager did: asyncua + # sends `name` as ApplicationName and " Session" as the + # session name, so each pod's session is recognizable in the server console. + client.name = self.pod_id + client.description = self.pod_id + pod_uri = self.pod_id.replace('-', ':') + client.product_uri = pod_uri + client.application_uri = pod_uri server_uri = server_cfg.get('server_uri') if server_uri: + # Must match the certificate's SAN URI (KEPServer: BadCertificateUriInvalid). client.application_uri = server_uri if server_cfg.get('use_security'): cert, key = await self.cert_provider.fetch() diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 2061073..0be1773 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -1,5 +1,6 @@ import traceback from copy import deepcopy +from os import getenv from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel @@ -14,9 +15,16 @@ from ingestor.connection.pool import OpcConnectionPool from ingestor.io.data_sink import DataSink from ingestor.io.slots import parse_slot from ingestor.managers.data_manager import DataManager +from ingestor.managers.opc_manager import OpcManager from ingestor.managers.resource_manager import ResourceManager from ingestor.managers.server_reader import ServerReader +# READ_ENGINE (SIENTIAPDE-1974): 'v2' = ServerReader/SubscriptionEngine over the shared +# pool, certificate from MinIO (default). 'legacy' = plan B: the untouched rc3 +# OpcManager, slot v1 only, certificate files from the Gitea clone (cert_path). Kept +# for validation until the end of 1974; remove once unused and the tests pass without it. +READ_ENGINES = ('v2', 'legacy') + class IngestorManager(SientiaMonitoring): """ @@ -77,6 +85,7 @@ class IngestorManager(SientiaMonitoring): notification_handler: NotificationHandler, metrics_controller: MetricsController, gateway_config: GatewayConfig | None = None, + read_engine: str | None = None, ): redis_host: str = redis_data['host'] redis_port: int = int(redis_data['port']) @@ -116,6 +125,9 @@ class IngestorManager(SientiaMonitoring): ) self.sink = DataSink(self.data_manager, None) self._server_cert_warned: set[str] = set() + self.read_engine = self._resolve_read_engine( + read_engine or getenv('READ_ENGINE') or 'v2', logger + ) self.resource_manager = ResourceManager( host=redis_host, port=redis_port, @@ -135,35 +147,67 @@ class IngestorManager(SientiaMonitoring): self.metadata = metadata - async def initialize_opc_from_config(self, server_config: dict) -> ServerReader | None: - """ - Initializes the reader of one OPC UA server from its slot v2 configuration. + @staticmethod + def _resolve_read_engine(value: str, logger: Logger) -> str: + engine = value.strip().lower() + if engine not in READ_ENGINES: + logger.warning(f'Invalid READ_ENGINE {value!r}, using v2 ({"|".join(READ_ENGINES)})') + engine = 'v2' + logger.info(f'OPC read engine: {engine}') + return engine - Creates a ServerReader and opens its pooled connection, so the server is ready - for tag subscriptions. With `use_security`, the certificate comes from MinIO. + def _build_reader(self, server_config: dict) -> ServerReader | OpcManager: + if self.read_engine == 'legacy': + # Plan B: the rc3 OpcManager, unchanged (own connection, cert files). + return OpcManager( + name=server_config['name'], + url=server_config['url'], + subscription_period_ms=server_config['subscription_period_ms'], + data_manager=self.data_manager, + 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, + ) + return ServerReader( + name=server_config['name'], + server_config=server_config, + pool=self.pool, + sink=self.sink, + logger=self.logger, + notification_handler=self.notification_handler, + metrics_controller=self.metrics_controller, + metadata=self.metadata, + ) + + async def initialize_opc_from_config( + self, server_config: dict + ) -> ServerReader | OpcManager | None: + """ + Initializes the reader of one OPC UA server from its slot configuration. + + With READ_ENGINE=v2 (default) creates a ServerReader and opens its pooled + connection; with `use_security`, the certificate comes from MinIO. With + READ_ENGINE=legacy (plan B) creates the rc3 OpcManager, which connects on its + own with the certificate files of the slot v1 (`cert_path`). Args: - server_config (dict): The server configuration of a parsed slot (parse_slot), - without the 'tags' key. Expected keys: 'name', 'server_id', 'url', - 'server_uri', 'use_security', 'subscription_period_ms'. + server_config (dict): The server configuration of the slot, without the + 'tags' key. v2: 'name', 'server_id', 'url', 'server_uri', + 'use_security', 'subscription_period_ms'. legacy: the slot v1 fields. Returns: - ServerReader | None: The connected reader, or None if the connection fails - (logged and notified, so the other servers keep working). + ServerReader | OpcManager | None: The connected reader, or None if the + connection fails (logged and notified, so the other servers keep working). """ try: self.logger.info(f'Initializing OPC reader at {server_config["url"]}') - manager = ServerReader( - name=server_config['name'], - server_config=server_config, - pool=self.pool, - sink=self.sink, - logger=self.logger, - notification_handler=self.notification_handler, - metrics_controller=self.metrics_controller, - metadata=self.metadata, - ) + manager = self._build_reader(server_config) manager.config = server_config await manager.connect() @@ -210,7 +254,11 @@ class IngestorManager(SientiaMonitoring): the shared connection stays for writes/gRPC until the idle sweep. """ if server in self.opc_managers: - await self.opc_managers[server].shutdown(evict=evict) + if self.read_engine == 'legacy': + # OpcManager owns its connection: shutdown always disconnects it. + await self.opc_managers[server].shutdown() + else: + await self.opc_managers[server].shutdown(evict=evict) del self.opc_managers[server] for slot, _config in self.managed_tags.items(): self.managed_tags[slot].pop(server, None) @@ -560,13 +608,18 @@ class IngestorManager(SientiaMonitoring): 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). + Orchestrator can be deployed in any order (SIENTIAPDE-1974). With + READ_ENGINE=legacy the raw slot v1 is kept as is (what OpcManager reads), and + a slot v2 is rejected. Returns: dict | None: The parsed slot, or None when the config is malformed (logged and notified). """ try: + if self.read_engine == 'legacy': + self._validate_slot_v1(raw) + return raw parsed = parse_slot(raw) except (AttributeError, KeyError, TypeError, ValueError) as e: self.send_notification( @@ -581,6 +634,18 @@ class IngestorManager(SientiaMonitoring): self._warn_server_cert_path(raw) return parsed + @staticmethod + def _validate_slot_v1(raw: dict) -> None: + """READ_ENGINE=legacy: OpcManager only understands slot v1 (`tag_name` + + `topics` per tag); a v2 slot (`names`) would fail in its datachange callback.""" + for name, server in raw.items(): + for address, tag in server['tags'].items(): + if 'topics' not in tag or 'tag_name' not in tag or 'frequency' not in tag: + raise ValueError( + f'READ_ENGINE=legacy needs a slot v1 (tag_name + topics); ' + f'server {name!r}, tag {address!r} is not' + ) + def _warn_server_cert_path(self, raw: dict) -> None: """Logs once per server when a slot still pins a server certificate (`server_cert_path`): connections now come from the pool with the platform diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py new file mode 100644 index 0000000..c08d5ab --- /dev/null +++ b/ingestor/managers/opc_manager.py @@ -0,0 +1,588 @@ +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/scripts/generate-and-push-opc-cert.sh b/scripts/generate-and-push-opc-cert.sh new file mode 100755 index 0000000..713965e --- /dev/null +++ b/scripts/generate-and-push-opc-cert.sh @@ -0,0 +1,95 @@ +#!/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_pool.py b/tests/unit/connection/test_pool.py index e1bf4fa..d23dc03 100644 --- a/tests/unit/connection/test_pool.py +++ b/tests/unit/connection/test_pool.py @@ -308,3 +308,37 @@ async def test_insecure_connect_failure_keeps_cert_cache(): await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False}) cert_provider.invalidate.assert_not_called() + + +@pytest.mark.asyncio +async def test_connect_identifies_the_pod_on_the_server(): + """Session/application name = pod id, as the legacy OpcManager did, so each pod's + session is recognizable in the KEPServer console.""" + pool = make_pool() + fake_client = MagicMock(connect=AsyncMock()) + with patch('ingestor.connection.pool.Client', return_value=fake_client): + await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False}) + + assert fake_client.name == pool.pod_id + assert fake_client.description == pool.pod_id + assert fake_client.product_uri == pool.pod_id.replace('-', ':') + assert fake_client.application_uri == pool.pod_id.replace('-', ':') + + +@pytest.mark.asyncio +async def test_connect_server_uri_still_wins_for_application_uri(): + pool = make_pool() + fake_client = MagicMock(connect=AsyncMock()) + with patch('ingestor.connection.pool.Client', return_value=fake_client): + await pool.get( + 'srv', + { + 'url': 'opc.tcp://h:4840', + 'use_security': False, + 'server_uri': 'urn:sientia:opc-ingestor', + }, + ) + + # must match the certificate SAN URI (BadCertificateUriInvalid otherwise) + assert fake_client.application_uri == 'urn:sientia:opc-ingestor' + assert fake_client.name == pool.pod_id diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index c7dfa23..df27f66 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -792,3 +792,133 @@ def test_parse_slot_config_warns_once_about_server_cert_path(ingestor_manager): c for c in ingestor_manager.logger.warning.call_args_list if 'server_cert_path' in str(c) ] assert len(warnings) == 1 + + +# --- READ_ENGINE=legacy: plan B with the untouched rc3 OpcManager ------------------- + + +def make_manager(read_engine=None): + with ( + patch('ingestor.managers.ingestor_manager.DataManager'), + patch('ingestor.managers.ingestor_manager.ResourceManager'), + ): + manager = IngestorManager( + redis_data={'host': 'localhost', 'port': 6379}, + lease_ttl=60, + heartbeat_ttl=60, + poll_interval=5, + mongo_connection_string='mongodb://localhost:27017', + mongo_database='sientia', + logger=MagicMock(), + notification_handler=MagicMock(), + metadata=metadata['metadata'], + metrics_controller=MagicMock(), + read_engine=read_engine, + ) + manager.send_notification = MagicMock() + manager.emit_metric = AsyncMock() + return manager + + +def test_read_engine_defaults_to_v2(monkeypatch): + monkeypatch.delenv('READ_ENGINE', raising=False) + + assert make_manager().read_engine == 'v2' + + +def test_read_engine_from_env(monkeypatch): + monkeypatch.setenv('READ_ENGINE', 'Legacy ') + + assert make_manager().read_engine == 'legacy' + + +def test_invalid_read_engine_falls_back_to_v2(monkeypatch): + monkeypatch.setenv('READ_ENGINE', 'v3') + + manager = make_manager() + + assert manager.read_engine == 'v2' + assert any('Invalid READ_ENGINE' in str(c) for c in manager.logger.warning.call_args_list) + + +@mark.asyncio +@patch('ingestor.managers.ingestor_manager.ServerReader') +@patch('ingestor.managers.ingestor_manager.OpcManager') +async def test_legacy_initialize_builds_rc3_opc_manager(opc_manager, server_reader): + manager = make_manager('legacy') + server_config = { + 'name': 'server1', + 'server_id': '1', + 'url': 'opc.tcp://localhost:4840', + 'subscription_period_ms': 1000, + 'server_uri': 'urn:sientia:opc-ingestor', + 'cert_path': '/app/certs/opc-ingestor-cert.pem', + 'private_key_path': '/app/certs/opc-ingestor-key.pem', + 'server_cert_path': None, + } + opc_manager.return_value = MagicMock(connect=AsyncMock()) + + result = await manager.initialize_opc_from_config(server_config) + + server_reader.assert_not_called() + opc_manager.assert_called_once_with( + name='server1', + url='opc.tcp://localhost:4840', + subscription_period_ms=1000, + data_manager=manager.data_manager, + logger=manager.logger, + server_uri='urn:sientia:opc-ingestor', + notification_handler=manager.notification_handler, + metadata=metadata['metadata'], + cert_path='/app/certs/opc-ingestor-cert.pem', + private_key_path='/app/certs/opc-ingestor-key.pem', + server_cert_path=None, + metrics_controller=manager.metrics_controller, + ) + assert result is opc_manager.return_value + result.connect.assert_awaited_once() + + +def test_legacy_keeps_raw_slot_v1(): + manager = make_manager('legacy') + raw = raw_slot_v1() + + # OpcManager reads tag_name/topics and the file cert paths straight from the slot + assert manager.parse_slot_config('1', raw) is raw + + +def test_legacy_rejects_slot_v2(): + manager = make_manager('legacy') + raw_v2 = { + 'server1': { + 'name': 'server1', + 'server_id': '1', + 'url': 'opc.tcp://localhost:4840', + 'use_security': True, + 'subscription_period_ms': 500, + 'tags': { + 'ns=2;i=2': { + 'frequency': '1000', + 'names': [{'collection': 'raw_a', 'tag_name': 'Counter', 'model_id': '1'}], + } + }, + } + } + + assert manager.parse_slot_config('1', raw_v2) is None + _, kwargs = manager.send_notification.call_args + assert kwargs['notification_id'] == 'OPC_SLOT_CONFIG_ERROR_1' + assert 'READ_ENGINE=legacy' in kwargs['message'] + + +@mark.asyncio +async def test_legacy_remove_server_calls_plain_shutdown(): + manager = make_manager('legacy') + opc_manager = MagicMock(shutdown=AsyncMock()) + manager.opc_managers = {'server1': opc_manager} + manager.managed_tags = {'slot1': {'server1': {}}} + + await manager.remove_server('server1', evict=False) + + opc_manager.shutdown.assert_awaited_once_with() + assert manager.opc_managers == {} diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py new file mode 100644 index 0000000..407c166 --- /dev/null +++ b/tests/unit/managers/test_opc_manager.py @@ -0,0 +1,627 @@ +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, + }, + ), + ] + )