OpenLineage — os scans do BeryX no lineage do cliente e o lineage do cliente no BeryX

Cada scan do BeryX é emitido como um run OpenLineage do job datavault-scan. Quem usa Marquez, DataHub, Astro (ou qualquer servidor que aceite o spec OpenLineage 2-0-x por HTTP) passa a ver os scans no seu grafo de lineage, ligados às tabelas que leram, com a qualidade medida em cada uma.

Read-only nas fontes mantém-se: o BeryX só envia eventos para o endpoint OpenLineage configurado. Só metadados e agregados (nomes de tabelas/colunas, contagens, score, resultado das regras) — nunca linhas nem amostras.

Eventos emitidos

• Job: namespace configurável por workspace (por omissão datavault), name = datavault-scan, facet jobType (BATCH / DATAVAULT / QUALITY_SCAN).
• runId determinístico: o correlation_id do scan (já é um UUID). START e COMPLETE caem no mesmo run, e as repetições do worker também — o mesmo id que aparece nos logs estruturados do BeryX.
• Datasets: namespace = <tipo>://<host[:porta]> da conexão (ex.: postgres://db.acme.pt:5432; credenciais nunca entram) e name = [base.]schema.tabela. Conexões sem host (ficheiros, SQLite) usam datavault://connection/<id>. Não há outputs: o scan só lê.

Facets de qualidade (em inputFacets)

O facet próprio segue a convenção OpenLineage de prefixar facets custom com o nome do produtor, para não colidir com facets futuros do spec.

Limites: 200 datasets por evento, 500 colunas no schema.

Comportamento

• Desligado por omissão. Sem configuração, ou com enabled: false, nada é enviado.
• Best-effort: timeout de 5 s e uma repetição (rede, 429, 5xx). Um servidor em baixo nunca falha nem atrasa significativamente o scan — o erro fica em last_event_status/last_event_message (GET /openlineage) e no log (openlineage_event_failed, com correlation_id).
• O API key (opcional, enviado como Authorization: Bearer …) é cifrado com Fernet, nunca devolvido pela API e retirado de qualquer mensagem de erro.
• Anti-SSRF: o URL é validado ao gravar e a cada envio (IP fixado, sem redirects). Destinos internos (localhost, redes privadas) e http só com o host em DATAVAULT_CONNECTOR_ALLOWED_HOSTS.

Configurar (API, admin)

PUT /api/v1/openlineage
{
  "endpoint": "https://marquez.acme.pt",   // raiz => /api/v1/lineage; ou URL completo
  "namespace": "acme-dq",                   // opcional (default "datavault")
  "api_key": "…",                           // opcional; omitido mantém, "" remove
  "enabled": true                           // default false
}
GET    /api/v1/openlineage   → configuração + estado do último evento (sem API key)
DELETE /api/v1/openlineage

Endpoints típicos:

Exemplo: Marquez local

1. Arrancar o Marquez (API em :5000, UI em :3000):

   git clone https://github.com/MarquezProject/marquez && cd marquez
   ./docker/up.sh
   

2. Autorizar o destino local no BeryX (dev) — sem isto o guard anti-SSRF recusa http://localhost:

   export DATAVAULT_CONNECTOR_ALLOWED_HOSTS=localhost
   

   (reiniciar a API e o worker — é o worker que emite os eventos.)

3. Ligar o OpenLineage no workspace:

   TOKEN=…  # JWT de um admin (POST /api/v1/auth/login)
   curl -X PUT http://localhost:8000/api/v1/openlineage \
     -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
     -d '{"endpoint": "http://localhost:5000", "namespace": "datavault-dev", "enabled": true}'
   

4. Correr um scan (UI ou POST /api/v1/scans). No Marquez (http://localhost:3000, namespace datavault-dev) aparece o job datavault-scan com o run, ligado aos datasets lidos; os facets de qualidade estão no painel do dataset / do run. Via API:

   curl -s http://localhost:5000/api/v1/namespaces/datavault-dev/jobs/datavault-scan/runs | jq '.runs[0].state'
   

5. Diagnóstico: GET /api/v1/openlineage mostra last_event_type, last_event_status e a mensagem do último envio.

Consumir: o lineage do cliente no grafo do BeryX (cartão #130)

O sentido inverso: o Airflow, Spark ou dbt do cliente envia os seus eventos OpenLineage para o BeryX. Cada input → output de um job vira uma aresta de lineage com a origem marcada (job + producer + estado do último run), e o RCA e a análise de impacto passam a ver o pipeline completo — também o que acontece fora do warehouse (S3, Kafka, APIs).

Ligar

1. Criar um token de API (POST /api/v1/auth/tokens ou Definições → Tokens). Tokens são read-only, com uma exceção explícita para POST /openlineage/events.

2. Apontar o transport HTTP do OpenLineage para o BeryX:

   # openlineage.yml (Airflow provider, Spark, dbt-ol)
   transport:
     type: http
     url: https://praxis.lbostuff.tech
     endpoint: api/v1/openlineage/events   # o cliente faz urljoin(url, endpoint)
     auth:
       type: api_key
       apiKey: dvk_…
   

   Airflow por variáveis: AIRFLOW__OPENLINEAGE__TRANSPORT='{"type": "http", "url": "https://praxis.lbostuff.tech", "endpoint": "api/v1/openlineage/events", "auth": {"type": "api_key", "apiKey": "dvk_…"}}'. Para enviar ao Marquez e ao BeryX, usar o transport composite.

3. GET /api/v1/openlineage/lineage mostra os nós recebidos (casados com um dataset ou externos) e as arestas por job.

O que acontece a cada evento

• Entre datasets nossos, a ligação entra no lineage como aresta openlineage — também quando passa por nós externos (pedidos → s3://stage/pedidos → relatorio dá pedidos → relatorio, até 4 nós externos). Impacto, consumidores dbt, Sankey e propagação de sensibilidade passam a vê-la.
• Dedupe: uma aresta declared, parsed (SQL das views) ou dbt do mesmo par ganha — não se cria a openlineage. A inferred (convenção de nomes) do mesmo par é substituída pela observada (guarda a coluna via_column) e os scans seguintes não a recriam.
• RCA: a explicação de uma anomalia traz external_upstream — fontes fora do warehouse com o job que as liga e o estado do último run; um job em FAIL aparece primeiro. O relatório de diagnóstico IA recebe o mesmo (só nomes e estados).
• Impacto (GET /issues/{id}/impact): external lista os destinos a jusante fora do warehouse (de qualquer dataset afetado) e o summary acrescenta «Fora do warehouse (OpenLineage): N destinos — …».
• Um scan que descobre tabelas novas volta a casar os nós externos (POST /openlineage/lineage/refresh força-o). DELETE /openlineage/lineage (admin) esquece tudo o que foi recebido.
• Limites: 2 MB e 100 eventos por pedido, 200 datasets por lado, 2 000 arestas por evento. Eventos repetidos (START/COMPLETE, retries) são idempotentes; eventos fora de ordem não recuam o estado do run.

Fora do âmbito (v1)

• UI de configuração (só API por agora).
• Ler lineage de um servidor Marquez/DataHub (pull) — o consumo é por push dos eventos.
• Facets por coluna (columnMetrics vai vazio) e lineage de colunas (columnLineage recebido é ignorado).
• Mapeamento manual de um nó externo para um dataset.