Usare lo strumento di stima dei feed di modifiche

SI APPLICA A: NoSQL

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

Perché è importante monitorare lo stato di avanzamento?

Il processore di feed di modifiche funge da puntatore che avanza nel feed di modifiche e distribuisce le modifiche a un'implementazione delegata.

La distribuzione del processore di feed di modifiche può elaborare le modifiche a una determinata frequenza 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 inizierà a ritardare.

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

Implementare lo strumento di stima dei feed di modifiche

.NET

Come modello push per le notifiche automatiche

Analogamente al processore di feed di modifiche, lo strumento di stima dei feed di modifiche può funzionare come modello push. Lo strumento misura la differenza tra l'ultimo elemento elaborato (definito dallo stato del contenitore di lease) e la modifica più recente nel contenitore ed effettua il push di 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 di 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();

Il modo corretto per inizializzare uno strumento di stima per misurare il processore consiste nell'uso di GetChangeFeedEstimatorBuilder, come indicato di seguito:

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

In cui sia il processore che lo strumento di stima condividono lo stesso leaseContainer e lo stesso nome.

Gli altri due parametri corrispondono al delegato, che riceverà un numero che rappresenta la quantità di modifiche in sospeso per la lettura da parte del processore e l'intervallo di tempo in cui si vuole che venga eseguita questa misurazione.

Di seguito è riportato un esempio di un 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 lo stato di avanzamento nel corso del 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 lease.
• L'istanza proprietaria e l'elaborazione di ogni lease, per identificare se si verifica un problema in un'istanza.

Se il processore di 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 strumento di stima con la stessa configurazione di lease:

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

Ogni ChangeFeedProcessorState conterrà le informazioni sul lease e sul ritardo e anche sull'istanza proprietaria corrente.

Distribuzione della stima

Non è necessario distribuire lo strumento di stima dei feed di modifiche come parte del processore di feed di modifiche né come parte dello stesso progetto. È consigliabile distribuire la stima in un'istanza indipendente e completamente diversa dai processori. Una singola istanza di stima può tenere traccia dello stato di avanzamento per tutti i lease e le istanze nella distribuzione del processore del feed di modifiche.

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

Java

L'esempio fornito rappresenta un'applicazione Java di esempio che illustra l'implementazione del processore dei feed di modifiche 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 esegue il pull dei documenti inseriti dal feed di modifiche del contenitore feed e li utilizza 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 di modifiche può essere eseguito in un'applicazione separata (ad esempio un monitoraggio dell'integrità) purché vengano usati gli stessi contenitori di input (feedContainer e leaseContainer) e lo stesso prefisso di lease (CONTAINER_NAME + "-lease"). Il codice di stima richiede che il processore dei feed di modifiche abbia la possibilità di inizializzare completamente i documenti di leaseContainer.

La 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 alla fine deve essere maggiore o uguale al numero di documenti creati. Il valore di ritardo totale può essere registrato o usato per ulteriori analisi, consentendo di monitorare le prestazioni dell'elaborazione del feed di modifiche e 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()");
    };
}

Risorse aggiuntive

• SDK di Azure Cosmos DB
• Esempi di utilizzo su GitHub (.NET)
• Esempi di utilizzo su GitHub (.Java)
• Altri esempi in GitHub

Passaggi successivi

È ora possibile procedere ad acquisire altre informazioni sul processore di feed di modifiche negli articoli seguenti:

• Panoramica del processore di feed di modifiche