gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

SIENTIAPDE-1974: feat(read): keep the legacy OpcManager as plan B (READ_ENGINE)

Browse Source 
The rc3 OpcManager is restored unchanged (with its tests and
scripts/generate-and-push-opc-cert.sh) for validation and as a fallback until
the end of SIENTIAPDE-1974; it is removed only once it is unused and the test
suite passes the same without it.

- READ_ENGINE=v2 (default): ServerReader/SubscriptionEngine over the shared
  pool, certificate from MinIO. READ_ENGINE=legacy: the rc3 OpcManager with
  its own connection, slot v1 only (a v2 slot is rejected and notified) and
  the certificate files of the Gitea clone. An invalid value falls back to v2
  with a warning. Writes/gRPC always use the pool.
- pool: OPC clients carry the pod id as application/session name and product
  URI (as OpcManager did), so each pod's session is recognizable in the
  server console; application_uri stays the server_uri (certificate SAN).
- README/CLAUDE.md document the plan B and how to remove it.

Tests: 275 unit tests pass (restored OpcManager tests unchanged); ruff clean.
Local e2e: both engines read the same slot v1 from an asyncua server and write
identical documents to the same collections.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Eduardo Rios
2026-10-06 14:12:58 -03:00
parent 3d8ee1c2ed
commit 0ac817a479
9 changed files with 1593 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
CLAUDE.md
View File

@@ -41,6 +41,7 @@ app.py (asyncio loop, signals, Prometheus server)
-> IngestorManager (managers/ingestor_manager.py) — central coordinator
-> ResourceManager (managers/resource_manager.py) — Redis slot leases, heartbeats, active-ingestor registry
-> ServerReader (managers/server_reader.py), one per OPC server — SubscriptionEngine over the shared OpcConnectionPool
(READ_ENGINE=legacy: OpcManager (managers/opc_manager.py), the rc3 reader kept as plan B until the end of SIENTIAPDE-1974)
-> OpcConnectionPool (connection/pool.py) + CertProvider (connection/certs.py) — pooled connections, cert from MinIO
-> DataManager (managers/data_manager.py) — MongoDB persistence + optional Kafka publish
```
@@ -57,7 +58,7 @@ app.py (asyncio loop, signals, Prometheus server)
Any change to slot/lease counts should be traced through `manage_leases`/`manage_no_slots` — that's where the rebalancing math lives, not in `ResourceManager` itself (which just does Redis primitives).
**ServerReader** (SIENTIAPDE-1974, replaced the legacy `OpcManager`) reads one OPC server through a `SubscriptionEngine`: the connection comes from the shared `OpcConnectionPool` (also used by NATS writes and gRPC), which fetches the platform certificate from MinIO via `CertProvider` and uses `SecurityPolicyBasic256Sha256` — no certificate files in the repo/image. Slots are parsed by `io/slots.py` (`parse_slot`), which accepts slot v2 (`names[]`) and converts the legacy v1 (`topics`). `datachange_notification` fans each value out to every `names` entry via `DataSink`. Subscribed connections are pinned in the pool so the idle sweep never drops them. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles` / `check_opc_listenning`); a lost server is shut down (pool connection evicted) and re-initialized on the next loop.
**ServerReader** (SIENTIAPDE-1974, default `READ_ENGINE=v2`; the legacy `OpcManager` stays as plan B with `READ_ENGINE=legacy` — slot v1 only, certificate files from the Gitea clone — until it is unused and the tests pass the same without it) reads one OPC server through a `SubscriptionEngine`: the connection comes from the shared `OpcConnectionPool` (also used by NATS writes and gRPC), which fetches the platform certificate from MinIO via `CertProvider` and uses `SecurityPolicyBasic256Sha256` — no certificate files in the repo/image. Slots are parsed by `io/slots.py` (`parse_slot`), which accepts slot v2 (`names[]`) and converts the legacy v1 (`topics`). `datachange_notification` fans each value out to every `names` entry via `DataSink`. Subscribed connections are pinned in the pool so the idle sweep never drops them. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles` / `check_opc_listenning`); a lost server is shut down (pool connection evicted) and re-initialized on the next loop.
**Metrics** (`ingestor/metrics.py`): all Prometheus metrics defined here, always labeled by `pod_id`. Emitted via `SientiaMonitoring.emit_metric(...)` (from `sientia_do`), not the prometheus client directly — new metrics should follow that pattern for consistency with logging/notifications.