Utilisation de l’estimateur de flux de modification

S’APPLIQUE À : NoSQL

Cet article décrit comment superviser la progression de vos instances de processeur de flux de modification au fur et à mesure qu’elles lisent le flux de modification.

Pourquoi la supervision de la progression est-elle importante ?

Le processeur de flux de modification agit comme un pointeur qui avance dans votre flux de modification et fournit les modifications à une implémentation de délégué.

Votre déploiement de processeur de flux de modification peut traiter les modifications à une vitesse particulière en fonction de ses ressources disponibles, telles que le CPU, la mémoire, le réseau, et ainsi de suite.

Si cette vitesse est inférieure à celle à laquelle vos modifications se produisent dans votre conteneur Azure Cosmos DB, votre processeur commencera à accumuler du retard.

L’identification de ce scénario aide à comprendre si nous devons mettre à l’échelle notre déploiement du processeur de flux de modification.

Utilisation de l’estimateur de flux de modification

.NET

En tant que modèle push pour les notifications automatiques

Comme le processeur de flux de modification, l’estimateur de flux de modification peut fonctionner en tant que modèle Push. L’estimateur mesure la différence entre le dernier élément traité (défini par l’état du conteneur de baux) et la dernière modification dans le conteneur, et il transmet cette valeur à un délégué. Vous pouvez également personnaliser l’intervalle de prise de cette mesure avec une valeur par défaut de cinq secondes.

Par exemple, si votre processeur de flux de modification est défini comme suit :

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

La méthode correcte pour initialiser un estimateur afin de mesurer ce processeur consisterait à utiliser GetChangeFeedEstimatorBuilder comme suit :

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

Où le processeur et l’estimateur partagent le même leaseContainer et le même nom.

Les deux autres paramètres sont le délégué, qui recevra un nombre représentant le nombre de modifications en attente de lecture par le processeur, et l’intervalle de temps auquel vous souhaitez que cette mesure soit prise.

Voici un exemple de délégué qui reçoit l’estimation :

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

Vous pouvez envoyer cette estimation à votre solution de supervision et l’utiliser pour comprendre le fonctionnement de votre progression au fil du temps.

Comme une estimation détaillée à la demande

Contrairement au modèle push, il existe une alternative qui vous permet d’obtenir l’estimation à la demande. Ce modèle fournit également des informations plus détaillées :

• Le décalage estimé par bail.
• L’instance possédant et traitant chaque bail, afin que vous puissiez déterminer s’il existe un problème sur une instance.

Si votre processeur de flux de modification est défini comme suit :

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

Vous pouvez créer l’estimateur avec la même configuration de bail :

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

À chaque fois que vous le souhaitez, avec la fréquence requise, vous pouvez obtenir l’estimation détaillée :

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

Chaque ChangeFeedProcessorState contient les informations relatives au bail et au décalage, ainsi que l’identité de l’instance actuelle qui en est la propriétaire.

Déploiement de l’estimateur

L’estimateur de flux de modification n’a pas besoin d’être déployé dans le cadre de votre processeur de flux de modification, ni de faire partie du même projet. Nous vous recommandons le déploiement de l’estimateur sur une instance indépendante et complètement différente de vos processeurs. Une unique instance de l’estimateur peut suivre la progression de l’ensemble des baux et des instances de votre déploiement du processeur de flux de modification.

Chaque estimation consomme des unités de requête de vos conteneurs surveillés et de bail. Une fréquence de 1 minute entre les deux est un bon point de départ ; une fréquence plus faible augmente le nombre d’unités de requête consommées.

Java

L’exemple fourni représente un exemple d’application Java qui illustre l’implémentation du processeur de flux de modification avec l’estimation du décalage dans un traitement des événements de flux de modification. Dans l’application, les documents sont insérés dans un conteneur (le « conteneur de flux »), et pendant ce temps, un autre thread de travail ou application de travail extrait les documents insérés du flux de modification du conteneur de flux et fonctionne sur ceux-ci d’une manière ou d’une autre.

Le processeur de flux de modification est généré et démarré comme suit :

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

La vérification du décalage du processeur de flux de modification peut être effectuée sur une application distincte (comme un moniteur d’intégrité) aussi longtemps que les mêmes conteneurs d’entrée (feedContainer et leaseContainer) et le même préfixe de bail (CONTAINER_NAME + "-lease") sont utilisés. Le code de l’estimateur exige que le processeur de flux de modification ait eu la possibilité d’initialiser complètement les documents de leaseContainer.

L’estimateur calcule le décalage cumulé en récupérant l’état actuel du processeur de flux de modification et en additionnant les valeurs de décalage estimées pour chaque événement en cours de traitement :

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

Le décalage total doit initialement être égal à zéro et, ensuite, doit être supérieur ou égal au nombre de documents créés. La valeur de décalage totale peut être consignée ou utilisée pour une analyse plus approfondie, ce qui permet de surveiller les performances du traitement du flux de modification et d’identifier des goulots d’étranglement ou des retards potentiels dans le système :

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

Voici un exemple de délégué qui reçoit des modifications et les gère avec un décalage :

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

Ressources supplémentaires

• Azure Cosmos DB SDK
• Exemples d’utilisation sur GitHub (.NET)
• Exemples d’utilisation sur GitHub (Java)
• Exemples supplémentaires sur GitHub

Étapes suivantes

Pour plus d’informations sur le processeur de flux de modification, consultez les articles suivants :

• Vue d’ensemble du processeur de flux de modification