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

17 KiB
Raw Blame History

OPC Gateway — Design (SIENTIAPDE-1972)

  • Épico: SIENTIAPDE-1972 — OPC Gateway: novo 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: SIENTIAPDE-1972-opc-gateway-novo-gateway-derivado-do-opc-ingestor-com-ajustes-no-orchestrator-e-no-laborious (no 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, cobrindo:

  • Gestão de certificados (via MinIO + MongoDB — reaproveitando o que o épico 1569 já construiu).
  • Leitura por subscription usando o sistema de slots já existente.
  • Leitura ativa: agendada (frequência múltipla do ciclo de operação) e on-demand via GRPC.
  • Escrita ativa via NATS como broker intermediário.
  • Métricas detalhadas de todas as operações.
  • Boas práticas / correções de bugs do card 1879.

O trabalho atravessa 3 repositórios. Este spec detalha o Gateway (repo atual) e define os contratos que o Orchestrator (1974) e o Laborious (1975) devem honrar. Cada um desses dois ganha spec próprio no seu repo.

2. Escopo

Dentro (este spec / este repo):

  • Arquitetura interna do Gateway e todos os seus componentes.
  • Consumo do banco de tags do 1569 e do slot v2 no Redis.
  • Escrita OPC via NATS; leitura ativa agendada e on-demand via GRPC.
  • Métricas e checklist 1879.
  • Contratos C1–C4 (interfaces com Orchestrator/Laborious/chamadores GRPC).

Fora (specs próprios, outros repos):

  • Implementação do 1974 no orchestrator_temporal.
  • Implementação do 1975 no laborious_temporal.
  • CRUD/UI do banco de tags (1569, já implementado).

3. Estado atual (fatos do código)

opc-ingestor (base)

  • asyncua 1.1.5. Leitura só por subscription. Sem escrita, sem NATS, sem GRPC.
  • Slots vivem no Redis (slot:opc_tags:{id}), com lease/heartbeat para balanceamento entre instâncias.
  • Arquitetura de managers: opc_manager, ingestor_manager, data_manager, resource_manager.
  • Egress: Mongo (obrigatório) + Kafka (opcional). Métricas Prometheus em :9090/metrics.
  • Conexão OPC hoje é criada por slot arrematado (só conecta nos servidores dos slots que possui).

dataops-library

  • MongoDBRepository (async: find/find_one/distinct/aggregate/insert/insert_many/update_one/update_many/delete_one/delete_many), MinioRepository, opc_ua_client, helpers de métricas Prometheus (sientia_do.observability). Métodos genéricos concluídos em feature/SIENTIAPDE-1590.
  • Não tem NATS (usa Kafka), GRPC nem modelo de slot.

Banco de tags (épico 1569) — confirmado local em sientia-projects-templates @ rc/SIENTIAPDE-1588

Coleções no Mongo sientia:

Coleção Campos
connectors (opc_ua) connector_id(int), type="opc_ua", name, url, server_uri, subscription_period_ms, use_security(bool), status, description
connectors (rest_api) connector_id, type="rest_api", name, base_url, auth_config{type,token}, headers_config(JSON string), request_timeout, verify_ssl, status, description
connector_tags tag_id(int), connector_id(FK int), address(NodeId), tag_name, category(Read/Write/Prediction Confidence), data_type(float/int/string/boolean), unit, status, description + atributos custom (top-level)
platform_certificates doc único: client_cert{minio_object_name, sha256_hash, uploaded_at, original_filename, expiration_date}, private_key{...}

Fatos relevantes:

  • Segurança já é Basic256Sha256 (hardcoded em _create_opc_ua_client, connectors.py), com certificado de plataforma único (compartilhado), baixado do MinIO (bucket streamlit-connectors, prefixo platform/certificates/) para arquivos temp e passado ao asyncua em runtime.
  • Multi-nome por endereço já é permitido: unicidade em (connector_id, tag_name), nunca em address.
  • Não existe metadata, model_id/model_name, collection nem frequency em connector_tags. Modelo/coleção/frequência vivem no lado models/pipelines (schema legado). O campo metadata é trabalho do subtask 1600 (pendente).
  • Bug local conhecido: projects-templates/app/src/operations/db_infra.py:16 insere ...\repos\sientia-dataops-library no sys.path (caminho sem o subdir software-engineer), inexistente nesta máquina → o app não importa a lib local. Questão em aberto (ver §12).

Escrita hoje (Laborious)

  • Escrita OPC direta via asyncua (laborious/activities/opc.py → opc_repository.write_data → node.write_value).
  • Status de escrita é gambiarra: sobrescreve prediction_confidence com sentinelas (12=erro escrita, 14=sessão ruim; PI=13) em process_confidence. É esse acoplamento que o 1975 remove.
  • Payload de escrita já contém: server_id, addr, value, data_type, tag_type(prediction|confidence), model_id, model_name, timestamp.

4. 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
                                          └─▶ Mongo `models`.write_status

Decisão-chave: conexões OPC deixam de ser atreladas a slot. Escrita (NATS) e leitura on-demand (GRPC) podem mirar qualquer servidor, então a conexão vira um pool por servidor compartilhado por todas as engines.

Componentes

  1. OpcConnectionPool (peça nova, núcleo compartilhado)

    • Conexões asyncua por connector_id/server, reusadas por todas as engines; criação sob demanda + idle eviction.
    • Segurança Basic256Sha256; secure_channel_timeout = int(session_timeout * 1.25) (1879 §3).
    • Certificados: baixa o platform certificate do MinIO (metadados em platform_certificates), espelhando o padrão _create_opc_ua_client do 1569.
    • Uma única referência time.monotonic() para latências/timeouts (1879 §2). Opção de sincronizar relógio via NTP.
    • Reconexão + métricas de conexão/desconexão e tipos de erro OPC.

  2. SubscriptionEngine (evolui ingestor_manager + opc_manager)

    • Arremata slots no Redis (reusa lease/heartbeat), cria subscription por (slot, servidor) via pool.
    • No datachange, faz fan-out multi-nome: para cada {collection, tag_name, model_name, model_id} do endereço, emite registro ao DataSink.

  3. ActiveReadEngine (nova)

    • Agendada: lê tags marcadas para leitura ativa em frequência múltipla do ciclo de operação; persiste via DataSink.
    • On-demand: atende o GrpcServer — lê agora, persiste E devolve valor+status+timestamp.
    • Mede lag de leitura (ts recebido − ts da máquina) (1879).

  4. WriteEngine (nova)

    • Escrita OPC via pool com DataValue + Variant + status_code Good + source_timestamp (1879 §1).
    • Flags: liga/desliga escrita de timestamp e de quality (quality Good_Local_Override 0xD8 se detectar problema de relógio).
    • Converter seguro: corrige BUG-1 (coerção de bool via string JSON) e BUG-2 (try/except em conversão de tipo), retornando "não escrito" + log em erro.
    • Rate-limit / pausa mínima entre escritas.
    • Diferencia tipos de erro OPC (métrica por erro).
    • Após escrita, atualiza models.write_status via DataSink.
    • (Futuro/flag) opção de escrever via método OPC em vez de tag direta (1879).

  5. Adapters de entrada

    • NatsWriteConsumer: subscribe no subject de escrita com queue group gateway-writers (exatamente 1 instância processa cada comando). Core NATS (fire-and-forget); migração para JetStream fica para depois sem mudar o Laborious.
    • GrpcServer: RPC de leitura on-demand → ActiveReadEngine.

  6. DataSink (evolui data_manager)

    • Persiste leituras no Mongo/Kafka com fan-out multi-nome.
    • update_model_write_status(model_id, status, ...) na coleção models.

  7. app — instancia pool + engines + adapters, sobe Prometheus, trata sinais e shutdown gracioso.

Coordenação: lease/heartbeat de slots no Redis seguem governando SubscriptionEngine e a leitura agendada. Escrita e leitura on-demand não dependem de lease — qualquer instância atende, conectando sob demanda pelo pool.

Meta de qualidade (1879 §Geral): ≤ ~500 linhas por arquivo, ≤ ~20 linhas por função.

5. Contratos

C1 — Slot v2 no Redis (slot:opc_tags:{id})

Orchestrator escreve, Gateway lê. Substitui topics:[str] por lista multi-nome por endereço, e carrega config de leitura ativa:

{
  "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"}
        ]
      }
    }
  }
}

O Orchestrator monta isso fazendo o JOIN entre:

  • Identidade da tag (banco de tags 1569): connector_tags.address/tag_name/data_type/category + connectors.url/server_uri/use_security/subscription_period_ms.
  • Config de modelo: model_id, model_name, collection, frequency (lado models/pipeline).

Multi-nome = múltiplas entradas em names para o mesmo address.

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.

{
  "msg_id": "uuid",
  "server_id": "server1",
  "addr": "ns=2;i=5",
  "value": 42.0,
  "data_type": "float",
  "tag_type": "prediction",
  "model_id": "1",
  "model_name": "m1",
  "timestamp": "2026-07-13T12:00:00Z",
  "write_timestamp": true,
  "write_quality": false
}

C3 — Campo de status de escrita (Mongo models)

Novo campo, substitui as sentinelas 12/13/14. Gateway grava após escrita OPC; Laborious grava nas escritas dele (PI/simples).

"write_status": {
  "status": "ok|error|session_bad",
  "opc_status_code": "Good",
  "message": "...",
  "source": "opc|pi",
  "updated_at": "2026-07-13T12:00:01Z"
}

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

6. Gestão de certificados

Reaproveita o que o 1569 já construiu: coleção platform_certificates (metadados) + MinIO (bucket streamlit-connectors, prefixo platform/certificates/). O OpcConnectionPool baixa client_cert/private_key para arquivos temporários e passa ao asyncua com Basic256Sha256, espelhando _create_opc_ua_client. Certificado de plataforma é único/compartilhado (não é por conector). Server cert opcional por conexão.

7. Métricas (1973 + 1879)

app_up; contadores de sucesso/erro de escrita e de leitura ativa; histogramas de latência de escrita e de leitura ativa; tags na subscription; slots possuídos; contagem de conexões/desconexões; status de escrita por modelo; valor de cada tag por tag/modelo; ciclos sem chegar dado em subscription; tipos de erro OPC diferenciados (contagem por erro); lag de leitura (ts recebido − ts da máquina). Base já existe em ingestor/metrics.py e em sientia_do.observability.

8. Checklist boas práticas (1879)

  • Basic256Sha256 (não Basic256) — já é o padrão do conector; garantir no pool.
  • time.monotonic() único para todo cálculo de tempo/latência.
  • Escrita com status_code Good + source_timestamp.
  • Flag liga/desliga escrita de timestamp.
  • Flag liga/desliga escrita de quality (Good_Local_Override 0xD8 em problema de relógio).
  • Opção de sincronizar relógio via NTP.
  • Rate-limit / pausa mínima entre escritas.
  • BUG-1: coerção de bool a partir de string JSON (str.lower() in ('false','0','no','off','')).
  • BUG-2: try/except na conversão de tipo do converter (retornar "não escrito" + log).
  • secure_channel_timeout = int(session_timeout * 1.25).
  • Diferenciar tipos de erro OPC nas métricas.
  • (Flag/futuro) escrita via método OPC em vez de tag direta.
  • ≤ ~500 linhas/arquivo, ≤ ~20 linhas/função.

9. Faseamento e dependências

  • F0 — Fundações: scaffolding na branch; OpcConnectionPool (cert MinIO+Mongo, Basic256Sha256, monotonic, channel>session); base de métricas.
  • F1 — SubscriptionEngine: consumo do slot v2 + fan-out multi-nome + DataSink. (depende de C1)
  • F2 — WriteEngine + NatsWriteConsumer + models.write_status. (depende de C2, C3)
  • F3 — ActiveReadEngine (agendada) + GrpcServer (on-demand). (depende de C4)
  • F4 — Métricas completas + hardening 1879 + testes + Dockerfile.

Dependências externas:

  • Banco de tags do 1569 disponível no Mongo alvo (schema connectors/connector_tags/platform_certificates).
  • Broker NATS provisionado.
  • Bucket de certs no MinIO (streamlit-connectors).
  • Versão da dataops-library com os métodos genéricos (feature/SIENTIAPDE-1590) publicada e pinada no gateway.

10. Trilhas paralelas (nível de contrato)

  • Orchestrator (1974): montar slot v2 a partir do banco de tags (JOIN connector_tags × modelo), propagar model_id/model_name/collection, suportar multi-nome por endereço. Arquivos-alvo hoje: orchestrator/utils/orchestrator_functions.py (gather_read_tags, build_tag_config), orchestrator/activities/formatters.py (process_slots), orchestrator/activities/slot_manager.py.
  • Laborious (1975): publicar comando de escrita no NATS (C2) em vez de node.write_value; gravar models.write_status (C3) nas escritas próprias (PI/simples); remover as sentinelas 12/13/14 de process_confidence/api.py, desacoplando status de escrita da confidence.

11. Testes

  • Unit por engine/pool (reusar padrão de tests/unit/managers): pool (reconexão, cert), subscription (fan-out multi-nome), write (converter bool/erro, status_code/timestamp, rate-limit), active read (lag), adapters (NATS queue group, GRPC read+persist).
  • Cada lógica não trivial deixa ao menos um teste executável (converter, fan-out, error mapping).
  • Funcional single-node contra o opc_simulator.

12. Questões em aberto

  1. Campo metadata em connector_tags (subtask 1600, pendente): formalizar como campo nested. Nas tags de teste do gateway, criar com metadata: {} (pedido do Vitor, para documentação).
  2. Durabilidade da escrita: começar com Core NATS puro; reavaliar JetStream (TTL curto p/ descartar comando obsoleto) se perda em restart se mostrar problema.
  3. db_infra.py:16 (projects-templates): sys.path aponta para caminho inexistente (...\repos\sientia-dataops-library, falta software-engineer). Corrigir para o app 1569 importar a lib local. Fora do escopo do gateway — decidir se entra aqui.
  4. Origem de collection/frequency/active_read no JOIN do 1974: confirmar de onde vêm (pipeline vs config de modelo) ao detalhar o spec do Orchestrator.
  5. Chamador do GRPC on-demand: confirmar quem chama (Laborious? Orchestrator? UI/API) para dimensionar auth/limites do GrpcServer.

13. Riscos

  • At-least-once futuro (JetStream): escrita OPC não é idempotente por natureza; se migrar, tratar dedupe por msg_id.
  • Relógio/OPC: timestamps e NTP são fonte recorrente de bug (1879) — priorizar monotonic e flags de quality.
  • Pool multi-instância: escrita/on-demand conectam sob demanda a qualquer servidor; vigiar limite de conexões e idle eviction para não estourar sessões no servidor OPC.
  • Dependência de schema do 1569 ainda em rc/: o schema do banco de tags pode mudar antes do merge; contrato C1 deve ser revalidado no merge.
Reference in New Issue View Git Blame Copy Permalink