From ca9f4ce188f4cce5454d202b0734b04ccd80c3b8 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 29 Oct 2026 16:38:03 -0300 Subject: [PATCH] SIENTIAPDE-2202 SIENTIAPDE-2202-escrita-opc-timestamp-qualidade: Enable granular control over OPC UA SourceTimestamp and StatusCode in write commands. This implements write_timestamp and write_quality flags (command > env > default true) for DataValue construction. Upgraded asyncua to 2.0+ to fix StatusCode field naming. Enhanced README.md with new gateway environment variables and detailed write command behavior. Added comprehensive unit and E2E tests for this feature, and removed obsolete planning documents. --- .gitignore | 3 + README.md | 40 +- ...IAPDE-1972-banco-de-tags-e-certificados.md | 354 ------------------ ...IENTIAPDE-1974-certificado-por-conector.md | 241 ------------ requirements-local.txt | 2 +- requirements.txt | 4 +- tests/functional/test_write_e2e.py | 173 +++++++++ tests/unit/engines/test_write.py | 30 ++ tests/unit/test_config.py | 2 + 9 files changed, 244 insertions(+), 605 deletions(-) delete mode 100644 docs/planning/SIENTIAPDE-1972-banco-de-tags-e-certificados.md delete mode 100644 docs/planning/SIENTIAPDE-1974-certificado-por-conector.md create mode 100644 tests/functional/test_write_e2e.py diff --git a/.gitignore b/.gitignore index 5f3a8bd..f44c5a2 100644 --- a/.gitignore +++ b/.gitignore @@ -188,3 +188,6 @@ docs/superpowers/ docs/e2e-test-guide.md docs/deploy-notes.md docs/HANDOFF.md + +openspec/ +.claude/ diff --git a/README.md b/README.md index 8021a09..9cadccb 100644 --- a/README.md +++ b/README.md @@ -67,9 +67,8 @@ Gateway for on-demand and command-driven interactions with OPC UA servers: (`opc_certificates`), named in the slot and downloaded from MinIO (`ingestor/connection/certs.py`) rather than shipped in the image. -See `docs/deploy-notes.md` for the deployment-facing details (env vars, -health probe, NetworkPolicy, cert cache) and the `# OPC Gateway` block in -`.env.example` for the full list of new environment variables. +See [Gateway environment variables](#gateway-environment-variables) and the +`# OPC Gateway` blocks in `.env.example` for the new environment variables. ### Write commands (C2) and `write_status` (C3) @@ -78,10 +77,16 @@ The Laborious publishes one command per tag (SIENTIAPDE-1975): ```json {"server_id": "1", "addr": "ns=2;s=Pred", "value": 1.25, "data_type": "float", "model_id": "2", "model_name": "sin-approx", "schema": "sientia_data", - "table_name": "predictions", "timestamp": "2026-10-13T12:00:00-03:00", - "write_timestamp": false, "write_quality": false} + "table_name": "predictions", "timestamp": "2026-10-13T12:00:00-03:00"} ``` +- **Written `DataValue`** (SIENTIAPDE-2202): the value, plus the prediction + `timestamp` as `SourceTimestamp` and a Good `StatusCode`. Timestamp and quality can be + turned off separately; the value is always written. Each flag is taken from the + command (`write_timestamp` / `write_quality`) when present, else from the env + (`WRITE_TIMESTAMP` / `WRITE_QUALITY`), else `true`. The envs apply to every server of + the gateway (per-server config is a future SIENTIAPDE-2276 card). + - **Server connection**: the command carries only `server_id`. The connection is resolved by `ServerConfigResolver` (`ingestor/connection/servers.py`): the live pooled connection opened by the reads, else the `connectors` document (`type: opc_ua`, @@ -342,9 +347,28 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring: | `MONGODB_PASSWORD` | MongoDB password | `sientia` | No | | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No | -See the `# OPC Gateway` block in `.env.example` (`NATS_URL`, `GRPC_ENABLED`, -`CERT_CACHE_DIR`, `WRITE_TIMESTAMP`, ...) and `docs/deploy-notes.md` for the -gateway-specific environment variables. +#### Gateway environment variables + +Read by `GatewayConfig` (`ingestor/config.py`). Booleans accept `true`/`1`/`yes` +(case-insensitive); anything else is `false`. + +| Variable | Description | Default | +|----------|-------------|---------| +| `NATS_URL` | NATS server for write commands (C2) | `nats://localhost:4222` | +| `NATS_WRITE_SUBJECT` | Subject of the write commands | `opc.write` | +| `NATS_QUEUE` | Queue group shared by the gateway pods | `gateway-writers` | +| `GRPC_ENABLED` | Starts the on-demand `ReadTag` gRPC server | `false` | +| `GRPC_PORT` | gRPC server port | `50051` | +| `CERT_CACHE_DIR` | Local cache of the client certificates downloaded from MinIO | `/tmp/opc-certs` | +| `WRITE_TIMESTAMP` | Writes the prediction `timestamp` as `SourceTimestamp` when the command has no `write_timestamp` | `true` | +| `WRITE_QUALITY` | Writes a Good `StatusCode` when the command has no `write_quality` | `true` | +| `WRITE_MIN_PAUSE_S` | Minimum pause between two OPC writes (seconds) | `0.0` | +| `IDLE_CONN_MAX_S` | Idle time before an unpinned pool connection is closed (seconds) | `300.0` | + +`WRITE_TIMESTAMP` / `WRITE_QUALITY` are independent and apply to every OPC server of +the gateway; the value is always written. See [Write commands](#write-commands-c2-and-write_status-c3) +for the precedence (command → env → `true`). MinIO (`MINIO_*`) and Postgres +(`POSTGRES_*`) settings are listed in `.env.example`. ### OPC Server Configuration diff --git a/docs/planning/SIENTIAPDE-1972-banco-de-tags-e-certificados.md b/docs/planning/SIENTIAPDE-1972-banco-de-tags-e-certificados.md deleted file mode 100644 index 821d0fd..0000000 --- a/docs/planning/SIENTIAPDE-1972-banco-de-tags-e-certificados.md +++ /dev/null @@ -1,354 +0,0 @@ -# SIENTIAPDE-1972 — Estado atual, pendências e plano: banco de tags + servidores OPC ↔ certificados (MinIO) - -> Levantamento em 28/10/2026, a partir do Jira (1972 e subtasks, 1568/1569/159x), do código do -> `sientia-dataops-opc-ingestor` (`release/SIENTIAPDE-1973` @ `d68e842`, 2.0.0-rc4), do -> `sientia-dataops-orchestrator_temporal` (`release/SIENTIAPDE-1974`), do Streamlit -> (`sientia-projects-templates` `rc/SIENTIAPDE-1588`), do `sientia-dataops-database-migrations` -> e do design `aig-docs/07 - Projects & R&D/sientia/opc-gateway/design.md`. - ---- - -## 1. Resumo executivo - -- **O gateway funciona no dev** (leitura v2, escrita via NATS, `write_status`) e as subtasks 1973, 1974, - 1975 e 1985 estão **Concluídas**. Mas **nada disso está na `main`/produção**: o PR #48 - (`release/SIENTIAPDE-1973` → `main`) está aberto, e a `release/SIENTIAPDE-1974` do orchestrator - nem PR para a `main` tem (5 commits à frente, 30 atrás). -- **Risco imediato**: a `main` do Laborious (1.6.0) já escreve OPC **só via NATS**. Se ela for para - um ambiente com o ingestor 1.4.x, toda escrita OPC fica em `pending`. -- **O banco de tags não existe em nenhum ambiente.** `connectors`/`connector_tags` só existiram no dev - durante o teste A2 da 1974 e foram apagadas. O orchestrator e o gateway já sabem ler o banco de tags - (por `tag_id`), mas sem dado, tudo roda pelo caminho legado (`opc-servers` + `server_id/tag_address`). -- **Certificado hoje é um só para a plataforma** (`platform_certificates`, `find_one({})`). Não existe - correlação servidor OPC ↔ certificado no Mongo. O `server_cert_path` (cert do servidor) é ignorado - com aviso. Bucket em uso: `streamlit-connectors` (o design pede `sientia-dataops`). -- **O CRUD (Streamlit 1568/1569) está On Hold** desde 04/2026 para migrar ao Angular; não há card - Angular aberto para isso (só a idea `SIENTIAPDI-84`, em parking lot). - -O que falta para fechar o 1972 é pouco (merges + 2202 + 2270). O que vocês tinham pensado — **banco de -tags + servidores OPC estruturados com certificados no MinIO** — é praticamente uma **fase 2 inteira** -que não tem card hoje. Este documento propõe o modelo e quebra em cards. - ---- - -## 2. Estado das subtasks do 1972 - -| Card | Status | O que entregou / o que falta | -|---|---|---| -| 1973 Gateway | ✅ Concluído (02/10) | Escrita NATS, gRPC ReadTag, pool, CertProvider. PRs empilhados #31–#33 fechados, código na `release/SIENTIAPDE-1973` | -| 1974 Orchestrator + leitura v2 | ✅ Concluído (09/10) | Gateway #41 (leitura v2, cert do MinIO), #42 (KeyError) mergeados na release. Orchestrator #45 (slot v2, `names[]`, banco de tags opcional por `tag_id`) mergeado na **`release/SIENTIAPDE-1974`**, não na `main`. G6/G7 (A6/A7) não serão feitos: não existe leitura ativa programada; leitura ativa é só on-demand via gRPC, sem persistir (G-AR) | -| 1975 Laborious NATS + `write_status` | ✅ Concluído (23/10) | migrations #37 (V16), gateway #43/#47, laborious #5 (main 1.6.0). Validado até o ProSYS (M0–M4) | -| 1985 NATS no cluster | ✅ Concluído (30/09) | NATS 2.14.6 via helm. `#108` (release/1.0.0) e `#126` (development) — conferir se ambos já foram mergeados | -| **2202** timestamp/qualidade na escrita | 🔄 Em andamento (Eduardo) | Decidir onde fica a config (sugestão: por servidor). **Encaixa direto no modelo de `connectors` deste plano** | -| **2270** mlops-bff `write_status` no Model Sustain | 🔄 Em andamento (Carla) | Depende de decisão do PO (falha de escrita deixa o modelo CRITICAL ou não) e da V16 aplicada antes do deploy do BFF | - ---- - -## 3. Pendências (consolidado dos comentários dos cards) - -### 3.1 Promoção para produção (bloqueia tudo o mais) - -| # | Pendência | Onde | Situação | -|---|---|---|---| -| P1 | PR #48 `release/SIENTIAPDE-1973` → `main` (gateway 2.x) | opc-ingestor | Aberto. Combinar número de versão (bump pode não gerar 2.0.0) | -| P2 | PR `release/SIENTIAPDE-1974` → `main` | orchestrator | **Não existe.** Release está 30 commits atrás da `main` (0.8.2), com conflito em 4 arquivos (`orchestrator_functions.py`, teste dele, `README.md`, `pyproject.toml`). **Desde 19/10 o dev roda o orchestrator 0.8.1 (`main`, sem slot v2)**: o deploy do PR #49 fez o bump de `GITHUB_BRANCH` 0.7.0-rc1 → 0.8.1 no chart (`development`, commit `07ffc99`) | -| P3 | Card do helm (H1–H4 do 1975): migrations com V16, Grafana sem 12/13 e com `write_status`, `docs/nats-broker.md` §11. NATS + envs já estão em todo profile com `opc-ingestor` ligado (desligado só em galvani, quartz-homologacao, gcp-hom, que não têm `opc-ingestor` nem `laborious-worker`) | sientia-helm-chart | **Card não criado** | -| P4 | Ordem de deploy por ambiente: migrations V16 → gateway 2.x → laborious 1.6.0 | todos | Documentar no runbook de release | -| P5 | Correção da `main` do gateway (SQLAlchemy 2.1 + senha do Mongo no log), branch `fix/sqlalchemy-21-mongo-uri-log` | opc-ingestor | Fica obsoleta se o #48 entrar; senão, abrir PR de hotfix | -| P6 | Servidores OPC de produção aceitarem `Basic256Sha256` e confiarem no certificado da plataforma (SAN URI = `server_uri`) | infra/cliente | Pré-requisito do deploy da leitura v2 em cada cliente | -| P7 | Certificado da plataforma cadastrado (`platform_certificates` + MinIO) em cada ambiente | por ambiente | Hoje só no dev, cadastrado à mão | - -### 3.2 Gateway (código) - -| # | Pendência | Referência | -|---|---|---| -| G-AR | **Leitura ativa = unitária e atômica: lê e devolve o valor, nada mais** (decisão 28/10). Sem leitura programada e sem persistência no gRPC. Hoje já se comporta assim (`slot_lookup=None` → `persisted=false`), falta remover o código morto: `scheduled_tick`/`_tick_tag`/`_should_read`/`_ticks` e o parâmetro `names` + `sink.publish_reading` de `read_now`/`_build_result` (`engines/active_read.py`); `slot_lookup` de `ReadTagServicer`/`GrpcServer` (`adapters/grpc_server.py`); `ponytail` de `gateway_runtime.py:10,93`; campo `persisted` do `ReadTagResponse` (marcar `reserved 5` no proto); `active_read` do slot v2 no orchestrator e no `parse_slot`; testes correspondentes | decisão 28/10 | -| G-DOC | Atualizar design §6 C4 e §14 ("Reads, persists to the pipeline, and returns the value") e os requisitos G3/G4/A6/A7 do 1974 para refletir a decisão | aig-docs `opc-gateway/design.md` | -| G-PB | Remover plano B `READ_ENGINE=legacy` (`OpcManager`, `scripts/generate-and-push-opc-cert.sh`) | critério do 1974; pode sair no #48 ou logo depois | -| G-RC | Reconectar ao receber `status_change_notification` (KEPServer derruba sessão a cada 2 h → 2,5–3 min sem dado) | achado do 1974 | -| G-SC | Suporte a `server_cert_path` (validar cert do servidor) — hoje só aviso | `ingestor/managers/ingestor_manager.py:708` — resolvido pela Fase 2 abaixo | - -### 3.3 Fora do gateway - -| # | Pendência | Situação | -|---|---|---| -| C1 | Consumidores dos códigos 12/13/14: `sientia-dataops-report`, `aignosi-scripts` (`sientia_status_report.sh`) | **Sem card** (só o BFF tem, 2270) | -| C2 | Laborious: remover `write_opc_data` antigo quando não houver execução anterior ao patch | sem card | -| C3 | Limpeza no dev: tabela `sientia_data.e2e_predictions`, pipeline `laborious-test-1975-write`, branches `test-SIENTIAPDE-197x-*` no Gitea | pendente de ok | -| C4 | Melhoria de baixa prioridade do scouter (timeout Redis, retries, catch-up) | sugerido, sem card | -| C5 | Usuário `aignosi` da VM ProSYS + senha em texto no README do `sientia-iac` | sem card | - -### 3.4 Banco de tags e certificados (bloqueios B1–B6 do 1974 — **nenhum resolvido**) - -| # | Bloqueio | Situação hoje | -|---|---|---| -| B1 | `connectors`/`connector_tags` não existem em nenhum ambiente | **escopo do orchestrator** (decisão 28/10) | -| B2 | Pipelines não gravam `tag_id` em `read_tags`/`write_tags` | **escopo do orchestrator** | -| B3 | ~~Onde fica `active_read`~~ | não se aplica: leitura ativa é só on-demand, sem persistir | -| B4 | Certificado por ambiente + bucket (`sientia-dataops` do design × `streamlit-connectors` do chart/D7) e correlação servidor ↔ certificado | **escopo do orchestrator**; bucket indefinido | -| B5 | Tela de certificados: 1599 entregou no **Streamlit** (rc, não em produção) | não existe no Angular | -| B6 | ~~gRPC persist / timestamp nulo~~ | não se aplica: gRPC não persiste; `source_timestamp` vazio só volta vazio na resposta | -| — | Renomear tag em `connector_tags` não reimplanta o scouter (só `pipelines.updated_at`) | limitação conhecida | -| — | Restos do 1569: 1598 (`tag_id` sequencial sem corrida), 1600 (`metadata`), 1602 (bucket por servidor), 1603 (Server URI), 1604, 1605 (CSV) | On Hold, "passíveis de cancelamento" | - ---- - -## 4. Diagnóstico do modelo atual (Mongo + MinIO) - -Hoje coexistem **dois registros de servidor OPC** e **um certificado global**: - -``` -opc-servers { id, name, url, uri, cert_path, private_key_path, server_cert_path, ... } ← legado, usado em todos os ambientes -connectors (opc_ua) { connector_id, type, name, url, server_uri, use_security, subscription_period_ms, status, description } -connector_tags { tag_id, connector_id, address, tag_name, category, data_type, unit, status, description, } -platform_certificates { client_cert: {minio_object_name, sha256_hash, expiration_date, ...}, private_key: {...} } ← 1 documento só -MinIO /platform/certificates/client_cert_, private_key_ -``` - -Problemas concretos: - -1. **Sem correlação servidor ↔ certificado.** `CertProvider._download_all` faz `find_one('platform_certificates', {})` - (`ingestor/connection/certs.py:47`). Um cliente com dois servidores que exigem certificados diferentes - (ou um servidor que exija validação do cert do servidor) não é atendível. -2. **`server_uri` é dado do certificado, não do servidor.** O `application_uri` do cliente precisa ser - igual à SAN URI do certificado da plataforma (comentário 06/10 do 1974). Hoje ele é digitado por - conector (`connectors.server_uri` / `opc-servers.uri`) — e o 1603 queria "chumbar" no backend. - Erro de digitação = conexão segura recusada sem pista clara. -3. **`server_cert_path` ignorado.** O cert do servidor não tem lugar no MinIO/Mongo. -4. **Dois espaços de id.** `ServerConfigResolver` procura `connectors.connector_id` e depois `opc-servers.id` - com o mesmo `server_id` (`ingestor/connection/servers.py:105-115`); o orchestrator trata "mesmos campos" - como mesmo servidor. Se os ids divergirem, um `server_id` aponta para servidores diferentes conforme a fonte. -5. **Ids por `$max + 1`** (Streamlit `tag_monitor.py:141`) — corrida em criação simultânea/import em lote. - O repo de migrations já tem o padrão certo (`model_id_counters` com `$inc` atômico). -6. **Sem schema/índices.** Nada impede `(connector_id, tag_name)` duplicado nem conector sem `url`. -7. **Sem visibilidade de expiração** do certificado (o Streamlit calcula `expiration_date`, ninguém alerta). - ---- - -## 5. Modelo proposto - -Princípio: **`connectors` vira o registro canônico de servidores OPC**, `opc-servers` vira legado -(somente leitura, depois removido); **certificados viram entidade própria** referenciada pelo conector. - -### 5.1 `opc_certificates` (nova; substitui `platform_certificates`) - -```js -{ - certificate_id: 1, // int, counter atômico - kind: "client", // client | server | ca - name: "sientia-platform-2026", - is_default: true, // no máximo 1 client default (índice parcial único) - status: "active", // active | retired - application_uri: "urn:sientia:opc-ingestor", // SAN URI extraída do cert no upload (kind=client) - subject: "CN=sientia-opc-ingestor,...", - thumbprint_sha1: "…", // o que aparece no console do KEPServer/servidor - not_before: ISODate(), not_after: ISODate(), - cert: { bucket, object_name, sha256, size, content_type, original_filename }, - private_key: { bucket, object_name, sha256, size }, // só kind=client - created_at, created_by, updated_at -} -``` - -- Índices: `certificate_id` único; `{is_default:1}` único parcial (`kind:"client", is_default:true`); - `not_after` (para alerta). -- `application_uri`, `thumbprint`, `not_after` **extraídos do arquivo no upload**, nunca digitados. - -### 5.2 `connectors` (opc_ua) — campos novos - -```js -{ - connector_id: 1, // = opc-servers.id no backfill (mantém server_id estável) - type: "opc_ua", name: "kepserver", url: "opc.tcp://…", status: "Active", - subscription_period_ms: 1000, - security: { - mode: "SignAndEncrypt", // None | Sign | SignAndEncrypt - policy: "Basic256Sha256", - client_certificate_id: null, // null → certificado default da plataforma - server_certificate_id: 7 // opcional: valida o cert do servidor (resolve G-SC) - }, - write_options: { write_timestamp: true, write_quality: true }, // SIENTIAPDE-2202 (Kepware: false/false) - legacy_opc_server_id: "1", // rastreio do backfill - created_at, updated_at -} -``` - -- `use_security` vira derivado (`security.mode != "None"`); manter o campo booleano durante a transição - para não quebrar orchestrator/gateway atuais. -- `server_uri` **sai do conector**: o `application_uri` vem do certificado cliente resolvido. - (Manter leitura do campo antigo como fallback na transição.) -- Índices: `connector_id` único; `{type, name}` único. - -### 5.3 `connector_tags` — ajustes - -```js -{ - tag_id: 9001, connector_id: 1, address: "ns=2;s=…", tag_name: "TargetSin", - category: "Read", // Read | Write | Prediction Confidence - data_type: "Double", unit: "", status: "Active", description: "", - metadata: {}, // objeto aninhado (decisão do design §14; fecha a dúvida do 1600) - created_at, updated_at -} -``` - -- Índices: `tag_id` único; `{connector_id, tag_name}` único; `{connector_id, address}` (não único — - multi-nome é permitido); `updated_at` (orchestrator detectar mudança). - -### 5.4 `connector_id_counters` (padrão `model_id_counters`) - -Documentos `{_id: "connectors"|"connector_tags"|"opc_certificates", seq}`, semeados com o `max` atual, -incremento por `find_one_and_update($inc)`. Resolve o 1598 sem `_get_next_tag_id_batch` com corrida -(batch = `$inc: N` e usa a faixa). - -### 5.5 MinIO - -``` -/opc-ua/certificates//cert.der -/opc-ua/certificates//private_key.pem -``` - -- Chave por `certificate_id` (não por nome de conector: renomear conector não move objeto; um cert serve vários conectores). -- Bucket: **decidir** (ver D1). Policies: escrita só para o usuário do front/BFF; leitura para o gateway. -- O `sha256` no Mongo é conferido pelo gateway no download (detecta objeto trocado/corrompido). - -### 5.6 Como o gateway resolve uma conexão (alvo) - -``` -server_id ──► connectors (connector_id) ──► security.client_certificate_id ?? default - │ │ - │ └─► opc_certificates ─► MinIO (cert+key) ─► cache por certificate_id - ├─► security.server_certificate_id ─► opc_certificates (kind=server) ─► MinIO - └─► application_uri := opc_certificates.application_uri -fallback (transição): opc-servers.id + certificado default -``` - ---- - -## 6. Plano em fases - -### Fase 0 — Orchestrator primeiro, depois levar a 2.x para produção (curto prazo) - -Ordem: **orchestrator antes do gateway**. Sem o slot v2 na `main` do orchestrator, a 2.x em produção só -roda o caminho legado (slot v1 convertido): sem multi-nome, sem banco de tags. O orchestrator também é dono -de popular o banco de tags, gravar `tag_id` nas pipelines e correlacionar servidores ↔ certificados (Fase 1). - -1. Atualizar `release/SIENTIAPDE-1974` com a `main` (0.8.2), resolver os 4 conflitos, rodar testes + e2e, - abrir PR → `main`, deploy no dev e **refazer A1, A2 e A8** (o dev perdeu o slot v2 em 19/10; os testes - da 1975 de 19–23/10 rodaram com slot v1). -2. Só então revisar/mergear **#48** (gateway → `main`); definir versão `2.0.0`. Decidir se o plano B sai aqui (G-PB). -3. **Criar card do helm (P3)** com H1–H4. -4. Runbook de deploy por ambiente (P4, P6, P7): checklist V16 → gateway → laborious; cert cadastrado; servidor - confia no cert; Kepware com `write_timestamp/quality=false`. -5. Concluir **2202** já usando `connectors.write_options` (fallback env `WRITE_TIMESTAMP/WRITE_QUALITY`) — é - o primeiro campo do modelo novo e não depende do resto. -6. **2270**: obter decisão do PO; criar cards de C1 (report, aignosi-scripts). -7. Limpeza do dev (C3). - -**Saída:** 1972 pode ser fechado. Tudo abaixo vira **nova história/épico** (sugestão: "Banco de tags e -certificados OPC estruturados", ligada à `SIENTIAPDI-84`). - -### Fase 1 — Modelo de dados + backfill — escopo do orchestrator (collections/índices via migrate-mongo no `sientia-dataops-database-migrations`) - -1. Migration `…-opc-tag-database.js`: cria `connectors`, `connector_tags`, `opc_certificates`, - `connector_id_counters` com `$jsonSchema` (validationLevel `moderate` na transição) e os índices de §5. -2. Migration `…-backfill-tag-database.js` (idempotente, só insere o que não existe): - - `opc-servers` → `connectors` com **`connector_id = int(opc-servers.id)`**, `security.mode` derivado de - `cert_path`, `legacy_opc_server_id`. - - `platform_certificates` → `opc_certificates` (id 1, `is_default: true`), copiando os metadados MinIO - existentes (sem mover objeto ainda). - - `pipelines.read_tags[]`/`write_tags[]` (`server_id`, `tag_address`, `tag_name`) → `connector_tags` - (dedupe por `(connector_id, tag_name)`), e **grava `tag_id` de volta na pipeline** (resolve B1 + B2 de uma vez). - - semeia os counters com os máximos. -3. Dry-run com dump do dev e de um cliente (relatório de conflitos: mesmo `tag_name` com endereços diferentes, - `server_id` sem servidor). - -### Fase 2 — Gateway (`sientia-dataops-opc-ingestor`) - -1. `CertProvider` → `CertificateStore`: `fetch(certificate_id | None)`, cache e lock **por `certificate_id`**, - verifica `sha256`; `None` = default (`is_default`), fallback `platform_certificates` durante a transição. -2. `ServerConfigResolver` + slot: conexão passa a carregar `client_certificate_id`, `server_certificate_id`, - `application_uri`, `write_options`. Pool usa esses campos em `apply_security` (passa `server_certificate`). -3. Validar no carregamento: SAN do cert == `application_uri`; cert não expirado → erro claro + notificação - (`OPC_CERTIFICATE_*`), em vez de falha opaca do asyncua. -4. Métrica `opc_certificate_expiry_seconds{certificate_id}` + alerta (ex.: < 30 dias). -5. `WriteEngine`: `write_timestamp/quality` = comando → `connectors.write_options` → env (fecha 2202 de vez). -6. G-AR: limpar leitura ativa para leitura unitária sem persistência (pode ir junto com o #48 ou logo depois) + G-DOC. - -### Fase 3 — Orchestrator (`sientia-dataops-orchestrator_temporal`) - -1. Fonte de servidor: `connectors` primeiro, `opc-servers` fallback (hoje é o contrário: só usa `connectors` - quando a pipeline tem `tag_id`). -2. Slot v2 por servidor: `connector_id`, `client_certificate_id`, `server_certificate_id`, `application_uri` - (no lugar de `use_security`/`server_uri` soltos — manter os antigos até o gateway novo estar em todo lugar). -3. Remover `active_read` do slot v2 (sem leitura ativa programada). -4. Reagir a mudança em `connector_tags.updated_at`/`connectors.updated_at` (reimplantar scouter/slot) — fecha a - limitação de renomear tag. - -### Fase 4 — CRUD no Angular (front `sientia` + BFF) - -1. Novo épico/história substituindo 1568/1569: telas de **Conectores**, **Tags** (com import CSV — escopo do 1605) - e **Certificados** (upload extrai SAN/thumbprint/validade; marca default; lista conectores que usam). -2. BFF é quem escreve no Mongo/MinIO (front não fala com MinIO direto); usa os counters. -3. Tela de pipeline passa a escolher tag por `tag_id` (pipelines novas já nascem com `tag_id`). -4. Cancelar no Jira os restos Streamlit não iniciados (1600, 1604, 1605, 1603) com link para o novo card. - -### Fase 5 — Desligar o legado - -1. Orchestrator e gateway sem fallback para `opc-servers`/`platform_certificates`; `server_id/tag_address` - nas pipelines viram só leitura histórica. -2. Remover `opc-servers`, `platform_certificates` (migration com backup) e o caminho Streamlit. -3. `$jsonSchema` em `validationLevel: strict`. - ---- - -## 7. Decisões necessárias - -| # | Decisão | Recomendação | Dono | -|---|---|---|---| -| D1 | Bucket dos certificados: `sientia-dataops` (design §7/§14, Streamlit rc) × `streamlit-connectors` (chart, D7 da 1974) | `sientia-dataops` (bucket por serviço, estratégia MinIO); migrar objetos na Fase 1 e trocar `CERT_BUCKET` no chart | Vitor / platform | -| D2 | Cert por servidor ou único da plataforma | Modelo suporta os dois; **default único** + override por conector só quando necessário | Vitor | -| D3 | `connector_id` = `opc-servers.id` no backfill | Sim — mantém `server_id` de slots, `write_tags` e métricas estáveis | Eduardo | -| D6 | `metadata` aninhado × chaves soltas (1600) | Aninhado (`metadata: {}`), como no design §14 | PO | -| D7 | `write_options` por servidor (2202) | Por servidor, padrão `true`; Kepware configurado `false` antes do deploy | Eduardo | -| D8 | Quem escreve banco de tags em produção até o Angular existir | Backfill por migration + edição via migration/script revisado (sem Streamlit em produção) | Bruno / PO | -| D9 | Plano B `READ_ENGINE=legacy` sai no #48? | Sim, se A1–A8 seguem ok no dev com a 2.0.0-rc4 | Eduardo | - ---- - -## 8. Cards a criar / ajustar - -| Card | Tipo | Repo | Fase | -|---|---|---|---| -| Helm: V16, NATS+gateway 2.x nos profiles, `OPC_WRITE_MODE`, Grafana `write_status`, docs (H1–H4) | Subtask do 1972 | sientia-helm-chart | 0 | -| PR `release/SIENTIAPDE-1974` → `main` (orchestrator) | Subtask do 1972 | orchestrator | 0 | -| Consumidores 12/13/14: report + aignosi-scripts | Subtask | report, aignosi-scripts | 0 | -| **[Nova história] Banco de tags e certificados OPC estruturados** | História (épico 1274 ou novo) | — | 1–5 | -| ↳ Orchestrator: popular banco de tags, gravar `tag_id` nas pipelines, correlação servidor ↔ certificado (collections/índices via migrate-mongo) | Subtask | orchestrator, database-migrations | 1 | -| ↳ Gateway: `CertificateStore` por `certificate_id`, validação SAN/expiração, métrica de expiração | Subtask | opc-ingestor | 2 | -| ↳ Gateway: leitura ativa unitária sem persistência (G-AR) + design (G-DOC) | Subtask | opc-ingestor, orchestrator, aig-docs | 2 | -| ↳ Gateway: reconexão por `status_change_notification` | Subtask | opc-ingestor | 2 | -| ↳ Orchestrator: `connectors` como fonte, slot com certificados, sem `active_read`, reação a `updated_at` | Subtask | orchestrator | 3 | -| ↳ Angular + BFF: CRUD conectores/tags/certificados (substitui 1568/1569) | História própria | sientia, mlops-bff | 4 | -| ↳ Remoção do legado (`opc-servers`, `platform_certificates`, `READ_ENGINE=legacy`) | Subtask | vários | 5 | - ---- - -## 9. Riscos - -- **Laborious 1.6.0 antes do gateway 2.x** em qualquer ambiente → escritas presas em `pending` (P4). -- **Backfill com ids divergentes** entre `opc-servers` e um `connectors` já existente em algum cliente (o - Streamlit rc pode ter rodado em algum lugar) → checar antes, migration aborta em conflito. -- **Troca de bucket/objeto do certificado** derruba conexões seguras se o gateway não invalidar o cache — - já existe `invalidate()` em falha; manter e testar por `certificate_id`. -- **Servidores de cliente com Basic256 apenas** (asyncua marca deprecated) → leitura v2 não conecta; levantar por cliente antes. - ---- - -## 10. Referências - -- Jira: SIENTIAPDE-1972, 1973, 1974, 1975, 1985, 2202, 2270, 1568, 1569, 1598, 1599, 1603, 1605; idea SIENTIAPDI-84. -- Código: `ingestor/connection/certs.py`, `ingestor/connection/servers.py`, `ingestor/gateway_runtime.py`, - `ingestor/managers/ingestor_manager.py:708`, `ingestor/io/slots.py`. -- Orchestrator: `release/SIENTIAPDE-1974` — `orchestrator/activities/formatters.py`, `orchestrator/utils/orchestrator_functions.py`. -- Streamlit: `sientia-projects-templates` `rc/SIENTIAPDE-1588` — `app/pages/connectors.py` (`_save_platform_certificate`), `app/pages/tag_monitor.py`. -- Migrations: `mongodb/migrations/20260908170557-model_id_counters_collection.js` (padrão de counter). -- Design: `aig-docs/07 - Projects & R&D/sientia/opc-gateway/design.md` §3, §7, §14, §15. diff --git a/docs/planning/SIENTIAPDE-1974-certificado-por-conector.md b/docs/planning/SIENTIAPDE-1974-certificado-por-conector.md deleted file mode 100644 index ce875c9..0000000 --- a/docs/planning/SIENTIAPDE-1974-certificado-por-conector.md +++ /dev/null @@ -1,241 +0,0 @@ -# SIENTIAPDE-1974 — Certificado OPC por conector (`opc_certificates`) - -> Plano de implementação. Atualizado em 2026-10-29. -> Repositórios: `sientia-dataops-database-migrations`, `sientia-helm-chart`, -> `sientia-dataops-orchestrator_temporal`, `sientia-dataops-opc-ingestor`. - -## Objetivo - -Ligar cada servidor OPC (`connectors`) ao certificado que ele usa (`opc_certificates`) por **um único caminho**, -sem transição, sem fallback e sem legado: - -``` -connectors.security.client_certificate_id ──► opc_certificates.certificate_id - │ -orchestrator: faz o join e grava no slot v2, por servidor, security + certificado (paths no MinIO + application_uri) - │ -ingestor: lê o slot, baixa cert + chave do MinIO e conecta. Sem certificado no slot → conecta sem segurança -``` - -O ingestor não usa certificado "default", `platform_certificates`, `opc-servers`, nem o caminho legado -(`READ_ENGINE=legacy`, `OpcManager`, slot v1). - -**Exceção consciente (D1):** a escrita mantém um fallback no Mongo. Quando o pod que recebe o comando não tem a -conexão no pool, ele resolve o servidor em `connectors` → `opc_certificates`, com a mesma regra do orchestrator. São -duas fontes da mesma regra, aceitas por enquanto. A otimização fica num card próprio (ver "Fora do escopo"). - -**Fora do escopo:** migração e correção de dados. As migrations deste plano tratam só de estrutura. - ---- - -## Estado atual (verificado) - -| Item | Estado | Evidência | -|---|---|---| -| `platform_certificates` | Criada pelo front Streamlit, sem migration | `sientia-projects-templates` `origin/rc/SIENTIAPDE-1588` `app/pages/connectors.py:27` | -| `opc_certificates` | Criada pela migration `20261028170000-opc_tag_database_collections.js`, com schema parcial. Tem dados no dev | Migration, consulta no dev | -| Vínculo connector → certificado | O campo existe no schema (`connectors.security.client_certificate_id`). No dev está `null` | Migration, consulta no dev | -| `security.mode` / `security.policy` | Estão no schema e nos dados do dev, mas o ingestor ignora os dois: usa `use_security` e fixa `Basic256Sha256` no código | `ingestor/connection/security.py`, `pool.py:130` | -| `application_uri` | Vem de `connectors.server_uri`, mas é dado do certificado (tem que bater com a SAN URI do `.pem`) | `pool.py:126-132`, `security.py:9` | -| Leitura do certificado | `CertProvider` faz `find_one('platform_certificates', {})` e baixa do `CERT_BUCKET` | `ingestor/connection/certs.py:10,47` | -| Bucket `opc-certificates` | Citado em `opc_certificates.cert.bucket`, mas não provisionado. O chart aponta `CERT_BUCKET=streamlit-connectors` | `sientia-helm-chart/values/minio.yaml:516-523`, `opc-ingestor/values.yaml:116` | -| Mudança de config de um servidor no slot | O ingestor compara a config inteira do servidor (sem `tags`) e reconecta se algo mudou | `ingestor/managers/ingestor_manager.py:293-307` | -| Escrita (NATS) | Core NATS, sem JetStream, entrega *at-most-once*. Queue group `gateway-writers`: cada comando vai para um só pod, que não tem como devolvê-lo. O Laborious publica e esquece; o resultado vai para `write_status` no Postgres | `sientia-helm-chart/docs/nats-broker.md:29-39`, `ingestor/adapters/nats_consumer.py`, `laborious/activities/nats_write.py` | - ---- - -## Decisões - -| # | Decisão | Status | -|---|---|---| -| D0 | `opc_certificates` é a oficial; o ingestor deixa de ler a `platform_certificates` (a collection fica) | **Decidido** | -| D0.1 | Migration da `opc_certificates` completa, documentada e auditável | **Decidido** (Etapa 1) | -| D0.2 | Bucket `opc-certificates` provisionado no `sientia-helm-chart`, branch `feature/SIENTIAPDE-1974-connectors-pi-web-api` | **Decidido** (Etapa 2) | -| D0.3 | Sem transição. Caminho único na leitura; a escrita tem a exceção de D1 | **Decidido** | -| D1 | Escrita: pool primeiro; sem conexão no pool, resolve no Mongo (`connectors` → `opc_certificates`) com a mesma regra do orchestrator. Roteamento ao pod dono vira card novo | **Decidido** (ver abaixo) | -| D2 | Zero legado: `READ_ENGINE=legacy`, `OpcManager`, slot v1, `opc-servers` | **Decidido** | -| D3 | Dados fora do plano; migrations só de estrutura; sem runbook | **Decidido** | -| D4 | `security.mode` + `security.policy` são a fonte. `use_security` e `server_uri` saem | **Decidido** | -| D5 | Sem certificado registrado, o servidor é tratado como um que não exige certificado: conecta sem segurança (e falha se o servidor exigir) | **Decidido** | -| D6 | Branch do ingestor: `release/SIENTIAPDE-1973` | **Decidido** | -| D7 | `created_at`, `updated_at`, `created_by`, `updated_by` obrigatórios. Rastro do backfill não é validado. Sem `additionalProperties: false` | **Decidido** | - -### D1 — Escrita: pool primeiro, Mongo como fallback (decidido) - -**Como funciona hoje:** o Laborious publica o comando (C2) em `opc.write` só com o `server_id`. No ingestor, o -`ServerConfigResolver` (`ingestor/connection/servers.py`) usa a conexão do pool, se existir. Senão, consulta o Mongo -(`connectors` e depois `opc-servers`), com cache de 60 s (10 s quando não encontra), e o pool abre a conexão. - -**O que muda:** -- Sai `opc-servers` (D2) e sai o override por dados de conexão no comando (o Laborious nunca envia). -- O lookup no Mongo passa a fazer o join `connectors` → `opc_certificates` e monta a config **no mesmo formato do - servidor no slot** (`server_id`, `url`, `security`, `certificate`). O pool e o `CertProvider` só conhecem esse formato. -- A regra do certificado (D5: certificado `active` e do tipo `client`, senão `certificate: None`) fica isolada numa - função do ingestor, com testes que espelham os do orchestrator. - -**Por que não só o pool:** o Core NATS entrega cada comando a um único pod, sem devolver nem reentregar. Rotear o -comando ao pod dono da conexão (subject por servidor, republicação ou JetStream) não cobre um servidor que nenhum pod -tem no pool (servidor só de escrita, ou pod recém-reiniciado) e ainda exige uma regra de expiração do `pending`. -O fallback no Mongo cobre todos esses casos hoje. - -**Custo aceito:** -- Duas fontes da mesma regra (orchestrator e ingestor). -- Sessões duplicadas: o pod que recebe uma escrita sem ter a conexão abre a sua própria sessão (já acontece hoje). - -### D5 — Sem certificado registrado (decidido) - -- O orchestrator grava `certificate: None` no slot quando `client_certificate_id` é `null`, aponta para um id que não - existe, ou aponta para um certificado que não é `active` ou não é `client`. Nos dois últimos casos (id preenchido, - mas sem certificado válido), o orchestrator registra um aviso em `issues`, porque provavelmente é erro de cadastro. - Mesmo assim o servidor entra no slot. -- O ingestor aplica segurança **só quando o slot tem `certificate`**. Nesse caso usa `security.mode` e - `security.policy`. Sem `certificate`, conecta sem segurança, e um servidor que exige certificado recusa a conexão. - O erro aparece no log e nas métricas de conexão do ingestor. -- Não existe certificado "default": `is_default` e o índice parcial `ux_opc_certificates_default_client` saem do schema. - ---- - -## Etapa 1 — Migrations de estrutura (`sientia-dataops-database-migrations`) - -**Nova migration** `mongodb/migrations/-opc_certificates_schema.js` (SIENTIAPDE-1974): - -1. **Schema da `opc_certificates`**, com o cabeçalho do arquivo descrevendo cada campo: - - | Campo | Tipo | Obrigatório | Descrição | - |---|---|---|---| - | `certificate_id` | int/long | sim | Id estável, referenciado por `connectors.security.client_certificate_id` | - | `kind` | `client` \| `server` \| `ca` | sim | `client`: cert + chave do ingestor. `server`/`ca`: só cert | - | `name` | string | sim | Nome legível, único por `kind` | - | `status` | `active` \| `retired` | sim | Só `active` entra no slot | - | `application_uri` | string | sim se `client` | SAN URI do certificado, usada como `client.application_uri` | - | `cert` | `{bucket, object_name, sha256}` | sim (`bucket`, `object_name`) | Objeto no MinIO. `sha256`: string (64 hex) ou null | - | `private_key` | `{bucket, object_name, sha256}` | sim se `client`, proibido nos outros | Chave privada no MinIO | - | `thumbprint_sha1` | string (40 hex) \| null | não | O que aparece no console do servidor OPC | - | `not_before` / `not_after` | date \| null | não | Validade (`not_after` já tem índice) | - | `created_at` / `updated_at` | date | sim | Auditoria | - | `created_by` / `updated_by` | string | sim | Auditoria: quem cadastrou ou trocou o certificado | - - - "Obrigatório se `client`" fica num `anyOf` do `$jsonSchema`. - - Campos fora da tabela (como o rastro do backfill) não são descritos nem validados: sem `additionalProperties: false`. - - Índices: novo `ux_opc_certificates_kind_name` em `{kind, name}`, único. Sai `ux_opc_certificates_default_client` (D5). Ficam `ux_opc_certificates_certificate_id` e `ix_opc_certificates_not_after`. - - `validationLevel: strict`. - - `down`: volta ao validator e aos índices da `20261028170000`. - -2. **Schema dos `connectors`** (parte do schema da `20261028190000-connectors_pi_web_api_types.js`): - - Saem `use_security` e `server_uri` (D4). - - `security` passa a ser obrigatório para `type: 'opc_ua'`, com `mode` e `policy` obrigatórios. - - `security.client_certificate_id` continua opcional/nullable (D5). - - `down` restaura o schema anterior. - -3. **Testes** em `tests/`, no padrão de `test_connectors_types.py` e `test_head.py`: - - Documentos válidos e inválidos por `kind` (com e sem `private_key` e `application_uri`). - - Campos de auditoria ausentes. - - Unicidade de `{kind, name}`. - - Connector `opc_ua` com e sem `security`, e com `client_certificate_id` nulo. - - Roundtrip `up` → `down`. - - Head esperado. - -## Etapa 2 — Provisionamento (`sientia-helm-chart`, branch `feature/SIENTIAPDE-1974-connectors-pi-web-api`) - -1. `values/minio.yaml`: bucket `opc-certificates` em `provisioning.buckets` (linhas 516-523). -2. Acesso: o ingestor usa as credenciais root do MinIO (`opc-ingestor/values.yaml:106-115`). Avaliar com quem cuida - da plataforma um usuário e uma policy de **somente leitura** em `opc-certificates`, como `model-training-user`. -3. `src/charts-internal/opc-ingestor/values.yaml:98-117`: remover a env `CERT_BUCKET` e o comentário sobre - `platform_certificates`. Remover `READ_ENGINE`, se estiver definida em algum profile (D2). -4. `docs/nats-broker.md:373`: trocar a menção à `platform_certificates` pelo fluxo novo. O NATS não muda (D1). - -## Etapa 3 — Orchestrator (`sientia-dataops-orchestrator_temporal`) - -1. **`orchestrator/workflows/orchestrator.py` (`DEFAULT_CONNECTORS_QUERY`)**: - - Projetar `security.mode`, `security.policy` e `security.client_certificate_id`. Saem `server_uri` e `use_security`. - - `$lookup` de `security.client_certificate_id` para `opc_certificates.certificate_id`, com `status: 'active'` e `kind: 'client'`. - - Projetar `certificate_id`, `application_uri`, `cert.{bucket,object_name}` e `private_key.{bucket,object_name}`. As datas não entram (BSON que o payload do Temporal não carrega). -2. **`orchestrator/utils/opc_slots.py`**: servidor do slot v2 (contrato C1): - ```python - { - 'server_id': ..., 'name': ..., 'url': ..., - 'subscription_period_ms': ..., - 'security': {'mode': 'SignAndEncrypt', 'policy': 'Basic256Sha256'}, - 'certificate': { # None: sem certificado registrado (D5) - 'certificate_id': 1, - 'application_uri': 'urn:sientia:opc-ingestor', - 'cert': {'bucket': 'opc-certificates', 'object_name': 'client/sientia-platform/opc-ingestor-cert.pem'}, - 'private_key': {'bucket': 'opc-certificates', 'object_name': 'client/sientia-platform/opc-ingestor-key.pem'}, - }, - 'tags': {...}, - } - ``` -3. **`orchestrator/utils/tag_database.py`**: - - `CONNECTOR_FIELDS` sem `server_uri`. - - `client_certificate_id` preenchido, mas sem certificado ativo do tipo `client` → `certificate: None` mais aviso em `issues` e em `formatters.py` (D5). -4. **Testes**: - - `test_opc_slots.py`, `test_tag_database.py`, `tag_database_fixtures.py`, `test_formatters.py`. - - `test_orchestrator.py` e a history de replay, se o input do workflow mudar. - - Cenários e2e. -5. **Docs**: README (seção "OPC slots", contrato C1) e CLAUDE.md. - -## Etapa 4 — Ingestor (`sientia-dataops-opc-ingestor`, branch `release/SIENTIAPDE-1973`) - -**Certificado e segurança** - -| Arquivo | Mudança | -|---|---| -| `ingestor/connection/certs.py` | Sai o Mongo (`COLLECTION_NAME`, `find_one`, o `mongo` do construtor). `fetch(certificate)` baixa o `bucket/object_name` vindo do slot. Cache, lock e pasta (`//`) separados por `certificate_id`. `invalidate(certificate_id)` | -| `ingestor/config.py` | Sai o `cert_bucket`; fica o `cert_cache_dir` | -| `ingestor/connection/security.py` | `apply_security(client, security, certificate, cert_path, key_path)`, com o `application_uri` do certificado e mode e policy do slot (mapa de string para classe do asyncua; desconhecida → erro) | -| `ingestor/connection/pool.py` | `_connect`: aplica segurança só quando `certificate` não é `None` (D5). Sai o `server_uri`. `_open` invalida o certificado da conexão. `_check_cfg` passa a comparar `security` e `certificate` (só aviso; a reconexão vem do `ingestor_manager`) | -| `ingestor/io/slots.py` | `SERVER_FIELDS`: `server_id`, `url`, `subscription_period_ms`, `security`, `certificate`. Sai a conversão do slot v1 | - -**Escrita (D1)** - -| Arquivo | Mudança | -|---|---| -| `ingestor/connection/servers.py` | `ServerConfigResolver`: pool primeiro (`config_for`); senão, Mongo, com o cache de TTL atual. `_lookup` faz uma aggregation em `connectors` (`type: 'opc_ua'`, `status: 'Active'`) com `$lookup` em `opc_certificates`, e devolve a config no formato do servidor no slot (`server_id`, `url`, `security`, `certificate`). Saem `opc-servers` (`OPC_SERVERS`, `_opc_server_cfg`, `_id_forms`, já que `connector_id` é slug string) e o override pelo comando | -| `ingestor/connection/certificates.py` (novo) ou dentro de `servers.py` | Função que, a partir do documento do connector e do certificado do `$lookup`, devolve `certificate` ou `None` com a regra de D5 (certificado `active` e `client`; id preenchido sem certificado válido → aviso no log). Testes que espelham os do orchestrator | -| `ingestor/engines/write.py` | `_server_cfg` sem o ramo de conexão vinda do comando. Sem mudança de comportamento: servidor não encontrado continua indo para erro no `write_status` | - -**Remoção do legado (D2)** - -| Arquivo | Mudança | -|---|---| -| `ingestor/managers/opc_manager.py` | Removido | -| `ingestor/managers/ingestor_manager.py` | Saem `READ_ENGINES`/`READ_ENGINE`, os ramos `legacy`, a checagem de slot v1 (`:698`) e `_warn_server_cert_path` (`:708`) | -| `ingestor/io/data_sink.py` | Sai o tratamento de slot v1 (`:56`) | -| `ingestor/gateway_runtime.py` | `CertProvider` sem Mongo. `ServerConfigResolver` continua com Mongo (D1) | -| `tests/` | Remover os testes do legado; adaptar `certs`, `pool`, `slots`, `servers`, `ingestor_manager` e `write` | -| `README.md`, `CLAUDE.md` | Sem plano B e sem `platform_certificates` (README:67 e :236) | - -## Etapa 5 — Deploy (por ambiente) - -1. Migrations (Etapa 1). Os dados precisam estar migrados antes (fora do escopo): o `strict` rejeita documento fora do schema. -2. Helm (Etapa 2). -3. Ingestor e orchestrator **juntos**. Sem transição, um slot antigo não tem `security`/`certificate`. -4. Validação no dev: - - O slot no Redis tem `security` e `certificate`. - - O ingestor conecta ao `kepserver` com o mode e a policy do banco. - - A troca de `client_certificate_id` reconecta com o certificado novo. - - Um servidor sem certificado conecta sem segurança. - - Um id de certificado inválido gera aviso. - - Uma escrita para um servidor que está no pool usa a conexão do pool. - - Uma escrita para um servidor fora do pool resolve pelo Mongo, com o mesmo certificado que o slot teria. - - Uma escrita para um `server_id` inexistente termina em erro no `write_status`. - -## Fora do escopo - -- Migração e correção de dados (`opc_certificates`, `connectors`, objetos no MinIO). -- Remoção da collection `platform_certificates` (o ingestor só deixa de lê-la). -- Validação do certificado do servidor (`security.server_certificate_id`, `kind: server`). -- Métrica e alerta de expiração (`not_after`). -- Conferência de `sha256` no download. -- Upload e CRUD de certificados no front Angular/BFF (quem preenche `created_by`/`updated_by`). -- **Card novo — Escrita OPC roteada ao pod dono da conexão:** subject por servidor (`opc.write.`), slot - cobrindo servidores só de escrita e tratamento de comando sem destino (expiração do `pending` ou `request` com - *no responders*). Quando estiver pronto, remove o fallback no Mongo do `ServerConfigResolver` e a segunda fonte da - regra de certificado (D1). - -## Referências - -- Plano geral: `docs/planning/SIENTIAPDE-1972-banco-de-tags-e-certificados.md` (§5.1, §5.2, Fases 1–3). -- Migrations: `mongodb/migrations/20261028170000-opc_tag_database_collections.js`, `20261028190000-connectors_pi_web_api_types.js`. -- Escrita: `sientia-dataops-laborious` `laborious/activities/nats_write.py`; ingestor `ingestor/connection/servers.py`, `ingestor/engines/write.py:148-160`; `sientia-helm-chart/docs/nats-broker.md`. -- Streamlit (origem da `platform_certificates`): `sientia-projects-templates` `origin/rc/SIENTIAPDE-1588` `app/pages/connectors.py`. diff --git a/requirements-local.txt b/requirements-local.txt index 063fdd4..1e85abd 100644 --- a/requirements-local.txt +++ b/requirements-local.txt @@ -1,4 +1,4 @@ -asyncua==1.1.5 +asyncua>=2.0 redis git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2 prometheus_client diff --git a/requirements.txt b/requirements.txt index 3e2fd88..116afa1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,6 @@ -asyncua +# >=2.0: WriteEngine builds DataValue(StatusCode=...); 1.x names the field StatusCode_, +# so every write would fail with TypeError (reported as bad_timestamp). +asyncua>=2.0 redis # 1.12.5 builds the Postgres engine with an explicit postgresql+psycopg2:// URL: SQLAlchemy 2.1 # defaults postgresql:// to psycopg (v3), which is not installed ("No module named 'psycopg'"). diff --git a/tests/functional/test_write_e2e.py b/tests/functional/test_write_e2e.py new file mode 100644 index 0000000..f57ad6f --- /dev/null +++ b/tests/functional/test_write_e2e.py @@ -0,0 +1,173 @@ +"""End-to-end OPC write (SIENTIAPDE-2202): env → GatewayConfig → WriteEngine → +OpcConnectionPool → a real asyncua server, in process. + +The server keeps the SourceTimestamp it receives, so the timestamp is checked on the +server. An omitted StatusCode decodes as Good on the server (OPC UA default), so the +quality is checked on the wire: the DataValue encoding mask of the WriteRequest the +client sent. +""" + +import socket +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest +import pytest_asyncio +from asyncua import Server, ua +from asyncua.client import ua_client +from asyncua.ua.ua_binary import Primitives, nodeid_to_binary + +from ingestor.config import GatewayConfig +from ingestor.connection.pool import OpcConnectionPool +from ingestor.engines.write import WriteEngine + +TIMESTAMP = '2026-07-13T12:00:00Z' + +# DataValue encoding mask bits (OPC UA Part 6, 5.2.2.17). +_STATUS_CODE_BIT = 0x02 +_SOURCE_TIMESTAMP_BIT = 0x04 + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(('127.0.0.1', 0)) + return s.getsockname()[1] + + +@pytest_asyncio.fixture +async def opc_server(): + url = f'opc.tcp://127.0.0.1:{_free_port()}' + server = Server() + await server.init() + server.set_endpoint(url) + idx = await server.register_namespace('urn:sientia:e2e') + var = await server.nodes.objects.add_variable(idx, 'Pred', 0.0, ua.VariantType.Float) + await var.set_writable() + async with server: + yield url, var + + +@pytest.fixture +def wire(monkeypatch): + """Serialized WriteRequests, as sent by the client.""" + sent: list[bytes] = [] + original = ua_client.struct_to_binary + + def capture(obj): + data = original(obj) + if isinstance(obj, ua.WriteRequest): + sent.append(data) + return data + + monkeypatch.setattr(ua_client, 'struct_to_binary', capture) + return sent + + +def _datavalue_mask(request: bytes, addr: str) -> int: + # WriteValue = NodeId, AttributeId, IndexRange, DataValue: the byte right after + # the first three is the DataValue encoding mask. The NodeId is parsed from the + # address as the client does (its binary encoding depends on how it was built). + prefix = ( + nodeid_to_binary(ua.NodeId.from_string(addr)) + + Primitives.UInt32.pack(ua.AttributeIds.Value) + + Primitives.String.pack(None) + ) + return request[request.index(prefix) + len(prefix)] + + +async def _engine(url: str, pools: list) -> WriteEngine: + config = GatewayConfig.from_env() + logger, notification_handler = MagicMock(), MagicMock() + # emit() is awaited, start()/shutdown_sync() are not. + metrics_controller = MagicMock(emit=AsyncMock(), shutdown=AsyncMock()) + pool = OpcConnectionPool(MagicMock(), logger, notification_handler, metrics_controller) + pools.append(pool) + servers = MagicMock() + servers.resolve = AsyncMock( + return_value={ + 'server_id': 'e2e', + 'url': url, + 'security': {'mode': 'None', 'policy': 'None'}, + 'certificate': None, + } + ) + return WriteEngine( + pool, + AsyncMock(), + logger, + notification_handler, + metrics_controller, + write_timestamp=config.write_timestamp, + write_quality=config.write_quality, + servers=servers, + ) + + +@pytest_asyncio.fixture +async def pools(): + created: list[OpcConnectionPool] = [] + yield created + for pool in created: + for client, _ in pool._connections.values(): + await client.disconnect() + + +@pytest.mark.integration +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'env,cmd_flags,expect_timestamp,expect_quality', + [ + pytest.param({}, {}, True, True, id='defaults-on'), + pytest.param({'WRITE_TIMESTAMP': 'false'}, {}, False, True, id='timestamp-off'), + pytest.param({'WRITE_QUALITY': 'false'}, {}, True, False, id='quality-off'), + pytest.param( + {'WRITE_TIMESTAMP': 'false', 'WRITE_QUALITY': 'false'}, {}, False, False, id='both-off' + ), + pytest.param( + {'WRITE_TIMESTAMP': 'true'}, + {'write_timestamp': False}, + False, + True, + id='command-overrides-env', + ), + pytest.param( + {'WRITE_TIMESTAMP': 'false', 'WRITE_QUALITY': 'false'}, + {'write_timestamp': True, 'write_quality': True}, + True, + True, + id='command-turns-on-over-env', + ), + ], +) +async def test_write_datavalue_parts( + monkeypatch, opc_server, wire, pools, env, cmd_flags, expect_timestamp, expect_quality +): + monkeypatch.delenv('WRITE_TIMESTAMP', raising=False) + monkeypatch.delenv('WRITE_QUALITY', raising=False) + for name, value in env.items(): + monkeypatch.setenv(name, value) + url, var = opc_server + engine = await _engine(url, pools) + cmd = { + 'server_id': 'e2e', + 'addr': var.nodeid.to_string(), + 'value': 1.5, + 'data_type': 'float', + 'model_id': 'm1', + 'timestamp': TIMESTAMP, + **cmd_flags, + } + + result = await engine.write(cmd) + + assert result.status == 'ok', result.message + stored = await var.read_data_value() + assert stored.Value.Value == 1.5 + if expect_timestamp: + assert stored.SourceTimestamp == datetime.fromisoformat(TIMESTAMP) + else: + assert stored.SourceTimestamp is None + assert len(wire) == 1 + mask = _datavalue_mask(wire[0], cmd['addr']) + assert bool(mask & _SOURCE_TIMESTAMP_BIT) is expect_timestamp + assert bool(mask & _STATUS_CODE_BIT) is expect_quality diff --git a/tests/unit/engines/test_write.py b/tests/unit/engines/test_write.py index 2d2aded..0e429f0 100644 --- a/tests/unit/engines/test_write.py +++ b/tests/unit/engines/test_write.py @@ -140,6 +140,36 @@ async def test_write_engine_flags_off_omit_statuscode_and_timestamp(): assert dv.SourceTimestamp is None +@pytest.mark.asyncio +async def test_write_engine_timestamp_off_keeps_statuscode(): + eng = make_engine(write_timestamp=False) + node = make_node() + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd()) + + dv = node.write_value.await_args.args[0] + assert dv.Value.Value == 42.0 + assert dv.SourceTimestamp is None + assert dv.StatusCode is not None + assert dv.StatusCode.is_good() + + +@pytest.mark.asyncio +async def test_write_engine_quality_off_keeps_timestamp(): + # The command has no write_quality, so the engine (env) default applies. + eng = make_engine(write_quality=False) + node = make_node() + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd()) + + dv = node.write_value.await_args.args[0] + assert dv.Value.Value == 42.0 + assert dv.StatusCode is None + assert dv.SourceTimestamp == datetime.fromisoformat('2026-07-13T12:00:00Z') + + @pytest.mark.asyncio async def test_per_message_flag_override_disables_timestamp(): # Engine default is True, but the command explicitly turns it off. diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index fbfe05b..6020059 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -41,6 +41,7 @@ def test_from_env_overrides(monkeypatch): monkeypatch.setenv('GRPC_ENABLED', 'true') monkeypatch.setenv('GRPC_PORT', '6000') monkeypatch.setenv('WRITE_MIN_PAUSE_S', '0.5') + monkeypatch.setenv('WRITE_TIMESTAMP', 'false') monkeypatch.setenv('WRITE_QUALITY', 'false') cfg = GatewayConfig.from_env() @@ -48,6 +49,7 @@ def test_from_env_overrides(monkeypatch): assert cfg.grpc_enabled is True assert cfg.grpc_port == 6000 assert cfg.write_min_pause_s == 0.5 + assert cfg.write_timestamp is False assert cfg.write_quality is False