Распределенные демоны

#алгоритм #синхронизация #распараллеливание

Всем привет!

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

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


Каждый час генерировать метрику для проекта
Каждые 10 минут выполнять обновление записей в таблице
Каждые 15 минут удалять записи из таблицы
Каждые 20 минут делать увеличение какого-то числа в определенных записях в таблицах


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

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

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

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

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



Я так понимаю есть следующие способы синхронизации процессов на разных серверах и
дата-центрах:

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

2) Автоматический запуск другого экземпляра демона или задачи через систему мониторинга
в случае падения основного демона.

3) Явная проверка существования другого демона (задачи) по открытому порту и распределение
весов. В демон прописываем открытый порт у другого демона и смотрим, работает ли он
сейчас и отвечает на запросы? Если нет, забираем корону и выполняем задачи(у). Такой
аналог VRRP.

Есть ли более элегантные способы решения задачи? Или какой из этих способов наиболее
оптимален и чаще всего используется по вашему опыту? 
Может есть какие-то распределенные планировщики? А может есть какие-нибудь книги
по архитектуре подобных демонов?
    

Ответы

Ответ 1

TL; DR: избежать дупликации работы нельзя, поэтому необходимо запускать идемпотентные
задачи. Если необходима синхронизация работы, необходимо брать распределенное решение,
позволяющее достигать консенсуса, например etcd, consul, riak, cassandra, zookeeper.
Redis и RabbitMQ использовать нельзя, потому что они к консенсусу не имеют абсолютно
никакого отношения. Вероятнее всего, если вы имеете дело с несколькими узлами, у вас
либо кто-то назначает один узел главным (и тогда просто надо выполнять все задачи на
нем), либо у вас есть какой-то оркестратор, в котором вероятнее всего есть понятие
one-time job и который стоит использовать, чтобы не набивать шишки самостоятельно.

Short story long

Перед ответом на этот вопрос сначала стоит обозначить, с чем вообще планируется борьба.

В распределенных системах есть несколько основных проблем, которые всплывают практически
всегда. Здесь упоминается "падение сервера" и "работа двух воркеров над одной задачей
одновременно". Первая на самом деле гораздо шире и называется разделением сети (network
partition) и обозначает просто то, что связь с узлом распределенной системы потеряна.
Это не значит, что узел прекратил работу или прекратил работу насовсем - это может
быть перезагрузка сетевого оборудование, сборка мусора в одном из передаточных звеньев
или на узле или просто товарищ майор не выдержал просмотра трафика в реалтайме и вышел
пока покурить. При этом разделение сети может носить совсем причудливые формы, вроде
того, что трафик от А к Б не идет, а наоборот - идет, или что А и Б свободно общаются
с В, но друг с другом - никак. Ровно здесь все три предлагаемых варианта отваливаются
- если работающий воркер не может продлить блокировку, если сервис мониторинга не может
увидеть рабочий процесс, если другой рабочий процесс не может увидеть первый рабочий
процесс - это все не значит, что работа не продолжается. И не значит, что она не оборвалась.
Это просто состояние неизвестности.

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

Кроме вышеописанных проблем необходимо еще упомянуть типы доставки сообщений. В идеальном
мире существует три режима доставки - at-most-once, exactly-once, at-least-once. К
сожалению, процесс приема и подтверждения сообщения - это две операции, которые не
могут сложиться в одну атомарную операцию, и из-за этого режим exactly-once технически
невозможен даже на холостом ходу: разделение сети всегда может произойти в тот момент,
когда узел принял сообщение, но не подтвердил его (при этом не имеет значения, подтвердил
он просто прием сообщения, или провел все необходимую обработку над ним - важно то,
что система теперь в состоянии коробки Шредингера, обработка либо запущена, либо нет,
но об этом нельзя узнать до восстановления сети, а время этого восстановления неизвестно).
Поэтому распределенная система может либо доставить сообщение не более одного раза,
либо доставлять его до тех пор, пока не получит подтверждение.

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

Но, конечно, решение у этой проблемы есть. Если задача будет идемпотентна и атомарна,
либо будет состоять из последовательности атомарных шагов, то ее можно будет запускать
произвольное количество раз, не опасаясь за результат. Фактически, это просто следование
ACI(D) 2.0.
Если же задачу невозможно сделать атомарной - например, если один шаг не может включать
в себя менее двух сайд-эффектов (нельзя одновременно отправить письмо и записать подтверждение
об отправке в базу - всегда есть вероятность, что одна операция провалится, а вторая
нет) - то можно попробовать минимизировать эти последствия.

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


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


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

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

Конечно, для этого всего бывают нужны внешние средства синхронизации. Обычно можно
использовать ту же базу данных, в которую пишет приложение - большинство решений, даже
распределенных, поддерживают CAS-операции, на основе которых можно построить ограниченную
по времени блокировку (пока воркер жив, он постоянно обновляет запись блокировки с
указанием нового времени истечения). Самые известные инструменты, которые могут позволить
это решить - это Riak, Cassandra, Zookeeper, Consul, Etcd. Что до реплики по поводу
других означенных инструментов:

RabbitMQ - это сервис очередей, и в то время как он действительно поддерживает работу
в режиме кластера и действительно на нем можно реализовать забавную схему блокировки
(до тех пор, пока сообщение-синхронизатор не будет проглочено упашим приложением),
он на самом деле не дает никаких гарантий по поводу консенсуса и без особых проблем
может несколько раз доставить одно и то же сообщение, разработчики прямо предупреждают
о split-brain. Кроме того, остается проблема с двумя режимами доставки (at-least-once
/ at-most-once), которые не могут гарантировать, что сообщение получит ровно один узел,
поэтому сам по себе - без атомизации/идемпотентизации задач - он никак делу не поможет.
Доставить до воркеров сообщение о том, что задача готова к исполнению - это легко.

Redis - это невероятный переоцененный инструмент, получивший свою популярность просто
за счет удобного интерфейса, являющийся едва ли не худшим инфраструктурным выбором
среди всех своих аналогов. Помимо всех остальных проблем, у него так же прямо прописано
в документации There is always a window of time when it is possible to lose writes
during partitions, что, в общем, является не единственным способом потерять данные
в этой волшебной системе, а потеря данных в данном кейсе ведет к дубликации ответственных
за задачу хостов и самой работы.

И, наконец, то, с чего стоило начинать ответ. Если вы добрались до распределенных
систем, то у вас с большой вероятностью уже есть либо инфраструктура, выбирающая мастер-сервер,
либо некоторый оркестратор (e.g. Kubernetes), в котором наверняка есть понятие one-time
job в этом случае все проблемы по поводу консенсуса стоит переложить на плечи этих
инструментов. В то же время, оркестратор сам будет следить за тем, чтобы процессы ваших
воркеров были запущены. Это, конечно, не спасет от невозможности гарантии однократного
выполнения задачи, но возведение инфраструктуры просто для запуска задач - тоже тот
еще геморрой.

Если же вы все-таки хотите управлять ставящими задачи демонами самостоятельно, то
вас спасет вышеупомянутая блокировка через consul/etcd/etc.

Мне правда надо это все держать в голове перед разработкой приложения?

Следует. На самом деле, как и было сказано, то же разделение сети - штука, с которой
вы обязательно встретитесь, но не обязательно большой число раз. Если это экономически
невыгодно - следует проанализировать вместе с руководителями возможные риски и уменьшить
объем работы. Тем не менее, провал фонового обработчика по вине разработчика - практически
ежедневная ситуация, поэтому я бы советовал все-таки держаться зеленой дорожки. Если
по каким-то причинам не хотите этого делать - можно раздавать задачи через брокер сообщений
(RabbitMQ в нормальных условиях должен корректно отрабатывать однократную посылку сообщения,
пока держится TCP-коннект к клиенту) и делать дополнительную синхронизацию на выполнение
задачи через один из вышеописанных инструментов.


Ответ 2

На мой взгляд вам подойдёт композиция publisher-subscriber, которая есть у большинства
систем сообщения, как то Akka, 0mq и др. В вашем случае придётся прописать немного
другую логику для основного (primary) рабочего от вторичного (secondary) если система
не позволяет создавать механизмы failsafe или broadcasting. Доступность рабочих будет
делаться самой системой, вам останется только уведомлять контролирующий процесс о прогрессе
работы, на который он будет реагировать, и в случае определённых условий обращаться
с запросом работы ко вторичному рабочему.