gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
3dcc59f65eb41adbeeb0665d7ebb629420904768
sientia-dataops-opc-ingestor/docs/superpowers/plans/2026-07-13-opc-gateway.md

44 KiB
Raw Blame History

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): (fill at F2 close-out) F3 — Active reads + gRPC (C4): (fill at F3 close-out) 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
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.

# 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:

# --- 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):

# 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:

"""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

# 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:

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.

# 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:

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.

# 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.

# 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)

  • Full pytest tests green + ruff clean (162 passed).
  • F0 commits recorded in the 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.

# 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.

# 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.

# 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)

  • Full pytest tests green (168) + ruff clean.
  • F1 commit SHAs recorded in the 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.

# 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=....

# 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).

# 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)

  • Full pytest tests green + ruff clean.
  • Record F2 commit SHAs (base + range) in the Commit Ledger.
  • Do NOT open the PR yet.

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.

# 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):

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.

# 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)

  • Full pytest tests green + ruff clean.
  • Record F3 commit SHAs (base + range) in the Commit Ledger.
  • Do NOT open the PR yet.

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)

  • Full pytest tests green + ruff clean; all F0–F4 phases complete, tested, working.
  • Record F4 commit SHAs (base + range) in the 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.
Reference in New Issue View Git Blame Copy Permalink