Usar o avaliador do feed de alterações

APLICA-SE A: NoSQL

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

Por que o progresso do monitoramento é importante?

O processador do feed de alterações funciona como um ponteiro que avança no feed de alterações e entrega as alterações a uma implementação de representante.

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

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

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

Implementar o avaliador do 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 medirá a diferença entre o último item processado (definido pelo estado do contêiner de concessões) e a alteração mais recente no contêiner e efetuará push desse valor a um representante. O intervalo no qual a medição é feita também pode ser personalizado com um valor padrão de 5 segundos.

Por exemplo, se o processador do feed de alterações 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 avaliador para medir esse processador é usar GetChangeFeedEstimatorBuilder desta 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 representante, que receberá um número que representa o número de alterações pendentes para serem lidas pelo processador e o intervalo de tempo no qual você deseja que essa medida seja feita.

Um exemplo de um representante 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);
}

Envie essa estimativa para sua solução de monitoramento e use-a para entender como o 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 retardo estimado por concessão.
• A instância proprietária de cada concessão e que as processa, de modo que você possa identificar se há um problema em uma instância.

Se o processador do feed de alterações 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();

Você poderá criar o avaliador com a mesma configuração de concessão:

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

Além disso, sempre que quiser, 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 conterá as informações de concessão e retardo, além de indicar a instância atual que é proprietária dela.

Implantação do avaliador

O avaliador do feed de alterações não precisa ser implantado como parte do processador do feed de alterações, nem fazer parte do mesmo projeto. É recomendável implantar o avaliador em uma instância independente e completamente diferente dos processadores. Uma única instância do avaliador pode acompanhar o progresso de todas as concessões e instâncias na implantação do processador do feed de alterações.

Cada estimativa consumirá unidades de solicitação de seus contêineres monitorados e de concessão. Uma frequência de 1 minuto entre eles é um bom começo, quanto menor a frequência, mais altas as unidades de solicitação consumidas.

Java

O exemplo fornecido representa um aplicativo Java de exemplo que demonstra a implementação do Processador do Feed de Alterações com a estimativa do lag no processamento de eventos do feed de alterações. No aplicativo, os documentos estão sendo inseridos em um contêiner (o "contêiner de feeds") e, enquanto isso, outro thread de trabalho ou aplicativo de trabalho está extraindo os documentos inseridos do Feed de Alterações do contêiner de feeds e operando neles de alguma forma.

O Processador do Feed de Alterações é construído e iniciado desta 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 lags do Processador do Feed de Alterações pode ser realizada em um aplicativo separado (como um monitor de integridade), desde que sejam usados os mesmos contêineres de entrada (feedContainer e leaseContainer) e exatamente o mesmo prefixo de concessão (CONTAINER_NAME + "-lease"). O código do estimador exige que o Processador do Feed de Alterações tenha tido a oportunidade de inicializar totalmente os documentos do leaseContainer.

O estimador calcula o lag acumulado recuperando o estado atual do Processador do Feed de Mudanças e somando os valores de lag 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 lag total inicialmente deve ser zero e, por fim, deve ser maior ou igual ao número de documentos criados. O valor total do lag 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 um representante que recebe alterações e lida com elas com um lag é:

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()");
    };
}

Recursos adicionais

• SDK do Azure Cosmos DB
• Exemplos de uso no GitHub (.NET)
• Exemplos de uso no GitHub (Java)
• Exemplos adicionais sobre o GitHub

Próximas etapas

Agora continue para saber mais sobre o processador do feed de alterações nos seguintes artigos:

• Visão geral do processador do feed de alterações