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

Это вторая часть из серии статей об обновлении кластера Elasticsearch без простоев и с минимальным воздействием на пользователей.

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

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

Всё меняется

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

Каждую секунду кластер получает 5000 новых документов и более 10000 обновлений. Всего в нем более 400 млрд документов, которые, вместе с репликами, «весят» более 3,5 ПБ.

Управлять этой горой данных очень непросто, даже если речь идет всего об одном кластере. К счастью, в Elasticsearch есть оптимистическая блокировка — и, смею вас заверить, ее мы использовали по полной. До начала миграции на новый кластер особых проблем с согласованностью в старом кластере не было (по крайней мере, мы ничего о них не знали).

Сценарии работы с обновлениями документов

Документы в нашей системе обновляются по ряду причин. Три распространенных примера: новая активность (например, лайк или комментарий к сообщению в соцсети), аннотации к документу, сделанные клиентом (например, метка или комментарий, прикрепленный к документу) или нормативные требования (приходится скрывать контент от поставщиков или удалять его на основании определенных правил).

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

• Upsert

  • Добавить новый документ или полностью заменить существующий.

  • Обратите внимание, что удаления также проводятся как Upsert'ы. То есть Upsert просто заменяет существующий документ «пустым».

• AddIfNotExists

  • Добавить новый документ, если его еще нет. В противном случае сохранить существующий и отбросить новый.

• SetField

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

• AppendToField / RemoveFromField

  • Для полей массива также поддерживается функция добавления/удаления на уровне полей. Она удобнее SetField при работе с наборами значений. Во всех остальных аспектах семантика ее операций аналогична SetField.

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

Добавление второго кластера

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

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

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

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

Непростая задача.

Архитектура, «заточенная» под согласованность

Почти сразу пришло понимание, что необходимо ограничить количество компонентов с прямым доступом к Elasticsearch. В исходном состоянии миграция на новый кластер была бы слишком сложной и требовала бы абсолютной координации из-за одномоментности переключения. Было решено остановиться на двух таких компонентах: один — для пакетной потоковой передачи данных в больших объемах и один — с REST API.

Главное, что предстояло сделать — убрать версионирование из Elasticsearch, поскольку было бы очень сложно обеспечить применение каждого отдельного обновления в одном и том же порядке в обоих кластерах. Также было решено отказаться от поддержки всех описанных выше операций в компонентах, которые взаимодействуют с Elasticsearch, а вместо этого реализовать две операции «низкого уровня»:

• GetDocument

  • Извлекает документ и номер его версии по идентификатору documentId. Возвращаемую здесь версию необходимо передать UpsertIfLatest.

• UpsertIfLatest

  • Делает Upsert для документа, если версия определена как последняя.

  • Если версия не новее существующей, выдать ошибку и сообщить об этом вызывающей стороне.

Например, высокоуровневая операция SetField может быть реализована так:

fun SetField(String field, String newValue, String docId) {
  var success = false
  while (!success) {
    val document = GetDocument(docId)
    document[field] = newValue
    document.version = GetNextVersion(document.version)
    success = UpsertIfLatest(docId, document)
  }
}

Это означает, что пришлось внедрить сервис документов (Document), который использовал компоненты хранилища для реализации API высокого уровня (AddIfNotExists / SetField / AppendToField / RemoveFromField). Сервис документов создает версионированные Upsert'ы, которые дублируются в компоненты хранилища, работающие с отдельными кластерами. Компоненты хранилища независимо для каждого кластера решают, последняя ли версия у входящей операции Upsert или нет.

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

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

Преимущества архитектуры, «заточенной» под согласованность

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

• Улучшены автоматические наборы тестов и мониторинг согласованности.

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

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

• Архитектуру «двух кластеров» можно использовать и в будущем:

  • например, для новых обновлений кластера;

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

Миграция данных между кластерами

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

Мы рассмотрели множество различных вариантов, от внедрения новых компонентов, которые считывали бы снапшоты Elasticsearch и записывали их на новый кластер, до индексации с удаленного сервера. В итоге было решено экспортировать все данные (1 ПБ) с помощью scan & scroll API и сохранять сжатые пакеты JSON-файлов в хранилище S3. После этого достаточно было пропустить эти JSON-документы через основной пайплайн.

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

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

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

Пиковая пропускная способность индексирования во время миграции составила 130 тыс. документов в секунду. Но, как видно из рисунка 5, по разным причинам ее не удавалось поддерживать длительное время. Кроме того, индексирование пришлось дополнительно замедлить после того, как основная часть запросов конечных пользователей начала направляться в новый кластер (чтобы не рисковать ростом поисковых задержек).

Стоит также отметить, что оценка средней пропускной способности оказалась очень точной. Весь процесс развертывания занял ~ 110 дней с момента начала индексирования до момента, когда все данные оказались на новом кластере.

На этом заканчивается вторая часть серии материалов об обновлении кластера Elasticsearch. Следите за обновлениями: очередная статья будет опубликована на следующей неделе.

P.S.

Читайте также в нашем блоге:

• «Логи в Kubernetes (и не только) сегодня: ожидания и реальность»;

• «elasticsearch-extractor — утилита для извлечения индексов из снапшотов Elasticsearch»;

• «Аварии как опыт #2. Как развалить Elasticsearch при переносе внутри Kubernetes».