Отказоустойчивость

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

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

• обеспечение максимальной доступности и надёжности;
• сохранeние прогресса уже выполненных вычислений в рамках исполняющихся операций.

Доступность

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

Планировщик внутри кластера представлен в виде нескольких запущенных процессов, которые, как правило, расположены на разных узлах. Каждый из процессов в цикле пытается захватить блокировку на узел //sys/scheduler/lock в Кипарисе. Процесс, которому удалось захватить блокировку, зачитывает мета-информацию про текущие операции из Кипариса, восстанавливает своё состояние и становится активным — то есть начинает обслуживать пользовательские запросы про операции, обрабатывать хартбиты от exec-нод и так далее.

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

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

• частота, с которой процесс планировщика пытается захватить блокировку;
• таймаут транзакции, в рамках которой захватывается блокировка;
• объём мета-информации про текущие операции — данная величина начинает играть роль в том случае, если в кластере исполняются тысячи операций одновременно и требуется существенное время, чтобы все их прочитать из Кипариса и восстановить состояние планировщика.

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

Надёжность

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

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

Сохранность операций достигается за счет того, что мета-информация об исполняющихся операциях надежно хранится в Кипарисе. Также при запуске операции создаются транзакции для работы с входными и выходными таблицами и файлами, и в рамках этих транзакций берутся блокировки в режиме snapshot. Это позволяет операции продолжать исполнение, если данные таблицы были удалены в ходе работы операции, или если сменится активный планировщик.

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

Сохранение прогресса вычислений

В отличие от real-time компонент кластера, исполнение одной операции требует значительного времени – от единиц минут до нескольких дней в зависимости от объёма обрабатываемых данных, доступной вычислительной квоты и запускаемого кода в джобах. Поэтому требования, предъявляемые к доступности системы обработки данных, достаточно расслабленные: предполагается, что задержки в единицы минут, а в большинстве случаев и в десятки минут, не скажутся значимым образом на SLO систем, которые запускают свои вычисления на YTsaurus кластерах.

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

Состояние операции можно разделить на 3 части:

1. Мета-информация операции: её спецификация, набор Кипарисных транзакций, блокировки входных, выходных таблиц и файлов, использующихся для запуска, различные runtime-настройки операции и текущая стадия операции (она же state). В планировщике фактически реализован конечный автомат, через который проходит каждая операция. Стадия является узлом этого автомата.
2. Состояние контроллера операции. Контроллер операции – это объект, который находится внутри процесса контроллер-агента и отвечает за процесс исполнения операции. Контроллер заведует информацией про то, какие джобы данной операции сейчас выполняются, какие закончились, какие данные уже обработаны, а какие ещё нет, и в целом в данном объекте инкапсулируется вся логика и состояние обработки данных, которые производятся данной операцией.
3. Набор выполняющихся и некоторых уже завершённых джобов на exec-нодах кластера.

Планировщик синхронно записывает мета-информацию операции в соответствующий узел Кипариса, который находится в поддереве //sys/operations. Контроллер операции умеет консистентно сохранять своё состояние в бинарном формате и записывать его в Кипарис в виде файла. Такой блоб называется снепшотом операции. Каждый контроллер-агент периодически строит снепшоты запущенных на нём операций.

Выполняющиеся джобы операции периодически сообщают своё состояние контроллеру операции. Это происходит с помощью хартбитов, которые присылают ноды кластера. Каждая exec-нода отправляет хартбиты со всеми релевантными джобами в каждый из контроллер-агентов.

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

• При выходе из строя exec-ноды кластера, контроллер-агенты и планировщик по истечении небольшого таймаута (обычно равного 1-2 минутам) обнаруживают, что нода перестала присылать хартбиты, и считают все аллокации и соответствующие джобы потерянными, после этого они будут запланированы на другие ноды кластера. Тут стоит отметить, что логически это будут новые джобы взамен потерянных.
• При выходе из строя контроллер-агента, теряется текущее состояние контроллеров операций, которые были запущены на данном агенте. При этом планировщик обнаруживает, что контроллер-агент отключился, и все операции, которые были на него назначены, переносит на другие контроллер-агенты. При назначении на новый агент контроллер операции восстанавливается из последнего сохранённого снепшота. Также по мере обработки хартбитов контроллер операции синхронизирует своё состояние, прочитанное из снепшота, с фактической ситуацией на кластере.
• При выходе из строя активного планировщика, автоматически прекращают работу все контроллер-агенты, подключается новый планировщик, он зачитывает текущую мета-информацию про операции, заново распределяет операции по контроллер-агентам и продолжает работу. Операции при этом восстанавливаются из последних записанных в Кипарис снепшотов.

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

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

Предположим, на ноде выполняется джоб J операции O. Тогда возможна следующая хронология событий:

Момент времени	Событие
X	Джоб J исполняется на ноде, контроллер-агент записал снепшот операции O. В рамках этого снепшота контроллер операции считает, что джоб J находится в процессе исполнения.
X+1	Джоб J завершился и контроллер операции обработал это событие.
X+2	Произошёл ревайв операции O (например, вследствие перезапуска контроллер-агента). После восстановления из снепшота контроллер операции O считает, что джоб J ещё исполняется.
X+3	При обработке хартбита от ноды, на которой исполнялся джоб J, контроллер операции обнаружит, что данного джоба нет среди исполняющихся джобов. При этом он запросит у ноды информацию о всех завершённых джобах в прошлом. При следующем хартбите нода пришлёт контроллеру информацию о том, что джоб J завершился, и контроллер обработает это событие и не будет запускать новый джоб взамен.

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

Резюмируя сказанное:

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

Согласованность состояний планировщика и контроллер-агентов

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

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

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

Контроллер-агенты действуют по следующей схеме:

• Сначала агент устанавливает связь с планировщиком и получает от него транзакцию, в рамках которой этот планировщик стал активным.
• Дальше агент заводит свою транзакцию, вложенную в транзакцию планировщика, и пытается под ней захватить блокировку на узел //sys/controller_agents/instances/<instance_address>/lock. В типичном случае агент не соревнуется с другими процессами за взятие блокировки, но может конфликтовать со своей предыдущей инкарнацией.
• Когда агенту удалось взять данную блокировку, он также становится активным.

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

Транзакция агента выполняет несколько функций:

• Позволяет планировщику управлять данным агентом и обеспечивает привязанность агента к текущему активному планировщику. С помощью этой транзакции планировщик может прерывать работу контроллер-агента в разных ситуациях, например, если с агентом прервалась связь. Для этого достаточно отменить транзакцию данного агента. Кроме того, если транзакция планировщика прервалась, то прерываются и все транзакции агентов.
• Используется агентом при взаимодействии с Кипарисом как пререквизитная. Это позволяет гарантировать, что изменять состояние операции в Кипарисе может только текущий активный агент, на который назначена данная операция.
• Используется как идентификатор сессии общения с планировщиком, в частности, передаётся в каждом сообщении. Если планировщик при обработке сообщения обнаружил, что текущий идентификатор агента отличается от пришедшего в сообщении, значит, это сообщение пришло от предыдущей инкарнации, и его необходимо игнорировать.

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