From 5318c826772b27780a7ea34d50b4bdb06ca02f03 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Wed, 28 Oct 2026 14:29:33 -0300 Subject: [PATCH] feat(SIENTIAPDE-2202): write timestamp and quality per OPC server - write_timestamp/write_quality come from the command, then the server document (connectors/opc-servers, read even when the reads have the connection open, same 60 s cache), then the WRITE_TIMESTAMP/WRITE_QUALITY env defaults. - The quality comes from the command's `quality` (the prediction status: Good/Bad/Uncertain, Good when absent); an unknown one is a bad_quality error. - A server that refuses them (BadWriteNotSupported, e.g. Kepware) gets the value written again alone and is remembered until restart: one warning and opc_write_value_only_fallback_total per server. Co-Authored-By: Claude Opus 5.5 --- .env.example | 1 + README.md | 14 ++- ingestor/connection/servers.py | 51 +++++++-- ingestor/engines/write.py | 106 +++++++++++++++--- ingestor/metrics.py | 6 + tests/unit/connection/test_servers.py | 64 +++++++++++ tests/unit/engines/test_write.py | 152 +++++++++++++++++++++++++- 7 files changed, 370 insertions(+), 24 deletions(-) diff --git a/.env.example b/.env.example index 27003d2..71021ce 100644 --- a/.env.example +++ b/.env.example @@ -23,6 +23,7 @@ GRPC_ENABLED="false" GRPC_PORT="50051" CERT_BUCKET="sientia-dataops" CERT_CACHE_DIR="/tmp/opc-certs" +# Defaults of the write options; the server document and the command override them. WRITE_TIMESTAMP="true" WRITE_QUALITY="true" WRITE_MIN_PAUSE_S="0.0" diff --git a/README.md b/README.md index ee7ea36..124ab61 100644 --- a/README.md +++ b/README.md @@ -79,9 +79,21 @@ The Laborious publishes one command per tag (SIENTIAPDE-1975): {"server_id": "1", "addr": "ns=2;s=Pred", "value": 1.25, "data_type": "float", "model_id": "2", "model_name": "sin-approx", "schema": "sientia_data", "table_name": "predictions", "timestamp": "2026-10-13T12:00:00-03:00", - "write_timestamp": false, "write_quality": false} + "quality": "Good"} ``` +- **Timestamp and quality** (SIENTIAPDE-2202): besides the value, the write carries the + prediction `timestamp` as `SourceTimestamp` and the `quality` (the prediction status: + `Good`, `Bad` or `Uncertain`; `Good` when absent) as `StatusCode`. Each one is on or + off by, in order: `write_timestamp`/`write_quality` in the command, the same boolean + fields on the server document (`connectors` or `opc-servers`, cached like the + connection), then the `WRITE_TIMESTAMP`/`WRITE_QUALITY` env vars (default `true`). + When a server refuses them (`BadWriteNotSupported`, e.g. Kepware), the value is + written again alone, the server is remembered until the gateway restarts (its next + writes go with the value alone), a warning is logged and + `opc_write_value_only_fallback_total` is incremented. Set the two fields to `false` on + such a server's document to skip the refused attempt. + - **Server connection**: the command carries only `server_id`. The connection is resolved by `ServerConfigResolver` (`ingestor/connection/servers.py`): `url`/ `server_uri`/`use_security` in the command when present (override), else the diff --git a/ingestor/connection/servers.py b/ingestor/connection/servers.py index fa4b431..b313700 100644 --- a/ingestor/connection/servers.py +++ b/ingestor/connection/servers.py @@ -4,6 +4,7 @@ The Laborious only knows the `server_id` of a write tag (pipeline `write_tags`); connection (url, server URI, security) belongs to the gateway, which owns every OPC session. It is resolved with the same rule the orchestrator uses to build the read slots (SIENTIAPDE-1974): the tag database `connectors` first, the legacy `opc-servers` otherwise. +The same document may also set the write options of the server (SIENTIAPDE-2202). """ from typing import Any @@ -15,6 +16,7 @@ from ingestor.connection.pool import OpcConnectionPool CONNECTORS = 'connectors' OPC_SERVERS = 'opc-servers' +WRITE_OPTIONS = ('write_timestamp', 'write_quality') def _id_forms(server_id: str) -> list[Any]: @@ -47,8 +49,15 @@ def _opc_server_cfg(server: dict[str, Any]) -> dict[str, Any] | None: } +def _write_options(document: dict[str, Any]) -> dict[str, bool]: + """The write options the server document sets. Only booleans count: a hand-edited + string such as "false" must not turn an option on.""" + return {key: document[key] for key in WRITE_OPTIONS if isinstance(document.get(key), bool)} + + class ServerConfigResolver: - """Resolve and cache the connection settings of an OPC server by `server_id`. + """Resolve and cache the connection settings (and write options) of an OPC server by + `server_id`. Order: settings carried by the command itself (an explicit override; the C2 command of the Laborious carries none), the live pooled connection (what the read @@ -91,17 +100,40 @@ class ServerConfigResolver: pooled = self.pool.config_for(server_id) if pooled is not None: return pooled + server = await self._server(server_id) + return dict(server['connection']) if server is not None else None + + async def write_options(self, server_id: str) -> dict[str, bool]: + """`write_timestamp`/`write_quality` set on the server document (SIENTIAPDE-2202), + only the ones it sets: the gateway defaults apply to the others. + + Read from MongoDB even when the read path has the connection open (`resolve` + skips MongoDB then), with the same cache. Never raises: when MongoDB cannot be + read, the gateway defaults apply and the write goes on. + """ + try: + server = await self._server(str(server_id)) + except Exception as e: + self.logger.warning( + f'Write options of OPC server {server_id} not read, using the gateway defaults: {e}' + ) + return {} + return dict(server['write_options']) if server is not None else {} + + async def _server(self, server_id: str) -> dict[str, Any] | None: now = self.clock.elapsed() cached = self._cache.get(server_id) if cached is not None: - cfg, cached_at = cached - if now - cached_at < (self.ttl_s if cfg is not None else self.missing_ttl_s): - return dict(cfg) if cfg is not None else None - cfg = await self._lookup(server_id) - self._cache[server_id] = (cfg, now) - return dict(cfg) if cfg is not None else None + server, cached_at = cached + if now - cached_at < (self.ttl_s if server is not None else self.missing_ttl_s): + return server + server = await self._lookup(server_id) + self._cache[server_id] = (server, now) + return server async def _lookup(self, server_id: str) -> dict[str, Any] | None: + """The connection settings and write options of the first document that knows + the server, or None.""" ids = {'$in': _id_forms(server_id)} connector = await self.mongo.find_one( CONNECTORS, @@ -110,11 +142,12 @@ class ServerConfigResolver: ) cfg = _connector_cfg(connector) if connector else None if cfg is not None: - return cfg + return {'connection': cfg, 'write_options': _write_options(connector)} server = await self.mongo.find_one(OPC_SERVERS, {'id': ids}, self.metadata) cfg = _opc_server_cfg(server) if server else None if cfg is None: self.logger.warning( f'OPC server {server_id} not found in {CONNECTORS} nor {OPC_SERVERS}' ) - return cfg + return None + return {'connection': cfg, 'write_options': _write_options(server)} diff --git a/ingestor/engines/write.py b/ingestor/engines/write.py index 959a7d8..e9371f7 100644 --- a/ingestor/engines/write.py +++ b/ingestor/engines/write.py @@ -9,7 +9,7 @@ import asyncio from dataclasses import dataclass from asyncua import ua -from asyncua.ua.uaerrors import UaStatusCodeError +from asyncua.ua.uaerrors import BadWriteNotSupported, UaStatusCodeError from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.observability.logger import Logger from sientia_do.observability.metrics_controller import MetricsController @@ -57,6 +57,23 @@ _SESSION_BAD_NAMES = frozenset( ) +# `quality` of the C2 command (the Laborious sends its prediction_status) -> OPC quality. +_QUALITY_CODES = { + 'good': ua.StatusCodes.Good, + 'uncertain': ua.StatusCodes.Uncertain, + 'bad': ua.StatusCodes.Bad, +} + + +class _InvalidFieldError(Exception): + """A command field needed by an option that is on (timestamp, quality) is bad.""" + + def __init__(self, error_type: str, message: str): + super().__init__(message) + self.error_type = error_type + self.message = message + + @dataclass class WriteResult: ok: bool @@ -85,11 +102,14 @@ class WriteEngine(SientiaMonitoring): # Resolves the server connection by server_id (SIENTIAPDE-1975 GW1). Without it, # only the connection carried by the command itself is used. self.servers = servers + # Defaults (env); the server document and the command override them (SIENTIAPDE-2202). self.write_timestamp = write_timestamp self.write_quality = write_quality self.min_pause_s = min_pause_s self.clock = MonotonicClock() self._last_write_at: float | None = None + # Servers that refused timestamp/quality: written with the value alone. + self._value_only_servers: set[str] = set() SientiaMonitoring.__init__( self, @@ -110,12 +130,6 @@ class WriteEngine(SientiaMonitoring): except ConversionError as e: return await self._emit_error(cmd, 'conversion', str(e), 'error') - try: - dv = self._build_datavalue(cmd, typed) - except (KeyError, TypeError, ValueError) as e: - # write_timestamp on with a missing or non-ISO timestamp. - return await self._emit_error(cmd, 'bad_timestamp', repr(e), 'error') - try: server_cfg = await self._server_cfg(cmd) except Exception as e: @@ -125,11 +139,18 @@ class WriteEngine(SientiaMonitoring): cmd, 'server_not_found', f'OPC server {cmd["server_id"]} is not configured', 'error' ) + server_id = str(cmd['server_id']) + write_timestamp, write_quality = await self._write_flags(server_id, cmd) + try: + dv = self._build_datavalue(cmd, typed, write_timestamp, write_quality) + except _InvalidFieldError as e: + return await self._emit_error(cmd, e.error_type, e.message, 'error') + start = self.clock.elapsed() try: client = await self.pool.get(cmd['server_id'], server_cfg) node = client.get_node(cmd['addr']) - await node.write_value(dv) + await self._write_node(node, dv, server_id) except Exception as e: status = 'session_bad' if self._is_session_bad(e) else 'error' return await self._emit_error(cmd, type(e).__name__, str(e), status) @@ -163,17 +184,76 @@ class WriteEngine(SientiaMonitoring): # direct index: unknown dtypes already became ConversionError in to_typed_value return _VARIANT_TYPES[data_type.strip().lower()] - def _build_datavalue(self, cmd: dict, typed_value) -> ua.DataValue: + async def _write_flags(self, server_id: str, cmd: dict) -> tuple[bool, bool]: + """`write_timestamp` and `write_quality` of this write (SIENTIAPDE-2202): the + command, then the server document, then the gateway env. A server that already + refused them (BadWriteNotSupported) gets the value alone.""" + if server_id in self._value_only_servers: + return False, False + options = await self.servers.write_options(server_id) if self.servers is not None else {} + + def flag(key: str, default: bool) -> bool: + value = cmd.get(key) + return bool(options.get(key, default) if value is None else value) + + return flag('write_timestamp', self.write_timestamp), flag( + 'write_quality', self.write_quality + ) + + def _build_datavalue( + self, cmd: dict, typed_value, write_timestamp: bool, write_quality: bool + ) -> ua.DataValue: + """The value, with the prediction timestamp and quality when they are on. + + Raises: + _InvalidFieldError: When an option is on and its command field is missing or bad. + """ variant = ua.Variant(typed_value, self._variant_type(cmd['data_type'])) - write_quality = cmd.get('write_quality', self.write_quality) - write_timestamp = cmd.get('write_timestamp', self.write_timestamp) # NOTE: DataValue.StatusCode default_factory populates a Good status # even when omitted, so "off" must pass StatusCode=None explicitly # or the flag silently no-ops. - status_code = ua.StatusCode(ua.UInt32(ua.StatusCodes.Good)) if write_quality else None - source_ts = ua.DateTime.fromisoformat(cmd['timestamp']) if write_timestamp else None + status_code = self._status_code(cmd) if write_quality else None + try: + source_ts = ua.DateTime.fromisoformat(cmd['timestamp']) if write_timestamp else None + except (KeyError, TypeError, ValueError) as e: + raise _InvalidFieldError('bad_timestamp', repr(e)) from e return ua.DataValue(variant, StatusCode=status_code, SourceTimestamp=source_ts) + @staticmethod + def _status_code(cmd: dict) -> ua.StatusCode: + """The OPC quality of the command's `quality` (the prediction status). Commands + without it (before SIENTIAPDE-2202) are written as Good.""" + quality = cmd.get('quality') or 'good' + code = _QUALITY_CODES.get(str(quality).strip().lower()) + if code is None: + raise _InvalidFieldError('bad_quality', f'unknown quality {quality!r}') + return ua.StatusCode(ua.UInt32(code)) + + async def _write_node(self, node, dv: ua.DataValue, server_id: str) -> None: + """Write `dv`. When the server refuses the timestamp/quality it carries + (BadWriteNotSupported, e.g. Kepware), write the value alone and remember the + server, so its next writes go without them.""" + try: + await node.write_value(dv) + except BadWriteNotSupported: + if dv.StatusCode is None and dv.SourceTimestamp is None: + raise + await node.write_value(ua.DataValue(dv.Value, StatusCode=None)) + await self._remember_value_only(server_id) + + async def _remember_value_only(self, server_id: str) -> None: + self._value_only_servers.add(server_id) + self.logger.warning( + f'OPC server {server_id} refused the timestamp/quality of a write ' + f'(BadWriteNotSupported): its writes go with the value alone until the gateway ' + f'restarts. Set write_timestamp/write_quality to false on its document to skip ' + f'the refused attempt.' + ) + await self.emit_metric( + metric_object=metrics.OPC_WRITE_VALUE_ONLY_FALLBACK, + tags={'pod_id': self.pod_id, 'server_name': server_id}, + ) + def _is_session_bad(self, exc: Exception) -> bool: return isinstance(exc, UaStatusCodeError) and type(exc).__name__ in _SESSION_BAD_NAMES diff --git a/ingestor/metrics.py b/ingestor/metrics.py index b7fcd76..c2087c8 100644 --- a/ingestor/metrics.py +++ b/ingestor/metrics.py @@ -156,6 +156,12 @@ OPC_WRITE_STATUS_BY_MODEL = Gauge( 'Last write status per model (1=ok,0=error)', ['pod_id', 'model_id', 'status'], ) +OPC_WRITE_VALUE_ONLY_FALLBACK = Counter( + 'opc_write_value_only_fallback_total', + 'Servers that refused the timestamp/quality of a write (BadWriteNotSupported); ' + 'their writes go with the value alone', + ['pod_id', 'server_name'], +) OPC_WRITE_STATUS_UPDATE_TOTAL = Counter( 'opc_write_status_update_total', 'write_status UPDATEs on the prediction rows', diff --git a/tests/unit/connection/test_servers.py b/tests/unit/connection/test_servers.py index 3f98131..acab1f7 100644 --- a/tests/unit/connection/test_servers.py +++ b/tests/unit/connection/test_servers.py @@ -149,3 +149,67 @@ async def test_mongo_errors_propagate_to_the_caller(): with pytest.raises(RuntimeError, match='mongo down'): await resolver.resolve('1', {}) + + +# --- SIENTIAPDE-2202: write options on the server document --- + + +@pytest.mark.asyncio +async def test_write_options_come_from_the_connector(): + resolver = make_resolver( + connector={**CONNECTOR, 'write_timestamp': False, 'write_quality': True}, + opc_server={**OPC_SERVER, 'write_timestamp': True}, + ) + + assert await resolver.write_options('1') == {'write_timestamp': False, 'write_quality': True} + + +@pytest.mark.asyncio +async def test_write_options_come_from_opc_servers_without_tag_database(): + resolver = make_resolver(opc_server={**OPC_SERVER, 'write_quality': False}) + + assert await resolver.write_options('1') == {'write_quality': False} + + +@pytest.mark.asyncio +async def test_only_boolean_write_options_count(): + resolver = make_resolver( + connector={**CONNECTOR, 'write_timestamp': 'false', 'write_quality': 0} + ) + + assert await resolver.write_options('1') == {} + + +@pytest.mark.asyncio +async def test_write_options_are_read_even_with_a_live_pooled_connection(): + pooled = {'url': 'opc.tcp://read:1', 'server_uri': 'urn:read', 'use_security': True} + resolver = make_resolver(connector={**CONNECTOR, 'write_timestamp': False}, pooled=pooled) + + assert await resolver.resolve('1', {}) == pooled + assert await resolver.write_options('1') == {'write_timestamp': False} + + +@pytest.mark.asyncio +async def test_write_options_share_the_lookup_cache(): + resolver = make_resolver(connector={**CONNECTOR, 'write_quality': False}) + + await resolver.resolve('1', {}) + await resolver.write_options('1') + + assert resolver.mongo.find_one.await_count == 1 + + +@pytest.mark.asyncio +async def test_write_options_of_an_unknown_server_are_empty(): + resolver = make_resolver() + + assert await resolver.write_options('9') == {} + + +@pytest.mark.asyncio +async def test_write_options_never_raise(): + resolver = make_resolver() + resolver.mongo.find_one = AsyncMock(side_effect=RuntimeError('mongo down')) + + assert await resolver.write_options('1') == {} + assert 'mongo down' in resolver.logger.warning.call_args.args[0] diff --git a/tests/unit/engines/test_write.py b/tests/unit/engines/test_write.py index a913151..4f61a0a 100644 --- a/tests/unit/engines/test_write.py +++ b/tests/unit/engines/test_write.py @@ -3,6 +3,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest from asyncua import ua +from asyncua.ua.uaerrors import BadTypeMismatch, BadWriteNotSupported import ingestor.metrics as metrics from ingestor.engines.write import WriteEngine @@ -10,9 +11,10 @@ from ingestor.engines.write import WriteEngine SERVER_CFG = {'url': 'opc.tcp://kepserver:49320', 'server_uri': 'urn:kep', 'use_security': True} -def make_resolver(cfg=SERVER_CFG): +def make_resolver(cfg=SERVER_CFG, write_options=None): resolver = MagicMock() resolver.resolve = AsyncMock(return_value=cfg) + resolver.write_options = AsyncMock(return_value=write_options or {}) return resolver @@ -367,3 +369,151 @@ async def test_write_bad_timestamp_records_error_without_raising(timestamp): assert result.opc_status_code == 'bad_timestamp' eng.pool.get.assert_not_awaited() eng.sink.update_write_status.assert_awaited_once() + + +# --- SIENTIAPDE-2202: timestamp/quality options, quality from the command, value-only retry --- + + +@pytest.mark.asyncio +async def test_server_options_override_the_gateway_defaults(): + eng = make_engine( + write_timestamp=True, + write_quality=True, + servers=make_resolver(write_options={'write_timestamp': False}), + ) + node = make_node() + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd()) + + eng.servers.write_options.assert_awaited_once_with('srv1') + dv = node.write_value.await_args.args[0] + assert dv.SourceTimestamp is None + assert dv.StatusCode is not None # not set on the server: the gateway default applies + + +@pytest.mark.asyncio +async def test_command_options_override_the_server_options(): + eng = make_engine( + servers=make_resolver(write_options={'write_timestamp': False, 'write_quality': False}) + ) + node = make_node() + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd(write_timestamp=True)) + + dv = node.write_value.await_args.args[0] + assert dv.SourceTimestamp == datetime.fromisoformat('2026-07-13T12:00:00Z') + assert dv.StatusCode is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ('quality', 'expected'), + [ + ('Good', ua.StatusCodes.Good), + ('Bad', ua.StatusCodes.Bad), + ('uncertain', ua.StatusCodes.Uncertain), + (None, ua.StatusCodes.Good), + ], +) +async def test_quality_comes_from_the_command(quality, expected): + eng = make_engine(write_quality=True) + node = make_node() + eng.pool.get.return_value = make_client(node) + + result = await eng.write(base_cmd(quality=quality)) + + assert result.status == 'ok' + assert node.write_value.await_args.args[0].StatusCode.value == expected + + +@pytest.mark.asyncio +async def test_unknown_quality_records_error_without_writing(): + eng = make_engine(write_quality=True) + + result = await eng.write(base_cmd(quality='Excellent')) + + assert result.status == 'error' + assert result.opc_status_code == 'bad_quality' + eng.pool.get.assert_not_awaited() + eng.sink.update_write_status.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_unknown_quality_is_ignored_when_quality_is_off(): + eng = make_engine(write_quality=False) + node = make_node() + eng.pool.get.return_value = make_client(node) + + result = await eng.write(base_cmd(quality='Excellent')) + + assert result.status == 'ok' + assert node.write_value.await_args.args[0].StatusCode is None + + +@pytest.mark.asyncio +async def test_refused_timestamp_and_quality_retry_with_the_value_alone(): + eng = make_engine(write_timestamp=True, write_quality=True) + node = make_node() + node.write_value.side_effect = [BadWriteNotSupported(), None] + eng.pool.get.return_value = make_client(node) + + result = await eng.write(base_cmd()) + + assert result.status == 'ok' + first, retry = (c.args[0] for c in node.write_value.await_args_list) + assert first.SourceTimestamp is not None and first.StatusCode is not None + assert retry.SourceTimestamp is None and retry.StatusCode is None + assert retry.Value.Value == 42.0 + assert 'srv1' in eng.logger.warning.call_args.args[0] + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_WRITE_VALUE_ONLY_FALLBACK, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1'}, + ) + eng.sink.update_write_status.assert_awaited_once_with(base_cmd(), result) + + +@pytest.mark.asyncio +async def test_server_that_refused_gets_the_value_alone_next_time(): + eng = make_engine(write_timestamp=True, write_quality=True) + node = make_node() + node.write_value.side_effect = [BadWriteNotSupported(), None, None] + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd()) + result = await eng.write(base_cmd()) + + assert result.status == 'ok' + assert node.write_value.await_count == 3 # refused + retry, then one value-only write + last = node.write_value.await_args.args[0] + assert last.SourceTimestamp is None and last.StatusCode is None + eng.servers.write_options.assert_awaited_once() # not looked up once remembered + assert eng.logger.warning.call_count == 1 + + +@pytest.mark.asyncio +async def test_refused_value_alone_is_an_error(): + eng = make_engine(write_timestamp=False, write_quality=False) + node = make_node() + node.write_value.side_effect = BadWriteNotSupported() + eng.pool.get.return_value = make_client(node) + + result = await eng.write(base_cmd()) + + assert result.status == 'error' + assert result.opc_status_code == 'BadWriteNotSupported' + assert node.write_value.await_count == 1 + + +@pytest.mark.asyncio +async def test_failed_value_only_retry_is_reported(): + eng = make_engine(write_timestamp=True, write_quality=True) + node = make_node() + node.write_value.side_effect = [BadWriteNotSupported(), BadTypeMismatch()] + eng.pool.get.return_value = make_client(node) + + result = await eng.write(base_cmd()) + + assert result.status == 'error' + assert result.opc_status_code == 'BadTypeMismatch'