Блог

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

Существует демонический процесс, который ежедневно запускает удаление старых данных. Это делается с помощью запроса delete_by_, содержащего идентификатор владельца и диапазон времени, в течение которого данные должны быть удалены. Запрос отправляется в наш эластичный кластер в асинхронном режиме ожидания, потому что теперь мы хотим дождаться завершения задачи. (удаление может занять от 1 минуты до нескольких часов в зависимости от случая)

Теперь мы столкнулись со случаем, когда несколько запросов были запущены с одним и тем же идентификатором владельца. (Мы не можем гарантировать, что этого не произойдет) Это приводит к нескольким задачам в Elastic (равным количеству запросов), поскольку один и тот же запрос выполняется в один и тот же период времени, некоторые задачи начинают создавать множество конфликтов версий, вероятно, из-за того, что документы уже помечены для удаления предыдущей задачей. В то же время мы видим, что наша сеть перегружена вызовами API с разных узлов кластера и на них.

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

Мы поиграли с параметрами conflict и refresh , но ничего не привело к правильному удалению документов и очистке созданных задач в Elastic. Единственное, что сработало, — это установить параметр конфликта abort on conflict равным . Хотя это отменяет все задачи, а не просто пропускает документы с конфликтом. proceed on conflict также это привело к сбою запросов и зависанию кластера на случай, если мы увеличим количество идентичных запросов.

Таким образом, у нас остается 2 вопроса:

• Как правильно удалить большое количество данных с помощью запроса, если вы не можете гарантировать, что запросы не перекрываются.
• Почему Elasticsearch не может обрабатывать несколько задач delete_by_query, не застревая в бесконечном цикле создания новых задач снова и снова.

Используемая версия Elastic составляет 6,8

Упрощенный фрагмент кода в Go с использованием gopkg.in/olivere/elastic.v6:

 for i := 0; i < 1000; i   {
    query := elastic.NewBoolQuery().Must(
        elastic.NewTermQuery("ownerId", ownerId),
        elastic.NewRangeQuery("timestamp").Lte(to).Gte(from),
    )
    _, err := elastic.NewDeleteByQueryService(esClient).
        Index("my-index").
        Query(query).
        ProceedOnVersionConflict().
        Refresh("true").
        DoAsync(context.Background())
    if err != nil {
        return err
    }
}