Usare lo strumento di valutazione del change feed

Questo articolo descrive come monitorare lo stato di avanzamento delle istanze del processore del feed di modifiche durante la lettura del feed di modifiche.

Perché il monitoraggio dello stato di avanzamento è importante?

Il processore del feed di modifiche funge da puntatore che si sposta in avanti nel feed di modifiche e trasmette le modifiche a un'implementazione delegata.

La distribuzione del processore di feed delle modifiche può elaborare le modifiche a una velocità specifica in base alle risorse disponibili, ad esempio CPU, memoria, rete e così via.

Se questa frequenza è più bassa di quella con cui si verificano le modifiche nel contenitore Azure Cosmos DB, il processore inizia a ritardare.

L'identificazione di questo scenario consente di comprendere se è necessario ridimensionare la distribuzione del processore dei feed di modifiche.

Implementare lo strumento di stima del feed di modifiche

.NET

Come modello push per le notifiche automatiche

Come il processore del feed di modifiche, lo stimatore del feed di modifiche può funzionare come modello push. Lo strumento di stima misura la differenza tra l'ultimo elemento elaborato (definito dallo stato del contenitore di lease) e la modifica più recente nel contenitore, e invia questo valore a un delegato. L'intervallo in cui viene eseguita la misurazione può anche essere personalizzato con un valore predefinito di 5 secondi.

Ad esempio, se il processore del feed di modifiche usa la modalità dell'ultima versione ed è definito come segue:

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

Il modo corretto per inizializzare uno strumento di stima per misurare quel processore sarebbe utilizzare GetChangeFeedEstimatorBuilder in questo modo:

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

Dove il processore e lo strumento di stima condividono lo stesso leaseContainer e lo stesso nome.

Gli altri due parametri sono il delegato, che riceve un numero che rappresenta il numero di modifiche in sospeso da leggere dal processore e l'intervallo di tempo in cui si desidera eseguire questa misurazione.

Un esempio di delegato che riceve la stima è:

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

È possibile inviare questa stima alla soluzione di monitoraggio e usarla per comprendere il comportamento dello stato di avanzamento nel tempo.

Come stima dettagliata su richiesta

A differenza del modello push, esiste un'alternativa che consente di ottenere la stima su richiesta. Questo modello fornisce anche informazioni più dettagliate:

• Il ritardo stimato per contratto di locazione.
• L'istanza che possiede ed elabora ogni lease, in modo da poter identificare se c'è un problema in un'istanza.

Se il processore del feed di modifiche è definito come segue:

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

È possibile creare lo stimatore con la stessa configurazione di locazione:

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

E ogni volta che lo si desidera, con la frequenza necessaria, è possibile ottenere la stima dettagliata:

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

Ognuno ChangeFeedProcessorState contiene le informazioni sul lease e sul ritardo e anche sull'istanza corrente proprietaria.

Distribuzione dello strumento di stima

Il valutatore del feed di modifiche non necessita di essere implementato come parte del processore del feed di modifiche, né deve far parte dello stesso progetto. È consigliabile distribuire lo strumento di stima in un'istanza indipendente dai processori. Una singola istanza di stimatore può tenere traccia dello stato di avanzamento per tutti i contratti e le istanze nella distribuzione del processore di feed di modifiche.

Ogni stima usa unità richiesta dai contenitori monitorati e di lease. Una frequenza compresa tra 1 minuto è un buon punto di partenza, minore è la frequenza, maggiore è l'utilizzo delle unità richiesta.

Java

In questo esempio viene usato il processore del feed di modifiche in modalità versione più recente con la stima del ritardo nell'elaborazione degli eventi del feed di modifiche. Nell'applicazione i documenti vengono inseriti in un contenitore (il "contenitore feed") e nel frattempo un altro thread di lavoro o un'applicazione di lavoro estrae i documenti inseriti dal feed di modifiche del contenitore feed e li elabora in qualche modo.

Il processore del feed di modifiche viene compilato e avviato come segue:

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;
}

Il controllo del ritardo del processore del feed delle modifiche può essere eseguito in un'applicazione separata (ad esempio un monitor di integrità) purché vengano usati gli stessi contenitori di input specificati (feedContainer e leaseContainer) e lo stesso prefisso di leasing CONTAINER_NAME + "-lease". Il codice dell'estimatore richiede che il processore del feed delle modifiche abbia la possibilità di inizializzare completamente i documenti del leaseContainer.

Lo strumento di stima calcola il ritardo accumulato recuperando lo stato corrente del processore del feed di modifiche e sommando i valori di ritardo stimati per ogni evento elaborato:

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

Il ritardo totale inizialmente deve essere zero e infine deve essere maggiore o uguale al numero di documenti creati. Il valore totale del ritardo può essere registrato o utilizzato per ulteriori analisi, consentendo di monitorare le prestazioni dell'elaborazione del feed delle modifiche e di identificare eventuali colli di bottiglia o ritardi nel 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());

Un esempio di delegato che riceve le modifiche e le gestisce con un ritardo è:

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

Modalità di feed di modifiche supportate

Lo stimatore del feed delle modifiche può essere utilizzato sia per la modalità della versione più recente che per la modalità di tutte le versioni ed eliminazioni. In entrambe le modalità, la stima fornita non è garantita come un conteggio esatto delle modifiche in sospeso da elaborare.

Risorse aggiuntive

• SDK di Azure Cosmos DB
• Esempi di utilizzo in GitHub (versione più recente di .NET)
• Esempi di utilizzo in GitHub (.NET tutte le versioni ed eliminazioni)
• Esempi di utilizzo in GitHub (Java)
• Altri esempi su GitHub

Passaggi successivi

È ora possibile scoprire di più sul processore del flusso delle modifiche nell'articolo seguente:

• Panoramica del processore di flusso delle modifiche