diff --git a/.gitignore b/.gitignore index 198783f..3f24065 100644 --- a/.gitignore +++ b/.gitignore @@ -181,4 +181,10 @@ git_log .git/ .ruff_cache/ -.mypy_cache/ \ No newline at end of file +.mypy_cache/ + +# Local planning/validation notes (not committed, not team docs) +docs/superpowers/ +docs/e2e-test-guide.md +docs/deploy-notes.md +docs/HANDOFF.md \ No newline at end of file diff --git a/docs/superpowers/plans/2026-07-13-opc-gateway.md b/docs/superpowers/plans/2026-07-13-opc-gateway.md deleted file mode 100644 index 812e52a..0000000 --- a/docs/superpowers/plans/2026-07-13-opc-gateway.md +++ /dev/null @@ -1,932 +0,0 @@ -# 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 -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:** _(fill at F4 close-out)_ - -### 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 . 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)) -- [ ] Full `pytest tests` green + `ruff` clean; all F0–F4 phases complete, tested, working. -- [ ] Record F4 commit SHAs (base + range) in the [Commit Ledger](#commit-ledger). -- [ ] **Now** open one PR per phase (F0→F4) from the recorded SHA ranges, in order — this is the only point where PRs are opened. - ---- - -## 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. diff --git a/docs/superpowers/specs/2026-07-13-opc-gateway-design.md b/docs/superpowers/specs/2026-07-13-opc-gateway-design.md deleted file mode 100644 index 0dee72f..0000000 --- a/docs/superpowers/specs/2026-07-13-opc-gateway-design.md +++ /dev/null @@ -1,251 +0,0 @@ -# OPC Gateway — Design (SIENTIAPDE-1972) - -- **Épico:** [SIENTIAPDE-1972](https://aignosi-wiki.atlassian.net/browse/SIENTIAPDE-1972) — OPC Gateway derivado do OPC Ingestor, com ajustes no Orchestrator e no Laborious. -- **Sub-tasks:** [1973](https://aignosi-wiki.atlassian.net/browse/SIENTIAPDE-1973) (Gateway), [1974](https://aignosi-wiki.atlassian.net/browse/SIENTIAPDE-1974) (Orchestrator), [1975](https://aignosi-wiki.atlassian.net/browse/SIENTIAPDE-1975) (Laborious). -- **Referências:** [1879](https://aignosi-wiki.atlassian.net/browse/SIENTIAPDE-1879) (boas práticas OPC), [1569](https://aignosi-wiki.atlassian.net/browse/SIENTIAPDE-1569) (banco de tags / CRUD conectores). -- **Branch:** `feature/SIENTIAPDE-1972-opc-gateway-...` (repo `sientia-dataops-opc-ingestor`). -- **Data:** 2026-07-13. - ---- - -## 1. Contexto e objetivo - -Evoluir o `opc-ingestor` (hoje read-only, leitura por subscription) para um **OPC Gateway** completo: - -- Gestão de certificados via MinIO + MongoDB (reaproveitando o épico 1569). -- Leitura por **subscription** usando o sistema de slots já existente. -- **Leitura ativa**: agendada (frequência múltipla do ciclo de operação) e **on-demand via GRPC**. -- **Escrita ativa** via **NATS** como broker intermediário. -- **Métricas detalhadas** de todas as operações. -- Boas práticas / correções do card 1879. - -O trabalho atravessa 3 repositórios. Este spec detalha o **Gateway** (repo atual) e define os **contratos** para Orchestrator (1974) e Laborious (1975); cada um ganha spec próprio no seu repo. - -## 2. Escopo - -**Dentro:** arquitetura interna do Gateway e componentes; consumo do banco de tags (1569) e do slot v2 no Redis; escrita OPC via NATS; leitura ativa agendada e on-demand via GRPC; métricas/observabilidade; checklist 1879; contratos C1–C4; integração de deploy (Helm/air-gapped) do serviço. - -**Fora (specs próprios):** implementação do 1974 (`orchestrator_temporal`) e do 1975 (`laborious_temporal`); CRUD/UI do banco de tags (1569, já feito); provisionamento do broker NATS na plataforma (dependência, ver §8). - -## 3. Documentos-fonte autoritativos (aig-docs) - -- **Brief oficial do serviço:** `07 - Projects & R&D/sientia/opc-gateway/topics.md` — escopo canônico (confirma tudo acima; define a coluna de status de escrita na tabela de predições — ver C3). -- **DataOps OPC (contrato de dados):** `.../technical-documentation/model-lifecycle/data-ingestion-opc-scouter.md` — coleções Mongo `opc-servers`, `pipelines` (`read_tags[]`: `tag_name/server_id/tag_address/aggr_func/frequency/data_range`, filtros, `tag_retention_minutes`), `orchestrated_schedules`; **gotcha:** `updated_at` deve ser BSON `Date` (não string ISO). Orchestrator sincroniza pipelines→Temporal em cron `*/30`. -- **Helm/air-gapped:** `.../helm-chart-air-gapped/**` — chart genérico `sientia-module`, namespace `sientia-core`, `values/.yaml` + perfis, init-containers, probes, netpol default-deny, image signing. Ver §10. -- **Observabilidade:** `.../helm-chart-air-gapped/07-observability-monitoring/**`. Ver §9. -- **Onboarding/dev:** `01 - Welcome & Onboarding/technical-onboarding.md`, `06 - Engineering & DevOps/**`. Ver §11. -- **Incidente OPC real (Samarco):** `04 - Clients & Deployments/Samarco/reports/2026-06-opc-maintance/report-samarco-2026-06-opc-timeout-kepserver.md` — `TimeoutError` em escrita `asyncua`; restrições do servidor (block size 512 vs 255; chars especiais em endereços `- _ / .`; degradação de sessão ao longo do tempo; publishing interval 2000ms; write timeout 30000ms). Motiva broker + retry + métricas de erro. - -## 4. Estado atual (fatos do código) - -### opc-ingestor (base) -- `asyncua` 1.1.5. Leitura **só por subscription**. Sem escrita, NATS ou GRPC. -- Slots no **Redis** (`slot:opc_tags:{id}`), com lease/heartbeat para balanceamento. -- Managers: `opc_manager`, `ingestor_manager`, `data_manager`, `resource_manager`. Egress: Mongo (obrigatório) + Kafka legado opcional (**removido no Gateway** para simplificar — decisão Vitor 2026-07-14). Métricas Prometheus em `:9090/metrics`. -- Conexão OPC hoje é **por slot arrematado** (só conecta nos servidores dos slots que possui). - -### dataops-library -- `MongoDBRepository` (async: `find/find_one/distinct/aggregate/insert/insert_many/update_one/update_many/delete_one/delete_many`), `MinioRepository`, `opc_ua_client`, `Postgres` (`sientia_do.temporal.activities.postgres`), helpers de métricas (`sientia_do.observability`). Métodos genéricos em `feature/SIENTIAPDE-1590`. **Sem** NATS/GRPC/modelo de slot. - -### Banco de tags (1569) — `sientia-projects-templates @ rc/SIENTIAPDE-1588` -Mongo `sientia`: - -| Coleção | Campos | -|---|---| -| `connectors` (opc_ua) | `connector_id`(int), `type`, `name`, `url`, `server_uri`, `subscription_period_ms`, `use_security`(bool), `status`, `description` | -| `connector_tags` | `tag_id`(int), `connector_id`(FK), `address`(NodeId), `tag_name`, `category`(`Read`/`Write`/`Prediction Confidence`), `data_type`, `unit`, `status`, `description` + atributos custom top-level | -| `platform_certificates` | doc único: `client_cert{minio_object_name,sha256_hash,uploaded_at,original_filename,expiration_date}`, `private_key{...}` | - -- **Segurança já é `Basic256Sha256`**; cert de plataforma **único**, baixado do MinIO (bucket `streamlit-connectors`, prefixo `platform/certificates/`) para temp e passado ao `asyncua`. -- **Multi-nome por endereço já é permitido** (unicidade em `(connector_id, tag_name)`, nunca em `address`). -- **Não existem** `model_id/model_name`, `collection`, `frequency` em `connector_tags` (vêm do lado modelo/pipeline). O subtask 1600 (pendente) adiciona `Category` + atributos custom como chaves flat na raiz. **Decisão Vitor:** cada tag também tem um campo **`metadata`** (objeto), deixado **vazio** por ora; tags de teste criadas com `metadata: {}` (para documentação). -- Fix aplicado: `projects-templates/app/src/operations/db_infra.py:16` corrigido para o caminho local correto da lib (import validado). - -### Escrita hoje (Laborious) -- Escrita OPC **direta** via `asyncua` (`laborious/activities/opc.py` → `opc_repository.write_data` → `node.write_value`). -- Status de escrita é gambiarra: sobrescreve `prediction_confidence` com sentinelas (`12`=erro escrita, `14`=sessão ruim; PI=`13`) em `process_confidence`. Predições persistidas em Postgres (`model_id, prediction, prediction_confidence, response_time, prediction_status, timestamp, created_at, comments`) via **pandas `to_sql`** (schema/tabela por-modelo, criados por inferência do DataFrame; `predictions_schema.predictions` é só fixture de teste). -- Payload de escrita já contém: `server_id`, `addr`, `value`, `data_type`, `tag_type`, `model_id`, `model_name`, `timestamp`. - -## 5. Arquitetura do Gateway (Abordagem B — engines por capacidade) - -``` -Orchestrator ─(Redis slot v2)─▶ SubscriptionEngine ─┐ -Laborious ────(NATS write)────▶ NatsWriteConsumer ──▶ WriteEngine ────┐ -caller ───────(GRPC read)─────▶ GrpcServer ─────────▶ ActiveReadEngine┤ -scheduler ────────────────────────────────────────▶ ActiveReadEngine┤ - ▼ - OpcConnectionPool (por servidor) ─▶ servidores OPC - │ - DataSink ◀────────────────────┘ - ├─▶ Mongo (dados) - └─▶ Postgres predictions.write_status -``` - -Decisão-chave: **conexões OPC deixam de ser atreladas a slot** (escrita e leitura on-demand miram qualquer servidor) → **pool por servidor** compartilhado. - -### Componentes -1. **`OpcConnectionPool`** *(novo)* — conexões `asyncua` por conector/servidor, reuso + *idle eviction*; `Basic256Sha256`; `secure_channel_timeout = int(session_timeout*1.25)`; certs do MinIO/Mongo (para `emptyDir`/tmpfs, ver §12); referência única `time.monotonic()`; opção NTP; reconexão + métricas de conexão e tipos de erro OPC. Tolera restrições do incidente Samarco (block size, timeouts, degradação de sessão). -2. **`SubscriptionEngine`** *(evolui `ingestor_manager`+`opc_manager`)* — arremata slots (lease/heartbeat), subscription por (slot, servidor) via pool; `datachange` faz **fan-out multi-nome** para cada `{collection, tag_name, model_name, model_id}`. -3. **`ActiveReadEngine`** *(nova)* — agendada (freq. múltipla do ciclo) e on-demand (GRPC: lê, **persiste E devolve**); mede *lag* de leitura (ts recebido − ts máquina). -4. **`WriteEngine`** *(nova)* — escrita OPC com `DataValue`+`Variant`+`status_code` Good+`source_timestamp`; flags timestamp/quality (`Good_Local_Override 0xD8`); **converter seguro** (BUG-1 bool, BUG-2 try/except); **rate-limit**; diferencia erros OPC; grava `write_status` na tabela de predições (C3). -5. **Adapters** — `NatsWriteConsumer` (subject de escrita, **queue group** `gateway-writers`, Core NATS fire-and-forget; JetStream depois) e `GrpcServer` (leitura on-demand). -6. **`DataSink`** *(evolui `data_manager`)* — persiste leituras (**Mongo**, fan-out; `updated_at` como BSON Date) e `update_write_status(model_id, timestamp, status,...)` na tabela de predições (Postgres). Kafka removido. -7. **`app`** — instancia pool+engines+adapters, sobe Prometheus/health, trata sinais e shutdown. - -Coordenação: lease/heartbeat governam SubscriptionEngine e leitura agendada; escrita/on-demand não dependem de lease. Meta 1879: ≤ ~500 linhas/arquivo, ≤ ~20/função. - -## 6. Contratos - -### C1 — Slot v2 no Redis (`slot:opc_tags:{id}`) -Orchestrator escreve, Gateway lê. Substitui `topics:[str]` por lista multi-nome por endereço + config de leitura ativa: -```json -{ - "server1": { - "server_id": "1", "name": "server1", "url": "opc.tcp://host:4840", - "server_uri": "urn:...", "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"} - ] - } - } - } -} -``` -Orchestrator monta via **JOIN**: identidade da tag (`connector_tags` + `connectors`, do banco 1569) × config de modelo (`model_id/model_name/collection/frequency`, lado `models`/pipeline). Multi-nome = múltiplas entradas em `names`. - -### C2 — Mensagem NATS de escrita -Laborious publica, Gateway consome. Core NATS, subject `opc.write` (avaliar `opc.write.{server_id}`), queue group `gateway-writers`, fire-and-forget. Deve carregar `timestamp`+`model_id` (correlação com C3). -```json -{ "msg_id":"uuid","server_id":"server1","addr":"ns=2;i=5","value":42.0, - "data_type":"float","tag_type":"prediction","model_id":"1","model_name":"m1", - "schema":"sientia_data","table_name":"predictions_", - "timestamp":"2026-07-13T12:00:00Z","write_timestamp":true,"write_quality":false } -``` -`schema`+`table_name` são necessários porque a tabela de predição é **por-modelo** — o Gateway usa-os para o `UPDATE` da coluna `write_status` (ver C3). - -### C3 — Coluna de status de escrita (Postgres, tabela de predições) ⭐ autoritativo em topics.md -Nova coluna no DDL padrão de predições (ex.: `write_status` + opcional `write_status_detail`/`write_status_at`). Em **escrita OPC**, o **Gateway** grava o status vinculando por **`timestamp` + `model_id`**. Em **escrita PI Web API**, o **Laborious** grava. Substitui as sentinelas `12/13/14` em `prediction_confidence`. -- **Onde:** schema/tabela de predição são **config por modelo** (`input_data['schema']`/`['table_name']` no Laborious) — não há tabela única fixa. A coluna faz parte do **DDL padrão** de predições; homologar no schema gerenciado pelo Flyway (**`sientia_data`**). Migração via `Aignosi/sientia-dataops-database-migrations` (`.sql` versionado, Job Helm hook), backward-compatible. -- **Provisionamento (LOCALIZADO):** **não há DDL/template** — as tabelas são criadas por **pandas `DataFrame.to_sql(if_exists='append')`** em `sientia_do/temporal/activities/postgres.py:439-464`, com o **schema inferido das colunas do DataFrame**. Logo: (a) **tabelas novas** ganham `write_status` automaticamente quando o DataFrame de predições do Laborious passar a incluir a coluna (trabalho do 1975); (b) **tabelas existentes** precisam de **backfill** `ALTER TABLE .
ADD COLUMN IF NOT EXISTS write_status ...` — como são dinâmicas/por-modelo, a migração Flyway enumera as tabelas existentes (ex.: via `information_schema`) ou usa migração repetível/callback. -- **Fluxo de gravação:** o Laborious grava a linha de predição (via `to_sql`) com `write_status` default (ex.: `pending`/null); depois o Gateway faz `UPDATE . SET write_status=... WHERE timestamp=... AND model_id=...`. Por isso a mensagem NATS (C2) carrega `schema`+`table_name` (a tabela é por-modelo). - -### C4 — GRPC leitura on-demand -``` -service OpcGateway { rpc ReadTag(ReadTagRequest) returns (ReadTagResponse); } -ReadTagRequest { string server_id; string addr; optional string data_type; } -ReadTagResponse { string value; string opc_status_code; string source_timestamp; - string server_timestamp; bool persisted; } -``` -Lê agora, persiste no pipeline e devolve. **Sem consumidor hoje** (talvez o frontend no futuro — Vitor, 2026-07-14): endpoint **interno ao cluster, sem auth** por ora; fase de menor prioridade, adiável sem afetar leitura/escrita. - -## 7. Gestão de certificados -Reaproveita o mecanismo do 1569: metadados em Mongo `platform_certificates` + arquivos no MinIO. O `OpcConnectionPool` baixa `client_cert`/`private_key` para **`emptyDir`/tmpfs** (root FS é read-only, §12) e passa ao `asyncua` com `Basic256Sha256`, espelhando `_create_opc_ua_client`. Cert de plataforma é único/compartilhado. - -**Bucket (DECIDIDO): `sientia-dataops`.** Padrão de MinIO (`03-database-storage/minio-strategy.md`, Bruno Domingues) é bucket + user scoped por serviço; DataOps/`sientia-core` usa **`sientia-dataops`**. O Gateway lê os certs de `sientia-dataops`. **A escrita dos certs do CRUD 1569 passa a ser em `sientia-dataops`** (hoje grava no bucket do frontend `streamlit-connectors`) — ajuste no app 1569 + policy do user do frontend para o bucket DataOps. - -## 8. NATS como nova dependência de plataforma -**Decisão: NATS** (definido pelo brief oficial `opc-gateway/topics.md` e confirmado pelo dono da feature, Vitor Pimentel — fonte mais atualizada). O padrão de subjects/queue groups do NATS casa com o comando de escrita ponto-a-ponto de baixa latência. - -Nota de plataforma: os docs air-gapped diziam *"do not include RabbitMQ/NATS unless a clear requirement arises from a new component"* (`helm-chart-air-gapped/03-database-storage/other-data-stores.md`; `requirements-checklist.md:106`). O **OPC Gateway é exatamente esse novo componente** que cria o requisito — logo o NATS entra na plataforma como dependência de primeira classe. Ônus de onboarding a cumprir: -- Subchart Helm + versão pinada (sem `latest`); imagem no bundle air-gapped (Trivy/Cosign). -- Entrada nos perfis de sizing; `NetworkPolicy` de egress no `sientia-core`. -- Credencial via `sientia-secrets-chart`. -- HA/persistência: **Core NATS** agora (fire-and-forget, sem storage); JetStream depois exige storage/persistência. - -## 9. Métricas e observabilidade -- **Métricas (1973+1879):** `app_up`; sucesso/erro de escrita e leitura ativa; latência (histograma) de escrita e leitura ativa; tags na subscription; slots possuídos; conexões/desconexões; **status de escrita por modelo**; **valor por tag/modelo**; ciclos sem dado em subscription; **tipos de erro OPC**; **lag de leitura**. Prefixo sugerido `opc_gateway_*`; manter `app_up`. -- **Exposição:** `/metrics` em `:9090`; **ServiceMonitor** com label `release: kube-prometheus-stack`. -- **Health probe (padrão DataOps):** exec `curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1'`. -- **Logs:** JSON estruturado (Loki+Promtail; indexado por namespace/pod/container). -- **Alerting:** AlertManager (já existe `SientiaAppDown`, `KubePodCrashLooping`, `DatabaseConnectionFailure`). Dashboard Grafana "OPC Ingestor" existe — estender/criar um para o Gateway. - -## 10. Deploy e integração de plataforma (Helm / air-gapped) -- **Chart:** consumidor do chart genérico **`sientia-module`** (app carrega só `values/opc-gateway.yaml`; sem templates). Namespace **`sientia-core`**, Phase 5. -- **Perfis de sizing:** adicionar entrada por perfil (`profiles/{small,medium,large}.yaml`) com requests/limits do Gateway (e do NATS). YAML estático, sem Go template em values. -- **Código:** workers clonam o app do **Gitea** via SSH no startup (`PYTHON_APP`); libs compartilhadas via Library Distribution Server (PyPI em MinIO). Base image `sientia-module`, Python 3.11+. -- **Init-containers:** `busybox nc -z` em loop sobre `waitForDependencies.services[]` — Gateway espera **Gitea, MongoDB, Redis, Postgres, MinIO, NATS**. -- **Segurança do pod (herdada do library chart):** `runAsNonRoot`, `readOnlyRootFilesystem: true`, `drop ALL`, writable só `/tmp` e `/.cache` → certs OPC em `emptyDir`/tmpfs. -- **NetworkPolicy default-deny:** liberar egress p/ PG/Mongo/Redis/MinIO/**NATS**/servidores OPC (4840) e ingress do Prometheus (9090); GRPC exposto é **net-new** (hoje o único gRPC da plataforma é o do Temporal) → adicionar regra. -- **Migração de banco (C3):** **Flyway** (repo `Aignosi/sientia-dataops-database-migrations`, schema `sientia_data`; substituiu Alembic/SQLAlchemy). Coluna nova = `.sql` versionado (V2+), imutável, aplicado por Job Helm hook `pre-upgrade`; backward-compatible. -- **Imagem:** assinada (Cosign) e escaneada (Trivy) no bundle; tags imutáveis (SemVer/SHA), nunca `latest`; `Chart.lock` commitado. -- **MinIO:** padrão é bucket + user scoped por serviço; o bucket do `sientia-core`/DataOps é **`sientia-dataops`**. Ver fronteira de certs em §7/§17. - -## 11. Convenções de desenvolvimento -- **Nome de branch dirige versionamento e é checado no CI:** `feature/`→minor, `fix/|hotfix/|bugfix/`→patch, `release/`→major, `rc/`, etc.; descrição em kebab-case; **branch fora do padrão falha o CI**. ✅ Branch já renomeada para `feature/SIENTIAPDE-1972-...`. -- **Ticket-first:** referência do Jira no branch e no commit (ex.: `feat(gateway): ... (SIENTIAPDE-1973)`). -- **Qualidade:** SonarQube (quality gate bloqueia merge — coverage, smells, security hotspots, duplicação, complexidade). Sem config numérica de tamanho de arquivo/função documentada além do 1879; qualidade delegada ao Sonar. Dockerfiles via `.hadolint.yaml`, centralizados em `sientia-container-images`, sem `latest`. -- **Testes:** unit + integration (pytest) + E2E (Playwright quando aplicável). -- **Docs-first, em inglês, um tópico por arquivo, com frontmatter** (regra de `aig-docs`). - -## 12. Segurança / secrets -- Certs OPC não podem ir para o root FS (read-only) → baixar para `emptyDir`/tmpfs em runtime. -- NetworkPolicy default-deny em `sientia-core`; adicionar allow-list para NATS e GRPC exposto + ServiceMonitor. -- ServiceAccount dedicada por app (default basta se não precisar de API do k8s). -- Secrets via `sientia-secrets-chart` (Sealed Secrets / `global.secrets.useExisting`); credencial NATS e certs seguem esse padrão (precedente: `pi-web-api-auth-token`). -- TLS via cert-manager (`global.ingress.tls.mode`); relevante só se o GRPC for exposto por ingress (interno fica cluster-DNS + NetworkPolicy). - -## 13. Checklist boas práticas (1879) -- [ ] `Basic256Sha256` (não `Basic256`) — já é padrão do conector; garantir no pool. -- [ ] `time.monotonic()` único para tempo/latência. -- [ ] Escrita com `status_code` Good + `source_timestamp`. -- [ ] Flag liga/desliga escrita de timestamp; flag liga/desliga quality (`Good_Local_Override 0xD8`). -- [ ] Opção NTP; rate-limit / pausa mínima entre escritas. -- [ ] BUG-1: coerção de bool via string JSON (`str.lower() in ('false','0','no','off','')`). -- [ ] BUG-2: try/except na conversão de tipo (retornar "não escrito" + log). -- [ ] `secure_channel_timeout = int(session_timeout*1.25)`. -- [ ] Diferenciar tipos de erro OPC nas métricas. -- [ ] (Flag/futuro) escrita via método OPC em vez de tag direta. -- [ ] ≤ ~500 linhas/arquivo, ≤ ~20/função. - -## 14. Faseamento e dependências -- **F0 — Fundações:** scaffolding; `OpcConnectionPool` (cert MinIO+Mongo→emptyDir, `Basic256Sha256`, monotonic, channel>session); base de métricas + health `app_up`. -- **F1 — SubscriptionEngine:** slot v2 + fan-out multi-nome + `DataSink`. *(C1)* -- **F2 — WriteEngine + NatsWriteConsumer + coluna de write_status (Postgres).** *(C2, C3 + migração)* -- **F3 — ActiveReadEngine (agendada) + GrpcServer (on-demand).** *(C4)* -- **F4 — Métricas completas + hardening 1879 + testes + Dockerfile + values/opc-gateway.yaml + ServiceMonitor/NetworkPolicy.** - -**Dependências externas:** banco de tags 1569 no Mongo alvo; broker NATS provisionado (§8); bucket de certs no MinIO; `dataops-library` com métodos genéricos (1590) publicada/pinada; migração da tabela de predições (C3). - -## 15. Trilhas paralelas (nível de contrato) -- **Orchestrator (1974):** montar slot v2 do banco de tags, propagar `model_id/model_name/collection`, multi-nome. **Integração é net-new** (o orchestrator hoje só usa `connectors_config` de conexão e `pipelines.read_tags` embutidas; não toca `connector_tags`). Alvos: `orchestrator/utils/orchestrator_functions.py` (`gather_read_tags`, `build_tag_config`), `orchestrator/activities/formatters.py` (`process_slots`), `orchestrator/activities/slot_manager.py`. - - **Mecânica do JOIN (DECIDIDO — chave = `tag_id`):** `connector_tags` **não tem** `model_id`/`collection`/`frequency`. Então: (a) **identidade da tag + conexão** (address, data_type, `tag_name`, `category`, url/server_uri/security) vêm de `connector_tags`+`connectors` — fonte única; (b) **associação modelo→tag + `collection` + `frequency`** vêm do lado modelo/pipeline. O pipeline passa a **referenciar `tag_id`** (não `(server+address)`, que é ambíguo sob multi-nome, já que `address` não é único no banco de tags). Cada `tag_id` = 1 nome → mapeia direto para uma entrada de `names` no slot v2. Requer migrar o schema do pipeline para guardar `tag_id`(s). -- **Laborious (1975):** publicar no NATS (C2) em vez de `node.write_value`; gravar `write_status` (C3) nas escritas próprias (PI/simples); remover sentinelas `12/13/14` de `process_confidence`/`api.py`. - -## 16. Testes -- Unit por engine/pool (padrão `tests/unit/managers`): pool (reconexão, cert, timeouts Samarco), subscription (fan-out multi-nome), write (converter bool/erro, status_code/timestamp, rate-limit, mapa de erros), active read (lag), adapters (NATS queue group, GRPC read+persist), DataSink (write_status por timestamp+model_id). -- Cada lógica não trivial deixa ≥1 teste executável. -- Funcional single-node contra o `opc_simulator`. - -## 17. Questões em aberto - -**Decididas:** -- **Broker = NATS** (Core NATS agora; §8). Definido pelo brief/Vitor. -- **Nome da branch:** renomeada para `feature/SIENTIAPDE-1972-...` ✅. - -**Resolvidas (docs/código + Vitor + Bruno):** -- **Migração = Flyway** — repo `Aignosi/sientia-dataops-database-migrations`, schema `sientia_data`. Coluna = `.sql` versionado, Job Helm hook. -- **Bucket de certs = `sientia-dataops`** (aprovado por Bruno). Ação: migrar a escrita dos certs do 1569 p/ esse bucket + policy (impl, não decisão). (§7) -- **JOIN do 1974 = chave `tag_id`** (o pipeline referencia `tag_id`; `(server+address)` é ambíguo sob multi-nome). (§15) -- **`metadata` = campo (objeto) na tag, default vazio** (decisão Vitor); tags de teste com `metadata: {}`. (§4) -- **Tabelas de predição = pandas `to_sql` (sem DDL/template)** — coluna nova via DataFrame do Laborious (tabelas novas) + backfill `ALTER TABLE` (existentes). (§6 C3) -- **Chamador do GRPC on-demand (Vitor, 2026-07-14):** ninguém hoje; talvez o frontend depois. Endpoint interno, sem auth por ora; fase adiável. (§6 C4) - -**Em aberto (precisam de humano):** -1. **Provisionamento do NATS no air-gapped:** subchart/versão/sizing/network policy (a *escolha* já é NATS). (§8) — **plataforma/Bruno** - -**Adiada por design (nós):** -- **Durabilidade da escrita:** Core NATS agora; reavaliar JetStream (TTL curto) se houver perda real em restart. - -## 18. Riscos -- **Introdução de NATS** (decidida): a diretriz antiga "não adicionar NATS sem requisito" é atendida — o Gateway é o requisito; resta cumprir o onboarding de plataforma (§8/§10). -- **Escrita OPC não idempotente:** se migrar p/ JetStream (at-least-once), dedupe por `msg_id`. -- **Relógio/OPC:** timestamps/NTP são fonte recorrente de bug (1879) — priorizar `monotonic` e flags de quality. -- **Restrições reais do servidor OPC (Samarco):** block size 512/255, chars especiais em endereços, degradação de sessão, timeouts (publishing 2000ms, write 30000ms) — o pool/WriteEngine devem tolerar e medir. -- **Pool multi-instância:** escrita/on-demand conectam sob demanda a qualquer servidor; vigiar limite de conexões/idle eviction para não estourar sessões. -- **Schema do 1569 ainda em `rc/`:** pode mudar antes do merge; revalidar C1 no merge. -- **GRPC como servidor é net-new na plataforma:** exige regra de rede e revisão (hoje só Temporal usa gRPC).