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

SIENTIAPDE-1972: enriquece design do OPC Gateway com padroes de plataforma (aig-docs) e corrige contrato de status de escrita

Browse Source
This commit is contained in:
PedroHMCosme
2026-07-13 16:47:59 -03:00
parent 6f6f3bf734
commit c980092d95
1 changed files with 136 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

305
docs/superpowers/specs/2026-07-13-opc-gateway-design.md
View File

@@ -1,75 +1,72 @@
# OPC Gateway — Design (SIENTIAPDE-1972)
- **Épico:** [SIENTIAPDE-1972](https://aignosi-wiki.atlassian.net/browse/SIENTIAPDE-1972) — OPC Gateway: novo gateway derivado do OPC Ingestor, com ajustes no Orchestrator e no Laborious.
- **É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:** `SIENTIAPDE-1972-opc-gateway-novo-gateway-derivado-do-opc-ingestor-com-ajustes-no-orchestrator-e-no-laborious` (no repo `sientia-dataops-opc-ingestor`).
- **Branch:** `SIENTIAPDE-1972-opc-gateway-...` (repo `sientia-dataops-opc-ingestor`). ⚠️ ver §11 (convenção de nomes de branch × CI).
- **Data:** 2026-07-13.
---
## 1. Contexto e objetivo
Evoluir o `opc-ingestor` (hoje read-only, leitura por subscription) para um **OPC Gateway** completo, cobrindo:
Evoluir o `opc-ingestor` (hoje read-only, leitura por subscription) para um **OPC Gateway** completo:
- Gestão de certificados (via MinIO + MongoDB — reaproveitando o que o épico 1569 já construiu).
- 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 de bugs do card 1879.
- 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** que o Orchestrator (1974) e o Laborious (1975) devem honrar. Cada um desses dois ganha spec próprio no seu repo.
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 (este spec / este repo):**
- Arquitetura interna do Gateway e todos os seus componentes.
- Consumo do banco de tags do 1569 e do slot v2 no Redis.
- Escrita OPC via NATS; leitura ativa agendada e on-demand via GRPC.
- Métricas e checklist 1879.
- Contratos C1–C4 (interfaces com Orchestrator/Laborious/chamadores GRPC).
**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, outros repos):**
- Implementação do 1974 no `orchestrator_temporal`.
- Implementação do 1975 no `laborious_temporal`.
- CRUD/UI do banco de tags (1569, já implementado).
**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. Estado atual (fatos do código)
## 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/<app>.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, sem NATS, sem GRPC.
- Slots vivem no **Redis** (`slot:opc_tags:{id}`), com lease/heartbeat para balanceamento entre instâncias.
- Arquitetura de managers: `opc_manager`, `ingestor_manager`, `data_manager`, `resource_manager`.
- Egress: Mongo (obrigatório) + Kafka (opcional). Métricas Prometheus em `:9090/metrics`.
- Conexão OPC hoje é criada **por slot arrematado** (só conecta nos servidores dos slots que possui).
- `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`.
- 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`, helpers de métricas Prometheus (`sientia_do.observability`). Métodos genéricos concluídos em `feature/SIENTIAPDE-1590`.
- **Não tem** NATS (usa Kafka), GRPC nem modelo de slot.
- `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 (épico 1569) — confirmado local em `sientia-projects-templates @ rc/SIENTIAPDE-1588`
Coleções no Mongo `sientia`:
### Banco de tags (1569) — `sientia-projects-templates @ rc/SIENTIAPDE-1588`
Mongo `sientia`:
| Coleção | Campos |
|---|---|
| `connectors` (opc_ua) | `connector_id`(int), `type="opc_ua"`, `name`, `url`, `server_uri`, `subscription_period_ms`, `use_security`(bool), `status`, `description` |
| `connectors` (rest_api) | `connector_id`, `type="rest_api"`, `name`, `base_url`, `auth_config{type,token}`, `headers_config`(JSON string), `request_timeout`, `verify_ssl`, `status`, `description` |
| `connector_tags` | `tag_id`(int), `connector_id`(FK int), `address`(NodeId), `tag_name`, `category`(`Read`/`Write`/`Prediction Confidence`), `data_type`(float/int/string/boolean), `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{...}` |
| `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{...}` |
Fatos relevantes:
- **Segurança já é `Basic256Sha256`** (hardcoded em `_create_opc_ua_client`, `connectors.py`), com **certificado de plataforma único** (compartilhado), baixado do MinIO (bucket `streamlit-connectors`, prefixo `platform/certificates/`) para arquivos temp e passado ao `asyncua` em runtime.
- **Multi-nome por endereço já é permitido**: unicidade em `(connector_id, tag_name)`, nunca em `address`.
- **Não existe** `metadata`, `model_id/model_name`, `collection` nem `frequency` em `connector_tags`. Modelo/coleção/frequência vivem no lado `models`/`pipelines` (schema legado). O campo `metadata` é trabalho do subtask 1600 (pendente).
- **Bug local conhecido:** `projects-templates/app/src/operations/db_infra.py:16` insere `...\repos\sientia-dataops-library` no `sys.path` (caminho sem o subdir `software-engineer`), inexistente nesta máquina → o app não importa a lib local. Questão em aberto (ver §12).
- **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** `metadata`, `model_id/model_name`, `collection`, `frequency` em `connector_tags`. `metadata` é trabalho do subtask 1600 (pendente).
- 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`. É esse acoplamento que o 1975 remove.
- Payload de escrita já contém: `server_id`, `addr`, `value`, `data_type`, `tag_type`(prediction|confidence), `model_id`, `model_name`, `timestamp`.
- 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 `predictions_schema.predictions` (`model_id, prediction, prediction_confidence, response_time, prediction_status, timestamp, created_at, comments`).
- Payload de escrita já contém: `server_id`, `addr`, `value`, `data_type`, `tag_type`, `model_id`, `model_name`, `timestamp`.
## 4. Arquitetura do Gateway (Abordagem B — engines por capacidade)
## 5. Arquitetura do Gateway (Abordagem B — engines por capacidade)
```
Orchestrator ─(Redis slot v2)─▶ SubscriptionEngine ─┐
@@ -81,66 +78,31 @@ scheduler ───────────────────────
│
DataSink ◀────────────────────┘
├─▶ Mongo (dados) / Kafka topics
└─▶ Mongo `models`.write_status
└─▶ Postgres predictions.write_status
```
Decisão-chave: **conexões OPC deixam de ser atreladas a slot**. Escrita (NATS) e leitura on-demand (GRPC) podem mirar qualquer servidor, então a conexão vira um **pool por servidor** compartilhado por todas as engines.
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/Kafka, fan-out; `updated_at` como BSON Date) e `update_write_status(model_id, timestamp, status,...)` na tabela de predições (Postgres).
7. **`app`** — instancia pool+engines+adapters, sobe Prometheus/health, trata sinais e shutdown.
1. **`OpcConnectionPool`** *(peça nova, núcleo compartilhado)*
- Conexões `asyncua` por `connector_id`/`server`, reusadas por todas as engines; criação sob demanda + *idle eviction*.
- Segurança `Basic256Sha256`; `secure_channel_timeout = int(session_timeout * 1.25)` (1879 §3).
- Certificados: baixa o **platform certificate** do MinIO (metadados em `platform_certificates`), espelhando o padrão `_create_opc_ua_client` do 1569.
- Uma única referência `time.monotonic()` para latências/timeouts (1879 §2). Opção de sincronizar relógio via NTP.
- Reconexão + métricas de conexão/desconexão e tipos de erro OPC.
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.
2. **`SubscriptionEngine`** *(evolui `ingestor_manager` + `opc_manager`)*
- Arremata slots no Redis (reusa lease/heartbeat), cria subscription por (slot, servidor) via pool.
- No `datachange`, faz **fan-out multi-nome**: para cada `{collection, tag_name, model_name, model_id}` do endereço, emite registro ao `DataSink`.
3. **`ActiveReadEngine`** *(nova)*
- **Agendada:** lê tags marcadas para leitura ativa em frequência múltipla do ciclo de operação; persiste via `DataSink`.
- **On-demand:** atende o `GrpcServer` — lê agora, **persiste E devolve** valor+status+timestamp.
- Mede *lag* de leitura (ts recebido − ts da máquina) (1879).
4. **`WriteEngine`** *(nova)*
- Escrita OPC via pool com `DataValue` + `Variant` + **`status_code` Good** + **`source_timestamp`** (1879 §1).
- **Flags**: liga/desliga escrita de timestamp e de quality (quality `Good_Local_Override 0xD8` se detectar problema de relógio).
- **Converter seguro**: corrige BUG-1 (coerção de bool via string JSON) e BUG-2 (try/except em conversão de tipo), retornando "não escrito" + log em erro.
- **Rate-limit** / pausa mínima entre escritas.
- Diferencia tipos de erro OPC (métrica por erro).
- Após escrita, atualiza `models.write_status` via `DataSink`.
- (Futuro/flag) opção de escrever via **método** OPC em vez de tag direta (1879).
5. **Adapters de entrada**
- **`NatsWriteConsumer`**: subscribe no subject de escrita com **queue group** `gateway-writers` (exatamente 1 instância processa cada comando). **Core NATS** (fire-and-forget); migração para JetStream fica para depois sem mudar o Laborious.
- **`GrpcServer`**: RPC de leitura on-demand → `ActiveReadEngine`.
6. **`DataSink`** *(evolui `data_manager`)*
- Persiste leituras no Mongo/Kafka com fan-out multi-nome.
- `update_model_write_status(model_id, status, ...)` na coleção `models`.
7. **`app`** — instancia pool + engines + adapters, sobe Prometheus, trata sinais e shutdown gracioso.
**Coordenação:** lease/heartbeat de slots no Redis seguem governando `SubscriptionEngine` e a leitura **agendada**. Escrita e leitura **on-demand** não dependem de lease — qualquer instância atende, conectando sob demanda pelo pool.
**Meta de qualidade (1879 §Geral):** ≤ ~500 linhas por arquivo, ≤ ~20 linhas por função.
## 5. Contratos
## 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, e carrega config de leitura ativa:
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,
"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",
@@ -154,110 +116,115 @@ Orchestrator escreve, Gateway lê. Substitui `topics:[str]` por lista multi-nome
}
}
```
O Orchestrator monta isso fazendo o **JOIN** entre:
- **Identidade da tag** (banco de tags 1569): `connector_tags.address/tag_name/data_type/category` + `connectors.url/server_uri/use_security/subscription_period_ms`.
- **Config de modelo**: `model_id`, `model_name`, `collection`, `frequency` (lado `models`/pipeline).
Multi-nome = múltiplas entradas em `names` para o mesmo `address`.
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.
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",
"timestamp": "2026-07-13T12:00:00Z",
"write_timestamp": true,
"write_quality": false
}
{ "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",
"timestamp":"2026-07-13T12:00:00Z","write_timestamp":true,"write_quality":false }
```
### C3 — Campo de status de escrita (Mongo `models`)
Novo campo, substitui as sentinelas `12/13/14`. Gateway grava após escrita OPC; Laborious grava nas escritas dele (PI/simples).
```json
"write_status": {
"status": "ok|error|session_bad",
"opc_status_code": "Good",
"message": "...",
"source": "opc|pi",
"updated_at": "2026-07-13T12:00:01Z"
}
```
### C3 — Coluna de status de escrita (Postgres, tabela de predições) ⭐ autoritativo em topics.md
Nova coluna em `predictions_schema.predictions` (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`. Requer **migração backward-compatible** via `sientia-dataops-database-migrations` (ver §10; confirmar Flyway vs Alembic).
### C4 — GRPC leitura on-demand
```
service OpcGateway {
rpc ReadTag(ReadTagRequest) returns (ReadTagResponse);
}
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 o valor.
Lê agora, persiste no pipeline e devolve.
## 6. Gestão de certificados
Reaproveita o que o 1569 já construiu: coleção `platform_certificates` (metadados) + MinIO (bucket `streamlit-connectors`, prefixo `platform/certificates/`). O `OpcConnectionPool` baixa `client_cert`/`private_key` para arquivos temporários e passa ao `asyncua` com `Basic256Sha256`, espelhando `_create_opc_ua_client`. Certificado de plataforma é **único/compartilhado** (não é por conector). Server cert opcional por conexão.
## 7. Gestão de certificados
Reaproveita o 1569: `platform_certificates` (metadados) + MinIO (`streamlit-connectors`, `platform/certificates/`). 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. **Questão de fronteira:** o bucket `streamlit-connectors` pertence ao app Streamlit; padrão de plataforma é bucket/secret scoped por serviço (§10/§12) — decidir se o Gateway lê desse bucket ou se ganha bucket `opc-gateway` próprio (ver §17).
## 7. Métricas (1973 + 1879)
`app_up`; contadores de sucesso/erro de **escrita** e de **leitura ativa**; histogramas de **latência** de escrita e de leitura ativa; tags na subscription; slots possuídos; contagem de conexões/desconexões; **status de escrita por modelo**; **valor de cada tag por tag/modelo**; ciclos sem chegar dado em subscription; **tipos de erro OPC diferenciados** (contagem por erro); **lag de leitura** (ts recebido − ts da máquina). Base já existe em `ingestor/metrics.py` e em `sientia_do.observability`.
## 8. NATS como nova dependência de plataforma ⚠️
Os cards e o brief exigem **NATS**. Porém os docs de plataforma dizem explicitamente: *"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`), e o **Kafka** já é o streaming adotado (o próprio opc-ingestor usa Kafka). Um revisor vai questionar **NATS vs Kafka**.
## 8. Checklist boas práticas (1879)
- [ ] `Basic256Sha256` (não `Basic256`) — já é o padrão do conector; garantir no pool.
- [ ] `time.monotonic()` único para todo cálculo de tempo/latência.
- **Justificativa a registrar:** o padrão request/response leve, subjects e queue groups do NATS casam com comando de escrita ponto-a-ponto de baixa latência; é o que o brief define. (Alternativa Kafka: possível, mas o card fixa NATS.)
- **Ônus de onboarding (o Gateway é o "novo componente" que justifica o NATS):** 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`, secrets de credencial via `sientia-secrets-chart`, e decisão de HA/persistência (Core agora; JetStream depois exige storage).
## 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):** coluna nova via `sientia-dataops-database-migrations`, automatizada, backward-compatible em 3 fases (initContainer ou Helm hook). Confirmar stack: docs helm citam Flyway; onboarding cita Alembic+SQLAlchemy.
- **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; existe bucket `opc-ingestor`. 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**. ⚠️ **A branch atual não tem prefixo `feature/`** → renomear para `feature/SIENTIAPDE-1972-...` antes de abrir PR (ver §17).
- **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 escrita de quality (`Good_Local_Override 0xD8` em problema de relógio).
- [ ] Opção de sincronizar relógio via NTP.
- [ ] Rate-limit / pausa mínima entre escritas.
- [ ] BUG-1: coerção de bool a partir de string JSON (`str.lower() in ('false','0','no','off','')`).
- [ ] BUG-2: try/except na conversão de tipo do converter (retornar "não escrito" + log).
- [ ] `secure_channel_timeout = int(session_timeout * 1.25)`.
- [ ] 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 linhas/função.
- [ ] ≤ ~500 linhas/arquivo, ≤ ~20/função.
## 9. Faseamento e dependências
## 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.**
- **F0 — Fundações:** scaffolding na branch; `OpcConnectionPool` (cert MinIO+Mongo, `Basic256Sha256`, monotonic, channel>session); base de métricas.
- **F1 — SubscriptionEngine:** consumo do slot v2 + fan-out multi-nome + `DataSink`. *(depende de C1)*
- **F2 — WriteEngine + NatsWriteConsumer + `models.write_status`.** *(depende de C2, C3)*
- **F3 — ActiveReadEngine (agendada) + GrpcServer (on-demand).** *(depende de C4)*
- **F4 — Métricas completas + hardening 1879 + testes + Dockerfile.**
**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).
**Dependências externas:**
- Banco de tags do 1569 disponível no Mongo alvo (schema `connectors`/`connector_tags`/`platform_certificates`).
- Broker NATS provisionado.
- Bucket de certs no MinIO (`streamlit-connectors`).
- Versão da `dataops-library` com os métodos genéricos (`feature/SIENTIAPDE-1590`) publicada e pinada no gateway.
## 15. Trilhas paralelas (nível de contrato)
- **Orchestrator (1974):** montar slot v2 do banco de tags (JOIN `connector_tags` × modelo), propagar `model_id/model_name/collection`, multi-nome. Alvos: `orchestrator/utils/orchestrator_functions.py` (`gather_read_tags`, `build_tag_config`), `orchestrator/activities/formatters.py` (`process_slots`), `orchestrator/activities/slot_manager.py`.
- **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`.
## 10. Trilhas paralelas (nível de contrato)
- **Orchestrator (1974):** montar slot v2 a partir do banco de tags (JOIN `connector_tags` × modelo), propagar `model_id/model_name/collection`, suportar multi-nome por endereço. Arquivos-alvo hoje: `orchestrator/utils/orchestrator_functions.py` (`gather_read_tags`, `build_tag_config`), `orchestrator/activities/formatters.py` (`process_slots`), `orchestrator/activities/slot_manager.py`.
- **Laborious (1975):** publicar comando de escrita no NATS (C2) em vez de `node.write_value`; gravar `models.write_status` (C3) nas escritas próprias (PI/simples); remover as sentinelas `12/13/14` de `process_confidence`/`api.py`, desacoplando status de escrita da confidence.
## 11. Testes
- Unit por engine/pool (reusar padrão de `tests/unit/managers`): pool (reconexão, cert), subscription (fan-out multi-nome), write (converter bool/erro, status_code/timestamp, rate-limit), active read (lag), adapters (NATS queue group, GRPC read+persist).
- Cada lógica não trivial deixa ao menos um teste executável (converter, fan-out, error mapping).
## 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`.
## 12. Questões em aberto
1. **Campo `metadata` em `connector_tags`** (subtask 1600, pendente): formalizar como campo nested. Nas tags de teste do gateway, criar com `metadata: {}` (pedido do Vitor, para documentação).
2. **Durabilidade da escrita:** começar com Core NATS puro; reavaliar JetStream (TTL curto p/ descartar comando obsoleto) se perda em restart se mostrar problema.
3. **`db_infra.py:16` (projects-templates):** `sys.path` aponta para caminho inexistente (`...\repos\sientia-dataops-library`, falta `software-engineer`). Corrigir para o app 1569 importar a lib local. Fora do escopo do gateway — decidir se entra aqui.
4. **Origem de `collection`/`frequency`/`active_read`** no JOIN do 1974: confirmar de onde vêm (pipeline vs config de modelo) ao detalhar o spec do Orchestrator.
5. **Chamador do GRPC on-demand:** confirmar quem chama (Laborious? Orchestrator? UI/API) para dimensionar auth/limites do `GrpcServer`.
## 17. Questões em aberto
1. **Nome da branch:** renomear para `feature/SIENTIAPDE-1972-...` (o padrão atual falha o CI de versionamento) — decidir antes do 1º PR.
2. **Bucket de certs:** Gateway lê do bucket `streamlit-connectors` (do app 1569) ou ganha bucket `opc-gateway` scoped? (§7/§10)
3. **Stack de migração de banco:** Flyway (docs helm) vs Alembic+SQLAlchemy (onboarding) para a coluna de write_status — confirmar.
4. **NATS vs Kafka:** registrar justificativa formal (§8); confirmar com plataforma o provisionamento do broker no air-gapped.
5. **Campo `metadata` em `connector_tags`** (subtask 1600): formalizar; tags de teste criadas com `metadata: {}` (pedido do Vitor).
6. **Durabilidade da escrita:** Core NATS agora; reavaliar JetStream (TTL curto p/ descartar comando obsoleto) se perda em restart for problema.
7. **Origem de `collection`/`frequency`/`active_read`** no JOIN do 1974: confirmar (pipeline vs config de modelo) ao detalhar o spec do Orchestrator.
8. **Chamador do GRPC on-demand:** confirmar quem chama (Laborious/Orchestrator/UI-API) p/ dimensionar auth/limites.
## 13. Riscos
- **At-least-once futuro (JetStream):** escrita OPC não é idempotente por natureza; se migrar, tratar dedupe por `msg_id`.
- **Relógio/OPC:** timestamps e NTP são fonte recorrente de bug (1879) — priorizar `monotonic` e flags de quality.
- **Pool multi-instância:** escrita/on-demand conectam sob demanda a qualquer servidor; vigiar limite de conexões e *idle eviction* para não estourar sessões no servidor OPC.
- **Dependência de schema do 1569 ainda em `rc/`:** o schema do banco de tags pode mudar antes do merge; contrato C1 deve ser revalidado no merge.
## 18. Riscos
- **Introdução de NATS** contraria a diretriz "não adicionar NATS sem requisito" — mitigado registrando o Gateway como o requisito e cumprindo 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).