SIENTIAPDE-1975: the gateway is the only writer that updates write_status
The Laborious no longer records OPC outcomes (it inserts the row as 'pending' and only publishes the commands), so the comments and docs that asked to keep a Laborious copy of the UPDATE in sync are updated. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -60,7 +60,7 @@ Any change to slot/lease counts should be traced through `manage_leases`/`manage
|
|||||||
|
|
||||||
**ServerReader** (SIENTIAPDE-1974, default `READ_ENGINE=v2`; the legacy `OpcManager` stays as plan B with `READ_ENGINE=legacy` — slot v1 only, certificate files from the Gitea clone — until it is unused and the tests pass the same without it) reads one OPC server through a `SubscriptionEngine`: the connection comes from the shared `OpcConnectionPool` (also used by NATS writes and gRPC), which fetches the platform certificate from MinIO via `CertProvider` and uses `SecurityPolicyBasic256Sha256` — no certificate files in the repo/image. Slots are parsed by `io/slots.py` (`parse_slot`), which accepts slot v2 (`names[]`) and converts the legacy v1 (`topics`). `datachange_notification` fans each value out to every `names` entry via `DataSink`. Subscribed connections are pinned in the pool so the idle sweep never drops them. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles` / `check_opc_listenning`); a lost server is shut down (pool connection evicted) and re-initialized on the next loop.
|
**ServerReader** (SIENTIAPDE-1974, default `READ_ENGINE=v2`; the legacy `OpcManager` stays as plan B with `READ_ENGINE=legacy` — slot v1 only, certificate files from the Gitea clone — until it is unused and the tests pass the same without it) reads one OPC server through a `SubscriptionEngine`: the connection comes from the shared `OpcConnectionPool` (also used by NATS writes and gRPC), which fetches the platform certificate from MinIO via `CertProvider` and uses `SecurityPolicyBasic256Sha256` — no certificate files in the repo/image. Slots are parsed by `io/slots.py` (`parse_slot`), which accepts slot v2 (`names[]`) and converts the legacy v1 (`topics`). `datachange_notification` fans each value out to every `names` entry via `DataSink`. Subscribed connections are pinned in the pool so the idle sweep never drops them. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles` / `check_opc_listenning`); a lost server is shut down (pool connection evicted) and re-initialized on the next loop.
|
||||||
|
|
||||||
**Writes** (SIENTIAPDE-1975): `NatsWriteConsumer` → `WriteEngine` → pool, then `DataSink.update_write_status`. Commands (C2) carry only `server_id`; `ServerConfigResolver` (`connection/servers.py`) resolves the connection (command override → live pooled connection → `connectors` → `opc-servers`). `write_status` keeps the worst outcome (`ok` < `error` < `session_bad`; same rule as the Laborious' `laborious/utils/write_status.py` — keep both in sync). The NATS first connect is bounded (it runs on the main loop); reconnection after it is unbounded.
|
**Writes** (SIENTIAPDE-1975): `NatsWriteConsumer` → `WriteEngine` → pool, then `DataSink.update_write_status`. Commands (C2) carry only `server_id`; `ServerConfigResolver` (`connection/servers.py`) resolves the connection (command override → live pooled connection → `connectors` → `opc-servers`). `write_status` keeps the worst outcome (`ok` < `error` < `session_bad`); the gateway is the only one that updates it (the Laborious inserts the row as `pending` and never records OPC outcomes). The NATS first connect is bounded (it runs on the main loop); reconnection after it is unbounded.
|
||||||
|
|
||||||
**Metrics** (`ingestor/metrics.py`): all Prometheus metrics defined here, always labeled by `pod_id`. Emitted via `SientiaMonitoring.emit_metric(...)` (from `sientia_do`), not the prometheus client directly — new metrics should follow that pattern for consistency with logging/notifications.
|
**Metrics** (`ingestor/metrics.py`): all Prometheus metrics defined here, always labeled by `pod_id`. Emitted via `SientiaMonitoring.emit_metric(...)` (from `sientia_do`), not the prometheus client directly — new metrics should follow that pattern for consistency with logging/notifications.
|
||||||
|
|
||||||
|
|||||||
@@ -91,8 +91,9 @@ The Laborious publishes one command per tag (SIENTIAPDE-1975):
|
|||||||
Lookups are cached for 60 s. An unknown server is a failed write (`server_not_found`).
|
Lookups are cached for 60 s. An unknown server is a failed write (`server_not_found`).
|
||||||
- **Outcome**: `UPDATE ... SET write_status` on the prediction row (by `timestamp` +
|
- **Outcome**: `UPDATE ... SET write_status` on the prediction row (by `timestamp` +
|
||||||
`model_id`), keeping the worst outcome: `ok` < `error` < `session_bad`, and
|
`model_id`), keeping the worst outcome: `ok` < `error` < `session_bad`, and
|
||||||
`pending`/NULL are replaced by any answer. The Laborious applies the same rule to
|
`pending`/NULL are replaced by any answer, so the order of the answers does not
|
||||||
the outcomes it records itself, so the order of the answers does not matter.
|
matter. The gateway is the only one that updates the column: the Laborious inserts the
|
||||||
|
row as `pending` (merged with its PI Web API outcome) and never records OPC outcomes.
|
||||||
The UPDATE never raises; its outcome (`updated`, `kept`, `row_not_found`,
|
The UPDATE never raises; its outcome (`updated`, `kept`, `row_not_found`,
|
||||||
`invalid`, `error`) is logged and counted in `opc_write_status_update_total`.
|
`invalid`, `error`) is logged and counted in `opc_write_status_update_total`.
|
||||||
The reason of a failed write is logged by `WriteEngine`.
|
The reason of a failed write is logged by `WriteEngine`.
|
||||||
|
|||||||
@@ -10,8 +10,8 @@ import ingestor.metrics as metrics
|
|||||||
_IDENTIFIER_RE = re.compile(r'\w+', re.ASCII)
|
_IDENTIFIER_RE = re.compile(r'\w+', re.ASCII)
|
||||||
|
|
||||||
# Worst-wins precedence of predictions.write_status (SIENTIAPDE-1975), from best to worst;
|
# Worst-wins precedence of predictions.write_status (SIENTIAPDE-1975), from best to worst;
|
||||||
# NULL, 'pending' and unknown values rank 0. Same rule as the Laborious
|
# NULL, 'pending' and unknown values rank 0. The gateway is the only writer that updates the
|
||||||
# (laborious/utils/write_status.py): keep both in sync.
|
# column: the Laborious only sets the value of the rows it inserts ('pending', PI outcome).
|
||||||
STATUS_RANK = {'ok': 1, 'error': 2, 'session_bad': 3}
|
STATUS_RANK = {'ok': 1, 'error': 2, 'session_bad': 3}
|
||||||
|
|
||||||
|
|
||||||
@@ -78,8 +78,9 @@ class DataSink:
|
|||||||
The row is found by timestamp + model_id. A prediction is written to several
|
The row is found by timestamp + model_id. A prediction is written to several
|
||||||
tags, each answered by its own command, so a better outcome never replaces a
|
tags, each answered by its own command, so a better outcome never replaces a
|
||||||
worse one: ``ok`` < ``error`` < ``session_bad``, and ``pending``/NULL are
|
worse one: ``ok`` < ``error`` < ``session_bad``, and ``pending``/NULL are
|
||||||
replaced by any answer. The Laborious runs the same statement for the outcomes it
|
replaced by any answer, so the order of the answers does not matter. Only the
|
||||||
records itself (SIENTIAPDE-1975), so the order of the answers does not matter.
|
gateway updates the column (SIENTIAPDE-1975): the Laborious publishes the commands
|
||||||
|
and sets ``pending`` (merged with its PI Web API outcome) on the row it inserts.
|
||||||
|
|
||||||
Never raises: a failure here must not reach the NATS callback nor hide the OPC
|
Never raises: a failure here must not reach the NATS callback nor hide the OPC
|
||||||
write itself, which already happened. The outcome is logged and counted instead.
|
write itself, which already happened. The outcome is logged and counted instead.
|
||||||
|
|||||||
Reference in New Issue
Block a user