Använda ändringsflödesestimatorn

Artikel
07/19/2023

GÄLLER FÖR: NoSQL

Den här artikeln beskriver hur du kan övervaka förloppet för dina instanser av ändringsflödesprocessorn när de läser ändringsflödet.

Varför är övervakningen av förloppet viktig?

Ändringsflödesprocessorn fungerar som en pekare som går framåt i ändringsflödet och levererar ändringarna till en delegatimplementering.

Distributionen av ändringsflödesprocessorn kan bearbeta ändringar med en viss hastighet baserat på dess tillgängliga resurser, till exempel CPU, minne, nätverk och så vidare.

Om den här hastigheten är långsammare än den hastighet med vilken dina ändringar sker i Azure Cosmos DB-containern börjar processorn släpa efter.

Genom att identifiera det här scenariot kan du förstå om vi behöver skala distributionen av ändringsflödesprocessorn.

Som push-modell för automatiska meddelanden

Precis som ändringsflödesprocessorn kan ändringsflödesberäknaren fungera som en push-modell. Skattaren mäter skillnaden mellan det senast bearbetade objektet (definierat av tillståndet för lånecontainern) och den senaste ändringen i containern och push-överför det här värdet till ett ombud. Intervallet för mätningen kan också anpassas med ett standardvärde på 5 sekunder.

Om till exempel ändringsflödesprocessorn definieras så här:

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

Rätt sätt att initiera en skattare för att mäta att processorn skulle använda GetChangeFeedEstimatorBuilder så här:

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

Där både processorn och skattaren delar samma leaseContainer och samma namn.

De andra två parametrarna är ombudet, som får ett tal som representerar hur många ändringar som ska läsas av processorn och det tidsintervall som du vill att mätningen ska utföras med.

Ett exempel på ett ombud som tar emot uppskattningen är:

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

Du kan skicka den här uppskattningen till din övervakningslösning och använda den för att förstå hur dina framsteg fungerar över tid.

Som en detaljerad uppskattning på begäran

Till skillnad från push-modellen finns det ett alternativ som gör att du kan få uppskattningen på begäran. Den här modellen innehåller också mer detaljerad information:

Den beräknade fördröjningen per lån.
Instansen äger och bearbetar varje lån, så att du kan identifiera om det finns ett problem på en instans.

Om ändringsflödesprocessorn definieras så här:

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

Du kan skapa uppskattningen med samma lånekonfiguration:

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

Och när du vill ha det, med den frekvens du behöver, kan du få den detaljerade uppskattningen:

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

Var ChangeFeedProcessorState och en innehåller låne- och fördröjningsinformationen, och även vem som är den aktuella instansen som äger den.

Beräkningsdistribution

Ändringsflödesberäknaren behöver inte distribueras som en del av ändringsflödesprocessorn eller vara en del av samma projekt. Vi rekommenderar att du distribuerar uppskattningen på en oberoende och helt annan instans än dina processorer. En enskild beräkningsinstans kan spåra förloppet för alla lån och instanser i distributionen av ändringsflödesprocessorn.

Varje uppskattning förbrukar enheter för begäranden från dina övervakade och leasade containrar. En frekvens på 1 minut däremellan är en bra startpunkt, desto lägre frekvens, desto högre förbrukas enheter för programbegäran.

Det angivna exemplet representerar ett Java-exempelprogram som visar implementeringen av ändringsflödesprocessorn med uppskattningen av fördröjningen i bearbetningen av ändringsflödeshändelser. I programmet - dokument infogas i en container ("feedcontainern") och under tiden hämtar en annan arbetstråd eller arbetsprogram infogade dokument från feedcontainerns ändringsflöde och arbetar på dem på något sätt.

Ändringsflödesprocessorn skapas och startas så här:

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

Fördröjningskontroll av ändringsflödesprocessor kan utföras i ett separat program (till exempel en hälsoövervakare) så länge samma indatacontainer (feedContainer och leaseContainer) och exakt samma låneprefix (CONTAINER_NAME + "-lease") används. Beräkningskoden kräver att ändringsflödesprocessorn hade möjlighet att helt initiera leaseContainer-dokumenten.

Beräknaren beräknar den ackumulerade fördröjningen genom att hämta det aktuella tillståndet för ändringsflödesprocessorn och summera de uppskattade fördröjningsvärdena för varje händelse som bearbetas:

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

Den totala fördröjningen bör inledningsvis vara noll och slutligen vara större eller lika med antalet dokument som skapats. Det totala fördröjningsvärdet kan loggas eller användas för ytterligare analys, vilket gör det möjligt att övervaka prestanda för ändringsflödesbearbetningen och identifiera eventuella flaskhalsar eller fördröjningar i systemet:

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

Ett exempel på ett ombud som tar emot ändringar och hanterar dem med en fördröjning är:

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

Ytterligare resurser

Nästa steg

Du kan nu fortsätta att lära dig mer om ändringsflödesprocessorn i följande artiklar:

Översikt över ändringsflödesprocessor