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, + }, + ), + ] + )