Индексация

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

Компоненты

Indexer сервис - в группе indexer добавлены по одному Consumer на сущность, через который идёт управление частичной индексацией. Индексация запускается с помощью REST API. Сервис позволяет осуществлять частичную индексацию:
- товаров, их остатков и категорий,
- цен и скидок на товары,
- медиа ассетов.
Core сервисы выполняют CRUD операции над сущностями, от которых зависит поиск. Сервисы выпускают события обновления сущностей в брокере Kafka.
Kafka брокер управляет событиями. Для централизации логики генерации событий используется библиотека event-generator.
Cервис-планировщик Scheduler используется для учёта периода валидности цен.
Проиндексированные продукты используются в ElasticSearch для ускорения поиска.

Сценарий

Core сервис обрабатывает входящий запрос на CRUD (Create, Update, Delete) операцию по сущности. После успешного выполнения операции, сервис генерирует новое событие, используя библиотеку event-generator, и публикует его в брокере сообщений Kafka. Если идёт создание / обновление / удаление сразу нескольких сущностей, отдельное событие генерируется на каждое изменение сущности.

Сервис Indexer обрабатывает событие изменения первоначальных данных - генерирует скрипт для обновления индекса, вызывает API для частичной переиндексации в ElasticSearch и сохраняет аудит по проведенной частичной индексации.

Публикация событий

Название топика: <название_сервиса>-change-<название_сущности>. Например, catalog-change-product.

Формат тела сообщения общий для всех событий и структурно состоит из 3 частей:

описание события - тип изменений и время. Описание строгое и для всех событий одинаковое.
описание сущности - ее тип и ID. Описание строгое и для всех событий одинаковое.
полезная нагрузка события. Описание в свободном формате, и задается в зависимости от события. При обновлении полезная нагрузка - это детали произошедшего изменения (какие свойства изменились), контекст события (например, товар, рынок, склад и т.д.). При создании полезная нагрузка - это данные сущности при создании.


      {
  "timestamp": 1721901863, // описание события
  "type": "CREATE or UPDATE or DELETE or READ", // описание события
  "entity": {
    "type": "PRODUCT or PRICE or PROMO, etc.", // описание сущности
    "id": "1" // описание сущности
  },
  "payload": {
     // полезная нагрузка события, в свободном формате
  }
}

Сервис Indexer позволяет настраивать реакции на события. В таких настройках задаются правила:

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

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

Создание нового документа происходит только при добавлении нового товара. При использовании операции POST {index}/_update/{id}, новый документ будет создан, если его еще не существует. Следовательно, создавать вручную документ нет необходимости, можно воспользоваться сценарием обновления документа.
Обновление документа. Соблюдены требования идемпотентности к операциям обновления. Для того, чтобы обновить документ, используется библиотека elasticsearch-java. Класс co.elastic.clients.elasticsearch.core.UpdateRequest позволяет сформировать запрос на обновление документа POST {index}/_update/{id}.
Обновление документов вскользь. Соблюдены требования идемпотентности к операциям обновления. Во время сложных сценариев, например, при изменении имени категории или изменении изображения, вместо обновления документа генерируется скрипт, который содержит правила обновления и фильтр для поиска документов. Сгенерированный скрипт отсылается вместе с запросом POST <index>/_update_by_query. Такие изменения соответствуют команде обновления продуктов по заданным правилам, у которых ID ассета совпадает.
Удаление документа в индексе происходит только при удалении продукта. Вызывается DELETE /{index}/_doc/{id}. Обрабатывается такой сценарий отдельно.

Какие события следует мониторить и отображение их на правила по частичной индексации приведины в таблице ниже. Формат name <- created.name говорит, что отслеживается свойство name в созданной сущности, и значение записывается в свойство документа name.

Событие	Имя сущности	Тип события	Тип операции	Мониторинг и правила по индексации
`catalog-change-product`	`product`	`create`	1	условие: `created.status in (300, 400)` создание новых документов с ID по значениям `created.variants[i].id`: `id <- created.variants[i].id productId <- entity.id catalogId <- created.catalogId name <- created.name description <- created.description attributeValues <- convert (created.variants[i].attributes) assets <- convert (created.variants[i].assets) categories <- convert (created.categoryId + GET /api/catalog/v1/categories)` Закешировать запрос GET /api/catalog/v1/categories, кэш обновляется в консьюмерах событий
		`update`	2	условие: `updated.stats in (300, 400)` обновление документов для каждого варианта с ID по значениям `updated.variants[i].id`: `id <- updated.variants[i].id productId <- entity.id catalogId <- updated.catalogId name <- updated.name description <- updated.description attributeValues <- convert (updated.variants[i].attributes) assets <- convert (updated.variants[i].assets) categories <- convert (updated.categoryId + GET /api/catalog/v1/categories)`
		`delete`	4	удаление документов для каждого варианта с ID по значениям `deleted.variants[i].id`: `id <- deleted.variants[i].id`
`catalog-change-media-asset`	`media_asset`	`update`	3	обновить `assets`во всех документах, в которых задан ассет`id` по значению `entity.id`: `id <- null assets[i].id <- entity.id assets[i].media.url <- updated.media.url assets[i].media.absoluteUrl <- updated.media.absoluteUrl assets[i].media.name <- updated.media.name assets[i].media.type <- updated.media.type assets[i].media.size <- updated.media.size assets[i].media.resolution <- updated.media.resolution assets[i].title <- updated.title assets[i].altAttr <- updated.altrAttr assets[i].description <- updated.description assets[i].sign <- updated.sign assets[i].tags <- updated.tags`
		`delete`	3	обновить `assets`во всех документах, в которых задан ассет `id` по значению `entity.id`: `id <- null assets[i].id <- entity.id assets[i] <- null`
`availability-change-stock`	`stock`	`create`	2	обновить документ с id по значению `created.articleId`, создать новый элемент в `stocks`: `id <- created.articleId stocks[i].id <- entity.id stocks[i].quantity <- created.quantity stocks[i].replenishmentTime <- created.replenishmentTime stocks[i].locationId <- created.locationId stocks[i].locationCode <- created.locationCode stocks[i].locationName <- created.locationName stockStatus <- max (stocks[i].quantity)` Перед реализацией нужно добавить массив stocks. Подробнее тут
		`update`	2	обновить документ с id по значению `updated.articleId`, обновить `stocks`в которых задан ассет `id` по значению `entity.id`: `id <- updated.articleId stocks[i].id <- entity.id stocks[i].quantity <- updated.quantity stocks[i].replenishmentTime <- updated.replenishmentTime stocks[i].locationId <- updated.locationId stocks[i].locationCode <- updated.locationCode stocks[i].locationName <- updated.locationName stockStatus <- max (stocks[i].quantity)`
			2	условие: `previous.articleId != updated.articleId` обновить документ с id по значению `previous.articleId`, удалить элемент в `stocks` для которого задан ассет `id` по значению `entity.id` `id <- previous.articleId stocks[i].id <- entity.id stocks[i] <- null`
		`delete`	2	обновить документ с id по значению `deleted.articleId`, удалить элемент в`stocks`в которых задан ассет `id` по значению `entity.id`: `id <- deleted.articleId stocks[i].id <- entity.id stocks[i] <- null`
`price-change-price`	`price`	`create`	2	условие: `created.active == true and created.validFrom >= now() and created.validTo <= now()` обновить документ с id по значению `created.articleId`, создать или обновить элемент в `prices[currencyId]` по значению `created.currency.id`: `id <- created.articleId prices[currencyId].priceId <- entity.id prices[currencyId].unitPrice <- created.unitPrice prices[currencyId].discountUnitPrice <- calculate (created + promos)`
		`update`	2	условие: `updated.active == true and updated.validFrom >= now() and updated.validTo <= now()` обновить документы с id по значению `updated.articleId`, обновить элемент в `prices` с индексом `currencyId` по значению `updated.currency.id`: `id <- updated.articleId prices[currencyId].priceId <- entity.id prices[currencyId].unitPrice <- created.unitPrice prices[currencyId].discountUnitPrice <- calculate (created + promos)`
			2	условие: `previous.articleId != updated.articleId` обновить документы с id по значению `previous.articleId`, удалить элемент в `prices[currencyId]` для которого задан ассет `priceId` по значению `entity.id` `id <- previous.articleId prices[currencyId].priceId <- entity.id prices[currencyId] <- null`
		`delete`	2	обновить документ с id по значению `deleted.articleId`, удалить элемент в `prices[currencyId]` в которых задан ассет `priceId` по значению `entity.id`: `id <- deleted.articleId prices[currencyId].priceId <- entity.id prices[currencyId] <- null`
`promo-change-promo`	`promo`	`create` `update` `delete`	2	обновить `prices`во всех документах, для которых действует скидка, в которых задан ассет `priceId` по значению `entity.id`. `id <- find (entity.id) prices[currencyId].discountUnitPrice <- calculated discountUnitPrice`

Планировщик

Для учета периода валидности цен используется сервис-планировщик Scheduler. В зависимости от временных границ (validFrom и validTo) создаются задачи на индексацию или удаление цены из индекса.

Отложенная индексация цены учтена:

при полной индексации;
при частичной индексации товаров (создание и удаление товара);
при частичной индексации цен.

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

Сервис, на котором запущен “планировщик”, подписывается на события изменения сущности Price;
При обработке события, сервис дает команду планировщику запланировать задачу, удалить или обновить ее на время, заданное в событии;
Сама запланированная задача выпускает событие о том, что изменилась цена на товаре, передавая новую цену;
Сервис Indexer подписывается на событие, которое создал планировщик, и индексирует новую цену;
Цена со скидкой может также расчитываться через один дополнительный запрос к промо движку в запланированной задаче.

Условия создания задачи

Если active = false то цена не индексируется и задачи не создаются.
Если текущее время < validFrom, создаётся задача на индексацию цены в момент времени validFrom.
Если текущее время < validTo, создаётся задача на удаление цены из индекса в момент времени validTo.

Процесс создания задачи

В сервисе Scheduler создаётся задача по вызову HTTP-эндпоинта POST <хост_сервиса_indexer>/api/indexer/v1/index/scheduled/price на заданное время. В ответе возвращается jobId.
В сервисе Indexer в таблице dc_scheduled_price_indexation создаётся запись с полями:
- job_id — идентификатор созданной задачи;
- price_id — идентификатор цены;
- launch_timestamp — время запуска задачи;
- action — действие (INDEX или DELETE);
- price — данные о цене в формате JSON.

Выполнение задачи

Когда наступает launchTimestamp, сервис Scheduler отправляет HTTP-запрос на эндпоинт <хост_сервиса_indexer>/api/indexer/v1/index/scheduled/price сервиса Indexer, передавая jobId.
Сервис Indexer находит соответствующую запись в таблице dc_scheduled_price_indexation и выполняет действие (INDEX или DELETE) в зависимости от значения action.
После выполнения запись удаляется из таблицы.

Настройки планировщика

Включена кластеризация (с 1 узлом на текущем этапе);
Количество одновременных потоков выполнения полезной нагрузки планировщика — 5;
Размер очереди на выполнение задач - нефиксированный, персистентное хранение описаний задач, триггеров и очередей в БД;
Схема БД для мета-данных инициализируется при запуске сервиса через утилиту Liquibase;
Необходимо вести историю (аудит) выполнения задач. Конфигурация повторных попыток в случае неудач не задана.

Сценарии по частичному обновлению индекса

Компонент Catalog

Создание продукта

Индексация для нового продукта нужна, только если:

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


      {
   "entity": {
      "type": "product",
      "id": id
   }
   "type": "create",
   "payload": ProductDto
}

Обновление продукта