Utilice el estimador de flujo de cambios

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 supervisar el progreso?

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 flujo de cambios puede procesar los cambios a un ritmo específico en función de sus 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 comienza a retardar.

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

Implementar el estimador de fuente de cambios

.NET

Como modelo de empuje para 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 estimador mide la diferencia entre el último elemento procesado (definido por el estado del contenedor de arrendamientos) y el último cambio en el contenedor, y envía este valor a un delegado. El intervalo en el que se toma la medida también se puede personalizar con un valor predeterminado de 5 segundos.

Por ejemplo, si el procesador de fuente de cambios usa el modo de versión más reciente y se define de esta 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 estimador para medir ese procesador estaría usando GetChangeFeedEstimatorBuilder de la siguiente manera:

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

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

Los otros dos parámetros son el delegado, que recibe 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 tome esta medida.

Un ejemplo de 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 enviar esta estimación a la solución de supervisión y usarla para comprender cómo se comporta el progreso con el tiempo.

Como estimación detallada a petición

A diferencia del modelo push, hay una alternativa que le permite obtener la estimación bajo demanda. 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 flujo 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 contiene la información de concesión y retraso, y también quién es la instancia actual que lo posee.

Implementación del estimador

No es necesario implementar el estimador de fuente de cambios como parte del procesador de fuente de cambios ni formar parte del mismo proyecto. Se recomienda implementar el estimador en una instancia independiente 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 consume 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

En este ejemplo se utiliza el procesador de 'change feed' en el modo de última versión, estimando el retraso en el procesamiento de eventos del 'change feed'. 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 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, por último, debe ser mayor o igual que el número de documentos creados. El valor total de retraso se puede registrar o usar para un análisis posterior, 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 los cambios y los controla con un 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()");
    };
}

Modos de fuente de cambios admitidos

El estimador de fuente de cambios se puede usar para modo de versión más reciente y todas las versiones y elimina el modo. En ambos modos, no se garantiza que la estimación proporcionada sea un recuento exacto de cambios pendientes en el proceso.

Recursos adicionales

• SDK de Azure Cosmos DB
• Ejemplos de uso en GitHub (versión más reciente de .NET)
• Ejemplos de uso en GitHub (.NET todas las versiones y eliminaciones)
• Ejemplos de uso en GitHub (Java)
• Ejemplos adicionales en GitHub

Pasos siguientes

Ahora puede continuar para obtener más información sobre el procesador de fuente de cambios en el siguiente artículo:

• Información general sobre el procesador de flujo de cambios