Usar o estimador de fluxo de alterações

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 mede 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 envia 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 usando o modo de versão mais recente e 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 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 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 a pedido

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 responsável por possuir e processar cada locação, permitindo que possas identificar se há um problema em uma instância.

Se o processador de feed de alterações estiver configurado desta forma:

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 contém 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 fluxo de alterações não precisa ser implementado como parte do processador de fluxo de alterações, nem precisa fazer parte do mesmo projeto. Recomendamos implantar o estimador em uma instância independente 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 consome unidades de solicitação dos seus contêineres monitorados e com arrendamento. 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.

Este exemplo usa o processador de feed de alterações em modo de última versão com a estimativa do tempo de atraso no processamento de eventos de feed de alterações. 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 feed de mudança do contêiner de alimentação 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ção 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 processador de alimentação de alterações tenha tido a oportunidade de inicializar totalmente os documentos do leaseContainer.

O estimador calcula o atraso acumulado recuperando o estado atual do processador de alimentação de alterações 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 lida com elas 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()");
    };
}