gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
941c6c002994aa7bdd4a92224a52f6d52e18240c
sientia-dataops-opc-ingestor/docs/superpowers/specs/2026-07-13-opc-gateway-design.md

24 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. O subtask 1600 (pendente) adiciona Category + atributos custom como chaves flat na raiz do doc (não um campo metadata nested) — Vitor pediu um campo metadata, o que diverge do 1600 (ver §17).
  • 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

  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.

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 mecanismo do 1569: metadados em Mongo platform_certificates + arquivos no MinIO. 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.

Bucket (default proposto): a estratégia de MinIO (03-database-storage/minio-strategy.md, Bruno Domingues) é bucket + user scoped por serviço, e o bucket do domínio DataOps/sientia-core é sientia-dataops. Hoje o CRUD do 1569 grava os certs no bucket do frontend streamlit-connectors. Default do spec: o Gateway lê os certs de sientia-dataops (domínio dele); reconciliar com a plataforma onde os certs passam a ser gravados — mover a escrita do 1569 para sientia-dataops, ou conceder ao Gateway policy de leitura no bucket do frontend. 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): Flyway (repo Aignosi/sientia-dataops-database-migrations, schema sientia_data; substituiu Alembic/SQLAlchemy). Coluna nova = .sql versionado (V2+), imutável, aplicado por Job Helm hook pre-upgrade; backward-compatible.
  • 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; o bucket do sientia-core/DataOps é sientia-dataops. 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, propagar model_id/model_name/collection, multi-nome. Integração é net-new (o orchestrator hoje só usa connectors_config de conexão e pipelines.read_tags embutidas; não toca connector_tags). Alvos: orchestrator/utils/orchestrator_functions.py (gather_read_tags, build_tag_config), orchestrator/activities/formatters.py (process_slots), orchestrator/activities/slot_manager.py.
    • Mecânica proposta do JOIN: connector_tags não tem model_id/collection/frequency. Então: (a) identidade da tag + conexão (address, data_type, tag_name, category, url/server_uri/security) vêm de connector_tags+connectors — fonte única; (b) associação modelo→tag + collection + frequency continua vindo do lado modelo/pipeline (hoje collection = raw_{schedule_name}, frequency no read config). Chave de junção a confirmar com Vitor: casar por (connector/server + address) (menos migração) ou o pipeline passar a referenciar connector_id/tag_id explicitamente.
  • 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-... ✅.

Resolvidas nesta rodada (lendo docs/código):

  • Migração = Flyway — repo Aignosi/sientia-dataops-database-migrations, schema Postgres sientia_data (substituiu Alembic/SQLAlchemy). Coluna de write_status = .sql versionado (V2+), Job Helm hook. Falta só confirmar o nome exato da tabela de predições em prod (sientia_data vs predictions_schema).
  • Bucket de certs (default) = sientia-dataops — padrão de MinIO é bucket scoped por serviço; DataOps usa sientia-dataops. Falta só a plataforma decidir onde os certs passam a ser gravados (mover escrita do 1569 ou dar policy de leitura). (§7)
  • JOIN do 1974: mecânica proposta em §15 (identidade/conexão do banco de tags; associação/collection/frequency do pipeline). Falta só a chave de junção (address+server vs tag_id) — pergunta ao Vitor.
  • metadata: o subtask 1600 salva atributos custom como chaves flat na raiz do doc da tag (não um campo nested). Vitor pediu um campo metadata — há discrepância a confirmar (nested metadata vs chaves flat). Pro gateway: tags de teste com metadata: {}.

Em aberto (precisam de humano):

  1. Provisionamento do NATS no air-gapped: subchart/versão/sizing/network policy com a plataforma (a escolha já é NATS). (§8) — plataforma/Bruno
  2. Escrita canônica dos certs: mover 1569→sientia-dataops ou policy cross-bucket. (§7) — plataforma
  3. Chave de junção do 1974: (server+address) vs connector_id/tag_id no pipeline. (§15) — Vitor
  4. metadata: nested vs chaves flat (discrepância 1600 × pedido do Vitor). — Vitor
  5. Chamador do GRPC on-demand: quem chama ReadTag e modelo de auth. — Vitor
  6. Tabela de predições em prod: schema/nome exato para a migração Flyway. — Vitor / dono do repo de migrations

Adiada por design (nós):

  • Durabilidade da escrita: Core NATS agora; reavaliar JetStream (TTL curto) se houver perda real em restart.

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).
Reference in New Issue View Git Blame Copy Permalink