Использование средства оценки канала изменений

ОБЛАСТЬ ПРИМЕНЕНИЯ: NoSQL

В этой статье описывается, как можно отслеживать ход выполнения экземпляров обработчика канала изменений, считывающих канал изменений.

Почему процесс отслеживания хода выполнения важен?

Обработчик канала изменений действует как указатель, который перемещается по каналу изменений и доставляет изменения в реализацию делегата.

Экземпляр развертывания обработчика канала изменений может обрабатывать изменения с определенной скоростью на основе доступных ресурсов, таких как ЦП, память, сеть и т. д.

Если эта скорость медленнее, чем скорость, с которой происходят изменения в контейнере Azure Cosmos DB, ваш процессор начнет отставать.

Определение этого сценария помогает понять, нужно ли масштабировать развертывание обработчика канала изменений.

Реализация средства оценки канала изменений

.NET

В качестве модели отправки автоматических уведомлений

Как и обработчик канала изменений, средство оценки канала изменений может использоваться как модель отправки уведомлений. Средство оценки будет измерять разницу между последним обработанным элементом (определяемым состоянием контейнера аренд) и последним изменением в контейнере и передавать это значение делегату. Интервал, с которым производится измерение, также можно настроить по умолчанию на 5 секунд.

Например, если обработчик канала изменений определен таким образом:

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

Правильным способом инициализации средства оценки для измерения этого обработчика будет использование GetChangeFeedEstimatorBuilder следующим образом:

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

Здесь и обработчик, и средство оценки имеют одно значение leaseContainer и одно и то же имя.

Два других параметра — это делегат, который получит число, представляющее, сколько изменений ожидают чтения обработчиком, и интервал времени, в течение которого вы хотите, чтобы это измерение было выполнено.

Пример делегата, который получает оценку:

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

Вы можете отправить эту оценку в решение для мониторинга и использовать ее, чтобы понять, как ход выполнения меняется с течением времени.

В качестве подробной оценки по запросу

Помимо модели отправки, существует альтернативный вариант, позволяющий получить оценку по запросу. Эта модель также предоставляет более подробные сведения:

• Предполагаемое отставание на аренду.
• Экземпляр владеет арендой и обрабатывает ее, что позволяет выяснить, есть ли в экземпляре проблемы.

Например, если обработчик канала изменений определен таким образом:

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

Вы можете создать оценщик с той же конфигурацией аренды:

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

И при необходимости можно получить подробную оценку с требуемой частотой:

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

Каждый объект ChangeFeedProcessorState будет содержать сведения об аренде и отставании, а также о текущем экземпляре-владельце.

Развертывание оценщика

Средство оценки канала изменений не нужно развертывать как часть обработчика канала изменений. Оно также не должно быть частью того же проекта. Рекомендуется развернуть оценщик на независимом экземпляре, совершенно отличном от процессоров. Один экземпляр средства оценки может отслеживать ход выполнения всех аренд и экземпляров в развертывании обработчика канала изменений.

Каждая оценка будет использовать единицы запроса из отслеживаемых контейнеров и контейнеров аренды. Частота между 1 минутой является хорошей отправной точкой: чем ниже частота, тем выше использованные единицы запроса.

Java

В приведенном примере представлен пример приложения Java, демонстрирующий реализацию обработчика канала изменений с оценкой задержки при обработке событий канала изменений. В приложении документы вставляются в один контейнер ("контейнер веб-канала"), а другой рабочий поток или приложение рабочей роли извлекают вставленные документы из канала изменений контейнера веб-канала и каким-то образом работают с ними.

Обработчик канала изменений создается и запускается следующим образом:

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

Проверка задержки обработчика канала изменений может выполняться в отдельном приложении (например, мониторе работоспособности), если используются те же контейнеры ввода (feedContainer и leaseContainer) и одинаковый префикс аренды (CONTAINER_NAME + "-lease"). Код оценщика требует, чтобы обработчик канала изменений имел возможность полностью инициализировать документы leaseContainer.

Оценщик вычисляет накопленные задержки, извлекая текущее состояние обработчика канала изменений и суммируя предполагаемые значения задержки для каждого обрабатываемого события:

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

Общая задержка изначально должна быть нулевой, а затем должна быть больше или равна количеству созданных документов. Значение общей задержки можно регистрировать или использовать для дальнейшего анализа, что позволяет отслеживать производительность обработки канала изменений и выявлять любые потенциальные узкие места или задержки в системе:

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

Пример делегата, который получает изменения и обрабатывает их с запаздыванием:

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

Дополнительные ресурсы

• Azure Cosmos DB
• Примеры использования на GitHub (.NET)
• Примеры использования на GitHub (Java)
• Дополнительные примеры на GitHub

Дальнейшие действия

Вы можете продолжить знакомство с обработчиком канала изменений, перейдя к следующим статьям:

• Обработчик канала изменений в Azure Cosmos DB