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

ОБЛАСТЬ ПРИМЕНЕНИЯ: 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

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

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

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"). Код оценки требует, чтобы обработчик канала изменений имел возможность полностью инициализировать документы арендыContainer.

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

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 (.NET все версии и удаления)
• Примеры использования на GitHub (Java)
• Дополнительные примеры на GitHub

Следующие шаги

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

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