что такое селери кэп
Celery: лучшие практики
Если вы работаете с Django, то на некотором этапе разработке вам может понадобиться фоновая обработка долго выполняющихся задач. Возможно, что для такого рода задач вы используете какой-либо инструмент для управления очередями задач. Celery — один из самых популярных проектов для решения подобных задач в мире python и Django на данный момент, но есть и другие проекты для этой цели.
Пока я работал над некоторыми проектами, использующими Celery для управления очередями задач, выявились некоторые лучшие практики, которые я решил задокументировать. Впрочем это громкие слова для того, что я думаю о правильном подходе к решению подобных задач, а также о некоторых недостаточно используемых возможностях, которые предлагает сообщество проекта Celery.
No.1: Не используйте СУБД как ваш AMQP брокер
Позвольте мне объяснить почему я считаю это неправильным (помимо тех ограничений что описаны в документации Celery).
СУБД не разрабатывались для тех задач, которые выполняют полноценный AMQP брокер такой как RabbitMQ. Она упадет в «боевых» условиях даже на проекте с не очень большим трафиком\пользовательской базой.
Я предполагаю, что самой популярной причиной того почему люди решают использовать СУБД в том что, как правило, у них уже есть одна СУБД для веб-приложения, так почему бы не воспользоваться ей еще раз. Начать работать с таким вариантом несложно и не надо беспокоиться о других компонентах (таких как RabbitMQ).
Предположим не такой уж гипотетический сценарий: у вас есть 4 фоновых воркера для обработки, которые вы помещаете в базу данных. Это значит что вы получаете 4 процесса, которые достаточно часто запрашивают базу о новых задачах, не говоря уже о том, что каждый из них может иметь собственные конкурирующие потоки. В некоторый момент времени вы понимаете, что растет задержка при обработке задач, а потому приходит больше новых задач чем завершается, необходимо увеличивать количество воркеров. Вдруг скорость вашей базы данных начинает «проседать» из-за огромного количества запросов воркеров к базе, дисковый ввод\вывод превышает заданные лимиты, а все это начинает влиять на ваше приложение, так как воркеры, фактически, устроили DDOS-атаку вашей базе.
Этого не произошло бы при использовании полноценного AMQP брокера, так как очередь размещается в памяти и таким образом устраняется высокая нагрузка на жесткий диск. Потребителям (воркерам) нет необходимости часто запрашивать информацию, так как очередь имеет механизм доставки новой задачи воркеру, и даже, если AMQP брокер будет перегружен по каким-либо иным причинам это не приведет к падению и тормозам того веб-приложения, которое взаимодействует с пользователем.
Я пойду еще дальше и скажу, что вы не должны использовать СУБД как брокера даже в процессе разработки, тогда когда есть такие вещи как Docker и множество преднастроенных образов, которые предоставляют настроенный RabbitMQ «из коробки».
No.2: Используйте больше очередей (т.е. не только одну, которая дается по умолчанию)
Celery очень легко начать использовать, и она предоставляет сразу же одну очередь по умолчанию, в которую и помещаются все задачи пока не будет явно предписано другое поведение Celery. Наиболее общий пример того, что вы можете увидеть:
Что происходит, если обе задачи будут размещены в одной очереди, если иное не определено в файле celeryconfig.py. Я полностью пониманию чем может оправдывать подобный подход, у вас есть один декоратор, который создает удобные фоновые задачи. Здесь я хотел бы обратить внимание, что taskA и taskB, находясь в одной очереди могут делать совершенно разные вещи и таким образом одна из них может быть куда важнее другой, так почему они находятся все в одной корзине? Даже, если у вас один воркер, то представьте такую ситуацию что менее важная задача taskB окажется настолько массовой, что более важной задаче taskA воркер не сможет уделить необходимого внимания.Это приводит нас к к следующему пункту.
No.3: Используйте приоритеты воркеров
Путем решения проблемы, указанной выше является размещение задачи taskA в одной очереди, а taskB в другой и после этого присвоить x воркеров обработке очередь Q1, а остальных на обработку Q2, так как в нее приходит больше задач. Таким образом вы можете быть уверены, что задача taskB получит достаточно воркеров, а остальные тем временем будут обрабатывать менее приоритетную задачу, когда она придет, не провоцируя длительного ожидания и обработки. Потому, определите ваши очереди сами:
И ваши роутеры, которые определять куда направлять задачу:
Это позволит выполнять воркеры для каждой задачи:
No.4: используйте механизмы Celery для обработки ошибок
Большинство задач, которые я видел не имеют механизмов обработки ошибок. Если в задаче произошла ошибка, то она просто падает. Это может быть удобно для некоторых задач, однако большинство задач, которые я видел взаимодействовали с внешними API и падали из-за некоторых видов сетевых ошибок или иных проблем «доступности ресурса». Самый простой подход к обработке таких ошибок перевыполнить код задачи, так как, возможно, проблемы взаимодействия с внешним API были уже устранены.
Я люблю определять по умолчанию для задачи время ожидания, которое она будет ждать прежде чем попытается выполниться снова и как много попыток перевыполнения она предпримет прежде чем окончательно выбросить ошибку(параметры default_retry_delay и max_retries соответственно). Это наиболее простая форма обработки ошибок, которую я могу представить, но я видел, что и она практически не применяется. Разумеется Celery имеет и более сложные методы обработки ошибок, они описаны в документации Celery.
No.5: используйте Flower
No.6: Отслеживайте статус задачи, только если вам это необходимо
Статус задачи это информация о том успешно или нет завершилась задача. Она может быть полезна для некоторых статистических показателей. Важная вещь, которую следует понимать в данном случае: статус задачи это не результирующие данные и той работы, которая она выполняла, такая информация наиболее похожа на неявные изменения, записываемые в базу данных (такие, например, как изменения списка друзей пользователя).
Celery: начинаем правильно
В этой статье мне хотелось бы поделиться с читателями своим опытом работы с таким замечательным инструментом в Python как Celery. Celery это ничто иное как распределённая очередь заданий, реализованная на языке Python. На момент написания этой статьи, самой последней версией является 3.1.20. Неосведомлённый читатель может не знать для чего вообще нужна система очередей задач наподобие Celery, поэтому кратко поясню этот момент.
Что такое Celery и зачем оно нам?
Часто ли вам приходилось сталкиваться с типовыми задачами в веб-приложениях вроде отправки электронного письма посетителю или обработки загруженных данных. Чаще всего такого рода манипуляции не требуют участия конечного пользователя вашего проекта, то есть их можно выполнять в фоновом режиме. Те из нас, кто реализует выполнение этих задач в одном из процессов веб-сервера, «тормозят» тем самым его работу, увеличивая время отклика и ухудшают user experience.
В данной заметке я опущу вводную информацию по установке и настройке Celery в вашем проекте. Кстати, Celery из коробки умеет работать с Django. Ранее был отдельный python пакет, соединяющий Django и Celery,именовался он django-celery. Сейчас он заброшен, так как последнее обновление было более года назад. Стоит отметить, что django-celery не работает Django 1.9 из-за изменений в работе cache backend. Исправленную версию можно посмотреть в моём форке. Одной из удобных фич django-celery является интеграция с Django Admin по части управления periodic tasks.
Советы по работе с Celery
Не используйте базу данных в качестве broker/backend
Разделяйте задачи по очередям
Это очень важный момент. По мере развития вашего приложения, в проекте будут появляться критичные для выполнения задачи: проверка статуса платежа, формирование отчёта, отправка электронных писем и так далее. Терять их недопустимо. Если все задачи складировать в одну очередь, то в один прекрасный момент она может забиться, поставив под угрозу выполнение критически важного кода. Мой подход: разделяйте очереди по приоритетам.
Несомненно очередей может быть больше, тут всё на усмотрение разработчика и архитектуры его приложения.
В базовых настройках Celery это выглядит следующим образом:
В данном конкретном примере объявлена очередь по-умолчанию под названием normal. То есть задачи явно не указанные в списке будут автоматически распределены в эту очередь. В high попадает задача под названием check_payment_status, а в low задача close_session.
Запускать исполнителей Celery для этих очередей необходимо следующим образом:
Здесь мы явно задаём имена исполнителей и названия очередей в которых необходимо мониторить задачи на исполнение.
ВАЖНО! Если вы явно указали для задачи очередь в которую ей нужно будет падать, и при этом запустили одного из исполнителей Celery без явного указания очереди, например вот так:
То при наступлении ситуации, когда все исполнители очереди high будут заняты, Celery автоматически перенаправит новую задачу исполнителям без конкретной очереди. Поэтому при использовании раздельных очередей задач, не запускайте исполнителей без указания для них явного наименования очереди.
Логгируйте ошибки
Важной опцией здесь является наличие CELERYD_HIJACK_ROOT_LOGGER = False. По-умолчанию значение этой переменной является True, что позволяет celery «перекрывать» все ранее объявленные кастомные обработчики logging.
При указанном выше подходе нет необходимости дополнительно в коде задач (task) логгировать ошибки/исключения отдельно. О том что такое Sentry, для чего оно используется и как его настроить я напишу отдельную статью немного позже.
Пишите задачи маленькими
При написании задач старайтесь придерживаться принципа минимализма кода. То есть не нужно в самом celery task описывать бизнес логику задачи. Например, если вам необходимо генерировать и отправлять отчёт, то не нужно в самом task писать код генерации и отправки. Разбейте его на 3 части:
Это, во-первых, позволит легче читать код (есть явное разделение на подзадачи). Во-вторых, тестировать такой код намного легче (привет модульным тестам!). В-третьих, отлавливать ошибки также будет намного легче и прозрачнее.
«Гасите» задачи вовремя
Явно указывайте лимит на выполнение задачи. Это можно сделать несколькими способами:
Указание таймлимита очень важно, так как в некоторых случаях его отсутствие попросту приведёт к «зависанию» исполнителя при выполнении неоднозначных задач (требующих длительного времени, коннект к внешнему сервису и так далее).
Не храните результаты исполнения без необходимости
В большинстве случаев результат выполнения вашей задачи вам не нужен (например, если происходит отправка письма). В такой ситуации вам нет необходимости хранить что-то. Если ваши задачи полностью попадают в эту категорию, то в настройках Celery можно задать глобальный параметр CELERY_IGNORE_RESULT = True, который будет игнорировать результат исполнения всех ваших task-функций.
Используйте Flower для мониторинга исполнения задач
Всегда используйте Flower при работе с Celery. Всегда! Данный инструмент это небольшое веб приложение, написанное с использованием микрофреймворка Flask, а также Tornado для поддержки веб-сокетов. Flower позволяет вам всегда быть в курсе того как исполняются ваши задачи. Немного скриншотов:
Не поленитесь и потратьте время на его изучение. Оно окупится многократно!
Не передавайте ORM объекты в качестве аргументов
Я пару раз попадался на этом хитром трюке, который потрепал мне изрядно нервы. Рассмотрим вот такой код:
Не самый лучший пример для демонстрации побочного эффекта при передаче ORM объекта, но всё же. В данной ситуации код, описанный в send_notification, сохранит объект, изменив лишь notified = True, но activated останется по-прежнему равен False. Лучшим решением будет передача идентификатора объекта в базе данных, а в самой task функции необходимо непосредственно обращаться к объекту через его id.
BROKER_TRANSPORT_OPTIONS и visibility_timeout
При использовании Celery нередко приходиться прибегать к помощи отложенных задач, используя apply_async и передавая аргументы eta или countdown. Но делать это нужно осторожно, так как даже здесь нас поджидают «подводные камни». О чём речь? Очень часто у разработчиков, начинающих использовать очередь задач вроде Celery, происходят аномалии вроде выполнения одного и того же таска несколькими воркерами одновременно. Согласитесь, нежелательный сценарий. Так может происходить по причине того, что время, через которое должна выполниться задача, превышает visibility_timeout. По умолчанию для Redis этот параметр равен 1 часу. То есть если вы укажете выполнение задачи через 2 часа, то демон celery подождёт 1 час, поймёт, что никто из доступных воркеров не откликнулся и насильно назначит всем воркерам её выполнение при наступлении дедлайна (eta/countdown). Поэтому не забывайте про этот параметр, если вы собираетесь использовать механизмы eta/countdown/retry, задайте visibility_timeout равным самому длительному eta/countdown в вашем проекте. Подробнее можно почитать тут.
UPD: С недавних пор у блога появился свой Telegram канал, где я стараюсь делиться со своими подписчиками интересными находками из сети на тему разработки программного обеспечения и смежных с этой областью материалов.
Long-running tasks
Старайтесь не использовать Celery для выполнения долгих задач. На этот аргумент есть ряд причин:
Если нет возможности использовать что-то другое, то при работе с long-running tasks в Celery знайте следующее:
По-умолчанию 1 воркер процесс будет забирать из очереди 4 задачи за раз. Это особенно актуально знать, если Celery масштабируется на кластере через центрального брокера. То есть, если у вас 3 отдельные машины и на каждой крутится по 10 воркеров на очередь, то каждая машина будет забирать по 40 задач. Отсюда очевидно возникает проблема равномерного распределения задач по кластеру. Такое поведение оправдано в некоторых случаях, т.к. оно уменьшает количество обращений к брокеру, увеличивая производительность при выполнении небольших тасков. Чтобы изменить это, переопределите параметр CELERYD_PREFETCH_MULTIPLIER. Например:
Долгоживущие процессы имеют тенденцию к пожиранию памяти, но вот назад её зачастую не возвращают, поэтому в контексте использования Celery с ними иногда имеет смысл перезагружать воркеры после выполнения заданного количества тасков. За это отвечает параметр CELERYD_MAX_TASKS_PER_CHILD
Настройка выше будет перезагружать воркер-процесс после выполнения 1 таска.
Полезные ссылки
💌 Присоединяйтесь к рассылке
Понравился контент? Пожалуйста, подпишись на рассылку.
Celery Python: основы и примеры
Все в сообществе Python слышали о Celery хотя бы один раз, и, возможно, уже работали с ним. По сути, это удобный инструмент, который помогает запускать отложенный или выделенный код в отдельном процессе или даже на отдельном компьютере или сервере. Это экономит время и усилия на многих уровнях.
Введение в Celery Python
Celery снижает нагрузку на производительность, выполняя часть функциональности в виде отложенных задач либо на том же сервере, что и другие задачи, либо на другом сервере. Чаще всего разработчики используют его для отправки электронных писем. Тем не менее, Celery может предложить гораздо больше. В этой статье я покажу вам некоторые основы Celery, а также пару лучших практик Python-Celery.
Основы Celery
Если вы уже работали с Celery, не стесняйтесь пропустить эту главу. Но если в Celery вы новичок, здесь вы узнаете, как включить Celery в своем проекте, и примите участие в отдельном руководстве по использованию Celery с Django. По сути, вам нужно создать экземпляр Celery и использовать его для пометки функций Python как задач.
Лучше создать экземпляр в отдельном файле, так как будет необходимо запустить Celery так же, как он работает с WSGI в Django. Например, если вы создадите два экземпляра Flask и Celery в одном файле в приложении Flask и запустите его, у вас будет два экземпляра, но вы будете использовать только один. То же самое, когда вы запускаете Celery.
Основные примеры Python Celery
Как я упоминал ранее, в случае использования Celery отправляется электронное письмо. Я буду использовать этот пример, чтобы показать вам основы использования Celery. Вот краткое руководство по Celery Python:
В этом коде используется Django, поскольку он является нашей основной средой для веб-приложений. Используя Celery, мы сокращаем время ответа клиенту, поскольку отделяем процесс отправки от основного кода, отвечающего за возврат ответа.
Теперь задача будет перезапущена через десять минут, если отправка не удалась. Кроме того, вы сможете установить количество повторных попыток.
Celery позволяет запускать задачи с помощью планировщиков, таких как crontab в Linux.
если вы не используете Django, вы должны использовать celery_app.conf.beat_schedule вместо CELERY_BEAT_SCHEDULE
Отложенное выполнение задачи в Celery
Давайте посмотрим, как это может выглядеть в коде:
Настройка Python Celery Queues
Celery может быть распределен, когда у вас есть несколько воркеров на разных серверах, которые используют одну очередь сообщений для планирования задач. Вы можете настроить дополнительную очередь для вашей задачи / воркера. Например, отправка электронных писем является важной частью вашей системы, и вы не хотите, чтобы какие-либо другие задачи влияли на отправку. Затем вы можете добавить новую очередь, назовем ее mail и использовать эту очередь для отправки электронных писем.
если вы не используете Django, используйте celery_app.conf.task_routes вместо CELERY_TASK_ROUTES
Запустите два отдельных воркера Celery для очереди по умолчанию и новой очереди:
Долгосрочные задачи Python Celery
Это очень простой пример того, как такая задача может быть реализована. В конце задачи мы проверяем, сколько пользователей мы нашли в базе данных. Если число равно пределу, то, вероятно, у нас есть новые пользователи для обработки. Поэтому мы снова запускаем задачу с новым смещением. Если количество пользователей меньше лимита, это означает, что это последний кусок, и нам не нужно продолжать. Но будьте осторожны: для реализации этой задачи каждый раз должен быть одинаковый порядок записей.
Celery: получение результатов задачи
Большинство разработчиков не записывают результаты, полученные после выполнения задачи. Представьте, что вы можете взять часть кода, назначить его для задачи и выполнить эту задачу независимо, как только вы получите запрос пользователя. Когда нам нужны результаты задания, мы либо сразу получаем результаты (если задание выполнено), либо ждем его завершения. Затем мы включаем результат в общий ответ. Используя этот подход, вы можете уменьшить время отклика, что очень хорошо для ваших пользователей и рейтинга сайта.
Мы используем эту функцию для запуска одновременных операций. В одном из наших проектов у нас много пользовательских данных и много поставщиков услуг. Чтобы найти лучшего поставщика услуг, мы делаем тяжелые расчеты и проверки. Чтобы сделать это быстрее, мы создаем задачи для пользователя с каждым поставщиком услуг, запускаем их и собираем результаты, чтобы показать их пользователю. Это очень легко сделать с целевыми группами Celery.
Вот пример того, как использовать этот подход в коде:
Здесь мы выполняем вычисления как можно скорее, ожидаем результатов в конце метода, затем готовим ответ и отправляем его пользователю.
Полезные советы
Крошечные данные
Я, наверное, уже упоминал, что я использую идентификаторы записей базы данных в качестве аргументов задачи вместо полных объектов. Это хороший способ уменьшить размер очереди сообщений. Но что более важно, это то, что при выполнении задачи данные в базе данных могут быть изменены. И когда у вас есть только идентификаторы, вы получите свежие данные, а не устаревшие данные, которые вы получаете при передаче объектов.
Операции
Иногда могут возникнуть проблемы, когда выполненная задача не может найти объект в базе данных. Почему это происходит? Например, в Django вы хотите запускать задачи после регистрации пользователя, например отправку приветственного письма, а ваши настройки Django заключают все запросы в транзакцию. В Celery, однако, задачи выполняются быстро, еще до того, как транзакция будет завершена. Поэтому, если вы используете Celery при работе в Django, вы можете увидеть, что пользователь не существует в базе данных (пока).
50 оттенков Celery
Вам сюда, если хотите знать, как приручить широкоизвестный в кругах Python-разработчиков фреймворк под названием Сelery. И даже, если в вашем проекте Celery уверенно выполняет базовые команды, то финтех опыт может открыть вам неизведанные стороны. Потому что финтех — это всегда Big Data, а с ней и необходимость фоновых задач, пакетной обработки, асинхронного API и т.д. 
Прелесть рассказа Олега Чуркина про Celery на Moscow Python Conf ++ помимо подробных инструкций, как настроить Celery под нагрузку и как его мониторить, в том, что можно позаимствовать полезные наработки.
О спикере и проекте: Олег Чуркин (Bahusss) 8 лет разрабатывает Python-проекты разной сложности, работал в многих известных компаниях: Яндексе, Рамблере, РБК, Лаборатории Касперского. Сейчас техлид в финтех-старапе StatusMoney.
Проект работает с большим количеством финансовых данных пользователей (1,5 терабайта): аккаунтами, транзакциями, мерчантами и т.д. В нем каждый день запускается до миллиона задач. Может быть, кому-то это число не покажется по-настоящему большим, но для маленького стартапа на скромных мощностях это существенный объем данных, и разработчикам пришлось столкнуться с разными проблемами на пути к стабильному процессу.
Олег рассказал о ключевых моментах работы:
Проблематика
Требовалось решить следующие задачи:
В фоновом режиме выполняются разные специфические проверки, например, проверка пользователя на фрод. В финансовых стартапах много сил и внимания уделяется именно безопасности данных, поскольку мы позволяем пользователям добавлять свои банковские аккаунты в нашу систему, и можем видеть все их транзакции. Мошенники могут попробовать воспользоваться нашим сервисом для чего-то нехорошего, например, для проверки баланса украденного аккаунта.
Последняя категория фоновых задач — это maintenance задачи: что-то подкрутить, посмотреть, поправить, замониторить и т.д.
Для рассылки уведомлений, только массовой, используется пакетная обработка данных. Большой объем данных, которые мы получаем от наших пользователей, приходится предрасчитывать и обрабатывать определенным образом, в т.ч. в пакетном режиме.
В это же понятие входит классический Extract, Transform, Load:
Требования к инструментам
Чтобы реализовать эти задачи, у нас были такие требования к инструментам:
Какой инструмент выбрать?
Какие есть варианты на рынке в 2018 году для решения этих задач?
Когда-то давно для менее амбициозных задач я написал удобную библиотеку которая все еще используется в некоторых проектах. Она проста в эксплуатации и выполняет задачи в фоне. Но при этом никакие брокеры не нужны (ни Celery, ни другие), только сервер приложений uwsgi, у которого есть спулер (spooler) — такая штука, которая запускается как отдельный воркер. Это очень простое решение — все задачи хранятся условно в файлах. Для простых проектов этого хватает, но для нашего было недостаточно.
Так или иначе мы рассмотрели:
Многообещающий кандидат 2018
Сейчас я бы обратил ваше внимание на Dramatiq. Это библиотека от адепта Celery, который познал все минусы Celery и решил все переписать, только очень красиво. Преимущества Dramatiq:
Но в 2016 году кроме Celery взять было особо нечего. Нам понравилась его богатая функциональность, и тогда он идеально подходил под наши задачи, потому что уже тогда был зрелый и функциональный:
Особенности проекта
Расскажу про наш контекст, чтобы дальнейший рассказ был понятнее.
Мы используем Redis как брокер сообщений. Я слышал много историй и слухов о том, что Redis теряет сообщения, что он не приспособлен быть брокером сообщений. На продакшен опыте это не подтверждается, а, как выясняется, Redis сейчас работает более производительно, чем RabbitMQ (именно с Celery, как минимум, видимо, проблема в коде интеграции с брокерами). В версии 4 починили брокер Redis, он действительно перестал терять задачи при рестартах и работает вполне стабильно. В 2016 году в Celery собирались отказаться от Redis и сконцентрироваться на интеграции с RabbitMQ, но, к счастью, этого не произошло.
В случае проблем с Redis, если нам потребуется серьезная high availability, то мы, поскольку используем мощности Amazon, переключимся на Amazon SQS или Amazon MQ.
Мы не используем result backend для хранения результатов, потому что предпочитаем хранить результаты сами где хотим, и проверять их так, как хотим. Мы не хотим, чтобы Celery за нас это делал.
Мы используем pefork pool, то есть процесс-воркеры, которые создают для дополнительного concurrency отдельные форки процессов.
Unit of work
Обсудим базовые элементы, чтобы ввести в курс дела тех, кто еще не пробовал Celery, а только собирается. Unit of work для Celery — это задача. Приведу пример простой задачи, которая посылает email.
Простая функция и декоратор:
Запуск задачи прост: либо вызываем функцию и задача выполнится в runtime (send_email(email=»python@example.com»)), либо в воркере, то есть тот самый эффект задачи в фоне:
За два года работы с Celery при высоких нагрузках мы вывели правила хорошего тона. Было много граблей, мы научились их обходить, и я поделюсь как.
Оформление кода
В задаче может находиться различная логика. Вообще Celery способствует тому, чтобы вы держали задачки в файлах или в packages tasks, или импортировали их откуда-то. Иногда получается нагромождение бизнес-логики в одном модуле. На наш взгляд, тут правильный подход с точки зрения модульности приложения — держать минимум логики в задаче. Мы используем задачки только как «запускаторы» кода. То есть задача не несет в себе логику, а триггерит запуск кода в Background.
Весь код мы выносим во внешние классы, которые используют еще какие-то классы. Все задачи по сути состоят из двух строчек.
Простые объекты в параметрах
В примере выше в задачу передается некий id. Во все задачи, которые мы используем, мы передаем только маленькие скалярные данные, id. Мы не сериализуем модели Django, чтобы их передавать. Даже в ETL, когда приходит из внешнего сервиса большой блоб данных, мы его сначала сохраняем и потом запускаем задачу, которая читает по id весь этот блоб и обрабатывает.
Если так не делать, то мы видели очень большие спайки потребляемой памяти у Redis. Сообщение начинает занимать больше памяти, сеть сильно загружается, количество обработанных задач (производительность) падает. Пока объект доходит до выполнения, задачи становятся не актуальными, объект уже удален. Данные нужно было сериализовать — не все хорошо сериализуется в JSON в Python. Нам нужна была возможность при retry задач как-то быстро решать, что делать с этими данными, получать их снова, запускать над ними какие-то проверки.
Если вы передаете большие данные в параметрах, одумайтесь! Лучше в задаче передавать маленький скаляр с малым количеством информации, и по этой информации в задаче получить все необходимое.
Идемпотентные задачи
Такой подход рекомендуют сами разработчики Celery. При повторном выполнении участка кода никаких побочных-эффектов произойти не должно, результат должен быть тот же. Не всегда этого просто добиться, особенно если идет взаимодействие с многими сервисами, или двухфазные коммиты.
Но когда вы все делаете локально, то всегда можете проверять, что входящие данные существуют и актуальны, над ними действительно можно совершить работу, и использовать транзакции. Если к одной задаче много запросов в базу и что-то может пойти не так во время выполнения — используйте транзакции, чтобы откатить ненужные изменения.
Обратная совместимость
Несколько интересных побочных эффектов было у нас при деплое приложения. Неважно, какой тип деплоя вы используете (blue+green или rolling update), всегда возникнет ситуация, когда старый код сервиса создает сообщения для нового кода воркера, и наоборот, старый воркер принимает сообщения от нового кода сервиса, потому что он раскатился «первее» и туда трафик пошел.
Мы ловили ошибки и теряли задачи, пока не научились поддерживать обратную совместимость между релизами. Обратная совместимость заключается в том, что между релизами задачи должны работать безопасно, независимо от того, какие параметры приходят в эту задачу. Поэтому во всех задачах мы сейчас делаем «резиновую» сигнатуру (**kwargs). Когда в следующем релизе вам потребуется добавить новый параметр, вы его из **kwargs возьмете в новом релизе, а в старом не возьмете — у вас ничего не сломается. Как только меняется сигнатура, а Celery об этом не знает, он падает и выдает ошибку, что такого параметра нет в задаче.
Более строгий способ избегать подобных проблем — это версионирование очередей задач между релизами, но он достаточно сложен в реализации и мы пока оставили его в беклоге.
Таймауты
Проблемы могут возникнуть из-за недостаточного количества или неправильных таймаутов.
Не ставить таймаут на задачу — это зло. Это значит, что вы не понимаете, что происходит в задаче, как должна работать бизнес-логика.
Поэтому все задачки у нас обвешаны таймаутами, в том числе глобальными для всех задач, и для каждой конкретной задачи тоже проставлены таймауты.
Обязательно должны быть проставлены: soft_limit_timeout и expires.
Expires — это сколько задача может жить в очереди. Нужно чтобы задачи не накапливались в очередях в случае проблем. Например, если мы сейчас хотим сообщить о чем-то пользователю, но что-то случилось, и задача может выполниться только завтра — в этом нет смысла, завтра сообщение уже будет неактуально. Поэтому на уведомления у нас достаточно маленький expires.
Обратите внимание на использование eta (countdown) + visibility_timeout. В FAQ описана такая проблема с Redis — так называемый visibility timeout у брокера Redis. По умолчанию его значение один час: если через час воркер видит, что задачу никто не взял к исполнению, то повторно добавляет ее в очередь. Таким образом, если countdown равен двум часам, уже через час брокер выяснит, что эта задача еще не выполнилась, и создаст еще одну такую же. А через два часа выполнится две одинаковых задачи.
Если estimation time или countdown превышают 1 час, то, скорее всего, при использовании Redis получится дублирование задач, если вы, конечно, не изменили значение visibility_timeout в настройках соединения с брокером.
Retry policy
Для тех задач, которые можно повторить, или которые могут выполниться с ошибками, мы используем Retry policy. Но используем аккуратно, чтобы не завалить внешние сервисы. Если быстро повторять задачи, не указывая exponential backoff, то внешний сервис, а может быть и внутренний, могут просто не выдержать.
Параметры retry_backoff, retry_jitter и max_retries хорошо бы указывать явно, особенно max_retries. retry_jitter — параметр, который позволяет внести немножко хаоса, чтобы задачи не начали повторятся одновременно.
Утечки памяти
К сожалению, утечки памяти возникают очень легко, а найти и исправить их сложно.
В целом работа с памятью у Python очень спорная. Вы потратите много времени и нервов, чтобы понять, почему происходит утечка, а потом выяснится, что она даже не в вашем коде. Поэтому всегда, начиная проект, проставляйте лимит памяти на воркер: worker_max_memory_per_child.
Это гарантирует, что однажды не придет OOM Killer, не убьет все воркеры, и вы не потеряете все задачи. Celery будет сам перезапускать воркеры, когда нужно.
Приоритет выполнения задач
Всегда есть задачи, которые нужно выполнять раньше всех, быстрее всех — они должны быть выполнены прямо сейчас! Есть задачи, которые не так важны — пусть выполнятся в течение дня. Для этого у задачи есть параметр priority. В Redis он работает достаточно интересно — создается новая очередь с именем, в которое добавляется priority.
Мы используем другой подход — отдельные воркеры для приоритетов, т.е. по старинке создаем воркеры для Celery с разными «важностями»:
Celery multi start — это хелпер, который помогает запустить всю конфигурацию Celery на одной машине и из одной командной строки. В этом примере мы создаем ноды (или воркеры): high_priority и low_priority, 2 и 6 — это concurrency.
Два воркера high_priority постоянно обрабатывают очередь urgent_notifications. Эти воркеры больше никто не займет, они будут только читать важные задачи из очереди urgent_notifications.
Для неважных задач есть low_priority очередь. Там 6 воркеров, которые принимают сообщения из всех остальных очередей. Также low_priority воркеры мы подписываем на urgent_notifications, чтобы они могли помочь, если воркеры с high_priority не будут справляться.
Мы используем эту классическую схему для приоритезации задач.
Extract, Transform, Load
Чаще всего ETL выглядит как цепочка задач, каждая из которых получает на вход данные из предыдущей задачи.
В примере три задачи. В Celery есть подход к distributed processing и несколько полезных утилит, в том числе функция chain, которая делает из трех таких задач один pipeline:
Celery сам разберет pipeline, выполнит по порядку сначала первую задачу, потом полученные данные передаст во вторую, данные, которые вернет вторая задача, передаст в третью. Так мы реализуем простые ETL pipelines.
Для более сложных цепочек приходится подключать дополнительную логику. Но важно иметь в виду, что если в этом chain возникнет проблема в одной задаче, то весь chain развалится. Если вы не хотите такого поведения, то обрабатывайте exception и продолжайте выполнение, либо останавливайте всю цепочку по исключению.
На самом деле эта цепочка внутри выглядит как одна большая задача, в которой содержатся все задачи со всеми параметрами. Поэтому если злоупотребить количеством задач в цепочке, то получится очень высокое потребление памяти и замедление общего процесса. Создание цепочек из тысяч задач — плохая идея.
Пакетная обработка задач
Теперь самое интересное: что происходит, когда нужно отправить письмо двум миллионам пользователей.
Вы пишите такую функцию обхода всех пользователей:
Правда, чаще всего функция будет получать не только id пользователей, но и вымывать вообще всю таблицу users. Для каждого пользователя будет запускаться своя задача.
В этой задаче есть несколько проблем:
Мы пробовали уменьшать конкурентность воркеров, это помогает в каком-то смысле — снижаются нагрузки на сервис. Или можно масштабировать внутренние сервисы. Но это не решит проблему задачи-генератора, которая все еще очень много на себя берет. И никак не влияет на зависимость от производительности внешних сервисов.
Генерация задач
Мы решили пойти по другому пути. Чаще всего нам не нужно запускать все 2 млн задач прямо сейчас. Нормально, что рассылка уведомлений всем пользователям займет, например, 4 часа, если эти письма не так важны.
Сначала мы попробовали использовать Celery.chunks:
Это не изменило ситуацию, потому что, несмотря на итератор, все user_id будут загружены в память. И все воркеры получают цепочки задач, и хотя воркеры будут немного отдыхать, мы остались не удовлетворены этим решением в итоге.
Мы пробовали выставить rate_limit на воркеры, чтобы они обрабатывали только определенное количество задач в секунду, и выяснили, что на самом деле rate_limit указанный как для задачи, это rate_limit для воркера. То есть если вы указываете rate_limit для задачи, это не значит, что задача будет выполняться 70 раз в секунду. Это значит, что воркер ее будет выполнять 70 раз в секунду, и в зависимости от того, что у вас с воркерами, этот лимит может меняться динамически, т.е. реальный лимит rate_limit * len(workers).
Если воркер запускается или останавливается, то суммарный rate_limit меняется. Более того, если у вас задачи медленные, то весь prefetch в очереди, который наполняет воркер, будет забит этими медленными задачами. Воркер смотрит: «О, у меня эта задача в rate_limit-е, я ее больше не могу исполнять. И все следующие задачи в очереди точно такие же — пусть они повисят!» — и ждет.
Chunkificator
В итоге мы решили, что напишем свое, и сделали маленькую библиотеку, которую назвали Chunkificator.
Она принимает sleep_timeout и initial_chunk, и вызывает сама себя с новым chunk. Chunk — это абстракция либо над integer-списками, либо над date или datetime списками. Мы передаем chunk в функцию, которая получает пользователей только с этим chunk, и запускает задачи только для этого chunk.
Таким образом генератор задач запускает только то количество задач, которое нужно, и не потребляет много памяти. Картина стала такой. 
Изюминки добавляет то, что мы используем sparse chunk, то есть мы используем в качестве chunk id инстансов в БД (некоторые из них могут быть пропущены, поэтому и задач может быть меньше). В итоге нагрузка получилась более равномерная, процесс стал дольше, но все живы-здоровы, база не напрягается.
Библиотека реализована для Python 3.6+ и доступна на GitHub. Есть нюанс, который я планирую исправить, но пока для datetime-chunk нужен pickle serializer — многие на это не смогут пойти.
Пара риторических вопросов — откуда вся эта информация взялась? Как мы узнали, что у нас были проблемы? Как узнать, что проблема скоро станет критична и её уже нужно начинать решать?
Ответ — это, конечно, мониторинг.
Мониторинг
Я очень люблю мониторинг, люблю все мониторить и держать руку на пульсе. Если вы не держите руку на пульсе, то вы будете постоянно наступать на грабли.
Стандартные вопросы мониторинга:
NB: нужен доступ к продакшен-брокеру, чтобы использовать интерфейс CLI.
Celery Flower позволяет сделать то же самое, что и CLI, только через веб-интерфейс, и то не всё. Зато строит некоторые простые графики и позволяет менять настройки «на лету».
В целом Celery Flower подходит, для того чтобы просто посмотреть, как все работает, в небольших сетапах. К тому же он поддерживает HTTP API, то есть удобен, если вы пишете автоматизацию.
Но мы остановились на Prometheus. Взяли текущий экспортер: пофиксили в нем утечки памяти; добавили метрики по типам exception; добавили метрики по количеству сообщений в очередях; интегрировали с aлертами в Grafana и радуемся. Он тоже выложен на GitHub, можно посмотреть здесь.
Примеры в Grafana
Чего не хватает в Celery?
Это развесистый фреймворк, в нем много всего, но нам не хватает! Не хватает маленьких фич, таких как:
Выводы
Несмотря на то, что Celery — это фреймворк, которым многие пользуется в продакшене, он состоит из 3 библиотек — Celery, Kombu и Billiard. Все эти три библиотеки разрабатывают соразработчики, и они могут релизнуть одну зависимость и сломать вашу сборку.
Поэтому, надеюсь, что вы уже как-то разобрались и сделали ваши сборки детерминистическими.
На самом деле выводы не настолько печальные. Celery справляется со своими задачами в нашем финтех-проекте под нашей нагрузкой. Мы наработали опыт, которым я с вами поделился, и вы можете применить наши решения или доработать их и тоже преодолеть все свои трудности.
Не забывайте, что мониторинг должен являться основной частью вашего проекта. Только с помощью мониторинга вы сможете узнать, где же у вас что-то не так, что нужно поправить, добавить, исправить.
Контакты спикера Олега Чуркина: Bahusss, facebook и github.
Следующая большая Moscow Python Conf++ пройдет в Москве 5 апреля. В этом году мы в экспериментальном режиме попробуем уместить всю пользу в один день. Докладов будет не меньше, целый поток выделим иностранным разработчикам известных библиотек и продуктов. К тому же пятница — идеальный день для афтерпати, которая, как известно, является неотъемлемой составляющей конференции про общение.
Присоединяйтесь к нашей профессиональной Python-конференции — подавать доклад здесь, бронировать билет здесь. А пока идет подготовка, здесь будут появляться статьи по Moscow Python Conf++ 2018.