940 lines
47 KiB
Markdown
940 lines
47 KiB
Markdown
# OPC Gateway Implementation Plan
|
||
|
||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||
|
||
**Goal:** Evolve `sientia-dataops-opc-ingestor` into an OPC Gateway that reads by subscription (multi-name fan-out), reads actively (scheduled + on-demand via gRPC), writes OPC via NATS, manages certs via MinIO+Mongo, and emits detailed metrics — following the OPC best practices of SIENTIAPDE-1879.
|
||
|
||
**Architecture:** Approach B — *engines per capability* over a shared `OpcConnectionPool` (connections keyed by server, not by slot). Input adapters (`NatsWriteConsumer`, `GrpcServer`) feed the engines; a `DataSink` handles egress (MongoDB reads + Postgres `write_status`; the legacy Kafka egress is removed). Coordination (slot lease/heartbeat) is reused from the existing Redis layer.
|
||
|
||
**Tech Stack:** Python 3.11, `asyncua`, `nats-py` (Core NATS), `grpcio`/`grpcio-tools`, `prometheus_client`, `sientia_do` dataops-library (Mongo/Redis/MinIO/Postgres), pytest + pytest-asyncio, ruff (line-length 100, single quotes).
|
||
|
||
**Spec:** `docs/superpowers/specs/2026-07-13-opc-gateway-design.md` (read it first — contracts C1–C4, metrics, §13 1879 checklist).
|
||
|
||
**Scope:** This plan is the **Gateway only** (repo `sientia-dataops-opc-ingestor`). Orchestrator (1974) and Laborious (1975) are parallel tracks with their own plans in their own repos; here they appear only as the contracts C1 (slot v2), C2 (NATS message), C3 (write_status).
|
||
|
||
**Conventions (match the repo):**
|
||
- Managers/engines extend `SientiaMonitoring` (`sientia_do.observability.sientia_monitoring`) → gives `self.logger`, `self.pod_id`, `await self.emit_metric(...)`, `await self.send_notification_async(...)`.
|
||
- Metrics defined in `ingestor/metrics.py` via `prometheus_client`.
|
||
- Package stays `ingestor/` (do **not** rename — the Helm health probe and Gitea `PYTHON_APP` entrypoint depend on `ingestor.app`). New capabilities go in new subpackages.
|
||
- Test command everywhere: `python -m pytest <path> -v`. Lint: `ruff check ingestor tests`.
|
||
|
||
---
|
||
|
||
## File Structure
|
||
|
||
**New subpackages under `ingestor/`** (one clear responsibility each; keep files < ~500 lines, functions < ~20 lines per 1879):
|
||
|
||
- `ingestor/connection/pool.py` — `OpcConnectionPool`: `asyncua` clients keyed by `server_id`, lazy open, reuse, idle-eviction, reconnect.
|
||
- `ingestor/connection/security.py` — `build_security(...)`: `Basic256Sha256`, `secure_channel_timeout = int(session_timeout*1.25)`, cert file paths.
|
||
- `ingestor/connection/certs.py` — `CertProvider`: read `platform_certificates` (Mongo) + download cert/key from MinIO (`sientia-dataops` bucket) to an `emptyDir`/tmpfs path.
|
||
- `ingestor/connection/clock.py` — `MonotonicClock`: single `time.monotonic()` reference + optional NTP; `elapsed()`, `now_wall()`.
|
||
- `ingestor/converters.py` — `to_variant_value(raw, data_type)`: safe type coercion (bool-string bug fix + `try/except`).
|
||
- `ingestor/engines/subscription.py` — `SubscriptionEngine`: slot v2 → subscription → multi-name fan-out to `DataSink`.
|
||
- `ingestor/engines/active_read.py` — `ActiveReadEngine`: scheduled reads + `read_now()` for on-demand.
|
||
- `ingestor/engines/write.py` — `WriteEngine`: OPC write with status_code/timestamp/quality flags, rate-limit, error mapping, → `DataSink.update_write_status`.
|
||
- `ingestor/io/slots.py` — `load_slot(redis, slot_id)`: parse slot v2 (C1) into typed structures.
|
||
- `ingestor/io/data_sink.py` — `DataSink`: `publish_reading(...)` (MongoDB fan-out) + `update_write_status(...)` (Postgres UPDATE by `timestamp`+`model_id`).
|
||
- `ingestor/adapters/nats_consumer.py` — `NatsWriteConsumer`: subscribe `opc.write` (queue group `gateway-writers`) → `WriteEngine`.
|
||
- `ingestor/adapters/grpc_server.py` — `GrpcServer`: `ReadTag` RPC → `ActiveReadEngine.read_now`.
|
||
- `ingestor/proto/opc_gateway.proto` (+ generated `*_pb2.py`, `*_pb2_grpc.py`).
|
||
- `ingestor/config.py` — extend with new env (NATS, gRPC, cert bucket, NTP).
|
||
- `ingestor/metrics.py` — extend with new metrics (§9 of spec).
|
||
|
||
**Reused as-is / lightly touched:** `ingestor/managers/resource_manager.py` (lease/heartbeat), `ingestor/managers/data_manager.py` (MongoDB egress — wrapped by `DataSink`; the legacy Kafka path is removed in Task 9b), `ingestor/app.py` (wire new engines in F4).
|
||
|
||
**Tests mirror source:** `tests/unit/connection/…`, `tests/unit/engines/…`, `tests/unit/io/…`, `tests/unit/adapters/…`, `tests/unit/test_converters.py`. Add an `__init__.py` to each new test dir (the repo's existing test dirs use them for collection, e.g. `tests/unit/managers/__init__.py`).
|
||
|
||
---
|
||
|
||
## Phase Delivery & Commit Ledger
|
||
|
||
**Delivery model: one PR per phase (F0…F4), opened only at the very end.** To avoid one giant PR, each phase is delivered and reviewed as its own PR. But **no PR is opened until every phase is complete, tested, and working** — we build all phases on this feature branch first, then split them into per-phase PRs at the end. This section is the ledger that makes that split possible.
|
||
|
||
### Phase close-out (the LAST task of every phase — Tasks 7.end, 10.end, 13.end, 16.end, 21.end)
|
||
At the end of each phase Fx, before starting the next:
|
||
1. Confirm the phase is green: full `python -m pytest tests -v` passes and `ruff check ingestor tests` is clean.
|
||
2. Record the phase's commit SHAs (in order) in the ledger below, with the phase's base (parent of its first commit).
|
||
3. **Do NOT open the PR yet** — PRs are opened only after F4 closes (see "Final: open per-phase PRs").
|
||
|
||
### Commit ledger
|
||
|
||
**F0 — Connection core** — base `498030d` (merge-base with `main`). Status: ✅ complete, 162 unit tests green, ruff clean.
|
||
- `0694f6a` chore(gateway): add nats-py, grpcio deps (Task 1)
|
||
- `3dd4f60` feat(gateway): add write/active-read/error metrics (Task 2)
|
||
- `55f6721` feat(gateway): safe type converter (1879 BUG-1/BUG-2) (Task 3)
|
||
- `d6cb1e4` feat(gateway): monotonic clock helper (1879) (Task 4)
|
||
- `b05c2fd` feat(gateway): Basic256Sha256 + channel>session lifetime (1879) (Task 5)
|
||
- `ec2a324` feat(gateway): cert provider from MinIO+Mongo (sientia-dataops) (Task 6)
|
||
- `5d30a5c` feat(gateway): OpcConnectionPool (per-server, reuse, idle-evict) (Task 7)
|
||
- `0b29857` fix(gateway): per-server lock in OpcConnectionPool to avoid duplicate connects (Task 7)
|
||
- `b57be9f` refactor(gateway): drop dead error_stack and unused metadata from OpcConnectionPool (ponytail) (Task 7)
|
||
- Also fixed a pre-existing test flake in `a08fd88` (gc.collect() in test_ingestor_manager) — include in the F0 code PR.
|
||
- ⚠️ The branch also carries **docs** commits (design/spec/plan) interleaved in time (`6f6f3bf`, `c980092`, `524f4d6`, `8a05ff7`, `941c6c0`, `7cffc26`, `bbe10a4`, `09e4290`, plus the gRPC/Kafka doc updates). Those are **not** F0 code — the aig-docs equivalents already ship in PR #76; the repo `docs/superpowers/**` commits can go in a separate small docs PR or ride with F0. Keep them out of the F0 **code** PR scope.
|
||
|
||
**F1 — Subscription reads (C1)** — base `00f919c`. Status: ✅ complete, 168 unit tests green, ruff clean.
|
||
- `44df022` feat(gateway): slot v2 loader (contract C1) (Task 8)
|
||
- `ae88ac6` feat(gateway): DataSink.publish_reading multi-name fan-out (Task 9)
|
||
- `323bf24` refactor(gateway): remove legacy Kafka egress (MongoDB-only) (Task 9b)
|
||
- `00c74e5` feat(gateway): SubscriptionEngine multi-name fan-out (C1) (Task 10)
|
||
- Carry-forward to F4/later (from F1 reviews):
|
||
- **Task 18 (wiring):** decide how many `SubscriptionEngine` instances the app builds. Current code keeps a single engine-wide `non_receive_count` and `check_opc_listenning(server_name)` takes a param — safe only if **one engine per OPC server**. If one engine is shared across servers, a datachange on server A masks silence on server B → must partition cycle/non-receive state per server. Make this deliberate at wiring time.
|
||
- **Task 12 (DataSink):** drop the currently-unused `metadata` param from `DataSink.__init__` while adding `update_write_status`.
|
||
- **F4 cleanup:** `README.md` and `tests/functional/test_single_node.py`/`conftest.py` still reference Kafka (stale/commented dead code) — remove during F4 hardening (Task 21 sweep).
|
||
**F2 — Active write via NATS (C2, C3)** — base `2a9af05`. Status: ✅ complete, 192 unit tests green, ruff clean.
|
||
- `dcf8369` feat(gateway): WriteEngine (status_code/timestamp/flags, safe converter, rate-limit) (Task 11)
|
||
- `86d0f6c` fix(gateway): make WriteEngine + converter raise-safe against malformed write commands (Task 11)
|
||
- `905da51` feat(gateway): DataSink.update_write_status Postgres by timestamp+model_id (C3); drop unused metadata param (Task 12)
|
||
- `abff80e` fix(gateway): tighten SQL identifier validation to fullmatch (reject trailing newline) (Task 12)
|
||
- `165ac79` feat(gateway): NatsWriteConsumer (Core NATS, queue group) (C2) (Task 13)
|
||
- `f7f356f` refactor(gateway): ponytail F2 - drop dead metadata, unreachable variant fallback, redundant except member
|
||
- Carry-forward:
|
||
- **asyncua version hygiene:** requirements pin `asyncua==1.1.5` but the dev/test env runs 1.1.8. The DataValue/StatusCode_ behavior was verified identical on both, but align the pin (bump to 1.1.8) or the env during F4 (Task 21) so version-specific bugs can't slip through.
|
||
- **`_SESSION_BAD_NAMES` in write.py** duplicates a subset of Laborious's reconnectable-error list (cross-repo) — move to `sientia_do` if it drifts (ponytail note in code).
|
||
- **`_server_cfg` in write.py** is minimal (url/security off the cmd); full connector-DB lookup by `server_id` is Task 18 (F4 wiring) — ponytail note in code.
|
||
**F3 — Active reads + gRPC (C4)** — base `4855205`. Status: ✅ complete, 206 unit tests green, ruff clean.
|
||
- `99a715c` feat(gateway): ActiveReadEngine (scheduled + read_now) (Task 14)
|
||
- `1798b69` feat(gateway): gRPC proto + generated stubs (C4) (Task 15)
|
||
- `52223e6` feat(gateway): gRPC server ReadTag (C4) (Task 16)
|
||
- Carry-forward to Task 18 (wiring):
|
||
- Wire the gRPC `slot_lookup` seam (address→names from the current slot) so on-demand reads persist; if the lookup becomes I/O-bound, move it inside the servicer's try/except.
|
||
- gRPC server is internal-only, no auth (no consumer yet — likely frontend later, per feature owner).
|
||
- Edge case (minor): `read_now` with `SourceTimestamp=None` persists `timestamp=None`; add a guard if it matters in practice.
|
||
**F4 — Wiring, health, deploy, hardening** — base `e9ce9e5`. Status: ✅ complete, 222 unit tests green, ruff clean, `grep -i kafka` = 0.
|
||
- `ff605ee` feat(gateway): GatewayConfig env (NATS/gRPC/certs/write flags) (Task 17)
|
||
- `3912687` feat(gateway): wire engines+adapters via GatewayRuntime (write/gRPC live; subscription v2 gated on 1974) (Task 18)
|
||
- `3f413f8` fix(gateway): app survives runtime init failure; dedupe NATS-down notifications; fix stale slot_lookup note (Task 18)
|
||
- `21d0b37` docs(gateway): deploy notes for helm/images/platform repos (Task 19+20)
|
||
- `cebb521` chore(gateway): final sweep - README/functional Kafka cleanup, asyncua pin, 1879 checklist verification (Task 21)
|
||
- Deferred by design (verified in Task 21): NTP sync + `Good_Local_Override` quality downgrade (no flag exists; ties to a real clock-drift case), write-via-OPC-method (card marks it flag/future). Pre-existing legacy files >500 lines (`ingestor_manager.py` 578, `opc_manager.py` 507) flagged, untouched (predate the gateway).
|
||
- Integration gates (cross-repo): subscription v2 swap + gRPC persist → SIENTIAPDE-1974; write_status column (Flyway) + Laborious NATS publish → SIENTIAPDE-1975; NATS broker in cluster → SIENTIAPDE-1985. End-to-end validation (real NATS + OPC server machine) agreed with the feature owner for deployment time.
|
||
|
||
### Final: open per-phase PRs (only after F4 is complete, tested, working)
|
||
Since all phases live on one feature branch, split them at the end using the recorded SHA ranges: for each phase in order F0→F4, create a phase branch off `main` (or stacked on the previous phase branch) containing only that phase's code commits (cherry-pick the ledger's SHA range), then open one PR per phase targeting `main`, in phase order. This keeps each PR small and reviewable instead of one giant diff.
|
||
|
||
---
|
||
|
||
## Phase F0 — Foundations (deps, config, metrics, connection pool)
|
||
|
||
### Task 1: Add dependencies
|
||
|
||
**Files:**
|
||
- Modify: `requirements.txt` (and `pyproject.toml` if deps are declared there)
|
||
|
||
- [ ] **Step 1: Add libs**
|
||
|
||
Add to `requirements.txt`:
|
||
```
|
||
nats-py>=2.6
|
||
grpcio>=1.60
|
||
grpcio-tools>=1.60
|
||
```
|
||
(`asyncua`, `prometheus_client`, `sientia_do`, `pandas` already present. Bump `sientia_do` to the version that ships the generic Mongo methods — `feature/SIENTIAPDE-1590` / ≥ current pin.)
|
||
|
||
- [ ] **Step 2: Install & verify import**
|
||
|
||
Run: `pip install -r requirements.txt && python -c "import nats, grpc; print('ok')"`
|
||
Expected: `ok`
|
||
|
||
- [ ] **Step 3: Commit**
|
||
|
||
```bash
|
||
git add requirements.txt pyproject.toml
|
||
git commit -m "chore(gateway): add nats-py, grpcio deps"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 2: New metrics (spec §9)
|
||
|
||
**Files:**
|
||
- Modify: `ingestor/metrics.py`
|
||
- Test: `tests/unit/test_metrics.py` (extend existing)
|
||
|
||
- [ ] **Step 1: Write failing test** — assert the new metric objects exist and accept labels.
|
||
|
||
```python
|
||
# tests/unit/test_metrics.py
|
||
import ingestor.metrics as m
|
||
|
||
def test_gateway_metrics_exist():
|
||
m.OPC_WRITE_TOTAL.labels(pod_id='p', server_name='s', model_id='1', result='success')
|
||
m.OPC_WRITE_LATENCY.labels(pod_id='p', server_name='s')
|
||
m.OPC_ACTIVE_READ_TOTAL.labels(pod_id='p', server_name='s', result='success')
|
||
m.OPC_ACTIVE_READ_LATENCY.labels(pod_id='p', server_name='s')
|
||
m.OPC_WRITE_STATUS_BY_MODEL.labels(pod_id='p', model_id='1', status='ok')
|
||
m.OPC_TAG_VALUE.labels(pod_id='p', tag_name='t', model_id='1')
|
||
m.OPC_READ_LAG.labels(pod_id='p', server_name='s')
|
||
m.OPC_ERROR_BY_TYPE.labels(pod_id='p', server_name='s', error_type='BadTimeout')
|
||
m.OPC_DISCONNECTIONS_TOTAL.labels(pod_id='p', server_name='s')
|
||
```
|
||
|
||
- [ ] **Step 2: Run — expect FAIL** (`AttributeError`). Run: `python -m pytest tests/unit/test_metrics.py -v`
|
||
|
||
- [ ] **Step 3: Implement** — append to `ingestor/metrics.py`:
|
||
|
||
```python
|
||
# --- Gateway: Write metrics ---
|
||
OPC_WRITE_TOTAL = Counter(
|
||
'opc_write_total', 'OPC active write attempts',
|
||
['pod_id', 'server_name', 'model_id', 'result'], # result: success|error
|
||
)
|
||
OPC_WRITE_LATENCY = Histogram(
|
||
'opc_write_latency_seconds', 'OPC active write latency', ['pod_id', 'server_name'],
|
||
)
|
||
OPC_WRITE_STATUS_BY_MODEL = Gauge(
|
||
'opc_write_status_by_model', 'Last write status per model (1=ok,0=error)',
|
||
['pod_id', 'model_id', 'status'],
|
||
)
|
||
# --- Gateway: Active read metrics ---
|
||
OPC_ACTIVE_READ_TOTAL = Counter(
|
||
'opc_active_read_total', 'OPC active read attempts',
|
||
['pod_id', 'server_name', 'result'],
|
||
)
|
||
OPC_ACTIVE_READ_LATENCY = Histogram(
|
||
'opc_active_read_latency_seconds', 'OPC active read latency', ['pod_id', 'server_name'],
|
||
)
|
||
OPC_READ_LAG = Gauge(
|
||
'opc_read_lag_seconds', 'Lag between OPC source timestamp and machine time',
|
||
['pod_id', 'server_name'],
|
||
)
|
||
OPC_TAG_VALUE = Gauge(
|
||
'opc_tag_value', 'Last value read per tag/model', ['pod_id', 'tag_name', 'model_id'],
|
||
)
|
||
# --- Gateway: connection/error metrics ---
|
||
OPC_DISCONNECTIONS_TOTAL = Counter(
|
||
'opc_disconnections_total', 'OPC disconnections', ['pod_id', 'server_name'],
|
||
)
|
||
OPC_ERROR_BY_TYPE = Counter(
|
||
'opc_error_by_type_total', 'OPC errors by type', ['pod_id', 'server_name', 'error_type'],
|
||
)
|
||
```
|
||
|
||
- [ ] **Step 4: Run — expect PASS.** `python -m pytest tests/unit/test_metrics.py -v`
|
||
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): add write/active-read/error metrics"`
|
||
|
||
---
|
||
|
||
### Task 3: Safe type converter (1879 BUG-1 & BUG-2)
|
||
|
||
**Files:**
|
||
- Create: `ingestor/converters.py`
|
||
- Test: `tests/unit/test_converters.py`
|
||
|
||
- [ ] **Step 1: Write failing tests** (the bool-string bug is the whole point):
|
||
|
||
```python
|
||
# tests/unit/test_converters.py
|
||
import pytest
|
||
from ingestor.converters import to_typed_value, ConversionError
|
||
|
||
@pytest.mark.parametrize('raw,expected', [
|
||
('false', False), ('False', False), ('0', False), ('no', False),
|
||
('off', False), ('', False), ('true', True), ('1', True), (True, True),
|
||
])
|
||
def test_bool_from_json_string(raw, expected):
|
||
assert to_typed_value(raw, 'boolean') is expected
|
||
|
||
def test_float_and_int():
|
||
assert to_typed_value('42.5', 'float') == 42.5
|
||
assert to_typed_value('7', 'int') == 7
|
||
|
||
@pytest.mark.parametrize('raw,dtype', [
|
||
(None, 'int'), ('abc', 'float'), ('123.123', 'int'), (None, 'float'),
|
||
])
|
||
def test_invalid_raises_conversion_error(raw, dtype):
|
||
with pytest.raises(ConversionError):
|
||
to_typed_value(raw, dtype)
|
||
```
|
||
|
||
- [ ] **Step 2: Run — expect FAIL** (module missing). `python -m pytest tests/unit/test_converters.py -v`
|
||
|
||
- [ ] **Step 3: Implement** `ingestor/converters.py`:
|
||
|
||
```python
|
||
"""Safe coercion of JSON-transported values into typed OPC write values.
|
||
|
||
Fixes 1879 BUG-1 (bool('false') is True in Python) and BUG-2 (uncaught
|
||
TypeError/ValueError from int/float on bad input).
|
||
"""
|
||
|
||
_FALSE_STRINGS = {'false', '0', 'no', 'off', ''}
|
||
|
||
|
||
class ConversionError(ValueError):
|
||
"""Raised when a raw value cannot be coerced to the requested type."""
|
||
|
||
|
||
def _to_bool(raw) -> bool:
|
||
if isinstance(raw, bool):
|
||
return raw
|
||
if isinstance(raw, (int, float)):
|
||
return bool(raw)
|
||
return str(raw).strip().lower() not in _FALSE_STRINGS
|
||
|
||
|
||
def to_typed_value(raw, data_type: str):
|
||
"""Coerce raw (often a JSON string) to the given data_type.
|
||
|
||
data_type: 'boolean' | 'int' | 'float' | 'double' | 'string'.
|
||
Raises ConversionError on any invalid input (never a bare TypeError).
|
||
"""
|
||
dtype = data_type.strip().lower()
|
||
try:
|
||
if dtype in ('bool', 'boolean'):
|
||
return _to_bool(raw)
|
||
if dtype in ('int', 'integer'):
|
||
if isinstance(raw, str) and ('.' in raw or raw.strip() == ''):
|
||
raise ConversionError(f'invalid int: {raw!r}')
|
||
return int(raw)
|
||
if dtype in ('float', 'double', 'real'):
|
||
return float(raw)
|
||
if dtype in ('str', 'string'):
|
||
return str(raw)
|
||
except (TypeError, ValueError) as e:
|
||
raise ConversionError(f'cannot convert {raw!r} to {data_type}: {e}') from e
|
||
raise ConversionError(f'unknown data_type: {data_type!r}')
|
||
```
|
||
|
||
- [ ] **Step 4: Run — expect PASS.** `python -m pytest tests/unit/test_converters.py -v`
|
||
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): safe type converter (1879 BUG-1/BUG-2)"`
|
||
|
||
---
|
||
|
||
### Task 4: Monotonic clock helper (1879 §2)
|
||
|
||
**Files:**
|
||
- Create: `ingestor/connection/clock.py`, `ingestor/connection/__init__.py`
|
||
- Test: `tests/unit/connection/test_clock.py`
|
||
|
||
- [ ] **Step 1: Failing test**
|
||
|
||
```python
|
||
# tests/unit/connection/test_clock.py
|
||
from ingestor.connection.clock import MonotonicClock
|
||
|
||
def test_elapsed_is_non_negative_and_monotonic():
|
||
clk = MonotonicClock()
|
||
a = clk.elapsed()
|
||
b = clk.elapsed()
|
||
assert a >= 0 and b >= a # never negative even if wall clock jumps back
|
||
```
|
||
|
||
- [ ] **Step 2: Run — expect FAIL.**
|
||
|
||
- [ ] **Step 3: Implement** `ingestor/connection/clock.py`:
|
||
|
||
```python
|
||
import time
|
||
|
||
|
||
class MonotonicClock:
|
||
"""Single monotonic reference for latency/timeout math (1879 §2).
|
||
|
||
Wall-clock (`now_wall`) is only for human-facing timestamps; all elapsed/
|
||
timeout math must use `elapsed()` so NTP re-sync can't produce negatives.
|
||
"""
|
||
|
||
def __init__(self):
|
||
self._start = time.monotonic()
|
||
|
||
def elapsed(self) -> float:
|
||
return time.monotonic() - self._start
|
||
|
||
@staticmethod
|
||
def now_wall() -> float:
|
||
return time.time()
|
||
```
|
||
|
||
- [ ] **Step 4: Run — expect PASS.**
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): monotonic clock helper (1879)"`
|
||
|
||
> **Note (calibration):** NTP sync + wall-clock quality downgrade (`Good_Local_Override 0xD8`) are flag-driven and belong to the WriteEngine (Task 11). Leave a `# ponytail:` marker here noting NTP option is deferred until a real clock-drift case appears.
|
||
|
||
---
|
||
|
||
### Task 5: Security builder (1879 §1, §3)
|
||
|
||
**Files:**
|
||
- Create: `ingestor/connection/security.py`
|
||
- Test: `tests/unit/connection/test_security.py`
|
||
|
||
- [ ] **Step 1: Failing test** — assert channel timeout is 1.25× session and policy is Sha256.
|
||
|
||
```python
|
||
# tests/unit/connection/test_security.py
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
import pytest
|
||
from ingestor.connection.security import apply_security, SESSION_TIMEOUT_MS
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_apply_security_sets_sha256_and_channel_lifetime():
|
||
client = MagicMock()
|
||
client.set_security = AsyncMock()
|
||
await apply_security(client, 'urn:x', 'c.pem', 'k.pem', None)
|
||
assert client.session_timeout == SESSION_TIMEOUT_MS
|
||
assert client.secure_channel_timeout == int(SESSION_TIMEOUT_MS * 1.25)
|
||
args, kwargs = client.set_security.call_args
|
||
from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256
|
||
assert args[0] is SecurityPolicyBasic256Sha256
|
||
```
|
||
|
||
- [ ] **Step 2: Run — expect FAIL.**
|
||
|
||
- [ ] **Step 3: Implement** `ingestor/connection/security.py`:
|
||
|
||
```python
|
||
from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256
|
||
|
||
SESSION_TIMEOUT_MS = 600_000 # 10 min; tune per server (Samarco: long sessions degrade)
|
||
|
||
|
||
async def apply_security(client, server_uri, cert_path, private_key_path, server_cert_path):
|
||
"""Apply Basic256Sha256 + certificate auth (1879 §1) and set channel > session
|
||
lifetime (1879 §3: secure_channel_timeout = session_timeout * 1.25)."""
|
||
client.application_uri = server_uri
|
||
await client.set_security(
|
||
SecurityPolicyBasic256Sha256,
|
||
certificate=cert_path,
|
||
private_key=private_key_path,
|
||
server_certificate=server_cert_path,
|
||
)
|
||
client.session_timeout = SESSION_TIMEOUT_MS
|
||
client.secure_channel_timeout = int(SESSION_TIMEOUT_MS * 1.25)
|
||
```
|
||
|
||
- [ ] **Step 4: Run — expect PASS.**
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): Basic256Sha256 + channel>session lifetime (1879)"`
|
||
|
||
---
|
||
|
||
### Task 6: Certificate provider (MinIO + Mongo)
|
||
|
||
**Files:**
|
||
- Create: `ingestor/connection/certs.py`
|
||
- Test: `tests/unit/connection/test_certs.py`
|
||
|
||
- [ ] **Step 1: Failing test** — reads `platform_certificates` doc, downloads both objects, returns local paths.
|
||
|
||
```python
|
||
# tests/unit/connection/test_certs.py
|
||
from unittest.mock import AsyncMock
|
||
import pytest
|
||
from ingestor.connection.certs import CertProvider
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_fetch_returns_local_paths(tmp_path):
|
||
mongo = AsyncMock()
|
||
mongo.find_one.return_value = {
|
||
'client_cert': {'minio_object_name': 'platform/certificates/client_cert_c.pem'},
|
||
'private_key': {'minio_object_name': 'platform/certificates/private_key_k.pem'},
|
||
}
|
||
minio = AsyncMock()
|
||
provider = CertProvider(mongo, minio, cache_dir=str(tmp_path), bucket='sientia-dataops')
|
||
cert, key = await provider.fetch()
|
||
assert cert.endswith('c.pem') and key.endswith('k.pem')
|
||
assert minio.download_file.await_count == 2
|
||
```
|
||
|
||
- [ ] **Step 2: Run — expect FAIL.**
|
||
|
||
- [ ] **Step 3: Implement** `ingestor/connection/certs.py` — read the single `platform_certificates` doc via `mongo.find_one({}, 'platform_certificates')`, `await minio.download_file(bucket, object_name, local_path)` into `cache_dir` (must be a writable `emptyDir`/tmpfs — root FS is read-only in the pod, spec §12). Return `(cert_path, key_path)`. Keep it < 20 lines.
|
||
|
||
- [ ] **Step 4: Run — expect PASS.**
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): cert provider from MinIO+Mongo (sientia-dataops)"`
|
||
|
||
> Confirm the exact `MongoDBRepository.find_one`/`MinioRepository.download_file` signatures against the pinned `sientia_do`; adapt the calls if they differ.
|
||
|
||
---
|
||
|
||
### Task 7: OpcConnectionPool
|
||
|
||
**Files:**
|
||
- Create: `ingestor/connection/pool.py`
|
||
- Test: `tests/unit/connection/test_pool.py`
|
||
|
||
- [ ] **Step 1: Failing tests** — (a) same `server_id` returns the same client (reuse); (b) a connect failure increments `OPC_CONNECTIONS_FAILED` and re-raises.
|
||
|
||
```python
|
||
# tests/unit/connection/test_pool.py
|
||
from unittest.mock import AsyncMock, patch
|
||
import pytest
|
||
from ingestor.connection.pool import OpcConnectionPool
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_reuses_connection_per_server(monkeypatch):
|
||
pool = OpcConnectionPool(cert_provider=AsyncMock(), metadata={}, logger=AsyncMock())
|
||
fake_client = AsyncMock()
|
||
with patch('ingestor.connection.pool.Client', return_value=fake_client):
|
||
c1 = await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
|
||
c2 = await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
|
||
assert c1 is c2
|
||
assert fake_client.connect.await_count == 1
|
||
```
|
||
|
||
- [ ] **Step 2: Run — expect FAIL.**
|
||
|
||
- [ ] **Step 3: Implement** `ingestor/connection/pool.py` — `OpcConnectionPool` keyed by `server_id`:
|
||
- `async get(server_id, server_cfg)`: if a live client exists, return it; else create `Client(url, timeout=10)`, if `use_security` call `apply_security(...)` with certs from `cert_provider.fetch()`, `await client.connect()`, store, emit `OPC_CONNECTIONS_TOTAL`/`OPC_CONNECTION_STATUS`. On failure emit `OPC_CONNECTIONS_FAILED`, drop entry, re-raise.
|
||
- `async release_idle(max_idle_s)`: disconnect clients unused past TTL (track last-use via `MonotonicClock`), emit `OPC_DISCONNECTIONS_TOTAL`.
|
||
- `async close_all()`.
|
||
- Reuse the existing `OpcManager.disconnection_fallback` pattern for robust disconnect.
|
||
|
||
- [ ] **Step 4: Run — expect PASS.**
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): OpcConnectionPool (per-server, reuse, idle-evict)"`
|
||
|
||
---
|
||
|
||
### Task 7.end: F0 close-out (see [Phase Delivery](#phase-delivery--commit-ledger))
|
||
- [x] Full `pytest tests` green + `ruff` clean (162 passed).
|
||
- [x] F0 commits recorded in the [Commit Ledger](#commit-ledger).
|
||
- [ ] PR intentionally NOT opened yet — opened only after F4 closes.
|
||
|
||
---
|
||
|
||
## Phase F1 — Subscription reads with multi-name fan-out (contract C1)
|
||
|
||
### Task 8: Slot v2 loader
|
||
|
||
**Files:**
|
||
- Create: `ingestor/io/slots.py`, `ingestor/io/__init__.py`
|
||
- Test: `tests/unit/io/test_slots.py`
|
||
|
||
- [ ] **Step 1: Failing test** — parse the C1 structure incl. `names[]` and `active_read`.
|
||
|
||
```python
|
||
# tests/unit/io/test_slots.py
|
||
from ingestor.io.slots import parse_slot
|
||
|
||
RAW = {'server1': {'server_id': '1', 'url': 'opc.tcp://h:4840', 'use_security': True,
|
||
'subscription_period_ms': 1000, 'tags': {'ns=2;i=2': {'frequency': '15000',
|
||
'active_read': {'enabled': True, 'cycle_multiple': 3},
|
||
'names': [{'collection': 'raw_x', 'tag_name': 'Counter', 'model_name': 'm1', 'model_id': '1'},
|
||
{'collection': 'raw_y', 'tag_name': 'Contador', 'model_name': 'm2', 'model_id': '2'}]}}}}
|
||
|
||
def test_parse_slot_multi_name():
|
||
slot = parse_slot(RAW)
|
||
tag = slot['server1']['tags']['ns=2;i=2']
|
||
assert len(tag['names']) == 2
|
||
assert tag['active_read']['cycle_multiple'] == 3
|
||
```
|
||
|
||
- [ ] **Step 2–4:** implement `parse_slot(raw)` (validate shape, default `active_read={'enabled': False}`, ensure `names` is a list); run to green.
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): slot v2 loader (contract C1)"`
|
||
|
||
---
|
||
|
||
### Task 9: DataSink.publish_reading (MongoDB fan-out)
|
||
|
||
**Files:**
|
||
- Create: `ingestor/io/data_sink.py`
|
||
- Test: `tests/unit/io/test_data_sink_publish.py`
|
||
|
||
- [ ] **Step 1: Failing test** — one reading with N names publishes N documents (one per collection), each carrying `tag_name`/`model_id`.
|
||
|
||
```python
|
||
# tests/unit/io/test_data_sink_publish.py
|
||
from unittest.mock import AsyncMock
|
||
import pytest
|
||
from ingestor.io.data_sink import DataSink
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_publish_reading_fans_out_per_name():
|
||
dm = AsyncMock()
|
||
sink = DataSink(data_manager=dm, postgres=AsyncMock(), logger=AsyncMock())
|
||
names = [{'collection': 'raw_x', 'tag_name': 'Counter', 'model_id': '1', 'model_name': 'm1'},
|
||
{'collection': 'raw_y', 'tag_name': 'Contador', 'model_id': '2', 'model_name': 'm2'}]
|
||
await sink.publish_reading(tag='ns=2;i=2', value=42, timestamp='2026-07-13T00:00:00Z', names=names)
|
||
assert dm.publish.await_count == 2
|
||
```
|
||
|
||
- [ ] **Step 2–4:** implement `DataSink.publish_reading(tag, value, timestamp, names)` — loop `names`, build `{tag, name: tag_name, model_id, timestamp, value}`, call `self.data_manager.publish(collection, doc)` (reuses existing MongoDB egress). Emit `OPC_TAG_VALUE`. Run to green.
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): DataSink.publish_reading multi-name fan-out"`
|
||
|
||
---
|
||
|
||
### Task 9b: Remove the legacy Kafka egress
|
||
|
||
Simplification requested by the feature owner (2026-07-14): the Gateway persists reads to MongoDB only.
|
||
|
||
**Files:**
|
||
- Modify: `ingestor/managers/data_manager.py` (remove the `KafkaProducer`, the Kafka branch of `publish`, and `shutdown` of the producer)
|
||
- Modify: `ingestor/metrics.py` (remove `KAFKA_MESSAGES_SENT`, `KAFKA_MESSAGES_ERRORS`, `KAFKA_CONNECTION_STATUS`)
|
||
- Modify: `.env.example` (remove `EXPORT_TO_KAFKA`, `KAFKA_SERVERS`)
|
||
- Modify: affected tests (`tests/unit/managers/test_data_manager.py`, any metrics test referencing `kafka_*`)
|
||
|
||
- [ ] **Step 1:** grep for `kafka` across `ingestor/` and `tests/` to enumerate every reference.
|
||
- [ ] **Step 2:** remove the Kafka producer/branch/config/metrics; keep the MongoDB path intact.
|
||
- [ ] **Step 3:** update/trim the affected tests so they no longer assert Kafka behavior.
|
||
- [ ] **Step 4:** run the full suite (`python -m pytest tests -v`) and `ruff check` — all green.
|
||
- [ ] **Step 5: Commit** — `git commit -am "refactor(gateway): remove legacy Kafka egress (MongoDB-only)"`
|
||
|
||
---
|
||
|
||
### Task 10: SubscriptionEngine (fan-out on datachange)
|
||
|
||
**Files:**
|
||
- Create: `ingestor/engines/subscription.py`, `ingestor/engines/__init__.py`
|
||
- Test: `tests/unit/engines/test_subscription.py`
|
||
|
||
- [ ] **Step 1: Failing test** — a datachange for an address with 2 names calls `DataSink.publish_reading` once with both names; resets cycle/non-receive counters; records `OPC_READ_LAG`.
|
||
|
||
```python
|
||
# tests/unit/engines/test_subscription.py
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
import pytest
|
||
from ingestor.engines.subscription import SubscriptionEngine
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_datachange_fans_out():
|
||
sink = AsyncMock()
|
||
eng = SubscriptionEngine(pool=AsyncMock(), sink=sink, logger=AsyncMock(), metadata={})
|
||
eng.nodes['ns=2;i=2'] = {'names': [{'tag_name': 'a'}, {'tag_name': 'b'}],
|
||
'cycle_rule': {'cycle_increment': 1, 'cycle_count': 3}}
|
||
node = MagicMock(); node.__str__ = lambda self: 'ns=2;i=2'
|
||
data = MagicMock()
|
||
data.monitored_item.Value.Value.Value = 42
|
||
await eng.datachange_notification(node, 42, data)
|
||
assert sink.publish_reading.await_count == 1
|
||
_, kwargs = sink.publish_reading.call_args
|
||
assert len(kwargs['names']) == 2
|
||
```
|
||
|
||
- [ ] **Step 2: Run — expect FAIL.**
|
||
|
||
- [ ] **Step 3: Implement** `SubscriptionEngine` by adapting `ingestor/managers/opc_manager.py`:
|
||
- Subscriptions created via `pool.get(...)` + `client.create_subscription(period, self)` (engine is the callback handler).
|
||
- `datachange_notification` extracts value + `SourceTimestamp`, computes lag (`OPC_READ_LAG` = wall_now − source_ts) and calls `await self.sink.publish_reading(tag, value, ts, names=self.nodes[tag]['names'])` (fan-out — replaces the old per-`topics` loop).
|
||
- Keep `check_cycles`/`check_opc_listenning` behavior.
|
||
- Slot lease/heartbeat: reuse `ResourceManager` (unchanged); the engine consumes `parse_slot(...)` output.
|
||
|
||
- [ ] **Step 4: Run — expect PASS.**
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): SubscriptionEngine multi-name fan-out (C1)"`
|
||
|
||
---
|
||
|
||
### Task 10.end: F1 close-out (see [Phase Delivery](#phase-delivery--commit-ledger))
|
||
- [x] Full `pytest tests` green (168) + `ruff` clean.
|
||
- [x] F1 commit SHAs recorded in the [Commit Ledger](#commit-ledger) (base `00f919c`).
|
||
- [ ] PR intentionally NOT opened yet — opened only after F4 closes.
|
||
|
||
---
|
||
|
||
## Phase F2 — Active write via NATS (contracts C2, C3)
|
||
|
||
### Task 11: WriteEngine
|
||
|
||
**Files:**
|
||
- Create: `ingestor/engines/write.py`
|
||
- Test: `tests/unit/engines/test_write.py`
|
||
|
||
- [ ] **Step 1: Failing tests** — (a) writes a `DataValue` with `StatusCode` Good + `SourceTimestamp` when flags on; (b) a converter failure returns an error result (not an exception) and increments `OPC_ERROR_BY_TYPE`; (c) rate-limit enforces a min pause between writes.
|
||
|
||
```python
|
||
# tests/unit/engines/test_write.py
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
import pytest
|
||
from ingestor.engines.write import WriteEngine, WriteResult
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_write_ok_sets_status_and_timestamp():
|
||
node = AsyncMock()
|
||
client = MagicMock(); client.get_node.return_value = node
|
||
pool = AsyncMock(); pool.get.return_value = client
|
||
eng = WriteEngine(pool=pool, sink=AsyncMock(), logger=AsyncMock(), metadata={},
|
||
write_timestamp=True, write_quality=True, min_pause_s=0)
|
||
res = await eng.write({'server_id': 's', 'addr': 'ns=2;i=5', 'value': '42.5',
|
||
'data_type': 'float', 'model_id': '1', 'timestamp': '2026-07-13T00:00:00Z',
|
||
'schema': 'sientia_data', 'table_name': 't'})
|
||
assert isinstance(res, WriteResult) and res.ok
|
||
node.write_value.assert_awaited() # DataValue passed
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_write_bad_type_returns_error_not_raise():
|
||
client = MagicMock(); client.get_node.return_value = AsyncMock()
|
||
pool = AsyncMock(); pool.get.return_value = client
|
||
eng = WriteEngine(pool=pool, sink=AsyncMock(), logger=AsyncMock(), metadata={}, min_pause_s=0)
|
||
res = await eng.write({'server_id': 's', 'addr': 'ns=2;i=5', 'value': 'abc',
|
||
'data_type': 'int', 'model_id': '1', 'timestamp': 't', 'schema': 'x', 'table_name': 't'})
|
||
assert not res.ok and res.status == 'error'
|
||
```
|
||
|
||
- [ ] **Step 2: Run — expect FAIL.**
|
||
|
||
- [ ] **Step 3: Implement** `ingestor/engines/write.py`:
|
||
- `WriteResult` dataclass: `ok: bool`, `status: str` (`ok|error|session_bad`), `opc_status_code: str`, `message: str`.
|
||
- **Per-message flags (C2):** `write_timestamp`/`write_quality` arrive on each NATS message; resolve as `cmd.get('write_timestamp', self.write_timestamp)` (per-message wins; config is the fallback default). Do not hard-code engine-global-only behavior.
|
||
- `async write(cmd)`: rate-limit (min pause via `MonotonicClock`); `value = to_typed_value(cmd['value'], cmd['data_type'])` inside try/except → on `ConversionError` return error `WriteResult` + `OPC_ERROR_BY_TYPE`; build `ua.DataValue(ua.Variant(value, ua_type), StatusCode_=ua.StatusCode(ua.StatusCodes.Good) if write_quality else None, SourceTimestamp=parse(ts) if write_timestamp else None)`; `node = client.get_node(addr); await node.write_value(dv)`; map OPC exceptions → `error_type` for `OPC_ERROR_BY_TYPE`; emit `OPC_WRITE_TOTAL`/`OPC_WRITE_LATENCY`/`OPC_WRITE_STATUS_BY_MODEL`; then `await self.sink.update_write_status(cmd, res)`.
|
||
- Quality downgrade `Good_Local_Override (0xD8)` when a clock problem is detected — leave a `# ponytail:` marker; wire only when NTP/clock detection is added.
|
||
|
||
- [ ] **Step 4: Run — expect PASS.**
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): WriteEngine (status_code/timestamp/flags, safe converter, rate-limit)"`
|
||
|
||
---
|
||
|
||
### Task 12: DataSink.update_write_status (Postgres, contract C3)
|
||
|
||
**Files:**
|
||
- Modify: `ingestor/io/data_sink.py`
|
||
- Test: `tests/unit/io/test_data_sink_status.py`
|
||
|
||
- [ ] **Step 1: Failing test** — issues an `UPDATE <schema>.<table> SET write_status=... WHERE timestamp=... AND model_id=...`.
|
||
|
||
```python
|
||
# tests/unit/io/test_data_sink_status.py
|
||
from unittest.mock import AsyncMock
|
||
import pytest
|
||
from ingestor.io.data_sink import DataSink
|
||
from ingestor.engines.write import WriteResult
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_write_status_targets_model_row():
|
||
pg = AsyncMock()
|
||
sink = DataSink(data_manager=AsyncMock(), postgres=pg, logger=AsyncMock())
|
||
cmd = {'schema': 'sientia_data', 'table_name': 'pred_m1',
|
||
'model_id': '1', 'timestamp': '2026-07-13T00:00:00Z'}
|
||
await sink.update_write_status(cmd, WriteResult(ok=True, status='ok', opc_status_code='Good', message=''))
|
||
assert pg.execute.await_count == 1
|
||
sql = pg.execute.call_args[0][0]
|
||
assert 'update' in sql.lower() and 'write_status' in sql.lower()
|
||
```
|
||
|
||
- [ ] **Step 2–4:** implement `update_write_status(cmd, result)` — parametrized UPDATE on `cmd['schema']`.`cmd['table_name']` filtered by `timestamp`+`model_id`; use the `sientia_do` Postgres client (`self.postgres`). **Never f-string the values** (SQL injection) — bind params; the identifiers (schema/table) are validated against `[A-Za-z0-9_]+`. Run to green.
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): DataSink.update_write_status Postgres (C3)"`
|
||
|
||
> **Cross-repo note (1975):** the column only appears on *new* per-model tables once Laborious's predictions DataFrame includes `write_status`; *existing* tables need the Flyway backfill `ALTER TABLE ... ADD COLUMN IF NOT EXISTS write_status`. Both are 1975/migrations work — this task only writes the value.
|
||
|
||
---
|
||
|
||
### Task 13: NatsWriteConsumer (contract C2)
|
||
|
||
**Files:**
|
||
- Create: `ingestor/adapters/nats_consumer.py`, `ingestor/adapters/__init__.py`
|
||
- Test: `tests/unit/adapters/test_nats_consumer.py`
|
||
|
||
- [ ] **Step 1: Failing test** — an incoming NATS message is JSON-decoded and dispatched to `WriteEngine.write`; malformed JSON is logged and dropped (no raise).
|
||
|
||
```python
|
||
# tests/unit/adapters/test_nats_consumer.py
|
||
import json
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
import pytest
|
||
from ingestor.adapters.nats_consumer import NatsWriteConsumer
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_message_dispatched_to_write_engine():
|
||
we = AsyncMock()
|
||
consumer = NatsWriteConsumer(nats_url='nats://x:4222', write_engine=we, logger=AsyncMock())
|
||
msg = MagicMock(); msg.data = json.dumps({'server_id': 's', 'addr': 'ns=2;i=5'}).encode()
|
||
await consumer._handle(msg)
|
||
we.write.assert_awaited_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bad_json_does_not_raise():
|
||
consumer = NatsWriteConsumer(nats_url='x', write_engine=AsyncMock(), logger=AsyncMock())
|
||
msg = MagicMock(); msg.data = b'not-json'
|
||
await consumer._handle(msg) # must not raise
|
||
```
|
||
|
||
- [ ] **Step 2: Run — expect FAIL.**
|
||
|
||
- [ ] **Step 3: Implement** `NatsWriteConsumer`:
|
||
- `async start()`: `self.nc = await nats.connect(self.nats_url)`; `await self.nc.subscribe('opc.write', queue='gateway-writers', cb=self._handle)` (queue group → exactly one instance handles each message).
|
||
- `async _handle(msg)`: `json.loads` in try/except (drop + log on error); `await self.write_engine.write(payload)` (fire-and-forget — no reply).
|
||
- `async stop()`: drain/close.
|
||
|
||
- [ ] **Step 4: Run — expect PASS.**
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): NatsWriteConsumer (Core NATS, queue group) (C2)"`
|
||
|
||
---
|
||
|
||
### Task 13.end: F2 close-out (see [Phase Delivery](#phase-delivery--commit-ledger))
|
||
- [x] Full `pytest tests` green (192) + `ruff` clean.
|
||
- [x] F2 commit SHAs recorded in the [Commit Ledger](#commit-ledger) (base `2a9af05`).
|
||
- [ ] PR intentionally NOT opened yet — opened only after F4 closes.
|
||
|
||
---
|
||
|
||
## Phase F3 — Active reads: scheduled + on-demand gRPC (contract C4)
|
||
|
||
### Task 14: ActiveReadEngine
|
||
|
||
**Files:**
|
||
- Create: `ingestor/engines/active_read.py`
|
||
- Test: `tests/unit/engines/test_active_read.py`
|
||
|
||
- [ ] **Step 1: Failing test** — `read_now(server_id, addr)` reads via pool, persists via `DataSink.publish_reading`, and returns value+status+timestamps.
|
||
|
||
```python
|
||
# tests/unit/engines/test_active_read.py
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
import pytest
|
||
from ingestor.engines.active_read import ActiveReadEngine
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_read_now_persists_and_returns():
|
||
dv = MagicMock(); dv.Value.Value = 42
|
||
node = AsyncMock(); node.read_data_value.return_value = dv
|
||
client = MagicMock(); client.get_node.return_value = node
|
||
pool = AsyncMock(); pool.get.return_value = client
|
||
sink = AsyncMock()
|
||
eng = ActiveReadEngine(pool=pool, sink=sink, logger=AsyncMock(), metadata={})
|
||
res = await eng.read_now('s', 'ns=2;i=2', names=[{'collection': 'c', 'tag_name': 't', 'model_id': '1'}])
|
||
assert res['value'] == 42 and res['persisted'] is True
|
||
sink.publish_reading.assert_awaited_once()
|
||
```
|
||
|
||
- [ ] **Step 2: Run — expect FAIL.**
|
||
- [ ] **Step 3: Implement** — `read_now(...)` (read via `node.read_data_value()`, emit `OPC_ACTIVE_READ_*`, persist, return dict) and `scheduled_tick(slot)` (iterate tags with `active_read.enabled`, read every `cycle_multiple` ticks, persist). Run to green.
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): ActiveReadEngine (scheduled + read_now)"`
|
||
|
||
---
|
||
|
||
### Task 15: gRPC proto + codegen
|
||
|
||
**Files:**
|
||
- Create: `ingestor/proto/opc_gateway.proto`
|
||
- Generate: `ingestor/proto/opc_gateway_pb2.py`, `_pb2_grpc.py`
|
||
|
||
- [ ] **Step 1: Write proto** (contract C4):
|
||
|
||
```proto
|
||
syntax = "proto3";
|
||
package opcgateway;
|
||
|
||
service OpcGateway {
|
||
rpc ReadTag(ReadTagRequest) returns (ReadTagResponse);
|
||
}
|
||
message ReadTagRequest { string server_id = 1; string addr = 2; string data_type = 3; }
|
||
message ReadTagResponse {
|
||
string value = 1; string opc_status_code = 2;
|
||
string source_timestamp = 3; string server_timestamp = 4; bool persisted = 5;
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 2: Generate stubs**
|
||
|
||
Run: `python -m grpc_tools.protoc -I ingestor/proto --python_out=ingestor/proto --grpc_python_out=ingestor/proto ingestor/proto/opc_gateway.proto`
|
||
Expected: `opc_gateway_pb2.py` and `opc_gateway_pb2_grpc.py` created. Fix the generated import to be package-relative if needed.
|
||
|
||
- [ ] **Step 3: Commit** — `git commit -am "feat(gateway): gRPC proto + generated stubs (C4)"`
|
||
|
||
---
|
||
|
||
### Task 16: GrpcServer
|
||
|
||
**Files:**
|
||
- Create: `ingestor/adapters/grpc_server.py`
|
||
- Test: `tests/unit/adapters/test_grpc_server.py`
|
||
|
||
- [ ] **Step 1: Failing test** — `ReadTag` delegates to `ActiveReadEngine.read_now` and maps the dict to the response message.
|
||
|
||
```python
|
||
# tests/unit/adapters/test_grpc_server.py
|
||
from unittest.mock import AsyncMock
|
||
import pytest
|
||
from ingestor.adapters.grpc_server import ReadTagServicer
|
||
from ingestor.proto import opc_gateway_pb2 as pb
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_readtag_maps_response():
|
||
eng = AsyncMock()
|
||
eng.read_now.return_value = {'value': '42', 'opc_status_code': 'Good',
|
||
'source_timestamp': 'ts', 'server_timestamp': 'ts2', 'persisted': True}
|
||
servicer = ReadTagServicer(active_read=eng, slot_lookup=AsyncMock())
|
||
resp = await servicer.ReadTag(pb.ReadTagRequest(server_id='s', addr='ns=2;i=2'), None)
|
||
assert resp.value == '42' and resp.persisted is True
|
||
```
|
||
|
||
- [ ] **Step 2–4:** implement `ReadTagServicer` (async grpc.aio) + `GrpcServer.start()/stop()` binding `grpc.aio.server()` on the configured port; look up `names` for the address from the slot layer (or accept none). Run to green.
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): gRPC server ReadTag (C4)"`
|
||
|
||
---
|
||
|
||
### Task 16.end: F3 close-out (see [Phase Delivery](#phase-delivery--commit-ledger))
|
||
- [x] Full `pytest tests` green (206) + `ruff` clean.
|
||
- [x] F3 commit SHAs recorded in the [Commit Ledger](#commit-ledger) (base `4855205`).
|
||
- [ ] PR intentionally NOT opened yet — opened only after F4 closes.
|
||
|
||
---
|
||
|
||
## Phase F4 — Wiring, health, deploy, hardening
|
||
|
||
### Task 17: Config additions
|
||
|
||
**Files:**
|
||
- Modify: `ingestor/config.py` (or wherever env is read), `.env.example`
|
||
- Test: `tests/unit/test_config.py`
|
||
|
||
- [ ] **Step 1–4 (TDD):** add env: `NATS_URL`, `NATS_WRITE_SUBJECT` (default `opc.write`), `GRPC_PORT` (default `50051`), `GRPC_ENABLED`, `CERT_BUCKET` (default `sientia-dataops`), `CERT_CACHE_DIR` (default `/tmp/opc-certs`), `WRITE_TIMESTAMP`/`WRITE_QUALITY`/`WRITE_MIN_PAUSE_S`, `NTP_ENABLED`. Test defaults. Run to green.
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): config for NATS/gRPC/certs/write flags"`
|
||
|
||
---
|
||
|
||
### Task 18: Wire engines + adapters into app
|
||
|
||
**Files:**
|
||
- Modify: `ingestor/app.py`, `ingestor/ingestor.py`
|
||
- Test: `tests/unit/test_app.py` (extend)
|
||
|
||
- [ ] **Step 1: Failing test** — app startup constructs pool + engines + adapters and sets `APP_UP=1`; shutdown stops adapters, closes pool, sets `APP_UP=0`.
|
||
- [ ] **Step 2: Run — expect FAIL.**
|
||
- [ ] **Step 3: Implement** — in startup: build `OpcConnectionPool`, `DataSink`, `SubscriptionEngine`, `ActiveReadEngine`, `WriteEngine`, start `NatsWriteConsumer` and (if `GRPC_ENABLED`) `GrpcServer`; register them in the main loop and signal handlers. Reuse the existing `start_http_server(HTTP_METRICS_PORT)` for `/metrics`.
|
||
- [ ] **Step 4: Run — expect PASS.**
|
||
- [ ] **Step 5: Commit** — `git commit -am "feat(gateway): wire pool+engines+adapters into app"`
|
||
|
||
---
|
||
|
||
### Task 19: Health probe alignment (spec §9/§10)
|
||
|
||
**Files:**
|
||
- Modify: `ingestor/app.py` (ensure `APP_UP` is set on the `:9090/metrics` endpoint), deploy values (`values/opc-gateway.yaml` if present in this repo; otherwise note for the Helm repo).
|
||
|
||
- [ ] **Step 1:** confirm `app_up{pod_id=...} 1` is exposed on `:9090/metrics` (already emitted; verify the probe string `curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1'` matches).
|
||
- [ ] **Step 2:** if a `values/opc-gateway.yaml` lives here, set the exec probe + `waitForDependencies` (Gitea, MongoDB, Redis, Postgres, MinIO, NATS) + ServiceMonitor label `release: kube-prometheus-stack`. Otherwise add a `docs/deploy-notes.md` capturing these for the Helm repo.
|
||
- [ ] **Step 3: Commit** — `git commit -am "chore(gateway): app_up health probe + deploy notes"`
|
||
|
||
---
|
||
|
||
### Task 20: Dockerfile / entrypoint
|
||
|
||
**Files:**
|
||
- Modify: `Dockerfile` (or note for `sientia-container-images`)
|
||
|
||
- [ ] **Step 1:** ensure the image runs `ingestor.app`, non-root, `readOnlyRootFilesystem`-compatible (writable `/tmp` for `CERT_CACHE_DIR`); pin base image; no `latest`. Hadolint clean.
|
||
- [ ] **Step 2:** Run `hadolint Dockerfile` (or note).
|
||
- [ ] **Step 3: Commit** — `git commit -am "chore(gateway): dockerfile entrypoint + read-only-fs cert cache"`
|
||
|
||
---
|
||
|
||
### Task 21: Full test + lint sweep and 1879 checklist verification
|
||
|
||
**Files:** all
|
||
|
||
- [ ] **Step 1:** `python -m pytest tests -v` → all green.
|
||
- [ ] **Step 2:** `ruff check ingestor tests` → clean.
|
||
- [ ] **Step 3:** Walk spec §13 (1879) and check each box in code: Sha256, monotonic, status_code+timestamp, timestamp/quality flags, rate-limit, BUG-1/BUG-2, channel>session, OPC error typing, file/function size. Add any missing micro-test.
|
||
- [ ] **Step 4: Commit** — `git commit -am "test(gateway): full sweep + 1879 checklist verification"`
|
||
|
||
---
|
||
|
||
### Task 21.end: F4 close-out + open per-phase PRs (see [Phase Delivery](#phase-delivery--commit-ledger))
|
||
- [x] Full `pytest tests` green (222) + `ruff` clean; all F0–F4 phases complete and unit-tested.
|
||
- [x] F4 commit SHAs recorded in the [Commit Ledger](#commit-ledger) (base `e9ce9e5`).
|
||
- [ ] **Now** open one PR per phase (F0→F4) from the recorded SHA ranges, in order — this is the only point where PRs are opened. *(Pending: user go-ahead — requires push; also note the epic's end-to-end validation with a real NATS broker + OPC server machine is agreed for deployment time, per the feature owner.)*
|
||
|
||
---
|
||
|
||
## Deferred / out of this plan (tracked in spec §17)
|
||
- NATS provisioning in air-gapped Helm (platform).
|
||
- Moving 1569 cert write path to `sientia-dataops` (platform + 1569 app).
|
||
- gRPC caller identity / auth model (Vitor) → may add auth to `GrpcServer` later.
|
||
- Flyway `write_status` column migration + Laborious DataFrame change (1975 plan).
|
||
- JetStream durability (revisit if Core NATS loss becomes real).
|
||
|
||
## Open validation notes for the implementer
|
||
- Verify exact `sientia_do` signatures (`MongoDBRepository.find_one`, `MinioRepository.download_file`, Postgres `execute`) against the pinned version; adapt calls.
|
||
- `asyncua` `ua.DataValue`/`ua.Variant`/`StatusCode` construction: confirm the constructor kwargs on the pinned `asyncua` (1.1.5) before Task 11.
|
||
- The functional test against `sientia-dataops-opc_simulator` (sibling repo) is a good end-to-end smoke once F0–F3 land.
|