diff --git a/.gitignore b/.gitignore index 7cd11b3..c35c913 100644 --- a/.gitignore +++ b/.gitignore @@ -51,4 +51,6 @@ catboost_info/ .ruff_cache/ .mypy_cache/ -mlruns/ \ No newline at end of file +mlruns/ + +relatorio* \ No newline at end of file diff --git a/README.md b/README.md index b63be57..0de066f 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,7 @@ A comprehensive, Temporal-based ML orchestration system for industrial data proc - [Prediction Operation Metrics](#prediction-operation-metrics) - [OPC Export Metrics](#opc-export-metrics) - [Data Quality Metrics](#data-quality-metrics) +- [OPC UA Communication](#opc-ua-communication) - [Configuration](#configuration-1) - [Environment Variables](#environment-variables) - [OPC Configuration](#opc-configuration) @@ -164,7 +165,7 @@ Laborious uses a Temporal-based architecture with strong separation of concerns - `connectors_config.py`: Env-driven configuration builders - `models/minio_dataframe_payload.py`: MinIO-offloaded DataFrame payload model - `repository/model_repository.py`: MLFlow operations and retraining -- `repository/opc_repository.py`: OPC communication and writes +- `repository/opc_repository.py`: OPC UA client, writes, session recovery (see [OPC UA Communication](#opc-ua-communication)) - `repository/minio_manager.py`: MinIO object storage operations - `filters/conditional_filters.py` and `filters/mlflow_filters.py` @@ -779,6 +780,15 @@ The Laborious system exposes comprehensive Prometheus metrics for operational vi - Labels: `pod_id`, `model_name`, `workflow_name`, `opc_server_id` - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] +OPC UA session and write diagnostics (Prometheus, `laborious/metrics.py`): + +- `opc_connections_initiated_total`, `opc_connections_failed_total`, `opc_connection_status` +- `opc_session_created_total`, `opc_session_closed_total`, `opc_session_revised_timeout_milliseconds` +- `opc_write_attempts_total` (label `result`: `OK` or exception name, e.g. `BadSessionIdInvalid`) +- `opc_write_inter_arrival_over_session_timeout_total` + +See [OPC UA Communication](#opc-ua-communication) for semantics, concurrency, and confidence codes **12** / **14**. + ### Data Quality Metrics - Filter pass/fail rates through notification system - MLFlow API response validation metrics @@ -810,7 +820,7 @@ The Laborious system exposes comprehensive Prometheus metrics for operational vi | `OPC_CERT_PATH` | OPC client certificate path | `None` | No | | `OPC_PRIVATE_KEY_PATH` | OPC private key path | `None` | No | | `OPC_SERVER_CERT_PATH` | OPC server certificate path | `None` | No | -| `OPC_RECONNECTION_INTERVAL` | OPC reconnection interval (ms) | `120` | No | +| `OPC_RECONNECTION_INTERVAL` | Minimum seconds between OPC reconnects | `120` | No | | `PI_WEB_API_BASE_URL` | PI Web API server base URL | `None` | No | | `PI_WEB_API_AUTH_TYPE` | PI Web API authentication type (basic/bearer) | `None` | No | | `PI_WEB_API_AUTH_TOKEN` | PI Web API authentication token | `None` | No | @@ -903,6 +913,12 @@ Legacy MinIO object layout (relative key): `training_datasets/{model_name}/{object_prefix}_{timestamp}.parquet` where `object_prefix` is sanitized (slashes replaced by underscores) to keep a stable model-level directory. +## OPC UA Communication + +Full reference: **[docs/opc-communication.md](docs/opc-communication.md)** (connection lifecycle, Tier-1 `Bad*` reconnect, connection lock / session readiness, metrics, PostgreSQL confidence **12** vs **14**, tests). + +Implementation plan: [`.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md`](.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md). + ### OPC Configuration For multiple OPC servers, use the `OPC_CONFIG` environment variable: @@ -1126,8 +1142,10 @@ laborious/ - Ensure proper connection pool configuration 4. **OPC Connection Failures** - - Verify OPC server is accessible + - See [docs/opc-communication.md](docs/opc-communication.md) + - Verify OPC server is accessible and `OPC_RECONNECTION_INTERVAL` is appropriate - Check certificate and key file paths + - Correlate `opc_write_attempts_total` with `opc_session_*` metrics; count session errors via `prediction_confidence = 14` - Review OPC server logs for connection issues 5. **PI Web API Connection Failures** diff --git a/docs/opc-communication.md b/docs/opc-communication.md new file mode 100644 index 0000000..3084d05 --- /dev/null +++ b/docs/opc-communication.md @@ -0,0 +1,162 @@ +# OPC UA communication (Laborious) + +Laborious exports predictions to OPC UA servers through `OpcRepository` ([`laborious/utils/repository/opc_repository.py`](../laborious/utils/repository/opc_repository.py)) and the Temporal activity layer in [`laborious/activities/opc.py`](../laborious/activities/opc.py). + +Implementation plan for session/channel recovery on Tier-1 `Bad*` errors: [`.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md`](../.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md). + +## Architecture + +```text +Worker (long-lived) + └── OpcRepository per OPC server id (from OPC_CONFIG / env) + ├── connect / disconnect / validate_connection (read-only) + ├── _connect_locked / _reconnect_locked (under _connection_lock) + ├── write_data (single attempt per call) + └── background reconnect on Tier-1 Bad* + +Temporal activity write_opc_data + └── OPC.manage_output_tags → write_data per tag (sequential per activity) +``` + +One worker process holds one `OpcRepository` instance per configured server. Multiple Temporal activities can call `write_data` concurrently on the same repository. + +## Connection lifecycle + +| Phase | Behavior | +|-------|----------| +| Startup | `init_opc()` creates repositories and calls `connect()` → `_connect_locked()` | +| Steady state | `validate_connection()` is read-only (`protocol.state` only); `_session_ready` is checked in `write_data` | +| Tier-1 Bad* | `_start_reconnect_on_bad` → `_run_reconnect_on_bad` → `_reconnect_locked()` (respects `reconnection_interval`) | +| Write | `write_data()` checks `_session_ready`, validates, then one `get_node` + `write_value` | +| Shutdown | `close()` disconnects all repositories | + +### Session and channel timeouts + +Requested session and secure-channel lifetime: **10 minutes** (`OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS` in `opc_repository.py`). The server may revise these values; negotiated values are logged after connect and exposed as `opc_session_revised_timeout_milliseconds`. + +### Reconnection interval + +`OPC_RECONNECTION_INTERVAL` is in **seconds** (default `120`). It gates **background** reconnect after Tier-1 `Bad*` (`last_reconnection_time` is updated only in `_reconnect_locked()`). It limits load on the OPC server when many workflows fail at once. + +## Concurrency: connection lock and session readiness + +To allow **multiple concurrent writes** when the session is healthy, but **block all writes** while the connection is being torn down or re-established: + +| Primitive | Role | +|-----------|------| +| `_connection_lock` (`asyncio.Lock`) | Held for the entire `disconnect` → `connect` path. Only one connection-maintenance task at a time. | +| `_session_ready` (`asyncio.Event`) | Set when a session is ready for writes; cleared before reconnect starts and set again after a successful connect. | + +**Connection methods (caller holds `_connection_lock` for `_*_locked` helpers):** + +| Method | Role | +|--------|------| +| `_create_client()` | Create asyncua `Client` + optional `set_security`; raises if `client` already exists | +| `_open_session()` | `client.connect()` + metrics; raises if session already open or client missing | +| `_connect_locked()` | `_create_client()` (when needed) + `_open_session()`; raises if already connected | +| `_disconnect_locked()` | Teardown session and clear `client` | +| `_reconnect_locked()` | `_disconnect_locked()` + `_connect_locked()`; sets `last_reconnection_time` | + +Public `connect()` / `disconnect()` acquire the lock and call `_connect_locked()` / `_disconnect_locked()`. + +**Write path (`write_data`):** + +1. If `_session_ready` is cleared → **fail immediately** (`opc_error_kind=reconnect_in_progress`). +2. `validate_connection()` checks `protocol.state` only (read-only). +3. Single `get_node` + `write_value` (no retry). + +**Reconnect path (`_run_reconnect_on_bad`):** + +1. `_start_reconnect_on_bad` clears `_session_ready` and schedules the task when the interval allows. +2. `async with _connection_lock:` → `_reconnect_locked()`. +3. `_session_ready` is set on successful `_open_session()`. + +A second `_connect_locked()` while a session is already open raises `OpcSessionAlreadyConnectedError` (disconnect first). + +**asyncua note:** Concurrent `write_value` on the same session is only safe if the stack tolerates it. If production shows issues, serialize writes with an optional `asyncio.Semaphore(1)` while keeping the connection lock semantics above. + +**Future threads:** replace `asyncio.Lock` / `Event` with `threading` primitives or route all OPC I/O through one dedicated loop. + +## Tier-1 `Bad*` errors and reconnect + +When the server invalidates the session (e.g. `BadSessionIdInvalid`) but the client still sees transport as open, `write_data` fails once, records the OPC status in metrics, and **schedules** reconnect if: + +- The exception is a `UaStatusCodeError` whose name is in `RECONNECTABLE_OPC_BAD_NAMES` (see plan), and +- `reconnection_interval` has elapsed since `last_reconnection_time`, and +- No reconnect task is already running. + +There is **no write retry**: the failed export is not sent again in the same activity. + +## Prediction confidence and PostgreSQL comments + +| `prediction_confidence` | Meaning | +|-------------------------|---------| +| (unchanged) | Successful OPC export | +| **12** | Generic OPC write failure (`OPC_WRITTING_ERROR_CONFIDENCE`) | +| **14** | Tier-1 session/channel `Bad*` on export (`OPC_SESSION_BAD_CONFIDENCE`) | +| **14** | Write while reconnect in progress (`OPC_SESSION_BAD_CONFIDENCE`, comment `OPC UA reconnect in progress`) | +| **13** | PI Web API write failure (separate path) | + +Session/channel errors use a stable comment for counting: + +```text +OPC UA session/channel error: BadSessionIdInvalid +``` + +Reconnect-in-progress exports use: + +```text +OPC UA reconnect in progress +``` + +Example SQL: + +```sql +SELECT count(*) FROM predictions WHERE prediction_confidence = 14; +SELECT count(*) FROM predictions WHERE comments LIKE 'OPC UA session/channel error:%'; +``` + +## Prometheus metrics (`opc_*`) + +Defined in [`laborious/metrics.py`](../laborious/metrics.py). Do not rename in production without a dashboard migration. + +| Metric | Purpose | +|--------|---------| +| `opc_connections_initiated_total` | Connection attempts | +| `opc_connections_failed_total` | Failed connects | +| `opc_connection_status` | Gauge 1=connected, 0=disconnected | +| `opc_session_created_total` | Session established after connect | +| `opc_session_closed_total` | Disconnect initiated | +| `opc_session_revised_timeout_milliseconds` | Negotiated session timeout (ms) | +| `opc_write_attempts_total` | Per write; label `result` = `OK` or exception name | +| `opc_write_inter_arrival_over_session_timeout_total` | Successful writes spaced longer than revised session timeout | + +Legacy activity metrics: `laborious_prediction_opc_writing_count`, `laborious_prediction_opc_writing_response_time_monitor`. + +## Environment variables + +| Variable | Default | Description | +|----------|---------|-------------| +| `OPC_CONFIG` | — | JSON map of server configs (overrides single-server env) | +| `OPC_ID` | `1` | Server id | +| `OPC_URL` | `opc.tcp://localhost:4840` | Endpoint | +| `OPC_SERVER_NAME` | `default_server` | Label for metrics/logs | +| `OPC_SERVER_URI` | same as URL | Application URI / cert SAN | +| `OPC_CERT_PATH` | — | Client certificate (secure mode) | +| `OPC_PRIVATE_KEY_PATH` | — | Client private key | +| `OPC_SERVER_CERT_PATH` | — | Server certificate | +| `OPC_RECONNECTION_INTERVAL` | `120` | Minimum seconds between reconnects | + +## Operations checklist + +- Correlate `BadSessionIdInvalid` in `opc_write_attempts_total` with `opc_session_closed_total` / `opc_session_created_total` (reconnect may finish after the row is stored with confidence 14). +- Use confidence **14** and comment prefix for session invalidation rates; use **12** for other OPC failures. +- Respect `OPC_RECONNECTION_INTERVAL` under parallel load; bursts of confidence 14 are expected until the next successful cycle. + +## Related tests + +- Unit: [`tests/laborious/utils/repository/test_opc_repository.py`](../tests/laborious/utils/repository/test_opc_repository.py) +- Unit: [`tests/laborious/activities/test_opc.py`](../tests/laborious/activities/test_opc.py) +- E2E (mock OPC): [`e2e/test_predictions_batch_format_export.py`](../e2e/test_predictions_batch_format_export.py) +- E2E (in-process asyncua server + real `OpcRepository`): [`e2e/test_opc_real_server.py`](../e2e/test_opc_real_server.py) — scenarios 3.1.2, 3.2.2, 3.2.4, 3.2.5 +- Scenarios: [`e2e/scenarios.md`](../e2e/scenarios.md) diff --git a/e2e/conftest.py b/e2e/conftest.py index 7825335..cc49d9b 100644 --- a/e2e/conftest.py +++ b/e2e/conftest.py @@ -2,7 +2,14 @@ Pytest configuration and fixtures for E2E tests. """ +import sys from unittest.mock import AsyncMock, MagicMock, patch + +# E2E workflows under test do not run ModelAnalysis; stub before Activities import. +_model_analysis_module = MagicMock() +_model_analysis_module.ModelAnalysis = MagicMock +sys.modules.setdefault('sientia', MagicMock()) +sys.modules.setdefault('sientia.ModelAnalysis', _model_analysis_module) from io import BytesIO import pandas as pd @@ -14,6 +21,7 @@ from testcontainers.postgres import PostgresContainer from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker +from e2e.opc_test_server import OpcE2ETestServer from laborious.activities.activities import Activities from laborious.workflows.predictions_batch import PredictionsBatch from laborious.workflows.sub_workflows.prediction_process import PredictionProcess @@ -297,6 +305,19 @@ def mock_opc_repository(): mock_repo.disconnect = AsyncMock() return mock_repo + +@pytest_asyncio.fixture +async def opc_e2e_server(): + """ + In-process asyncua OPC UA server for E2E tests against OpcRepository. + """ + server = OpcE2ETestServer() + await server.start() + try: + yield server + finally: + await server.stop() + @pytest_asyncio.fixture def patch_create_engine(postgres_engine): """Patch create_engine to return test postgres_engine.""" @@ -577,3 +598,84 @@ async def temporal_worker_real_minio(temporal_test_env, test_activities_real_min activities=_worker_activity_list(test_activities_real_minio), ) as worker: yield worker + + +@pytest_asyncio.fixture(scope='function') +async def test_activities_real_opc( + postgres_engine, + postgres_container, + opc_e2e_server: OpcE2ETestServer, + mock_logger, + notification_handler, + metrics_controller, + patch_create_engine, + patch_minio_repository, + patch_mlflow, + patch_pi_web_api_repository, +): + """ + Activities with a real OpcRepository connected to the in-process OPC UA server. + """ + activities = Activities( + postgres_config={ + 'host': 'localhost', + 'port': postgres_container.get_exposed_port(5432), + 'user': 'test', + 'password': 'test', + 'dbname': 'test', + 'min_connections': 1, + 'max_connections': 5, + }, + mlflow_config={ + 'host': 'http://localhost', + 'port': '5000', + 'username': 'test', + 'password': 'test', + }, + minio_config={ + 'endpoint_url': 'localhost:9000', + 'access_key': 'test', + 'secret_key': 'test', + 'default_bucket': 'test-bucket', + 'retention_hours': 24, + 'secure': False, + }, + opc_config={ + '1': { + 'id': '1', + 'server_name': 'e2e-opc', + 'url': opc_e2e_server.url, + 'server_uri': opc_e2e_server.url, + 'cert_path': None, + 'private_key_path': None, + 'server_cert_path': None, + 'reconnection_interval': 0, + } + }, + pi_web_api_config={ + 'base_url': 'http://localhost:8080', + 'auth_type': 'bearer', + 'auth_token': 'test_token', + }, + logger=mock_logger, + notification_handler=notification_handler, + ) + await activities.init_opc() + repo = activities.opc_repository['1'] + assert repo._session_ready.is_set(), 'OPC E2E server connection failed during init_opc' + try: + yield activities + finally: + await activities.shutdown() + + +@pytest_asyncio.fixture(scope='function') +async def temporal_worker_real_opc(temporal_test_env, test_activities_real_opc): + """Temporal worker backed by Activities using the in-process OPC UA server.""" + async with Worker( + temporal_test_env.client, + task_queue='test-queue', + workflows=[PredictionsBatch, PredictionProcess, FormatAndExportPrediction], + activities=_worker_activity_list(test_activities_real_opc), + ) as worker: + yield worker diff --git a/e2e/helpers.py b/e2e/helpers.py index 1aaa4f2..bb404ca 100644 --- a/e2e/helpers.py +++ b/e2e/helpers.py @@ -64,7 +64,8 @@ def assert_prediction( prediction: float = 0.5, prediction_confidence: int | Decimal = 0, prediction_status: str = 'Good', - comments: str = '', + comments: str | None = None, + comments_contains: str | None = None, ) -> None: """ Assert exactly one prediction row exists for model_id with expected columns. @@ -75,7 +76,8 @@ def assert_prediction( prediction: Expected prediction value. prediction_confidence: Expected confidence (int or Decimal for numeric column). prediction_status: Expected status string. - comments: Expected comments string. + comments: Expected exact comments string (optional). + comments_contains: Substring expected in comments when queued (optional). """ import pytest @@ -98,7 +100,12 @@ def assert_prediction( str(prediction_confidence) ), f'Expected prediction_confidence={prediction_confidence}, got {row[2]}' assert row[3] == prediction_status, f"Expected prediction_status='{prediction_status}', got {row[3]}" - assert row[4] == comments, f"Expected comments='{comments}', got {row[4]}" + if comments is not None: + assert row[4] == comments, f"Expected comments='{comments}', got {row[4]}" + if comments_contains is not None: + assert comments_contains in row[4], ( + f"Expected comments to contain '{comments_contains}', got {row[4]}" + ) def assert_continue( diff --git a/e2e/opc_test_server.py b/e2e/opc_test_server.py new file mode 100644 index 0000000..69b02c8 --- /dev/null +++ b/e2e/opc_test_server.py @@ -0,0 +1,189 @@ +""" +In-process OPC UA server for E2E tests (asyncua). + +Provides writable prediction/confidence nodes and optional write faults +(Tier-1 BadSessionIdInvalid via PreWrite callback). +""" + +from __future__ import annotations + +import socket +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from asyncua import Server, ua +from asyncua.common.callback import CallbackType +from asyncua.common.utils import ServiceError + +if TYPE_CHECKING: + from asyncua.common.node import Node + + +UNKNOWN_NODE_ID = 'ns=99;i=9999' + + +@dataclass(frozen=True) +class OpcE2ENodeIds: + """NodeId strings used in opc_output_config for E2E workflows.""" + + prediction: str + confidence: str + unknown: str = UNKNOWN_NODE_ID + + +class OpcE2ETestServer: + """ + Ephemeral asyncua server with Laborious E2E variables and controllable faults. + + Args: + host: Bind address (default 127.0.0.1). + """ + + def __init__(self, host: str = '127.0.0.1') -> None: + self._host = host + self._server: Server | None = None + self._prediction_node: Node | None = None + self._confidence_node: Node | None = None + self._session_bad_on_write = False + self._url: str | None = None + self._node_ids: OpcE2ENodeIds | None = None + + @property + def url(self) -> str: + if self._url is None: + raise RuntimeError('OPC E2E server is not started') + return self._url + + @property + def node_ids(self) -> OpcE2ENodeIds: + if self._node_ids is None: + raise RuntimeError('OPC E2E server is not started') + return self._node_ids + + def set_session_bad_on_write(self, enabled: bool) -> None: + """ + When enabled, every client Write is rejected with BadSessionIdInvalid. + + Args: + enabled (bool): Turn Tier-1 session fault injection on or off. + """ + self._session_bad_on_write = enabled + + async def start(self) -> OpcE2ENodeIds: + """ + Start the OPC UA server on a free TCP port. + + Return: + OpcE2ENodeIds: NodeId strings for prediction and confidence tags. + """ + port = _free_port(self._host) + self._url = f'opc.tcp://{self._host}:{port}/freeopcua/server/' + + server = Server() + server.set_endpoint(self._url) + await server.init() + server.iserver.callback_service.addListener( + CallbackType.PreWrite, + self._pre_write_callback, + ) + + idx = await server.register_namespace('http://sientia.test/laborious-e2e') + e2e_object = await server.nodes.objects.add_object(idx, 'LaboriousE2E') + prediction = await e2e_object.add_variable( + idx, + 'Prediction', + ua.Variant(0.0, ua.VariantType.Float), + ) + confidence = await e2e_object.add_variable( + idx, + 'Confidence', + ua.Variant(0.0, ua.VariantType.Float), + ) + await prediction.set_writable() + await confidence.set_writable() + + await server.start() + self._server = server + self._prediction_node = prediction + self._confidence_node = confidence + self._node_ids = OpcE2ENodeIds( + prediction=prediction.nodeid.to_string(), + confidence=confidence.nodeid.to_string(), + ) + return self._node_ids + + async def stop(self) -> None: + """Stop the OPC UA server and release the listening port.""" + if self._server is not None: + await self._server.stop() + self._server = None + self._prediction_node = None + self._confidence_node = None + self._url = None + self._node_ids = None + self._session_bad_on_write = False + + async def read_prediction(self) -> float: + """ + Read the current prediction variable value from the address space. + + Return: + float: Stored prediction value. + """ + if self._prediction_node is None: + raise RuntimeError('OPC E2E server is not started') + value = await self._prediction_node.read_value() + return float(value) + + async def read_confidence(self) -> float: + """ + Read the current confidence variable value from the address space. + + Return: + float: Stored confidence value. + """ + if self._confidence_node is None: + raise RuntimeError('OPC E2E server is not started') + value = await self._confidence_node.read_value() + return float(value) + + async def _pre_write_callback(self, _event, _service) -> None: + if self._session_bad_on_write: + raise ServiceError(ua.StatusCodes.BadSessionIdInvalid) + + +def _free_port(host: str) -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind((host, 0)) + return int(sock.getsockname()[1]) + + +def build_opc_output_config( + node_ids: OpcE2ENodeIds, + *, + prediction_tag: str | None = None, + confidence_tag: str | None = None, + prediction_only: bool = False, + server_key: str = '1', +) -> dict[str, dict]: + """ + Build opc_output_config for PredictionsBatch using real server NodeIds. + + Args: + node_ids (OpcE2ENodeIds): Node ids from OpcE2ETestServer. + prediction_tag (str | None): Override prediction NodeId (default: node_ids.prediction). + confidence_tag (str | None): Override confidence NodeId (default: node_ids.confidence). + prediction_only (bool): When True, omit confidence_tags (single write per activity). + server_key (str): OPC server id key in opc_output_config. + + Return: + dict: opc_output_config payload for workflow input. + """ + pred = prediction_tag if prediction_tag is not None else node_ids.prediction + conf = confidence_tag if confidence_tag is not None else node_ids.confidence + server_config: dict = { + 'prediction_tags': {pred: {'data_type': 'float'}}, + } + if not prediction_only: + server_config['confidence_tags'] = {conf: {'data_type': 'float'}} + return {server_key: server_config} diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 5732cd3..8841424 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -8,6 +8,8 @@ This document describes all possible test scenarios for the `predictions_batch` - **Dependencies**: install dev requirements (includes `testcontainers[postgres,minio]`). - **Invocation**: run only integration-marked tests, for example: `pytest e2e/ -m integration`. - **MinIO tests**: `e2e/test_minio_offload.py` exercises real S3 uploads; other E2E modules continue to mock MinIO on the worker used by most scenarios. +- **OPC tests (real server)**: `e2e/test_opc_real_server.py` uses an in-process **asyncua** server and real `OpcRepository` (`test_activities_real_opc`). Scenarios 3.1.2, 3.2.2, 3.2.4, and 3.2.5 are covered there. Other E2E modules keep the OPC mock. +- Run only OPC real-server tests: `pytest e2e/test_opc_real_server.py -m "integration and opc"`. ## Workflow Overview @@ -495,6 +497,48 @@ These paths do **not** rely on Temporal activity retries for export failures: th --- +#### Scenario 3.2.4: OPC Session / Channel Bad* (Tier-1) +**Description**: OPC write fails with a Tier-1 session or channel status (e.g. `BadSessionIdInvalid`) while transport may still appear open on the client + +**Input**: +- Valid prediction and OPC output config +- Mock or server returning Tier-1 `UaStatusCodeError` on write (no write retry in the same activity) + +**Expected Behavior**: +- `write_opc_data` fails forward for affected tags; background reconnect may be scheduled if `OPC_RECONNECTION_INTERVAL` allows +- Workflow **completes** +- PostgreSQL row uses **`prediction_confidence` 14** and comment prefix `OPC UA session/channel error:` (including OPC status name) +- `opc_write_attempts_total` records `result=BadSessionIdInvalid` (or matching status); no second write attempt in the same activity + +**Assertions**: +- Workflow completes +- `prediction_confidence = 14` +- `comments` matches `OPC UA session/channel error:%` +- Generic OPC error confidence **12** is not used for this case + +**Reference**: [docs/opc-communication.md](../docs/opc-communication.md), plan `.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md` + +--- + +#### Scenario 3.2.5: OPC Write Blocked During Reconnect +**Description**: A write is attempted while the repository is reconnecting (session not ready) + +**Input**: +- Valid prediction +- Simulated slow reconnect (e.g. delayed `connect`) or concurrent writes where the first triggers reconnect + +**Expected Behavior**: +- Second write (or parallel write) is rejected **immediately** when reconnect is in progress or `_session_ready` is cleared — **without** calling `write_value` +- No wait/sleep on the write path; no duplicate `connect` from parallel writers (connection lock) +- `prediction_confidence = 14`, `comments = OPC UA reconnect in progress` (distinguish from Tier-1 `Bad*` via comment prefix in SQL) + +**Assertions**: +- At most one reconnect sequence (`disconnect` + `connect`) for the overlapping window +- No write retry after failure +- Tests in `test_opc_repository` (unit) and optional e2e in `test_predictions_batch_format_export.py` + +--- + #### Scenario 3.2.3: PI Web API Partial Write Error **Description**: Two prediction tags attempt to be written to PI Web API, but only one succeeds diff --git a/e2e/test_opc_real_server.py b/e2e/test_opc_real_server.py new file mode 100644 index 0000000..200a8a4 --- /dev/null +++ b/e2e/test_opc_real_server.py @@ -0,0 +1,194 @@ +""" +E2E tests for OPC export using an in-process asyncua server and real OpcRepository. + +Covers scenarios 3.1.2, 3.2.2, 3.2.4, and 3.2.5 from e2e/scenarios.md. +Mock-based OPC tests remain in test_predictions_batch_format_export.py. +""" + +import asyncio + +import pytest +from temporalio.testing import WorkflowEnvironment +from temporalio.worker import Worker + +from e2e.helpers import assert_prediction, insert_sample_data, make_workflow_id, start_and_await_workflow +from e2e.opc_test_server import UNKNOWN_NODE_ID, OpcE2ETestServer, build_opc_output_config +from e2e.test_predictions_batch_format_export import get_base_input_data +from laborious.activities.activities import Activities +from laborious.activities.opc import OPC_RECONNECT_IN_PROGRESS_COMMENT +from laborious.utils.repository.opc_repository import OpcRepository +from laborious.workflows.predictions_batch import PredictionsBatch + + +async def _slow_reconnect_under_lock(repo: OpcRepository, hold_seconds: float = 0.75) -> None: + """ + Hold the connection lock briefly so concurrent writes see reconnect_in_progress. + + Args: + repo (OpcRepository): Connected repository. + hold_seconds (float): Time to keep the lock before reconnecting. + """ + async with repo._connection_lock: + await asyncio.sleep(hold_seconds) + await repo._reconnect_locked() + + +@pytest.mark.asyncio +@pytest.mark.integration +@pytest.mark.opc +async def test_scenario_3_1_2_export_with_opc_only_real_server( + temporal_test_env: WorkflowEnvironment, + temporal_worker_real_opc: Worker, + test_activities_real_opc: Activities, + opc_e2e_server: OpcE2ETestServer, + postgres_engine, +): + """ + Scenario 3.1.2 (real OPC): connect, write prediction and confidence, verify server values. + """ + client = temporal_test_env.client + model_id = 412 + insert_sample_data(postgres_engine, model_id, [23.5, 78.2]) + + input_data = get_base_input_data(model_id) + input_data['opc_output_config'] = build_opc_output_config(opc_e2e_server.node_ids) + input_data['pi_web_api_output_config'] = None + + await start_and_await_workflow( + client, + PredictionsBatch.run, + input_data, + make_workflow_id('test-opc-real-happy'), + ) + + test_activities_real_opc.pi_web_api_client.write_value.assert_not_called() + assert await opc_e2e_server.read_prediction() == pytest.approx(0.5) + assert await opc_e2e_server.read_confidence() == pytest.approx(0.0) + assert_prediction(postgres_engine, model_id) + + +@pytest.mark.asyncio +@pytest.mark.integration +@pytest.mark.opc +async def test_scenario_3_2_2_opc_write_error_real_server( + temporal_test_env: WorkflowEnvironment, + temporal_worker_real_opc: Worker, + test_activities_real_opc: Activities, + opc_e2e_server: OpcE2ETestServer, + postgres_engine, +): + """ + Scenario 3.2.2 (real OPC): unknown NodeId yields generic write failure (confidence 12). + """ + client = temporal_test_env.client + model_id = 422 + insert_sample_data(postgres_engine, model_id, [23.5, 78.2]) + + node_ids = opc_e2e_server.node_ids + input_data = get_base_input_data(model_id) + input_data['opc_output_config'] = build_opc_output_config( + node_ids, + prediction_tag=UNKNOWN_NODE_ID, + confidence_tag=UNKNOWN_NODE_ID, + ) + input_data['pi_web_api_output_config'] = None + + await start_and_await_workflow( + client, + PredictionsBatch.run, + input_data, + make_workflow_id('test-opc-real-bad-node'), + ) + + assert_prediction( + postgres_engine, + model_id, + prediction_confidence=12, + comments='Some data could not be written to OPC servers', + ) + + +@pytest.mark.asyncio +@pytest.mark.integration +@pytest.mark.opc +async def test_scenario_3_2_4_opc_session_bad_real_server( + temporal_test_env: WorkflowEnvironment, + temporal_worker_real_opc: Worker, + test_activities_real_opc: Activities, + opc_e2e_server: OpcE2ETestServer, + postgres_engine, +): + """ + Scenario 3.2.4 (real OPC): server PreWrite fault injects BadSessionIdInvalid (confidence 14). + """ + client = temporal_test_env.client + model_id = 424 + insert_sample_data(postgres_engine, model_id, [23.5, 78.2]) + + opc_e2e_server.set_session_bad_on_write(True) + try: + input_data = get_base_input_data(model_id) + input_data['opc_output_config'] = build_opc_output_config( + opc_e2e_server.node_ids, + prediction_only=True, + ) + input_data['pi_web_api_output_config'] = None + + await start_and_await_workflow( + client, + PredictionsBatch.run, + input_data, + make_workflow_id('test-opc-real-session-bad'), + ) + finally: + opc_e2e_server.set_session_bad_on_write(False) + + assert_prediction( + postgres_engine, + model_id, + prediction_confidence=14, + comments_contains='OPC UA session/channel error: BadSessionIdInvalid', + ) + + +@pytest.mark.asyncio +@pytest.mark.integration +@pytest.mark.opc +async def test_scenario_3_2_5_opc_write_blocked_during_reconnect_real_server( + temporal_test_env: WorkflowEnvironment, + temporal_worker_real_opc: Worker, + test_activities_real_opc: Activities, + opc_e2e_server: OpcE2ETestServer, + postgres_engine, +): + """ + Scenario 3.2.5 (real OPC): writes rejected while reconnect holds the connection lock. + """ + client = temporal_test_env.client + model_id = 425 + insert_sample_data(postgres_engine, model_id, [23.5, 78.2]) + + repo = test_activities_real_opc.opc_repository['1'] + repo._session_ready.clear() + reconnect_task = asyncio.create_task(_slow_reconnect_under_lock(repo)) + + input_data = get_base_input_data(model_id) + input_data['opc_output_config'] = build_opc_output_config(opc_e2e_server.node_ids) + input_data['pi_web_api_output_config'] = None + + try: + await start_and_await_workflow( + client, + PredictionsBatch.run, + input_data, + make_workflow_id('test-opc-real-reconnect-block'), + ) + finally: + await reconnect_task + + assert_prediction( + postgres_engine, + model_id, + prediction_confidence=14, + comments_contains=OPC_RECONNECT_IN_PROGRESS_COMMENT, + ) diff --git a/e2e/test_predictions_batch_format_export.py b/e2e/test_predictions_batch_format_export.py index 35543f8..5582d0c 100644 --- a/e2e/test_predictions_batch_format_export.py +++ b/e2e/test_predictions_batch_format_export.py @@ -153,7 +153,6 @@ async def test_scenario_3_1_1_default_prediction_export( 'addr_1', 0, 'float', - ANY, { 'model_id': 311, 'model_name': 'test_model', @@ -165,7 +164,6 @@ async def test_scenario_3_1_1_default_prediction_export( 'addr_2', 2, 'float', - ANY, { 'model_id': 311, 'model_name': 'test_model', @@ -252,20 +250,28 @@ async def test_scenario_3_1_2_export_with_opc_only( opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data) opc_write_data.assert_has_calls( [ - call('addr_1', 0.5, 'float', ANY, - { - 'model_id': 312, - 'model_name': 'test_model', - 'schedule_name': 'test-schedule', - 'workflow_name': 'predictions_batch', - }), - call('addr_2', 0, 'float', ANY, - { - 'model_id': 312, - 'model_name': 'test_model', - 'schedule_name': 'test-schedule', - 'workflow_name': 'predictions_batch', - }), + call( + 'addr_1', + 0.5, + 'float', + { + 'model_id': 312, + 'model_name': 'test_model', + 'schedule_name': 'test-schedule', + 'workflow_name': 'predictions_batch', + }, + ), + call( + 'addr_2', + 0, + 'float', + { + 'model_id': 312, + 'model_name': 'test_model', + 'schedule_name': 'test-schedule', + 'workflow_name': 'predictions_batch', + }, + ), ] ) @@ -500,20 +506,28 @@ async def test_scenario_3_1_5_export_without_transformed_data( opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data) opc_write_data.assert_has_calls( [ - call('addr_1', 0.5, 'float', ANY, - { - 'model_id': 315, - 'model_name': 'test_model', - 'schedule_name': 'test-schedule', - 'workflow_name': 'predictions_batch', - }), - call('addr_2', 0, 'float', ANY, - { - 'model_id': 315, - 'model_name': 'test_model', - 'schedule_name': 'test-schedule', - 'workflow_name': 'predictions_batch', - }), + call( + 'addr_1', + 0.5, + 'float', + { + 'model_id': 315, + 'model_name': 'test_model', + 'schedule_name': 'test-schedule', + 'workflow_name': 'predictions_batch', + }, + ), + call( + 'addr_2', + 0, + 'float', + { + 'model_id': 315, + 'model_name': 'test_model', + 'schedule_name': 'test-schedule', + 'workflow_name': 'predictions_batch', + }, + ), ] ) @@ -646,7 +660,68 @@ async def test_scenario_3_2_2_opc_write_error( prediction_confidence=12, comments='Some data could not be written to OPC servers', ) - + + +@pytest.mark.asyncio +@pytest.mark.integration +async def test_scenario_3_2_4_opc_session_bad_error( + temporal_test_env: WorkflowEnvironment, + temporal_worker: Worker, + test_activities: Activities, + postgres_engine, +): + """ + Scenario 3.2.4: OPC session/channel Tier-1 Bad* (e.g. BadSessionIdInvalid). + + PostgreSQL stores prediction_confidence 14 and a stable session error comment. + """ + client = temporal_test_env.client + + model_id = 324 + + insert_sample_data(postgres_engine, model_id, [23.5, 78.2]) + + opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data) + opc_write_data.return_value = ( + False, + { + 'notification_id': 'OPC_WRITE_DATA_ERROR_1', + 'message': 'BadSessionIdInvalid', + 'block': 'opc_repository', + 'level': NotificationLevel.ERROR, + 'attachment_content': 'BadSessionIdInvalid', + 'opc_error_kind': 'session_bad', + 'opc_status': 'BadSessionIdInvalid', + }, + ) + + input_data = get_base_input_data(model_id) + input_data['opc_output_config'] = { + '1': { + 'prediction_tags': { + 'addr_1': { + 'data_type': 'float', + } + }, + 'confidence_tags': { + 'addr_2': { + 'data_type': 'float', + } + }, + } + } + input_data['pi_web_api_output_config'] = None + + await start_and_await_workflow( + client, PredictionsBatch.run, input_data, make_workflow_id('test-opc-session-bad') + ) + + assert_prediction( + postgres_engine, + model_id, + prediction_confidence=14, + comments='OPC UA session/channel error: BadSessionIdInvalid', + ) @pytest.mark.asyncio diff --git a/inter_arrival.py b/inter_arrival.py new file mode 100644 index 0000000..bce9cee --- /dev/null +++ b/inter_arrival.py @@ -0,0 +1,25 @@ +# %% + +# Load logs.txt +with open('logs.txt', 'r') as file: + lines = file.readlines() + +# %% +import re +# Grep "inter-arrival_s=number" with regex +intervals = [] +for line in lines: + match = re.search(r'inter-arrival_s=([0-9.]+)', line) + if match: + intervals.append(float(match.group(1))) +# %% + +print(intervals) +# %% +import matplotlib.pyplot as plt +plt.plot(intervals) +plt.ylabel('Inter-arrival time (s)') +plt.xlabel('Sample') +plt.title('Inter-arrival time distribution') +plt.show() +# %% diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index da36793..be5a941 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -15,6 +15,45 @@ with workflow.unsafe.imports_passed_through(): from laborious.utils.repository.opc_repository import OpcRepository OPC_WRITTING_ERROR_CONFIDENCE = 12 +OPC_SESSION_BAD_CONFIDENCE = 14 +OPC_SESSION_BAD_COMMENT_PREFIX = 'OPC UA session/channel error:' +OPC_WRITTING_ERROR_MESSAGE = 'Some data could not be written to OPC servers' +OPC_RECONNECT_IN_PROGRESS_COMMENT = 'OPC UA reconnect in progress' +OPC_COMMENT_SEPARATOR = ' | ' + + +def _opc_session_bad_comment(opc_status: str | None) -> str: + status = opc_status or 'Unknown' + return f'{OPC_SESSION_BAD_COMMENT_PREFIX} {status}' + + +def _apply_opc_write_error( + error_info: dict[str, Any] | None, + session_bad_seen: bool, + session_bad_status: str | None, + reconnect_in_progress_seen: bool, +) -> tuple[bool, str | None, bool]: + """ + Update session/reconnect flags from an OPC write error payload. + + Args: + error_info: Repository error details, or None when the write succeeded. + session_bad_seen: Whether a session_bad error was seen so far. + session_bad_status: Last known OPC status for session errors. + reconnect_in_progress_seen: Whether reconnect_in_progress was seen so far. + + Return: + Updated (session_bad_seen, session_bad_status, reconnect_in_progress_seen). + """ + if not error_info: + return session_bad_seen, session_bad_status, reconnect_in_progress_seen + + kind = error_info.get('opc_error_kind') + if kind == 'session_bad': + return True, error_info.get('opc_status', session_bad_status), reconnect_in_progress_seen + if kind == 'reconnect_in_progress': + return session_bad_seen, session_bad_status, True + return session_bad_seen, session_bad_status, reconnect_in_progress_seen class OPC(SientiaMonitoring): @@ -118,29 +157,18 @@ class OPC(SientiaMonitoring): data_type: str, tag_type: str, metadata: dict[str, Any], - ) -> float | None: + ) -> tuple[float | None, dict[str, Any] | None]: """ Write data to a specific OPC server tag with comprehensive error handling. - This method provides a secure and reliable way to write data to OPC servers - with automatic error handling, notification integration, and detailed logging. - It validates server availability before attempting write operations and - provides comprehensive error reporting for operational monitoring. - - Args: - - server_id (str): The id of the OPC server. - - tag (str): The tag to write to. - - data (Any): The data to write. - - data_type (str): The data type. - - tag_type (str): The tag type. - - Returns: - - bool: True if the data was written successfully, False otherwise. + Return: + tuple[float | None, dict[str, Any] | None]: Response time on success, or + (None, error info_data) on repository failure. """ try: is_success, info_data = await self.opc_repository[server_id].write_data( - tag, data, data_type, self.logger, metadata + tag, data, data_type, metadata ) if not is_success: await self.send_notification_async( @@ -151,8 +179,8 @@ class OPC(SientiaMonitoring): level=info_data.get('level', NotificationLevel.ERROR), attachment_content=info_data.get('attachment_content', None), ) - return None - return info_data['response_time'] + return None, info_data + return info_data['response_time'], None except Exception as e: trace = traceback.format_exc() await self.send_notification_async( @@ -199,13 +227,69 @@ class OPC(SientiaMonitoring): return False return True + async def _write_tags_from_config( + self, + server_id: str, + tags_config: dict[str, dict[str, Any]], + data: DataFrame, + data_column: str, + tag_type: str, + log_label: str, + metadata: dict[str, Any], + ) -> tuple[dict[str, float | None], bool, str | None, bool]: + """ + Write a group of OPC tags and collect response times and error flags. + + Args: + server_id: Target OPC server identifier. + tags_config: Tag name to configuration mapping. + data: DataFrame with prediction/confidence columns. + data_column: Column name whose first row value is written. + tag_type: Tag category passed to write_data ('prediction' or 'confidence'). + log_label: Human-readable label for success logs. + metadata: Context metadata for logging and notifications. + + Return: + (response_times, session_bad_seen, session_bad_status, reconnect_in_progress_seen) + """ + response_times: dict[str, float | None] = {} + session_bad_seen = False + session_bad_status: str | None = None + reconnect_in_progress_seen = False + + for tag, tag_config in tags_config.items(): + response_time, error_info = await self.write_data( + server_id=server_id, + tag=tag, + data=data.head(1)[data_column].values[0], + data_type=tag_config['data_type'], + tag_type=tag_type, + metadata=metadata, + ) + session_bad_seen, session_bad_status, reconnect_in_progress_seen = ( + _apply_opc_write_error( + error_info, + session_bad_seen, + session_bad_status, + reconnect_in_progress_seen, + ) + ) + if response_time is not None: + self.info( + f'{log_label} written to OPC server {server_id} for tag {tag}.', + metadata, + ) + response_times[tag] = response_time + + return response_times, session_bad_seen, session_bad_status, reconnect_in_progress_seen + async def manage_output_tags( self, server_id: str, config: dict[str, Any], data: DataFrame, metadata: dict[str, Any], - ) -> tuple[bool, dict[str, float | None]]: + ) -> tuple[bool, dict[str, float | None], bool, str | None, bool]: """ Manage the writing of prediction and confidence data to OPC server tags. @@ -232,46 +316,47 @@ class OPC(SientiaMonitoring): - overall_success: True if all configured tags were written successfully - total_tags_written: Count of successfully written tags """ - response_times: dict[str, float | None] = {} + session_bad_seen = False + session_bad_status: str | None = None + reconnect_in_progress_seen = False - if 'prediction_tags' in config: - for tag, tag_config in config['prediction_tags'].items(): - response_time = await self.write_data( - server_id=server_id, - tag=tag, - data=data.head(1)['prediction'].values[0], - data_type=tag_config['data_type'], - tag_type='prediction', - metadata=metadata, - ) - if response_time is not None: - self.info( - f'Prediction data written to OPC server {server_id} for tag {tag}.', - metadata, - ) - response_times[tag] = response_time - - if 'confidence_tags' in config: - for tag, tag_config in config['confidence_tags'].items(): - response_time = await self.write_data( - server_id=server_id, - tag=tag, - data=data.head(1)['prediction_confidence'].values[0], - data_type=tag_config['data_type'], - tag_type='confidence', - metadata=metadata, - ) - if response_time is not None: - self.info( - f'Confidence data written to OPC server {server_id} for tag {tag}.', - metadata, - ) - response_times[tag] = response_time + tag_groups = ( + ('prediction_tags', 'prediction', 'prediction', 'Prediction data'), + ('confidence_tags', 'prediction_confidence', 'confidence', 'Confidence data'), + ) + for config_key, data_column, tag_type, log_label in tag_groups: + if config_key not in config: + continue + ( + group_times, + group_session_bad, + group_status, + group_reconnect, + ) = await self._write_tags_from_config( + server_id=server_id, + tags_config=config[config_key], + data=data, + data_column=data_column, + tag_type=tag_type, + log_label=log_label, + metadata=metadata, + ) + response_times.update(group_times) + if group_session_bad: + session_bad_seen = True + session_bad_status = group_status or session_bad_status + if group_reconnect: + reconnect_in_progress_seen = True success = None not in response_times.values() - - return success, response_times + return ( + success, + response_times, + session_bad_seen, + session_bad_status, + reconnect_in_progress_seen, + ) @activity.defn(name='write_opc_data') async def write_opc_data( @@ -301,6 +386,9 @@ class OPC(SientiaMonitoring): self.info(f'Data to write: {data.size} rows', metadata) success = True + session_bad_seen = False + session_bad_status: str | None = None + reconnect_in_progress_seen = False metrics: dict[str, dict[str, float | None]] = {} @@ -309,22 +397,48 @@ class OPC(SientiaMonitoring): success = False continue - local_success, local_response_times = await self.manage_output_tags( - server_id, config, data, metadata - ) + ( + local_success, + local_response_times, + local_session_bad, + local_status, + local_reconnect_in_progress, + ) = await self.manage_output_tags(server_id, config, data, metadata) metrics[server_id] = local_response_times local_count = len(local_response_times) success = success and local_success + if local_session_bad: + session_bad_seen = True + session_bad_status = local_status or session_bad_status + if local_reconnect_in_progress: + reconnect_in_progress_seen = True self.info( f'Process completed for OPC server {server_id}: {local_count} of {len(config.get("prediction_tags", []))} prediction tags and {len(config.get("confidence_tags", []))} confidence tags', metadata, ) - return self.process_confidence(data, success, metadata), metrics + return ( + self.process_confidence( + data, + success, + metadata, + session_bad=session_bad_seen, + opc_status=session_bad_status, + reconnect_in_progress=reconnect_in_progress_seen, + ), + metrics, + ) def process_confidence( - self, data: DataFrame, success: bool, metadata: dict[str, Any] + self, + data: DataFrame, + success: bool, + metadata: dict[str, Any], + *, + session_bad: bool = False, + opc_status: str | None = None, + reconnect_in_progress: bool = False, ) -> dict[Hashable, Any]: """ Process prediction confidence based on OPC write operation success. @@ -352,16 +466,26 @@ class OPC(SientiaMonitoring): This allows downstream systems to handle data quality appropriately. """ - message = 'Some data could not be written to OPC servers' - if not success: - data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE - data['comments'] = message + comment_parts: list[str] = [] + confidence = OPC_WRITTING_ERROR_CONFIDENCE + + if session_bad: + comment_parts.append(_opc_session_bad_comment(opc_status)) + confidence = OPC_SESSION_BAD_CONFIDENCE + if reconnect_in_progress: + comment_parts.append(OPC_RECONNECT_IN_PROGRESS_COMMENT) + confidence = OPC_SESSION_BAD_CONFIDENCE + if not comment_parts: + comment_parts.append(OPC_WRITTING_ERROR_MESSAGE) + + comments = OPC_COMMENT_SEPARATOR.join(comment_parts) + data['prediction_confidence'] = confidence + data['comments'] = comments self.debug( - f'{message}, setting confidence to {OPC_WRITTING_ERROR_CONFIDENCE}.', + f'OPC write issues, confidence={confidence}, comments={comments}', metadata, ) - else: self.debug('Data written to OPC servers successfully.', metadata) diff --git a/laborious/metrics.py b/laborious/metrics.py index 776b826..74a9189 100644 --- a/laborious/metrics.py +++ b/laborious/metrics.py @@ -89,6 +89,40 @@ OPC_CONNECTION_STATUS = Gauge( ['pod_id', 'server_name', 'server_url'], ) +_OPC_SESSION_DEBUG_LABELS = ['pod_id', 'server_name', 'runtime', 'opc_server_id', 'session_id'] + +OPC_SESSION_CREATED_TOTAL = Counter( + 'opc_session_created_total', + 'OPC UA sessions established (after successful connect)', + _OPC_SESSION_DEBUG_LABELS, +) + +OPC_SESSION_CLOSED_TOTAL = Counter( + 'opc_session_closed_total', + 'OPC UA client disconnects completed (session tear-down initiated)', + _OPC_SESSION_DEBUG_LABELS, +) + +OPC_SESSION_REVISED_TIMEOUT_MS = Gauge( + 'opc_session_revised_timeout_milliseconds', + 'Server-revised OPC UA session timeout (RevisedSessionTimeout) in ms after connect', + _OPC_SESSION_DEBUG_LABELS, +) + +OPC_WRITE_ATTEMPT_LABELS = [*_OPC_SESSION_DEBUG_LABELS, 'model_id', 'model_name', 'result'] + +OPC_WRITE_ATTEMPTS_TOTAL = Counter( + 'opc_write_attempts_total', + 'OPC UA write attempts with session and outcome (result=OK or exception class name)', + OPC_WRITE_ATTEMPT_LABELS, +) + +OPC_WRITE_INTER_ARRIVAL_OVER_SESSION_TIMEOUT_TOTAL = Counter( + 'opc_write_inter_arrival_over_session_timeout_total', + 'Successful writes where seconds since the previous successful write exceeded RevisedSessionTimeout (ms)', + _OPC_SESSION_DEBUG_LABELS, +) + # ================== Model metrics ================== MODEL_READ_LAG = Histogram( diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 385b256..b164ce1 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -9,6 +9,7 @@ from typing import Any from asyncua import Client from asyncua.crypto.security_policies import SecurityPolicyBasic256 from asyncua.ua import DataValue, Variant, VariantType +from asyncua.ua.uaerrors import UaStatusCodeError from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger @@ -17,6 +18,119 @@ from sientia_do.observability.sientia_monitoring import SientiaMonitoring from laborious import metrics +# Requested session and secure channel lifetime (ms) before server revision; 10 minutes. +OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS = 10 * 60 * 1000 + + +class OpcClientAlreadyExistsError(RuntimeError): + """Raised when _create_client is called while self.client is already set.""" + + +class OpcSessionAlreadyConnectedError(RuntimeError): + """Raised when _open_session is called while a UA session is already open.""" + + +class OpcClientNotInitializedError(RuntimeError): + """Raised when _open_session is called before _create_client.""" + + +RECONNECTABLE_OPC_BAD_NAMES: frozenset[str] = frozenset( + { + 'BadSessionIdInvalid', + 'BadSessionClosed', + 'BadSessionNotActivated', + 'BadSecureChannelIdInvalid', + 'BadSecureChannelClosed', + 'BadSecureChannelTokenUnknown', + 'BadTcpSecureChannelUnknown', + 'BadServerNotConnected', + 'BadConnectionClosed', + 'BadDisconnect', + 'BadConnectionRejected', + 'BadCommunicationError', + 'BadRequestInterrupted', + 'BadUnknownResponse', + 'BadTimeout', + 'BadRequestTimeout', + 'BadSequenceNumberInvalid', + 'BadSequenceNumberUnknown', + 'BadSecurityModeInsufficient', + 'BadRequestHeaderInvalid', + 'BadInvalidState', + } +) + + +def _opc_authentication_token_str(client: Client | None) -> str: + """ + Serialize the current OPC UA authentication token (session handle) for logging and metrics. + + Return: + str: Token string, or "unknown" if unavailable. + """ + if client is None: + return 'unknown' + try: + proto = client.uaclient.protocol + if proto is None: + return 'unknown' + tok = getattr(proto, 'authentication_token', None) + if tok is None: + return 'unknown' + return str(tok) + except Exception: + return 'unknown' + + +def _opc_status_from_exception(exc: BaseException) -> str: + """ + Resolve OPC UA status name from an exception, including chained UaStatusCodeError causes. + + Args: + exc (BaseException): Raised error from asyncua. + + Return: + str: Status class name or generic Python exception name. + """ + current: BaseException | None = exc + while current is not None: + if isinstance(current, UaStatusCodeError): + return type(current).__name__ + current = current.__cause__ + return type(exc).__name__ + + +def is_reconnectable_opcua_bad(exc: BaseException) -> bool: + """ + Return whether the exception is a Tier-1 OPC UA Bad* that should trigger reconnect. + + Args: + exc (BaseException): Raised error from get_node or write_value. + + Return: + bool: True if reconnect should be scheduled. + """ + return _opc_status_from_exception(exc) in RECONNECTABLE_OPC_BAD_NAMES + + +def _model_labels_from_write_metadata(metadata: dict[str, Any] | None) -> dict[str, str]: + """ + Extract model_id and model_name from write metadata for Prometheus labels. + + Args: + metadata (dict[str, Any] | None): Context passed into write_data; may omit keys. + + Return: + dict[str, str]: Labels model_id and model_name, defaulting to "unknown". + """ + if not metadata: + return {'model_id': 'unknown', 'model_name': 'unknown'} + return { + 'model_id': str(metadata.get('model_id', 'unknown')), + 'model_name': str(metadata.get('model_name', 'unknown')), + } + + data_type_map = { 'float': { 'converter': float, @@ -63,8 +177,6 @@ class OpcRepository(SientiaMonitoring): self.cert_path = cert_path self.private_key_path = private_key_path self.server_cert_path = server_cert_path - self.logger = logger - self.error_count = 0 self.reconnection_interval = reconnection_interval self.last_reconnection_time: None | datetime = None self.disconnection_interval = 10.0 @@ -79,27 +191,63 @@ class OpcRepository(SientiaMonitoring): 'workflow_name': 'opc_repository', 'schedule_name': '-', } + self._last_write_mono: float | None = None + self._connection_lock = asyncio.Lock() + self._session_ready = asyncio.Event() + self._reconnect_task: asyncio.Task[None] | None = None - async def set_security(self): + def _opc_debug_tags(self, session_id: str) -> dict[str, str]: + return { + 'pod_id': str(getattr(self, 'pod_id', 'unknown')), + 'server_name': self.server_name, + 'runtime': str(getattr(self, 'runtime', 'unknown')), + 'opc_server_id': self.id, + 'session_id': session_id, + } + + def _is_session_open(self) -> bool: """ - 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. + Return whether the asyncua client has an open transport session. + + Return: + bool: True when protocol exists and is not closed. + """ + if self.client is None: + return False + try: + proto = self.client.uaclient.protocol + return proto is not None and proto.state != 'closed' + except Exception: + return False + + def _reconnection_window_elapsed(self) -> bool: + """ + Return whether enough time has passed since the last reconnect attempt. + + Return: + bool: True if a new reconnect is allowed. + """ + if self.last_reconnection_time is None: + return True + return ( + datetime.now() - self.last_reconnection_time + ).total_seconds() > self.reconnection_interval + + def _not_connected_error(self) -> dict[str, Any]: + return { + 'notification_id': f'OPC_CONNECTION_NOT_READY_{self.id}', + 'message': f'OPC server {self.id} is not connected', + 'block': 'opc_repository', + 'level': NotificationLevel.WARNING, + } + + async def set_security(self) -> None: + """ + Configure certificates and timeouts on the asyncua client. + Raises: - ValueError: If either the certificate path or private key path is not provided. - Attributes: - - cert_path (str): Path to the client's certificate file. - - private_key_path (str): Path to the client's private key file. - - server_cert_path (str, optional): Path to the server's certificate file. - - server_uri (str): The URI of the server to be used as the application URI. - - client (opcua.Client): The OPC UA client instance. - - logger (logging.Logger): Logger instance for logging information. - Security Settings: - - Security Policy: Basic256 - - Secure Channel Timeout: 10,000,000 ms - - Session Timeout: 10,000,000 ms + ValueError: If cert paths or client are missing. """ - if self.cert_path is None or self.private_key_path is None: raise ValueError( 'Certificate and private key paths must be provided for secure connection.' @@ -113,91 +261,107 @@ class OpcRepository(SientiaMonitoring): raise ValueError('Client must be initialized before setting security') self.client.application_uri = self.server_uri - self.logger.custom_info('Setting security...', self.metadata) + self.info('Setting security...', self.metadata) await self.client.set_security( SecurityPolicyBasic256, certificate=str(cert), private_key=str(private_key), server_certificate=str(server_cert) if server_cert else None, ) - self.client.secure_channel_timeout = 10000000 - self.client.session_timeout = 10000000 + self.client.secure_channel_timeout = OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS + self.client.session_timeout = OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS - async def connect(self) -> tuple[bool, dict[str, Any]]: + async def _create_client(self) -> None: """ - 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. + Instantiate the asyncua Client and apply security when configured. + + Caller must hold _connection_lock. Does not open a UA session. + Raises: - Exception: If the connection to the OPC server fails. + OpcClientAlreadyExistsError: If self.client is already set. """ + if self.client is not None: + raise OpcClientAlreadyExistsError( + f'OPC client already exists for server {self.id}; ' + 'call disconnect() before creating a new client' + ) - self.client = Client(self.url, timeout=10, watchdog_intervall=3600000) # type: ignore[attr-defined] - + self.client = Client(self.url, timeout=10, watchdog_intervall=50) # type: ignore[attr-defined] self.client.name = self.pod_id self.client.application_name = 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.custom_info( - f'Starting connection to OPC server {self.id}:{self.server_name}...', self.metadata - ) - return await self.try_connect() - async def try_connect(self) -> tuple[bool, dict[str, Any]]: + async def _open_session(self) -> tuple[bool, dict[str, Any]]: """ - Attempt to establish connection to the OPC server. + Open the OPC UA session on the existing client. - This method performs the actual connection attempt to the OPC server - and handles connection failures with comprehensive error reporting. - It updates reconnection timing and provides detailed error information - for operational monitoring and debugging. + Caller must hold _connection_lock. - Returns: - tuple[bool, dict[str, Any]]: Connection result - - bool: True if connection successful, False otherwise - - dict: Error information if connection failed + Raises: + OpcClientNotInitializedError: If self.client is None. + OpcSessionAlreadyConnectedError: If a session is already open. + + Return: + tuple[bool, dict[str, Any]]: Success flag and error payload on connect failure. """ + if self.client is None: + raise OpcClientNotInitializedError( + f'OPC client is not initialized for server {self.id}; ' + 'call _create_client() before opening a session' + ) + if self._is_session_open(): + raise OpcSessionAlreadyConnectedError( + f'OPC session already connected for server {self.id}; ' + 'call disconnect() before connecting again' + ) tags = { 'pod_id': self.pod_id, 'server_name': self.server_name, } await self.emit_metric(metrics.OPC_CONNECTIONS_TOTAL, tags) + try: - self.last_reconnection_time = datetime.now() - if self.client is None: - return False, { - 'notification_id': f'OPC_CONNECTION_ERROR_{self.id}', - 'message': 'Client is not initialized', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - } await self.client.connect() + session_id = _opc_authentication_token_str(self.client) + revised_session_timeout_ms = int(self.client.session_timeout) + revised_secure_channel_timeout_ms = int(self.client.secure_channel_timeout) + self.info( + f'OPC new session connected opc_server_id={self.id} session_id={session_id} ' + f'revised_session_timeout_ms={revised_session_timeout_ms} ' + f'revised_secure_channel_timeout_ms={revised_secure_channel_timeout_ms}', + self.metadata, + ) + await self.emit_metric( + metrics.OPC_SESSION_CREATED_TOTAL, self._opc_debug_tags(session_id) + ) + await self.emit_metric( + metric_object=metrics.OPC_SESSION_REVISED_TIMEOUT_MS, + method='set', + tags=self._opc_debug_tags(session_id), + value=revised_session_timeout_ms, + ) await self.emit_metric( metric_object=metrics.OPC_CONNECTION_STATUS, method='set', - tags={ - **tags, - 'server_url': self.url, - }, + tags={**tags, 'server_url': self.url}, value=1, ) + self._last_write_mono = None + self._session_ready.set() return True, {} + except Exception as e: - await self.disconnect() - + await self._disconnect_locked() trace = traceback.format_exc() - self.logger.custom_error(trace, self.metadata) - + self.error(trace, self.metadata) await self.emit_metric(metrics.OPC_CONNECTIONS_FAILED, tags) - return False, { 'notification_id': f'OPC_CONNECTION_ERROR_{self.id}', 'message': f'Failed to connect to OPC server: {e}', @@ -206,21 +370,45 @@ class OpcRepository(SientiaMonitoring): 'attachment_content': trace, } - async def disconnection_fallback(self) -> list: - """ - Tries 5 times to disconnect from the OPC UA server, with a delay of 100ms x try. + async def _connect_locked(self) -> tuple[bool, dict[str, Any]]: """ + Create the client when absent, then open a UA session. + Caller must hold _connection_lock. + + Raises: + OpcSessionAlreadyConnectedError: If a session is already open. + + Return: + tuple[bool, dict[str, Any]]: Result from _open_session on connect failure. + """ + if self._is_session_open(): + raise OpcSessionAlreadyConnectedError( + f'OPC session already connected for server {self.id}; ' + 'call disconnect() before connecting again' + ) + if self.client is None: + await self._create_client() + return await self._open_session() + + async def _disconnection_fallback(self) -> list[dict[str, Any]]: + """ + Try up to five times to disconnect from the OPC UA server. + """ assert self.client is not None - error_stack = [] + error_stack: list[dict[str, Any]] = [] for i in range(5): try: - self.logger.info(f'Disconnecting from OPC UA server, attempt {i + 1} of 5') + self.info( + f'Disconnecting from OPC UA server, attempt {i + 1} of 5', + self.metadata, + ) await self.client.disconnect() return [] except Exception as e: - self.logger.error( - f'Failed to disconnect from OPC UA server in attempt {i + 1} of 5: {e}' + self.error( + f'Failed to disconnect from OPC UA server in attempt {i + 1} of 5: {e}', + self.metadata, ) error_stack.append( { @@ -232,18 +420,26 @@ class OpcRepository(SientiaMonitoring): await asyncio.sleep(self.disconnection_interval * i) return error_stack - async def disconnect(self): + async def _disconnect_locked(self) -> None: """ - Gracefully disconnect from the OPC server. + Tear down the current session and client. - This method safely terminates the connection to the OPC server - and cleans up client resources. It handles disconnection errors - gracefully and ensures proper resource cleanup. + Caller must hold _connection_lock. """ + self._last_write_mono = None + self._session_ready.clear() + if self.client is None: return - errors = await self.disconnection_fallback() + session_id = _opc_authentication_token_str(self.client) + self.info( + f'OPC disconnecting opc_server_id={self.id} session_id={session_id}', + self.metadata, + ) + await self.emit_metric(metrics.OPC_SESSION_CLOSED_TOTAL, self._opc_debug_tags(session_id)) + + errors = await self._disconnection_fallback() if errors: await self.send_notification_async( metadata=self.metadata, @@ -254,7 +450,8 @@ class OpcRepository(SientiaMonitoring): attachment_content=json.dumps(errors, indent=4), ) else: - self.logger.warning(f'Disconnected from OPC server {self.id} successfully') + self.warning(f'Disconnected from OPC server {self.id} successfully', self.metadata) + await self.emit_metric( metric_object=metrics.OPC_CONNECTION_STATUS, method='set', @@ -265,186 +462,286 @@ class OpcRepository(SientiaMonitoring): }, value=0, ) - self.client = None + async def _reconnect_locked(self) -> tuple[bool, dict[str, Any]]: + """ + Close the current session and open a new one. + + Caller must hold _connection_lock. Records last_reconnection_time for interval gating. + + Return: + tuple[bool, dict[str, Any]]: Result from _connect_locked after teardown. + """ + self.last_reconnection_time = datetime.now() + await self._disconnect_locked() + return await self._connect_locked() + + async def connect(self) -> tuple[bool, dict[str, Any]]: + """ + Open an OPC UA session under the connection lock (worker initialization). + """ + async with self._connection_lock: + self.info( + f'Starting connection to OPC server {self.id}:{self.server_name}...', + self.metadata, + ) + return await self._connect_locked() + + async def disconnect(self) -> None: + """ + Gracefully disconnect from the OPC server under the connection lock. + """ + async with self._connection_lock: + await self._disconnect_locked() + async def validate_connection(self) -> tuple[bool, dict[str, Any]]: """ - Validate and maintain OPC server connection health. + Read-only check that the asyncua protocol is open. - This method performs comprehensive connection validation and - implements automatic reconnection logic for production reliability. - It handles various connection states and implements intelligent - reconnection strategies with error counting and timing controls. + Caller must ensure _session_ready before writing. Does not connect or reconnect. - Connection Validation: - 1. Checks client existence and connection state - 2. Implements error counting with automatic disconnection - 3. Enforces reconnection timing windows - 4. Provides detailed error reporting and notifications + Return: + tuple[bool, dict[str, Any]]: (True, {}) when open, otherwise (False, error). + """ + if self._is_session_open(): + return True, {} + self.error(f'OPC server {self.id} is not connected', self.metadata) + return False, self._not_connected_error() - Reconnection Strategy: - - Error Count Threshold: Disconnects after 5 consecutive errors - - Reconnection Window: Enforces minimum intervals between attempts - - Automatic Recovery: Attempts reconnection when conditions allow - - State Monitoring: Continuously monitors connection health + async def _start_reconnect_on_bad(self, opc_status: str, session_id: str) -> None: + """ + Schedule a background reconnect if interval and task state allow it. Args: - None - - Returns: - tuple[bool, dict[str, Any]]: Connection validation result - - bool: True if connection is healthy, False otherwise - - dict: Error information if validation fails + opc_status (str): OPC UA status name that triggered reconnect. + session_id (str): Session token before failure. """ - if self.client is None: - return await self.connect() + if not self._reconnection_window_elapsed(): + self.warning( + f'OPC reconnect skipped reason=reconnection_window opc_server_id={self.id} ' + f'opc_status={opc_status}', + self.metadata, + ) + return + if self._reconnect_task is not None and not self._reconnect_task.done(): + self.warning( + f'OPC reconnect skipped reason=in_progress opc_server_id={self.id} ' + f'opc_status={opc_status}', + self.metadata, + ) + return - # if self.error_count > 5: # NOSONAR - # self.logger.custom_warning( - # f'OPC server {self.id} will be disconnected due to multiple errors', self.metadata - # ) - # try: - # await self.disconnect() - # except Exception as e: - # trace = traceback.format_exc() - # self.logger.custom_error( - # f'Failed to disconnect from OPC server: {e}', self.metadata - # ) - # self.logger.custom_error(trace, self.metadata) - # self.logger.custom_info( - # f'Attempting to reconnect to OPC server {self.id}...', self.metadata - # ) - # return await self.connect() + self._session_ready.clear() + self.info( + f'OPC reconnect scheduled after opc_status={opc_status} opc_server_id={self.id} ' + f'old_session_id={session_id}', + self.metadata, + ) + self._reconnect_task = asyncio.create_task( + self._run_reconnect_on_bad(opc_status, session_id) + ) - # Check if client is connected using asyncua's connection state + async def _run_reconnect_on_bad(self, opc_status: str, session_id: str) -> None: + """ + Tear down and re-establish the OPC UA session under the connection lock. + + Args: + opc_status (str): OPC UA status that triggered reconnect. + session_id (str): Previous session token string. + """ try: - if ( - self.client.uaclient.protocol is None - or self.client.uaclient.protocol.state == 'closed' - ): - # OPC server is not connected - self.logger.custom_error(f'OPC server {self.id} is not connected', self.metadata) - if ( - self.last_reconnection_time is None - or (datetime.now() - self.last_reconnection_time).total_seconds() - > self.reconnection_interval - ): - await self.disconnect() - self.logger.custom_info( - f'Trying to reconnect to OPC server {self.id}...', self.metadata + async with self._connection_lock: + self.info( + f'OPC reconnect started opc_status={opc_status} opc_server_id={self.id} ' + f'old_session_id={session_id}', + self.metadata, + ) + await self._reconnect_locked() + except Exception: + self.error( + f'OPC reconnect task failed opc_server_id={self.id} opc_status={opc_status}', + self.metadata, + ) + self.error(traceback.format_exc(), self.metadata) + + async def _log_write_inter_arrival(self, session_id: str, node: str) -> None: + """ + Log elapsed wall time since the previous successful OPC write on this repository. + + Args: + session_id (str): Current OPC UA session token string. + node (str): Node id written in this operation. + """ + now = time.monotonic() + if self._last_write_mono is not None: + delta_s = now - self._last_write_mono + self.info( + f'OPC write inter-arrival_s={delta_s:.6f} opc_server_id={self.id} ' + f'session_id={session_id} node={node}', + self.metadata, + ) + if self.client is not None: + session_timeout_ms = float(self.client.session_timeout) + if session_timeout_ms > 0 and delta_s > (session_timeout_ms / 1000.0): + await self.emit_metric( + metrics.OPC_WRITE_INTER_ARRIVAL_OVER_SESSION_TIMEOUT_TOTAL, + self._opc_debug_tags(session_id), ) - return await self.connect() + self._last_write_mono = now - return False, { - 'notification_id': f'OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{self.id}', - 'message': f'OPC server {self.id} is not connected, waiting for next reconnection window...', - 'block': 'opc_repository', - 'level': NotificationLevel.WARNING, - } - return True, {} - except Exception as e: - trace = traceback.format_exc() - message = f'Failed to validate connection to OPC server: {e}' - self.logger.custom_error(message, self.metadata) - return False, { - 'notification_id': f'OPC_CONNECTION_CHECK_ERROR_{self.id}', - 'message': message, - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': trace, - } + async def _emit_opc_write_metric( + self, session_id: str, result: str, metadata: dict[str, Any] | None + ) -> None: + await self.emit_metric( + metrics.OPC_WRITE_ATTEMPTS_TOTAL, + { + **self._opc_debug_tags(session_id), + **_model_labels_from_write_metadata(metadata), + 'result': result, + }, + ) - async def write_data( - self, node: str, value: Any, data_type: str, logger: Logger, metadata: dict[str, Any] + def _write_failure_payload( + self, + notification_id: str, + message: str, + level: NotificationLevel = NotificationLevel.ERROR, + attachment_content: str | None = None, + opc_error_kind: str | None = None, + opc_status: str | None = None, + ) -> dict[str, Any]: + payload: dict[str, Any] = { + 'notification_id': notification_id, + 'message': message, + 'block': 'opc_repository', + 'level': level, + } + if attachment_content is not None: + payload['attachment_content'] = attachment_content + if opc_error_kind is not None: + payload['opc_error_kind'] = opc_error_kind + if opc_status is not None: + payload['opc_status'] = opc_status + return payload + + async def _handle_tier1_bad( + self, + exc: BaseException, + session_id: str, + node: str, + metadata: dict[str, Any], + phase: str, ) -> tuple[bool, dict[str, Any]]: """ - Write data to OPC server with comprehensive validation and monitoring. - - This method provides secure and reliable data writing to OPC servers - with automatic connection validation, data type conversion, and - comprehensive error handling. It implements performance monitoring - and metrics collection for operational visibility. - - Data Writing Process: - 1. Connection validation and automatic reconnection - 2. Node validation and error handling - 3. Data type conversion and validation - 4. OPC data writing with timestamp - 5. Performance metrics collection - 6. Error handling and notification + Record metrics/logs and schedule reconnect after a Tier-1 Bad* error. Args: - node (str): OPC node identifier to write data to - value (Any): Data value to write to the OPC node - data_type (str): Data type for OPC conversion - logger (Logger): Logger instance for operation logging - metadata (dict[str, Any]): Context metadata for logging and metrics + exc (BaseException): Tier-1 OPC UA error. + session_id (str): Session token at failure time. + node (str): Node id being written. + metadata (dict[str, Any]): Write context. + phase (str): get_node or write_value. - Returns: - tuple[bool, dict[str, Any]]: Write operation result - - bool: True if write successful, False otherwise - - dict: Error information if write failed + Return: + tuple[bool, dict[str, Any]]: Always (False, error payload). """ + opc_status = _opc_status_from_exception(exc) + trace = traceback.format_exc() + self.error(trace, metadata) + await self._emit_opc_write_metric(session_id, opc_status, metadata) + self.error( + f'OPC write failed opc_status={opc_status} opc_server_id={self.id} ' + f'session_id={session_id} model_id={metadata.get("model_id", "unknown")} ' + f'model_name={metadata.get("model_name", "unknown")} node={node} phase={phase}', + metadata, + ) + await self._start_reconnect_on_bad(opc_status, session_id) + return False, self._write_failure_payload( + notification_id=f'OPC_WRITE_DATA_ERROR_{self.id}', + message=f'Failed to {phase} on OPC server: {exc} | metadata: {metadata}', + attachment_content=trace, + opc_error_kind='session_bad', + opc_status=opc_status, + ) + + async def write_data( + self, node: str, value: Any, data_type: str, metadata: dict[str, Any] + ) -> tuple[bool, dict[str, Any]]: + """ + Write data to OPC server with a single attempt and Tier-1 Bad* reconnect scheduling. + """ + if not self._session_ready.is_set(): + await self._emit_opc_write_metric('unknown', 'ReconnectInProgress', metadata) + self.warning( + f'OPC write rejected reconnect_in_progress opc_server_id={self.id} ' + f'model_id={metadata.get("model_id", "unknown")} ' + f'model_name={metadata.get("model_name", "unknown")}', + metadata, + ) + return False, { + 'notification_id': f'OPC_WRITE_RECONNECT_IN_PROGRESS_{self.id}', + 'message': f'OPC write skipped: reconnect in progress | metadata: {metadata}', + 'block': 'opc_repository', + 'level': NotificationLevel.WARNING, + 'opc_error_kind': 'reconnect_in_progress', + } is_connected, error = await self.validate_connection() - if not is_connected: + await self._emit_opc_write_metric('unknown', 'NotConnected', metadata) return False, error start_time = time.time() + session_id = _opc_authentication_token_str(self.client) try: - # ignored because self.validate_connection is called before, so we know self.client is not None node_obj = self.client.get_node(node) # type: ignore[union-attr] except Exception as e: + if is_reconnectable_opcua_bad(e): + return await self._handle_tier1_bad(e, session_id, node, metadata, 'get_node') trace = traceback.format_exc() - logger.custom_error(trace, metadata.get('schedule_name', 'N/A')) - self.error_count += 1 - return False, { - 'notification_id': f'OPC_WRITE_GET_NODE_ERROR_{self.id}', - 'message': f'Failed to get node from OPC server: {e} | metadata: {metadata}', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': trace, - } + self.error(trace, metadata) + await self._emit_opc_write_metric( + session_id, f'GetNodeError:{type(e).__name__}', metadata + ) + return False, self._write_failure_payload( + notification_id=f'OPC_WRITE_GET_NODE_ERROR_{self.id}', + message=f'Failed to get node from OPC server: {e} | metadata: {metadata}', + attachment_content=trace, + ) if data_type not in data_type_map: - return False, { - 'notification_id': f'OPC_WRITE_DATA_TYPE_ERROR_{self.id}', - 'message': f'Unsupported data type: {data_type} | metadata: {metadata}', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - } + await self._emit_opc_write_metric(session_id, 'UnsupportedDataType', metadata) + return False, self._write_failure_payload( + notification_id=f'OPC_WRITE_DATA_TYPE_ERROR_{self.id}', + message=f'Unsupported data type: {data_type} | metadata: {metadata}', + ) data = data_type_map[data_type]['converter'](value) - logger.custom_info(f'Writing {data} - {type(data)} to {node}', metadata) - # now = datetime.now() # NOSONAR + self.info(f'Writing {data} - {type(data)} to {node}', metadata) ua_data = DataValue( Variant(data, data_type_map[data_type]['opc_type']), - # SourceTimestamp=DateTime( # NOSONAR - # now.year, now.month, now.day, now.hour, now.minute, now.second, now.microsecond # NOSONAR - # ), # NOSONAR ) try: await node_obj.write_value(ua_data) - end_time = time.time() response_time = end_time - start_time - except Exception as e: + if is_reconnectable_opcua_bad(e): + return await self._handle_tier1_bad(e, session_id, node, metadata, 'write_value') trace = traceback.format_exc() - logger.custom_error(trace, metadata) - self.error_count += 1 - return False, { - 'notification_id': f'OPC_WRITE_DATA_ERROR_{self.id}', - 'message': f'Failed to write data to OPC server: {e} | metadata: {metadata}', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': trace, - } - self.error_count = 0 + self.error(trace, metadata) + await self._emit_opc_write_metric(session_id, type(e).__name__, metadata) + return False, self._write_failure_payload( + notification_id=f'OPC_WRITE_DATA_ERROR_{self.id}', + message=f'Failed to write data to OPC server: {e} | metadata: {metadata}', + attachment_content=trace, + ) + + await self._emit_opc_write_metric(session_id, 'OK', metadata) + await self._log_write_inter_arrival(session_id, node) return True, { 'response_time': response_time, diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index fa09259..fe78753 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -63,7 +63,7 @@ with workflow.unsafe.imports_passed_through(): ) from laborious.workflows.sub_workflows.prediction_process import PredictionProcess -POD_ID = os.getenv('POD_ID') +POD_ID = os.getenv('HOSTNAME') SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091')) diff --git a/pyproject.toml b/pyproject.toml index 5123123..a878882 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -121,6 +121,7 @@ markers = [ "asyncio: marks tests as async", "integration: marks tests as integration tests", "unit: marks tests as unit tests", + "opc: marks tests that use the in-process OPC UA server (OpcRepository E2E)", ] [tool.coverage.run] diff --git a/requirements-dev.txt b/requirements-dev.txt index c4f4ef6..4d0f89d 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -18,3 +18,4 @@ testcontainers[postgres,minio] # PostgreSQL and MinIO containers for E2E tests # Development Tools ipython>=8.12.0 # Enhanced Python shell ipdb>=0.13.13 # IPython debugger +ipykernel==6.30.1 # IPython kernel for Jupyter notebooks diff --git a/tests/laborious/activities/test_opc.py b/tests/laborious/activities/test_opc.py index 0da8214..c2d37dd 100644 --- a/tests/laborious/activities/test_opc.py +++ b/tests/laborious/activities/test_opc.py @@ -5,7 +5,16 @@ from pandas import DataFrame from pytest import mark from sientia_do.notifications.models import NotificationLevel -from laborious.activities.opc import OPC +from laborious.activities.opc import ( + OPC, + OPC_COMMENT_SEPARATOR, + OPC_RECONNECT_IN_PROGRESS_COMMENT, + OPC_SESSION_BAD_COMMENT_PREFIX, + OPC_SESSION_BAD_CONFIDENCE, + OPC_WRITTING_ERROR_CONFIDENCE, + OPC_WRITTING_ERROR_MESSAGE, + _apply_opc_write_error, +) metadata = { 'metadata': { @@ -206,7 +215,7 @@ WRITE_DATA_CASES = [ async def test_write_data_success(opc, tag, data_type, data): opc.opc_repository['server1'].write_data.return_value = (True, {'response_time': 0.1}) - result = await opc.write_data( + response_time, error_info = await opc.write_data( server_id='server1', tag=tag, data=data, @@ -214,10 +223,9 @@ async def test_write_data_success(opc, tag, data_type, data): tag_type='prediction', metadata=metadata, ) - assert result == 0.1 - opc.opc_repository['server1'].write_data.assert_called_once_with( - tag, data, data_type, opc.logger, metadata - ) + assert response_time == 0.1 + assert error_info is None + opc.opc_repository['server1'].write_data.assert_called_once_with(tag, data, data_type, metadata) @mark.asyncio @@ -233,7 +241,7 @@ async def test_write_data_failed(opc): }, ) - result = await opc.write_data( + response_time, error_info = await opc.write_data( server_id='server1', tag='tag1', data=50, @@ -241,7 +249,8 @@ async def test_write_data_failed(opc): tag_type='prediction', metadata=metadata, ) - assert result is None + assert response_time is None + assert error_info is not None opc.send_notification_async.assert_called_once_with( metadata=metadata, @@ -281,17 +290,211 @@ async def test_write_data_exception(opc): raise AssertionError('Expected an exception to be raised') +@mark.parametrize( + 'error_info,initial_seen,initial_status,initial_reconnect,expected', + [ + (None, False, None, False, (False, None, False)), + ({}, False, None, False, (False, None, False)), + ( + {'opc_error_kind': 'session_bad', 'opc_status': 'BadSessionIdInvalid'}, + False, + None, + False, + (True, 'BadSessionIdInvalid', False), + ), + ( + {'opc_error_kind': 'session_bad', 'opc_status': 'NewStatus'}, + True, + 'OldStatus', + False, + (True, 'NewStatus', False), + ), + ( + {'opc_error_kind': 'session_bad'}, + True, + 'KeptStatus', + False, + (True, 'KeptStatus', False), + ), + ( + {'opc_error_kind': 'reconnect_in_progress'}, + False, + None, + False, + (False, None, True), + ), + ( + {'opc_error_kind': 'other'}, + True, + 'Status', + True, + (True, 'Status', True), + ), + ], +) +def test_apply_opc_write_error( + error_info, initial_seen, initial_status, initial_reconnect, expected +): + result = _apply_opc_write_error( + error_info, + initial_seen, + initial_status, + initial_reconnect, + ) + assert result == expected + + +@mark.asyncio +async def test_write_tags_from_config_prediction_success(opc): + opc.write_data = AsyncMock(return_value=(0.1, None)) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + tags_config = {'tag1': {'data_type': 'float'}} + + response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config( + server_id='server1', + tags_config=tags_config, + data=data, + data_column='prediction', + tag_type='prediction', + log_label='Prediction data', + metadata=metadata['metadata'], + ) + + assert response_times == {'tag1': 0.1} + assert session_bad is False + assert opc_status is None + assert reconnect is False + opc.write_data.assert_called_once_with( + server_id='server1', + tag='tag1', + data=0.75, + data_type='float', + tag_type='prediction', + metadata=metadata['metadata'], + ) + + +@mark.asyncio +async def test_write_tags_from_config_confidence_success(opc): + opc.write_data = AsyncMock(return_value=(0.2, None)) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + tags_config = {'tag2': {'data_type': 'float'}} + + response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config( + server_id='server1', + tags_config=tags_config, + data=data, + data_column='prediction_confidence', + tag_type='confidence', + log_label='Confidence data', + metadata=metadata['metadata'], + ) + + assert response_times == {'tag2': 0.2} + assert session_bad is False + assert opc_status is None + assert reconnect is False + opc.write_data.assert_called_once_with( + server_id='server1', + tag='tag2', + data=0.95, + data_type='float', + tag_type='confidence', + metadata=metadata['metadata'], + ) + + +@mark.asyncio +async def test_write_tags_from_config_write_failure(opc): + opc.write_data = AsyncMock(return_value=(None, {})) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + + response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config( + server_id='server1', + tags_config={'tag1': {'data_type': 'float'}}, + data=data, + data_column='prediction', + tag_type='prediction', + log_label='Prediction data', + metadata=metadata['metadata'], + ) + + assert response_times == {'tag1': None} + assert session_bad is False + assert opc_status is None + assert reconnect is False + + +@mark.asyncio +async def test_write_tags_from_config_session_bad(opc): + opc.write_data = AsyncMock( + return_value=( + None, + { + 'opc_error_kind': 'session_bad', + 'opc_status': 'BadSessionIdInvalid', + }, + ) + ) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + + response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config( + server_id='server1', + tags_config={'tag1': {'data_type': 'float'}}, + data=data, + data_column='prediction', + tag_type='prediction', + log_label='Prediction data', + metadata=metadata['metadata'], + ) + + assert response_times == {'tag1': None} + assert session_bad is True + assert opc_status == 'BadSessionIdInvalid' + assert reconnect is False + + +@mark.asyncio +async def test_write_tags_from_config_reconnect_in_progress(opc): + opc.write_data = AsyncMock( + return_value=( + None, + {'opc_error_kind': 'reconnect_in_progress'}, + ) + ) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + + response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config( + server_id='server1', + tags_config={'tag1': {'data_type': 'float'}}, + data=data, + data_column='prediction', + tag_type='prediction', + log_label='Prediction data', + metadata=metadata['metadata'], + ) + + assert response_times == {'tag1': None} + assert session_bad is False + assert opc_status is None + assert reconnect is True + + @mark.asyncio async def test_manage_output_tags_success(opc): - opc.write_data = AsyncMock(return_value=0.1) - + opc._write_tags_from_config = AsyncMock( + side_effect=[ + ({'tag1': 0.1}, False, None, False), + ({'tag2': 0.1}, False, None, False), + ] + ) data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) config = { 'prediction_tags': {'tag1': {'data_type': 'float'}}, 'confidence_tags': {'tag2': {'data_type': 'float'}}, } - output_data, opc_metrics = await opc.manage_output_tags( + output_data, opc_metrics, session_bad, opc_status, reconnect = await opc.manage_output_tags( server_id='server1', config=config, data=data, @@ -300,83 +503,53 @@ async def test_manage_output_tags_success(opc): assert output_data is True assert opc_metrics == {'tag1': 0.1, 'tag2': 0.1} - opc.write_data.assert_has_calls( - [ - call( - server_id='server1', - tag='tag1', - data=0.75, - data_type='float', - tag_type='prediction', - metadata=metadata['metadata'], - ), - call( - server_id='server1', - tag='tag2', - data=0.95, - data_type='float', - tag_type='confidence', - metadata=metadata['metadata'], - ), - ] - ) + assert session_bad is False + assert opc_status is None + assert reconnect is False + assert opc._write_tags_from_config.await_count == 2 @mark.asyncio -@mark.parametrize('side_effect', [[0.1, None], [None, 0.2]]) -async def test_manage_output_tags_failed(opc, side_effect): - opc.write_data = AsyncMock(side_effect=side_effect) +async def test_manage_output_tags_failed(opc): + opc._write_tags_from_config = AsyncMock( + side_effect=[ + ({'tag1': 0.1}, False, None, False), + ({'tag2': None}, False, None, False), + ] + ) data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) config = { 'prediction_tags': {'tag1': {'data_type': 'float'}}, 'confidence_tags': {'tag2': {'data_type': 'float'}}, } - output_data, opc_metrics = await opc.manage_output_tags( + + output_data, opc_metrics, _, _, _ = await opc.manage_output_tags( server_id='server1', config=config, data=data, metadata=metadata['metadata'], ) + assert output_data is False - assert opc_metrics == {'tag1': side_effect[0], 'tag2': side_effect[1]} - opc.write_data.assert_has_calls( - [ - call( - server_id='server1', - tag='tag1', - data=0.75, - data_type='float', - tag_type='prediction', - metadata=metadata['metadata'], - ), - call( - server_id='server1', - tag='tag2', - data=0.95, - data_type='float', - tag_type='confidence', - metadata=metadata['metadata'], - ), - ] - ) + assert opc_metrics == {'tag1': 0.1, 'tag2': None} @mark.asyncio async def test_manage_output_tags_do_nothing(opc): - opc.write_data = AsyncMock(return_value=0.1) + opc._write_tags_from_config = AsyncMock() data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) - config = { - '_invalid_key': {'tag1': {'data_type': 'float'}}, - } - output_data, opc_metrics = await opc.manage_output_tags( + config = {'_invalid_key': {'tag1': {'data_type': 'float'}}} + + output_data, opc_metrics, _, _, _ = await opc.manage_output_tags( server_id='server1', config=config, data=data, metadata=metadata['metadata'], ) + assert output_data is True assert opc_metrics == {} - opc.write_data.assert_not_called() + opc._write_tags_from_config.assert_not_called() @mark.asyncio @@ -395,7 +568,9 @@ async def test_write_opc_data_success(mock_dataframe, opc): } # Act - opc.manage_output_tags = AsyncMock(return_value=(True, {'tag1': 0.1, 'tag2': 0.2})) + opc.manage_output_tags = AsyncMock( + return_value=(True, {'tag1': 0.1, 'tag2': 0.2}, False, None, False) + ) opc.process_confidence = MagicMock(return_value={'data': 'data'}) output_data, opc_metrics = await opc.write_opc_data(input_data) @@ -413,6 +588,9 @@ async def test_write_opc_data_success(mock_dataframe, opc): mock_dataframe.return_value, True, metadata['metadata'], + session_bad=False, + opc_status=None, + reconnect_in_progress=False, ) @@ -462,13 +640,91 @@ async def test_write_opc_data_no_validate_server(opc): ], ) def test_process_confidence(opc, data, success, expected): - # Act - result = opc.process_confidence(data, success, metadata) - - # Assert + result = opc.process_confidence(data, success, metadata['metadata']) assert result['prediction_confidence'][0] == expected +def test_process_confidence_session_bad(opc): + data = DataFrame({'prediction_confidence': [0.9]}) + result = opc.process_confidence( + data, + False, + metadata['metadata'], + session_bad=True, + opc_status='BadSessionIdInvalid', + ) + assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE + assert result['comments'][0].startswith(OPC_SESSION_BAD_COMMENT_PREFIX) + assert 'BadSessionIdInvalid' in result['comments'][0] + + +def test_process_confidence_generic_failure(opc): + data = DataFrame({'prediction_confidence': [0.9]}) + result = opc.process_confidence(data, False, metadata['metadata']) + assert result['prediction_confidence'][0] == OPC_WRITTING_ERROR_CONFIDENCE + assert result['comments'][0] == OPC_WRITTING_ERROR_MESSAGE + + +@mark.asyncio +async def test_manage_output_tags_merges_error_flags(opc): + opc._write_tags_from_config = AsyncMock( + side_effect=[ + ({'tag1': None}, True, 'BadSessionIdInvalid', False), + ({'tag2': 0.2}, False, None, True), + ] + ) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + config = { + 'prediction_tags': {'tag1': {'data_type': 'float'}}, + 'confidence_tags': {'tag2': {'data_type': 'float'}}, + } + + ( + success, + metrics, + session_bad_seen, + opc_status, + reconnect_in_progress, + ) = await opc.manage_output_tags('server1', config, data, metadata['metadata']) + + assert success is False + assert session_bad_seen is True + assert reconnect_in_progress is True + assert opc_status == 'BadSessionIdInvalid' + assert metrics == {'tag1': None, 'tag2': 0.2} + + +def test_process_confidence_reconnect_in_progress(opc): + data = DataFrame({'prediction_confidence': [0.9]}) + result = opc.process_confidence( + data, + False, + metadata['metadata'], + reconnect_in_progress=True, + ) + assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE + assert result['comments'][0] == OPC_RECONNECT_IN_PROGRESS_COMMENT + + +def test_process_confidence_concatenates_multiple_comments(opc): + data = DataFrame({'prediction_confidence': [0.9]}) + session_comment = f'{OPC_SESSION_BAD_COMMENT_PREFIX} BadSessionIdInvalid' + + result = opc.process_confidence( + data, + False, + metadata['metadata'], + session_bad=True, + opc_status='BadSessionIdInvalid', + reconnect_in_progress=True, + ) + + assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE + assert result['comments'][0] == OPC_COMMENT_SEPARATOR.join( + [session_comment, OPC_RECONNECT_IN_PROGRESS_COMMENT] + ) + + @mark.asyncio async def test_validate_server(opc): assert await opc.validate_server('server1', metadata) is True diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index 8be96a7..f1fd81a 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/tests/laborious/utils/repository/test_opc_repository.py @@ -1,12 +1,20 @@ +import asyncio import json from datetime import datetime from unittest.mock import ANY, AsyncMock, MagicMock, Mock, patch import pytest from asyncua.crypto.security_policies import SecurityPolicyBasic256 +from asyncua.ua.uaerrors import BadNodeIdUnknown, BadSessionIdInvalid from sientia_do.notifications.models import NotificationLevel -from laborious.utils.repository.opc_repository import OpcRepository +from laborious.utils.repository.opc_repository import ( + OpcClientAlreadyExistsError, + OpcClientNotInitializedError, + OpcRepository, + OpcSessionAlreadyConnectedError, + is_reconnectable_opcua_bad, +) @pytest.fixture @@ -33,6 +41,11 @@ def opc_repository(mock_logger): repository.send_notification = MagicMock() repository.send_notification_async = AsyncMock() repository.emit_metric = AsyncMock() + repository.info = MagicMock() + repository.error = MagicMock() + repository.warning = MagicMock() + repository.debug = MagicMock() + repository._session_ready.set() return repository @@ -65,7 +78,6 @@ def test_init(opc_repository): assert opc_repository.reconnection_interval == 60 assert opc_repository.client is None assert opc_repository.last_reconnection_time is None - assert opc_repository.error_count == 0 @pytest.mark.asyncio @@ -80,8 +92,8 @@ async def test_set_security(opc_repository, mock_client): private_key='/path/to/key.pem', server_certificate='/path/to/server_cert.pem', ) - assert mock_client.secure_channel_timeout == 10000000 - assert mock_client.session_timeout == 10000000 + assert mock_client.secure_channel_timeout == 600_000 + assert mock_client.session_timeout == 600_000 @pytest.mark.asyncio @@ -106,48 +118,95 @@ async def test_set_security_missing_client(opc_repository): @pytest.mark.asyncio async def test_connect_with_security(opc_repository, mock_client): - opc_repository.try_connect = AsyncMock(return_value=(True, {})) + opc_repository._create_client = AsyncMock() + opc_repository._open_session = AsyncMock(return_value=(True, {})) result = await opc_repository.connect() - opc_repository.try_connect.assert_called_once() - assert opc_repository.client == mock_client + opc_repository._create_client.assert_called_once() + opc_repository._open_session.assert_called_once() assert result == (True, {}) @pytest.mark.asyncio async def test_connect_without_security(opc_repository, mock_client): opc_repository.cert_path = None - opc_repository.try_connect = AsyncMock(return_value=(True, {})) + opc_repository._create_client = AsyncMock() + opc_repository._open_session = AsyncMock(return_value=(True, {})) opc_repository.set_security = AsyncMock() result = await opc_repository.connect() - opc_repository.try_connect.assert_called_once() + opc_repository._create_client.assert_called_once() + opc_repository._open_session.assert_called_once() opc_repository.set_security.assert_not_called() - assert opc_repository.client == mock_client assert result == (True, {}) @pytest.mark.asyncio -async def test_try_connect_success(opc_repository): - opc_repository.last_reconnection_time = None +async def test_connect_raises_when_session_already_open(opc_repository, mock_client): + opc_repository.client = mock_client + proto = MagicMock() + proto.state = 'open' + mock_client.uaclient = MagicMock(protocol=proto) + + with pytest.raises(OpcSessionAlreadyConnectedError, match='disconnect'): + await opc_repository.connect() + + +@pytest.mark.asyncio +async def test_create_client_raises_when_client_exists(opc_repository, mock_client): + opc_repository.client = mock_client + + with pytest.raises(OpcClientAlreadyExistsError, match='already exists'): + await opc_repository._create_client() + + +@pytest.mark.asyncio +async def test_open_session_success(opc_repository): + closed_proto = MagicMock() + closed_proto.state = 'closed' opc_repository.client = AsyncMock() - result = await opc_repository.try_connect() + opc_repository.client.uaclient = MagicMock(protocol=closed_proto) + opc_repository.client.session_timeout = 600_000 + opc_repository.client.secure_channel_timeout = 600_000 + + open_proto = MagicMock() + open_proto.state = 'open' + open_proto.authentication_token = 'tok' + + async def connect_side_effect(): + opc_repository.client.uaclient.protocol = open_proto + + opc_repository.client.connect = AsyncMock(side_effect=connect_side_effect) + + result = await opc_repository._open_session() opc_repository.client.connect.assert_called_once() - assert opc_repository.last_reconnection_time is not None + assert opc_repository.last_reconnection_time is None assert result == (True, {}) + assert opc_repository._session_ready.is_set() @pytest.mark.asyncio -async def test_try_connect_fail(opc_repository): - opc_repository.last_reconnection_time = None - opc_repository.disconnect = AsyncMock() +async def test_open_session_raises_when_already_connected(opc_repository, mock_client): + opc_repository.client = mock_client + proto = MagicMock() + proto.state = 'open' + mock_client.uaclient = MagicMock(protocol=proto) + + with pytest.raises(OpcSessionAlreadyConnectedError, match='disconnect'): + await opc_repository._open_session() + + +@pytest.mark.asyncio +async def test_open_session_fail(opc_repository): + opc_repository._disconnect_locked = AsyncMock() opc_repository.client = MagicMock() - opc_repository.client.connect.side_effect = Exception('Test error') + opc_repository.client.uaclient = MagicMock(protocol=MagicMock(state='closed')) + opc_repository.client.connect = AsyncMock(side_effect=Exception('Test error')) - is_connected, error_data = await opc_repository.try_connect() + is_connected, error_data = await opc_repository._open_session() - opc_repository.disconnect.assert_called_once() + opc_repository._disconnect_locked.assert_called_once() opc_repository.client.connect.assert_called_once() assert is_connected is False assert error_data['notification_id'] == f'OPC_CONNECTION_ERROR_{opc_repository.id}' @@ -158,25 +217,18 @@ async def test_try_connect_fail(opc_repository): @pytest.mark.asyncio -async def test_try_connect_no_client(opc_repository): +async def test_open_session_raises_when_no_client(opc_repository): opc_repository.client = None - result = await opc_repository.try_connect() - assert result == ( - False, - { - 'notification_id': f'OPC_CONNECTION_ERROR_{opc_repository.id}', - 'message': 'Client is not initialized', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - }, - ) + + with pytest.raises(OpcClientNotInitializedError, match='not initialized'): + await opc_repository._open_session() @pytest.mark.asyncio async def test_disconnection_fallback_success(opc_repository, mock_client): opc_repository.client = mock_client mock_client.disconnect.return_value = True - result = await opc_repository.disconnection_fallback() + result = await opc_repository._disconnection_fallback() mock_client.disconnect.assert_called_once() assert result == [] @@ -186,7 +238,7 @@ async def test_disconnection_fallback_success(opc_repository, mock_client): async def test_disconnection_fallback_fail(opc_repository, mock_client): opc_repository.client = mock_client mock_client.disconnect.side_effect = Exception('Test error') - result = await opc_repository.disconnection_fallback() + result = await opc_repository._disconnection_fallback() assert result == [ {'attempt': 1, 'error': 'Test error', 'traceback': ANY}, {'attempt': 2, 'error': 'Test error', 'traceback': ANY}, @@ -200,10 +252,10 @@ async def test_disconnection_fallback_fail(opc_repository, mock_client): @pytest.mark.asyncio async def test_disconnect(opc_repository, mock_client): opc_repository.client = mock_client - opc_repository.disconnection_fallback = AsyncMock(return_value=[]) + opc_repository._disconnection_fallback = AsyncMock(return_value=[]) await opc_repository.disconnect() - opc_repository.disconnection_fallback.assert_called_once() + opc_repository._disconnection_fallback.assert_called_once() assert opc_repository.client is None @@ -216,12 +268,12 @@ async def test_disconnect_no_client(opc_repository): @pytest.mark.asyncio async def test_disconnect_error(opc_repository, mock_client): opc_repository.client = mock_client - opc_repository.disconnection_fallback = AsyncMock( + opc_repository._disconnection_fallback = AsyncMock( return_value=[{'attempt': 1, 'error': 'Test error', 'traceback': 'text'}] ) await opc_repository.disconnect() - opc_repository.disconnection_fallback.assert_called_once() + opc_repository._disconnection_fallback.assert_called_once() opc_repository.send_notification_async.assert_called_once_with( metadata=opc_repository.metadata, notification_id=f'OPC_DISCONNECTION_ERROR_{opc_repository.id}', @@ -238,91 +290,24 @@ async def test_disconnect_error(opc_repository, mock_client): @pytest.mark.asyncio async def test_validate_connection_none_client(opc_repository): opc_repository.client = None - opc_repository.connect = AsyncMock(return_value=(True, {})) response = await opc_repository.validate_connection() - assert response == (True, {}) - opc_repository.connect.assert_called_once() - - -# @pytest.mark.asyncio -# async def test_validate_connection_error_count_disconnect_error(opc_repository): -# opc_repository.error_count = 6 -# opc_repository.client = AsyncMock() -# opc_repository.disconnect = AsyncMock(side_effect=Exception('Test error')) -# opc_repository.connect = AsyncMock(return_value=(True, {})) - -# response = await opc_repository.validate_connection() -# assert response == opc_repository.connect.return_value -# opc_repository.disconnect.assert_called_once() -# opc_repository.connect.assert_called_once() -# opc_repository.logger.custom_error.assert_has_calls( -# [ -# call('Failed to disconnect from OPC server: Test error', ANY), -# ] -# ) + assert response == (False, opc_repository._not_connected_error()) @pytest.mark.asyncio -async def test_validate_connection_error_validate_connection_error(opc_repository): - opc_repository.client = MagicMock(uaclient=Exception('Test error')) - opc_repository.error_count = 0 - - response = await opc_repository.validate_connection() - - assert response == ( - False, - { - 'notification_id': f'OPC_CONNECTION_CHECK_ERROR_{opc_repository.id}', - 'message': "Failed to validate connection to OPC server: 'Exception' object has no attribute 'protocol'", - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': ANY, - }, - ) - - -@pytest.mark.asyncio -@patch('laborious.utils.repository.opc_repository.datetime') -async def test_validate_connection_lost_not_time_to_reconnect(_mock_datetime, opc_repository): - _mock_datetime.now = MagicMock(return_value=datetime(2025, 1, 1, 0, 0, 0)) - opc_repository.error_count = 0 +async def test_validate_connection_session_not_open(opc_repository): opc_repository.client = MagicMock() opc_repository.client.uaclient.protocol = None - opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) - opc_repository.connect = MagicMock(return_value=(True, {})) response = await opc_repository.validate_connection() - opc_repository.connect.assert_not_called() - assert response == ( - False, - { - 'notification_id': f'OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{opc_repository.id}', - 'message': f'OPC server {opc_repository.id} is not connected, waiting for next reconnection window...', - 'block': 'opc_repository', - 'level': NotificationLevel.WARNING, - }, - ) - -@pytest.mark.asyncio -@patch('laborious.utils.repository.opc_repository.datetime') -async def test_validate_connection_lost_time_to_reconnect(mock_datetime, opc_repository): - mock_datetime.now = MagicMock(return_value=datetime(2025, 1, 1, 1, 0, 0)) - opc_repository.error_count = 0 - opc_repository.client = AsyncMock() - opc_repository.client.uaclient.protocol = None - opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) - opc_repository.connect = AsyncMock(return_value=(True, {})) - - response = await opc_repository.validate_connection() - opc_repository.connect.assert_called_once() - assert response == opc_repository.connect.return_value + assert response == (False, opc_repository._not_connected_error()) + opc_repository.error.assert_called_once() @pytest.mark.asyncio async def test_validate_connection_success(opc_repository): opc_repository.client = MagicMock() - opc_repository.error_count = 0 opc_repository.client.uaclient.protocol = MagicMock() opc_repository.client.uaclient.protocol.state = 'open' @@ -337,9 +322,7 @@ async def test_write_data_validate_connection_do_nothing(opc_repository): mock_node = AsyncMock() opc_repository.client.get_node.return_value = mock_node - result = await opc_repository.write_data( - 'ns=2;s=TestNode', 42.0, 'float', opc_repository.logger, metadata['metadata'] - ) + result = await opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) opc_repository.validate_connection.assert_called_once() opc_repository.client.get_node.assert_called_once_with('ns=2;s=TestNode') @@ -350,11 +333,8 @@ async def test_write_data_validate_connection_do_nothing(opc_repository): async def test_write_data_validate_connection_failed(opc_repository): opc_repository.validate_connection = AsyncMock(return_value=(False, {})) opc_repository.client = AsyncMock() - opc_repository.error_count = 0 - result = await opc_repository.write_data( - 'ns=2;s=TestNode', 42.0, 'float', opc_repository.logger, metadata['metadata'] - ) + result = await opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) opc_repository.validate_connection.assert_called_once() opc_repository.client.get_node.assert_not_called() @@ -365,11 +345,10 @@ async def test_write_data_validate_connection_failed(opc_repository): async def test_write_data_get_node_failed(opc_repository): opc_repository.validate_connection = AsyncMock(return_value=(True, {})) opc_repository.client = AsyncMock() - opc_repository.error_count = 0 opc_repository.client.get_node = MagicMock(side_effect=Exception('Test error')) is_success, error_data = await opc_repository.write_data( - 'ns=2;s=TestNode', 42.0, 'float', opc_repository.logger, metadata['metadata'] + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] ) opc_repository.validate_connection.assert_called_once() @@ -393,7 +372,7 @@ async def test_write_data_invalid_data_type(opc_repository, mock_client): mock_client.get_node = MagicMock(return_value=mock_node) is_success, error_data = await opc_repository.write_data( - 'ns=2;s=TestNode', 42.0, 'invalid_type', opc_repository.logger, metadata['metadata'] + 'ns=2;s=TestNode', 42.0, 'invalid_type', metadata['metadata'] ) opc_repository.validate_connection.assert_called_once() @@ -417,9 +396,7 @@ async def test_write_data(opc_repository, mock_client): mock_node = AsyncMock() mock_client.get_node = MagicMock(return_value=mock_node) - result = await opc_repository.write_data( - 'ns=2;s=TestNode', 42.0, 'float', opc_repository.logger, metadata['metadata'] - ) + result = await opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) mock_client.get_node.assert_called_once_with('ns=2;s=TestNode') mock_node.write_value.assert_called_once() @@ -431,12 +408,11 @@ async def test_write_data_write_value_failed(opc_repository, mock_client): opc_repository.validate_connection = AsyncMock(return_value=(True, {})) opc_repository.client = mock_client mock_node = AsyncMock() - opc_repository.error_count = 0 mock_client.get_node = MagicMock(return_value=mock_node) mock_node.write_value.side_effect = Exception('Test error') is_success, error_data = await opc_repository.write_data( - 'ns=2;s=TestNode', 42.0, 'float', opc_repository.logger, metadata['metadata'] + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] ) opc_repository.validate_connection.assert_called_once() @@ -451,3 +427,108 @@ async def test_write_data_write_value_failed(opc_repository, mock_client): assert error_data['block'] == 'opc_repository' assert error_data['level'] == NotificationLevel.ERROR assert error_data['attachment_content'] is not None + + +def test_is_reconnectable_opcua_bad(): + assert is_reconnectable_opcua_bad(BadSessionIdInvalid()) is True + assert is_reconnectable_opcua_bad(BadNodeIdUnknown()) is False + assert is_reconnectable_opcua_bad(Exception('other')) is False + + +@pytest.mark.asyncio +async def test_write_data_bad_session_id_invalid_schedules_reconnect(opc_repository, mock_client): + opc_repository.validate_connection = AsyncMock(return_value=(True, {})) + opc_repository.client = mock_client + opc_repository._start_reconnect_on_bad = AsyncMock() + mock_node = AsyncMock() + mock_client.get_node = MagicMock(return_value=mock_node) + mock_node.write_value.side_effect = BadSessionIdInvalid() + + is_success, error_data = await opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) + + mock_node.write_value.assert_called_once() + opc_repository._start_reconnect_on_bad.assert_called_once() + assert is_success is False + assert error_data['opc_error_kind'] == 'session_bad' + assert error_data['opc_status'] == 'BadSessionIdInvalid' + + +@pytest.mark.asyncio +async def test_write_data_reconnect_in_progress_immediate(opc_repository): + opc_repository._session_ready.clear() + opc_repository._reconnect_task = asyncio.create_task(asyncio.sleep(60)) + opc_repository.validate_connection = AsyncMock() + + is_success, error_data = await opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) + + opc_repository._reconnect_task.cancel() + with pytest.raises(asyncio.CancelledError): + await opc_repository._reconnect_task + opc_repository._reconnect_task = None + + opc_repository.validate_connection.assert_not_called() + assert is_success is False + assert error_data['opc_error_kind'] == 'reconnect_in_progress' + + +@pytest.mark.asyncio +async def test_start_reconnect_on_bad_skips_within_interval(opc_repository): + opc_repository.last_reconnection_time = datetime.now() + opc_repository.reconnection_interval = 3600 + + await opc_repository._start_reconnect_on_bad('BadSessionIdInvalid', 'tok') + + assert opc_repository._reconnect_task is None + + +@pytest.mark.asyncio +async def test_parallel_bad_writes_single_reconnect_task(opc_repository, mock_client): + opc_repository.validate_connection = AsyncMock(return_value=(True, {})) + opc_repository.client = mock_client + opc_repository.reconnection_interval = 0 + opc_repository.last_reconnection_time = None + mock_node = AsyncMock() + mock_client.get_node = MagicMock(return_value=mock_node) + mock_node.write_value.side_effect = BadSessionIdInvalid() + + connect_count = 0 + + async def slow_reconnect(): + nonlocal connect_count + connect_count += 1 + await asyncio.sleep(0.05) + opc_repository._session_ready.set() + return True, {} + + opc_repository._reconnect_locked = slow_reconnect + + results = await asyncio.gather( + opc_repository.write_data('ns=2;s=TestNode', 1.0, 'float', metadata['metadata']), + opc_repository.write_data('ns=2;s=TestNode2', 2.0, 'float', metadata['metadata']), + ) + await asyncio.sleep(0.15) + + assert connect_count <= 1 + assert 1 <= mock_node.write_value.call_count <= 2 + error_kinds = [r[1].get('opc_error_kind') for r in results] + assert error_kinds.count('session_bad') >= 1 + assert all(k in ('session_bad', 'reconnect_in_progress') for k in error_kinds) + + +@pytest.mark.asyncio +@patch('laborious.utils.repository.opc_repository.datetime') +async def test_reconnect_locked_sets_last_reconnection_time(mock_datetime, opc_repository): + mock_datetime.now = MagicMock(return_value=datetime(2025, 1, 1, 12, 0, 0)) + opc_repository._disconnect_locked = AsyncMock() + opc_repository._connect_locked = AsyncMock(return_value=(True, {})) + + result = await opc_repository._reconnect_locked() + + opc_repository._disconnect_locked.assert_called_once() + opc_repository._connect_locked.assert_called_once() + assert result == (True, {}) + assert opc_repository.last_reconnection_time == datetime(2025, 1, 1, 12, 0, 0) diff --git a/values.yaml b/values.yaml index 31dd4fa..b07d60b 100644 --- a/values.yaml +++ b/values.yaml @@ -17,8 +17,8 @@ image: imagePullSecrets: - name: docker-hub-secret # This is to override the chart name. -nameOverride: "sientia-laborious-worker" -fullnameOverride: "sientia-laborious-worker" +nameOverride: "sientia-laborious-legacy-worker" +fullnameOverride: "sientia-laborious-legacy-worker" namespace: sientia # This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/ @@ -31,7 +31,7 @@ serviceAccount: annotations: {} # The name of the service account to use. # If not set and create is true, a name is generated using the fullname template - name: "sientia-laborious-worker" + name: "sientia-laborious-legacy-worker" # This is for setting Kubernetes Annotations to a Pod. # For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/ @@ -157,7 +157,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git" - name: GITHUB_BRANCH - value: "feature/SIENTIAPDE-1712" + value: "fix/SIENTIAPDE-1811" - name: PYTHON_APP value: "laborious.worker.worker" @@ -307,7 +307,7 @@ ssh: # kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp -# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.6.0 +# helm upgrade --install sientia-laborious-legacy-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.6.0 # kubectl create secret generic git-ssh-key-sientia-laborious-worker \ # --namespace sientia \