gitea_admin/sientia-dataops-laborious_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1811

Browse Source 
Enhance OPC UA communication and metrics tracking

- Updated README.md to include new OPC UA Communication section and detailed metrics for session and write diagnostics.
- Added new metrics in laborious/metrics.py for tracking OPC UA session states and write attempts.
- Refactored OPC activity in laborious/activities/opc.py to handle session errors and improve error reporting.
- Updated e2e tests to cover new scenarios for OPC session/channel errors and reconnect handling.
- Modified .gitignore to include relatorio files and mlruns directory.
- Added ipykernel to requirements-dev.txt for Jupyter notebook support.
This commit is contained in:
vitor-aignosi
2026-05-15 15:28:28 -03:00
parent 473bd0b03f
commit 638d5b70b4
15 changed files with 1370 additions and 424 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
laborious/activities/opc.py
View File

@@ -15,6 +15,16 @@ 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}'
class OPC(SientiaMonitoring):
@@ -118,29 +128,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 +150,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(
@@ -205,7 +204,7 @@ class OPC(SientiaMonitoring):
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.
@@ -234,10 +233,13 @@ class OPC(SientiaMonitoring):
"""
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(
response_time, error_info = await self.write_data(
server_id=server_id,
tag=tag,
data=data.head(1)['prediction'].values[0],
@@ -245,6 +247,13 @@ class OPC(SientiaMonitoring):
tag_type='prediction',
metadata=metadata,
)
if error_info:
kind = error_info.get('opc_error_kind')
if kind == 'session_bad':
session_bad_seen = True
session_bad_status = error_info.get('opc_status', session_bad_status)
elif kind == 'reconnect_in_progress':
reconnect_in_progress_seen = True
if response_time is not None:
self.info(
f'Prediction data written to OPC server {server_id} for tag {tag}.',
@@ -254,7 +263,7 @@ class OPC(SientiaMonitoring):
if 'confidence_tags' in config:
for tag, tag_config in config['confidence_tags'].items():
response_time = await self.write_data(
response_time, error_info = await self.write_data(
server_id=server_id,
tag=tag,
data=data.head(1)['prediction_confidence'].values[0],
@@ -262,6 +271,13 @@ class OPC(SientiaMonitoring):
tag_type='confidence',
metadata=metadata,
)
if error_info:
kind = error_info.get('opc_error_kind')
if kind == 'session_bad':
session_bad_seen = True
session_bad_status = error_info.get('opc_status', session_bad_status)
elif kind == 'reconnect_in_progress':
reconnect_in_progress_seen = True
if response_time is not None:
self.info(
f'Confidence data written to OPC server {server_id} for tag {tag}.',
@@ -271,7 +287,13 @@ class OPC(SientiaMonitoring):
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 +323,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 +334,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 +403,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)

34
laborious/metrics.py
View File

@@ -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(

715
laborious/utils/repository/opc_repository.py
View File

@@ -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,

2
laborious/worker/worker.py
View File

@@ -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'))