Histórico da Página

SOBRE ESTA PÁGINA

Aqui você encontrará as informações necessárias para implementar o Smartlink - Upstream Data Sharing v2 (gRPC) utilizado para compartilhar os dados do ERP com os apps da plataforma TOTVS Apps.

Como requisito para compreensão deste documento é importante conhecer o protocolo de comunicação gRPC. Saiba mais: https://grpc.io/

ÍNDICE

01.ENVIO POR PACOTES (BATCHS)

Nesta versão, a sincronização de dados entre o ERP e a Carol irá seguir o conceito de envio por pacotes (batchs). Cada pacote possui um identificador que será utilizado para rastreio dos dados conforme são enviados pelo Smartlink, para a Carol e por fim até ao app.

1.1 Pacote (Batch)

Toda vez que o ERP rodar um job de envio de dados, esse envio será definido como um pacote de envio de dados. O pacote terá um identificador sequencial em int64 que será o timestamp de inicio do range de datas do delta (Unix Timestamp: segundos percorridos desde 1 de janeiro de 1970).

O batch pode ser composto de múltiplas mensagens que irão conter os registros das tabelas que deverão ser enviadas para o serviço de ingestão utilizando o protocolo de comunicação gRPC.

1.2 Mensagem (Table)

Cada mensagem corresponde a um conjunto de registros de uma determinada tabela do ERP e devem ser numeradas em ordem crescente, iniciando do numero 1 para cada pacote. É obrigatório que a sequência seja respeitada, não havendo envios de mensagens repetidas (1,2,2,3) ou com a ordem pulada (1,2,4).

Além dos registros a mensagem também possui a definição do schema que será utilizado para definir a chave primária dos registros na Carol.

1.2.1 Schema

O schema deve conter uma ou mais colunas listadas para compor a chave primária da tabela. Essa chave é importante pois ela será utilizada para definir se um registro deverá ser adicionado ou atualizado na Carol.

Outra propriedade importante do schema é a propriedade flexible = true, utilizada para que a criação das demais colunas na Carol sejam realizadas automaticamente de acordo com os dados enviados. Dessa forma versões diferentes do mesmo ERP que possuírem tabelas com mais ou menos colunas poderão conviver sem conflitos.

1.2.2 Estrutura da mensagem

1.2.2.1 Table

1.2.2.2 TableRow

1.2.2.3 RowColumn

1.2.2.4 Schema

1.2.2.5 SchemaColumn

1.2.2.6 SchemaPrimaryKey

1.3 Sumário

No final do job será necessário enviar um sumário do pacote, informando o numero de mensagens enviadas, o total de registros enviados e o range de datas que o pacote representa.

O sumário é o responsável por duas tarefas importantes:

• Permitir a observabilidade dos pacotes em relação ao número de mensagens enviadas. Permitindo a identificação de mensagens que podem ter sido perdidas e emissão de alertas.
• Serve como um sinalizador para a execução otimizada dos pipelines de dados na Carol. Garantindo que o processamento dos dados será, na maioria das vezes, executado quando pacotes completos estiverem disponíveis.  

1.4 Estrutura básica de um pacote

1.5 Métodos utilizados para envio

O envio por pacotes possui dois principais métodos, Write() que recebe os dados e o schema da tabela e o Summary() que recebe o sumário contendo as informações sobre a quantidade de mensagens assim como intervalo de datas utilizado para selecionar os dados.

1.5.1 Write

Método utilizado para envio da mensagem com os dados e o schema.

1.5.2 Summary

Método utilizado para envio do summary. Deve ser chamado quando todas as mensagens de um determinado pacote foram enviadas pelo método Write.

1.6 Proto

No contexto do gRPC a definição do serviço contendo as chamadas e as estruturas são escritas em arquivos .proto  (mais detalhes em: https://protobuf.dev/overview/).

Segue arquivo proto com as definições:

syntax = "proto3";

option csharp_namespace = "Ingestion.Streaming";

package com.totvs.ingestion.grpc;

import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";

service  TableStream {
    rpc Write(Table) returns (TableInputResponse);
    rpc Summary(BatchSummary) returns (BatchSummaryResponse);
}

message Table {
    string tableName = 1;
    string erp = 2;

    message RowColumn {
        string name = 1;
        string value = 2;
    }

    message TableRow {
        repeated RowColumn columns = 1;
    }

    repeated TableRow rows = 3;

    message Schema {
        message SchemaColumn {
            string name = 1;
            string type = 2;
        }

        bool flexible = 1;
        bool exportData = 2;
        repeated SchemaColumn columns = 3;

        message SchemaPrimaryKey {
            string name = 1;
            repeated string columns = 2;
        }

        SchemaPrimaryKey primaryKey = 4;
    }

    Schema schema = 4;

    int64 batchId = 6;
    int32 messageNumber = 7;
    int32 recordCount = 8;
    bool isBaseLoad = 9;
}

message TableInputResponse {
    bool success = 1;
    repeated string errors = 2;
}

message BatchSummary {
    string erp = 1;
    int64 batchId = 2;
    int32 messageCount = 3;
    int32 recordCount = 4;
    google.protobuf.Timestamp timestampStart = 5;
    google.protobuf.Timestamp timestampEnd = 6;
}

message BatchSummaryResponse {
    bool success = 1;
    repeated string errors = 2;
}

02.MECANISMO DE ENVIO (JOB)

Mecanismo de job que deverá rodar em um período configurável de tempo, executando a sequência de chamadas de APIs descritas a seguir:

1. Registry (eventual)
2. Token (eventual)
3. Entidades (eventual)
4. Write
5. Summary

Ou seja, se todos os mecanismos recomendados forem implementados, será possível em alguns momentos de execução do job, chamar apenas a API Intake da Carol para enviar os dados.

O framework deverá implementar uma rotina de controle dos registros baseado em timestamp, este controle poderá ser nativo/prévio ou habilitado somente para as tabelas que forem retornadas pela chamada da api Entities.

2.1 Registry

Serviço responsável por centralizar todas as URLs acessadas pelo ERP. Não possui autenticação e deve ser a única URL configurada diretamente.

Os serviços cadastrados são retornados usando a seguinte estrutura:

[{
    "service": "rac-token",
    "endpoints": [
        {
            "version": "1",
            "address": "https://admin.rac.dev.totvs.app/totvs.rac/connect/token"
        }
    ]
}]

2.2 Autenticação

As chamadas disponíveis na solução (Write e Summary) são autenticadas pelo RAC. Cada cliente provisionado recebe um client id e um client secret que são utilizados para gerar o token de serviço onde além de autenticar também é utilizado para identificar o cliente.

Segue um exemplo da chamada para geração do token:

• A url responsável pela geração do token de serviço para o tenant deve ser recuperada pela chave rac-token na listagem dos serviços do Registry (2.1)

curl --location 'https://admin.rac.dev.totvs.app/totvs.rac/connect/token' \
--header 'Content-Type: application/x-www-form-urlencoded' \
--data-urlencode 'grant_type=client_credentials' \
--data-urlencode 'scope=authorization_api' \
--data-urlencode 'client_id=xxxxxxxxx' \
--data-urlencode 'client_secret=xxxxxxxxx'

2.3 Entidades

As entidades consistem em uma parte fundamental do processo, elas determinam todas as entidades que o ERP deve subir de acordo com os produtos contratados e o ERP utilizado.

Quando o ERP inicia um ciclo de envio, após gerar as credenciais conforme o passo 2.2, é necessário fazer um solicitação para a API de entidades, onde serão retornadas no seguinte formato

• A url responsável pela geração do token de serviço para o tenant deve ser recuperada pela chave provisioning-carol-definitions-entities na listagem dos serviços do Registry (2.1)

{
    "uploadVersion": 2,
    "uploadVersionInfo": "Gestão de Upload de Dados via gRPC",
    "queries": [
        {
            "table": "SE4",
            "filter": null,
            "forceReload": false,
            "metadata": null
        },
        {
            "table": "FKD",
            "filter": " FKD_DTBAIX > '20200101' ",
            "forceReload": false,
            "metadata": null
        },
        {
            "table": "CTT",
            "filter": null,
            "forceReload": false,
            "metadata": null
        }
    ]
}

2.3.1 Migração progressiva

Para habilitar uma migração progressiva dos clientes é necessário enviar no no endpoint provisioning-carol-definitions-entities, o header http x-upload-version,esse header vai nos informar qual versão da subida de dados é suportada pela versão do ERP instalada no cliente.

• Versão 1: Utiliza rest para enviar direto para a Carol (x-upload-version: 1)
• Versão 2: Utiliza o framework gRPC para envio e permite observabilidade (x-upload-version: 2)

2.3.2 Características:

1. Filtro de dados (filter): clausula SQL ANSI obrigatória para entidades que possuem dados históricos ou de movimentação, atualmente a plataforma trabalha com dados no máximo até de 2 anos para trás. Deve ser utilizado sempre que o ciclo de envio estiver enviando a tabela pela primeira vez (carga inicial) ou quando uma solicitação de recarga dessa tabela (force reload) seja sinalizada para o ERP.

2. Recarga de uma tabela específica (forceReload): é um campo do tipo booleano que indicará para o ERP efetuar a recarga desta tabela. Uma vez que a recarga foi solicitada, na próxima chamada feita para o serviço de Entities ela virá como 'true' para sinalizar o reenvio e as chamadas subsequentes irão trazer ela imediatamente como 'false'. Recurso importante para tratar cenários com problema sem a necessidade de acessar o ambiente do cliente.

   1. Quando efetuar uma recarga de dados é obrigatório que as mensagens referentes as tabelas envolvidas na recarga sejam enviadas com a flag isBaseLoad igual a true, para evitar que grandes quantidades de dados entrem na mesma fila que os dados incrementais (delta) gerados no durante uso das funcionalidades que afetam dados no ERP.
3. Campo genérico para utilização conforme a necessidade de cada ERP (metadata): Campo do tipo texto livre para cadastro de qualquer informação que possa ser utilizada para auxiliar os ERPs no ciclo de envio para cada tabela.

2.4 Tratamento de erros para a rotina de envio

• A rotina deverá trabalhar sob o controle de transação READ_COMMITED, não pode realizar leitura suja de registros, ou seja, registros gerados por transações que ainda não foram commitadas na base.
• Caso o framework permita a remoção física de registros, deverá ser implementada uma rotina para garantir a notificação da deleção. Pois no caso de delete lógico o registro é enviado à Carol com o atributo deleted = true.
• Somente poderão subir para a Carol, tabelas que possuam chave primária, seja ela simples ou composta.
• Controle de Overlapping - sobreposição do timestamp de início de execução para garantir que transações lentas não sejam perdidas.
• Thread Carga Inicial - thread isolada para execução de carga inicial de uma tabela nova ou de um forceReload quando o cliente já estiver em delta.