gitea_admin/sientia-dataops-opc-ingestor

Fork 0

Files

PedroHMCosme 524f4d6d3f SIENTIAPDE-1972: fixa NATS como broker decidido (brief/Vitor) e reflete rename da branch

2026-07-13 16:55:27 -03:00

22 KiB

Raw Blame History

OPC Gateway — Design (SIENTIAPDE-1972)

Épico: SIENTIAPDE-1972 — OPC Gateway derivado do OPC Ingestor, com ajustes no Orchestrator e no Laborious.
Sub-tasks: 1973 (Gateway), 1974 (Orchestrator), 1975 (Laborious).
Referências: 1879 (boas práticas OPC), 1569 (banco de tags / CRUD conectores).
Branch: feature/SIENTIAPDE-1972-opc-gateway-... (repo sientia-dataops-opc-ingestor).
Data: 2026-07-13.

1. Contexto e objetivo

Evoluir o opc-ingestor (hoje read-only, leitura por subscription) para um OPC Gateway completo:

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 do card 1879.

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: 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): 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. 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, 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, 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 (1569) — `sientia-projects-templates @ rc/SIENTIAPDE-1588`

Mongo sientia:

Coleção	Campos
`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{...}`

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

5. Arquitetura do Gateway (Abordagem B — engines por capacidade)

Orchestrator ─(Redis slot v2)─▶ SubscriptionEngine ─┐
Laborious ────(NATS write)────▶ NatsWriteConsumer ──▶ WriteEngine ────┐
caller ───────(GRPC read)─────▶ GrpcServer ─────────▶ ActiveReadEngine┤
scheduler ────────────────────────────────────────▶ ActiveReadEngine┤
                                                                      ▼
                                        OpcConnectionPool (por servidor) ─▶ servidores OPC
                                                                      │
                                        DataSink ◀────────────────────┘
                                          ├─▶ Mongo (dados) / Kafka topics
                                          └─▶ Postgres predictions.write_status

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

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).
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}.
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).
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).
Adapters — NatsWriteConsumer (subject de escrita, queue group gateway-writers, Core NATS fire-and-forget; JetStream depois) e GrpcServer (leitura on-demand).
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).
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.

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 + config de leitura ativa:

{
  "server1": {
    "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",
        "active_read": { "enabled": true, "cycle_multiple": 3 },
        "names": [
          {"collection": "raw_x", "tag_name": "Counter",  "model_name": "m1", "model_id": "1"},
          {"collection": "raw_y", "tag_name": "Contador", "model_name": "m2", "model_id": "2"}
        ]
      }
    }
  }
}

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. Deve carregar timestamp+model_id (correlação com C3).

{ "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 — 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); }
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.

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

8. NATS como nova dependência de plataforma

Decisão: NATS (definido pelo brief oficial opc-gateway/topics.md e confirmado pelo dono da feature, Vitor Pimentel — fonte mais atualizada). O padrão de subjects/queue groups do NATS casa com o comando de escrita ponto-a-ponto de baixa latência.

Nota de plataforma: os docs air-gapped diziam "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). O OPC Gateway é exatamente esse novo componente que cria o requisito — logo o NATS entra na plataforma como dependência de primeira classe. Ônus de onboarding a cumprir:

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.
Credencial via sientia-secrets-chart.
HA/persistência: Core NATS agora (fire-and-forget, sem storage); JetStream depois exige storage/persistência.

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. ✅ Branch já renomeada para feature/SIENTIAPDE-1972-....
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 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/função.

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.

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

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.

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.

17. Questões em aberto

Decididas:

Broker = NATS (Core NATS agora; §8). Definido pelo brief/Vitor.
Nome da branch: renomeada para feature/SIENTIAPDE-1972-... ✅.

Em aberto:

Bucket de certs: Gateway lê do bucket streamlit-connectors (do app 1569) ou ganha bucket opc-gateway scoped? Default do spec: reutilizar o cert de plataforma do 1569 (MinIO+Mongo, conforme brief); confirmar scoping com plataforma. (§7/§10)
Provisionamento do NATS no air-gapped: alinhar com plataforma o subchart/versão/sizing (a escolha já é NATS; falta o provisionamento). (§8)
Stack de migração de banco: Flyway (docs helm) vs Alembic+SQLAlchemy (onboarding) para a coluna de write_status — confirmar.
Campo metadata em connector_tags (subtask 1600): formalizar; tags de teste criadas com metadata: {} (pedido do Vitor).
Durabilidade da escrita: Core NATS agora; reavaliar JetStream (TTL curto p/ descartar comando obsoleto) se perda em restart for problema.
Origem de collection/frequency/active_read no JOIN do 1974: confirmar (pipeline vs config de modelo) ao detalhar o spec do Orchestrator.
Chamador do GRPC on-demand: confirmar quem chama (Laborious/Orchestrator/UI-API) p/ dimensionar auth/limites.

18. Riscos

Introdução de NATS (decidida): a diretriz antiga "não adicionar NATS sem requisito" é atendida — o Gateway é o requisito; resta cumprir 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).

22 KiB Raw Blame History Unescape Escape