SIENTIAPDE-1972: remover egress Kafka do Gateway (Mongo-only) no spec/plano
This commit is contained in:
PedroHMCosme
@@ -4,7 +4,7 @@
|
||||
|
||||
**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 (Mongo/Kafka reads + Postgres `write_status`). Coordination (slot lease/heartbeat) is reused from the existing Redis layer.
|
||||
**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).
|
||||
|
||||
@@ -33,14 +33,14 @@
|
||||
- `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(...)` (Mongo/Kafka fan-out) + `update_write_status(...)` (Postgres UPDATE by `timestamp`+`model_id`).
|
||||
- `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` (Kafka/Mongo egress — wrapped by `DataSink`), `ingestor/app.py` (wire new engines in F4).
|
||||
**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`).
|
||||
|
||||
@@ -451,7 +451,7 @@ def test_parse_slot_multi_name():
|
||||
|
||||
---
|
||||
|
||||
### Task 9: DataSink.publish_reading (Mongo/Kafka fan-out)
|
||||
### Task 9: DataSink.publish_reading (MongoDB fan-out)
|
||||
|
||||
**Files:**
|
||||
- Create: `ingestor/io/data_sink.py`
|
||||
@@ -475,11 +475,29 @@ async def test_publish_reading_fans_out_per_name():
|
||||
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 Kafka/Mongo egress). Emit `OPC_TAG_VALUE`. Run to green.
|
||||
- [ ] **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:**
|
||||
|
||||
@@ -41,7 +41,7 @@ O trabalho atravessa 3 repositórios. Este spec detalha o **Gateway** (repo atua
|
||||
### 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 (opcional). Métricas Prometheus em `:9090/metrics`.
|
||||
- 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
|
||||
@@ -77,7 +77,7 @@ scheduler ───────────────────────
|
||||
OpcConnectionPool (por servidor) ─▶ servidores OPC
|
||||
│
|
||||
DataSink ◀────────────────────┘
|
||||
├─▶ Mongo (dados) / Kafka topics
|
||||
├─▶ Mongo (dados)
|
||||
└─▶ Postgres predictions.write_status
|
||||
```
|
||||
|
||||
@@ -89,7 +89,7 @@ Decisão-chave: **conexões OPC deixam de ser atreladas a slot** (escrita e leit
|
||||
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/Kafka, fan-out; `updated_at` como BSON Date) e `update_write_status(model_id, timestamp, status,...)` na tabela de predições (Postgres).
|
||||
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.
|
||||
|
||||
Reference in New Issue
Block a user