Usar o estimador de feed de alterações

APLICA-SE A: NoSQL

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

Por que razão é importante acompanhar os progressos?

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

A implantação do processador de feed de alterações pode processar alterações em 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 acontecem 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 alimentação de alterações

.NET

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

Como o processador de alimentação de alterações, o estimador de alimentação de alterações pode funcionar como um modelo push. O estimador medirá a diferença entre o último item processado (definido pelo estado do contêiner de concessão) e a última alteração no contêiner e enviará esse valor para um delegado. O intervalo em que a medição é feita também pode ser personalizado com um valor padrão de 5 segundos.

Por exemplo, se o processador de alimentação 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();

A maneira correta de inicializar um estimador para medir esse processador estaria usando GetChangeFeedEstimatorBuilder assim:

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

Onde o processador e o estimador compartilham o mesmo leaseContainer e o mesmo nome.

Os outros dois parâmetros são o delegado, que receberá 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 medição seja feita.

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 push, há uma alternativa que permite obter a estimativa sob demanda. Este modelo também fornece informações mais detalhadas:

• A defasagem estimada por locação.
• A instância que possui e processa cada locação, para que você possa identificar se há um problema em uma instância.

Se o processador de alimentação 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 concessão:

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

E sempre que quiser, com a frequência que necessitar, 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 um conterá as informações de concessão e atraso e também quem é a instância atual proprietária dele.

Implantação do estimador

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

Cada estimativa consumirá unidades de solicitação de seus contêineres monitorados e alugados. Uma frequência de 1 minuto no intervalo é um bom ponto de partida, quanto menor a frequência, maior a quantidade de unidades de solicitação consumidas.

Java

O exemplo fornecido representa um aplicativo Java de exemplo que demonstra a implementação do Change Feed Processor com a estimativa do atraso no processamento de eventos de feed de alteração. No aplicativo - os documentos estão sendo inseridos em um contêiner (o "contêiner de alimentação") e, enquanto isso, outro thread de trabalho ou aplicativo de trabalho está puxando documentos inseridos do Change Feed do contêiner de feed e operando neles de alguma forma.

O processador de feed de alteração é construído e iniciado assim:

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 de alimentação 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 o mesmo prefixo de concessão (CONTAINER_NAME + "-lease") sejam usados. O código do estimador requer que o Change Feed Processor tenha tido a oportunidade de inicializar totalmente os documentos do leaseContainer.

O estimador calcula o atraso acumulado recuperando o estado atual do Change Feed Processor e somando os valores de atraso 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());

A defasagem total inicialmente deve ser zero e, finalmente, deve ser maior ou igual ao número de documentos criados. O valor de defasagem total pode ser registrado ou usado para análise posterior, permitindo monitorar o desempenho do processamento de alimentação de mudança 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 delegado que recebe alterações e as lida com um 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()");
    };
}

Recursos adicionais

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

Próximos passos

Agora você pode continuar para saber mais sobre o processador de alimentação de alterações nos seguintes artigos:

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