Uso del calculador de la fuente de cambios

SE APLICA A: NoSQL

En este artículo se describe cómo puede supervisar el progreso de las instancias del procesador de la fuente de cambios a medida que leen la fuente de cambios.

¿Por qué es importante el progreso de la supervisión?

El procesador de la fuente de cambios actúa como un puntero que avanza por la fuente de cambios y entrega los cambios a una implementación de delegado.

La implementación del procesador de la fuente de cambios puede procesar los cambios a una velocidad determinada en función de los recursos disponibles, como la CPU, la memoria, la red, etc.

Si esta velocidad es más lenta que la velocidad a la que se producen los cambios en el contenedor de Azure Cosmos DB, el procesador comenzará a retrasarse.

La identificación de este escenario ayuda a comprender si es necesario escalar la implementación del procesador de la fuente de cambios.

Implementación del calculador de la fuente de cambios

.NET

Como modelo de inserción para las notificaciones automáticas

Al igual que el procesador de la fuente de cambios, el estimador de la fuente de cambios funciona como un modelo de inserción. El calculador medirá la diferencia entre el último elemento procesado (definido por el estado del contenedor de concesiones) y el cambio más reciente en el contenedor, e insertará este valor en un delegado. El intervalo en el que se toma la medida también puede personalizarse con un valor predeterminado de 5 segundos.

Por ejemplo, si el procesador de la fuente de cambios se define de la siguiente manera:

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 manera correcta de inicializar un calculador para medir ese procesador sería usar GetChangeFeedEstimatorBuilder como se explica a continuación:

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

Donde tanto el procesador como el calculador comparten el mismo leaseContainer y el mismo nombre.

Los otros dos parámetros son el delegado, que recibirá un número que representa cuántos cambios están pendientes de ser leídos por el procesador y el intervalo de tiempo en el que desea que se realice esta medida.

Un ejemplo de un delegado que recibe la estimación es:

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

Puede enviarla a su solución de supervisión y usarla para saber cómo se comporta el progreso con el tiempo.

Como una estimación detallada a petición

A diferencia del modelo de inserción, hay una alternativa que le permite obtener la estimación a petición. Este modelo también proporciona información más detallada:

• Retardo estimado por concesión.
• Instancia que posee y procesa cada concesión, de modo que puede identificar si hay algún problema en una instancia.

Si el procesador de la fuente de cambios se define de la siguiente manera:

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

Puede crear el estimador con la misma configuración de concesión:

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

Y siempre que quiera, con la frecuencia que necesite, puede obtener la estimación detallada:

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 contendrá la información de la concesión y del retardo, y también quién es la instancia actual propietaria.

Implementación del estimador

No es necesario implementar el calculador de la fuente de cambios como parte del procesador de la fuente de cambios ni que formen parte del mismo proyecto. Se recomienda implementar el estimador en una instancia independiente y completamente diferente de los procesadores. Una sola instancia del estimador puede realizar un seguimiento del progreso de todas las concesiones e instancias de la implementación del procesador de fuente de cambios.

Cada estimación consumirá unidades de solicitud de los contenedores supervisados y de concesión. Una frecuencia de 1 minuto entre sí es un buen punto de partida; cuanto menor sea la frecuencia, mayor será la cantidad de unidades de solicitud consumidas.

Java

El ejemplo proporcionado representa una aplicación Java de muestra que demuestra la implementación del procesador de fuente de cambios con la estimación del retraso en el procesamiento de eventos de fuente de cambios. En la aplicación: los documentos se insertan en un contenedor (el "contenedor de fuente") y, mientras tanto, otro subproceso de trabajo o aplicación de trabajo extrae documentos insertados de la fuente de cambios del contenedor de fuente y opera con ellos de alguna manera.

El procesador de fuente de cambios se compila y se inicia de la siguiente manera:

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 comprobación del retraso del procesador de fuente de cambios se puede realizar en una aplicación independiente (como un monitor de estado) siempre que se usen los mismos contenedores de entrada (feedContainer y leaseContainer) y el mismo prefijo de concesión (CONTAINER_NAME + "-lease"). El código del estimador requiere que el procesador de fuente de cambios tenga la oportunidad de inicializar completamente los documentos de leaseContainer.

El estimador calcula el retraso acumulado recuperando el estado actual del procesador de fuente de cambios y sumando los valores de retraso estimados para cada evento que se está procesando:

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

El retraso total inicialmente debe ser cero y finalmente debe ser mayor o igual que el número de documentos creados. El valor total de retraso se puede registrar o usar para su posterior análisis, lo que permite supervisar el rendimiento del procesamiento de la fuente de cambios e identificar posibles cuellos de botella o retrasos en el 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 ejemplo de delegado que recibe cambios y los controla con retraso es:

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

Recursos adicionales

• SDK de Azure Cosmos DB
• Ejemplos de uso en GitHub (.NET)
• Ejemplos de uso en GitHub (Java)
• Ejemplos adicionales en GitHub

Pasos siguientes

Puede obtener más información sobre el procesador de la fuente de cambios en los siguientes artículos:

• Introducción a la fuente de cambios