Usar o estimador do feed de alterações

Este artigo descreve como você pode monitorar o progresso das instâncias do processador de feed de alterações enquanto elas leem o feed de alterações.

Por que o monitoramento do progresso é importante?

O processador do feed de alterações atua como um ponteiro que avança pelo feed de alterações e entrega as alterações a uma implementação de delegado.

A implantação do processador do feed de alterações pode processar mudanças a uma velocidade específica com base nos seus recursos disponíveis, como CPU, memória, rede e assim por diante.

Se essa taxa for mais lenta do que a taxa em que as alterações ocorrem no contêiner do Azure Cosmos DB, o processador começará a ficar para trás.

Identificar esse cenário ajuda a entender se precisamos dimensionar nossa implantação do processador de feed de alterações.

Implementar o estimador de feed de alterações

.NET

Como um modelo de push para notificações automáticas

Assim como o processador do feed de alterações, o avaliador do feed de alterações pode funcionar como um modelo de push. O avaliador mede a diferença entre o último item processado (definido pelo estado do contêiner de concessões) e a última alteração no contêiner e envia esse valor para um delegado. O intervalo no qual a medida é tomada também pode ser personalizado com um valor padrão de 5 segundos.

Por exemplo, se o processador do feed de alterações estiver usando o modo de versão mais recente e for definido assim:

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

A maneira correta de inicializar um estimador para medir esse processador seria usar GetChangeFeedEstimatorBuilder da seguinte forma:

ChangeFeedProcessor changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimatorBuilder("changeFeedEstimator", Program.HandleEstimationAsync, TimeSpan.FromMilliseconds(1000))
    .WithLeaseContainer(leaseContainer)
    .Build();

Em que o processador e o avaliador compartilham o mesmo leaseContainer e o mesmo nome.

Os outros dois parâmetros são o delegado, que recebe um número que representa quantas alterações estão pendentes para serem lidas pelo processador e o intervalo de tempo no qual você deseja que essa medida seja realizada.

Um exemplo de um delegado que recebe a estimativa é:

static async Task HandleEstimationAsync(long estimation, CancellationToken cancellationToken)
{
    if (estimation > 0)
    {
        Console.WriteLine($"\tEstimator detected {estimation} items pending to be read by the Processor.");
    }

    await Task.Delay(0);
}

Você pode enviar essa estimativa para sua solução de monitoramento e usá-la para entender como seu progresso está se comportando ao longo do tempo.

Como uma estimativa detalhada sob demanda

Em contraste com o modelo de push, há uma alternativa que permite obter a estimativa sob demanda. Esse modelo também fornece informações mais detalhadas:

• O atraso estimado por concessão.
• A instância que possui e gerencia cada aluguel, para que você possa identificar se há um problema em determinada instância.

Se o processador de fluxo de alterações estiver definido assim:

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

Você pode criar o estimador com a mesma configuração de arrendamento.

ChangeFeedEstimator changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimator("changeFeedEstimator", leaseContainer);

E sempre que desejar, com a frequência necessária, você poderá obter a estimativa detalhada:

Console.WriteLine("Checking estimation...");
using FeedIterator<ChangeFeedProcessorState> estimatorIterator = changeFeedEstimator.GetCurrentStateIterator();
while (estimatorIterator.HasMoreResults)
{
    FeedResponse<ChangeFeedProcessorState> states = await estimatorIterator.ReadNextAsync();
    foreach (ChangeFeedProcessorState leaseState in states)
    {
        string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
        Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
    }
}

Cada ChangeFeedProcessorState contém as informações de locação e atraso, além de quem é a atual instância que a detém.

Implantação do estimador

O estimador do feed de alterações não precisa ser implantado junto com o processador do feed de alterações nem ser parte do mesmo projeto. Recomendamos implantar o avaliador em uma instância independente de seus processadores. Uma única instância do estimador pode acompanhar o progresso de todas as locações e instâncias na implantação do processador de fluxo de alterações.

Cada estimativa consome unidades de solicitação de seus contêineres monitorados e de concessão. Uma frequência de 1 minuto entre as requisições é um bom ponto de partida; quanto menor a frequência, maior o consumo de unidades de solicitação.

Java

Este exemplo usa o processador de feed de alterações na versão mais recente, juntamente com a estimativa do atraso no processamento dos eventos do feed de alterações. No aplicativo - documentos estão sendo inseridos em um contêiner (o "contêiner de feed"), e, enquanto isso, outro thread de execução ou aplicativo de trabalho está retirando documentos inseridos do feed de alterações do contêiner de feed e processando-os de alguma forma.

O processador de feed de alterações é criado e iniciado da seguinte forma:

ChangeFeedProcessor changeFeedProcessorMainInstance = new ChangeFeedProcessorBuilder()
    .hostName("SampleHost_1")
    .feedContainer(feedContainer)
    .leaseContainer(leaseContainer)
    .handleChanges(handleChangesWithLag())
    .buildChangeFeedProcessor();

try {
    changeFeedProcessorMainInstance
        .start()
        .subscribeOn(Schedulers.boundedElastic())
        .timeout(Duration.ofSeconds(10))
        .then(Mono.just(changeFeedProcessorMainInstance)
                  .delayElement(Duration.ofSeconds(10))
                  .flatMap(value -> changeFeedProcessorMainInstance
                      .stop()
                      .subscribeOn(Schedulers.boundedElastic())
                      .timeout(Duration.ofSeconds(10))))
        .subscribe();
} catch (Exception ex) {
    logger.error("Change feed processor did not start and stopped in the expected time", ex);
    throw ex;
}

A verificação de atraso do processador do feed de alterações pode ser executada em um aplicativo separado (como um monitor de integridade), desde que os mesmos contêineres de entrada (feedContainer e leaseContainer) e exatamente o mesmo prefixo de concessão (CONTAINER_NAME + "-lease") sejam usados. O código do estimador requer que o processador do feed de alterações tenha tido a oportunidade de inicializar totalmente os documentos do leaseContainer.

O avaliador calcula o atraso acumulado recuperando o estado atual do processador do feed de alterações e somando os valores de retardo estimados para cada evento que está sendo processado:

Mono<List<ChangeFeedProcessorState>> currentState = changeFeedProcessorMainInstance.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

totalLag.set(0);
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

O atraso total inicialmente deve ser zero e, por fim, deve ser maior ou igual ao número de documentos criados. O valor de retardo total pode ser registrado ou usado para análise posterior, permitindo monitorar o desempenho do processamento do feed de alterações e identificar possíveis gargalos ou atrasos no sistema:

totalLag.set(0);
changeFeedProcessorSideCart.start().block();
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).block();

// Finally, totalLag should be greater or equal to the number of documents created
logger.info("Finally total lag is : {}", totalLag.get());

Um exemplo de delegado que recebe alterações e as manipula com atraso é:

private static Consumer<List<JsonNode>> handleChangesWithLag() {
    return (List<JsonNode> docs) -> {
        logger.info("Start handleChangesWithLag()");

        try {
            Thread.sleep(Duration.ofSeconds(5).toMillis());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        for (JsonNode document : docs) {
            try {
                //Change Feed hands the document to you in the form of a JsonNode
                //As a developer you have two options for handling the JsonNode document provided to you by Change Feed
                //One option is to operate on the document in the form of a JsonNode, as shown below. This is great
                //especially if you do not have a single uniform data model for all documents.
                logger.info("Document received: " + OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                                                                 .writeValueAsString(document));

                //You can also transform the JsonNode to a POJO having the same structure as the JsonNode,
                //as shown below. Then you can operate on the POJO.
                CustomPOJO2 pojo_doc = OBJECT_MAPPER.treeToValue(document, CustomPOJO2.class);
                logger.info("id: " + pojo_doc.getId());

            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }
        logger.info("End handleChangesWithLag()");
    };
}

Modos de feed de alterações com suporte

O avaliador do feed de alterações pode ser usado para o modo de versão mais recente e modo de todas as versões e exclusões. Em ambos os modos, a estimativa fornecida não tem a garantia de ser uma contagem exata de alterações pendentes a serem processadas.

Recursos adicionais

• SDK do Azure Cosmos DB
• Exemplos de uso no GitHub (versão mais recente do .NET)
• Exemplos de uso no GitHub (.NET, todas as versões e exclusões)
• Exemplos de uso no GitHub (Java)
• Exemplos adicionais no GitHub

Próximas etapas

Agora você pode prosseguir para saber mais sobre o processador de feed de alterações no seguinte artigo:

• Visão geral do Processador de Feed de Alterações