Flower is a web based tool for monitoring and administrating Celery clusters.
Features
- Real-time monitoring using Celery Events
- Task progress and history
- Ability to show task details (arguments, start time, runtime, and more)
- Graphs and statistics
- View worker status and statistics
- Shutdown and restart worker instances
- Control worker pool size and autoscale settings
- View and modify the queues a worker instance consumes from
- View currently running tasks
- View scheduled tasks (ETA/countdown)
- View reserved and revoked tasks
- Apply time and rate limits
- Configuration viewer
- Revoke or terminate tasks
- View statistics for all Celery queues
- Queue length graphs
Installation
Installing flower with pip _ is simple ::
$ pip install flower
Development version can be installed with ::
Miley Cyrus — Flowers / Перевод на русский
$ pip install https://github.com/mher/flower/zipball/master#egg=flower
Usage
Important Please note that from version 1.0.0 Flower uses Celery 5 and has to be invoked in the same style as celery commands do.
The key takeaway here is that the Celery app’s arguments have to be specified after the celery command and Flower’s arguments have to be specified after the flower sub-command.
This is the template to follow::
celery [celery args] flower [flower args]
Core Celery args that you may want to set::
-A, —app -b, —broker —result-backend
Usage Examples
Launch the Flower server at specified port other than default 5555 (open the UI at http://localhost:5566): ::
$ celery flower —port=5566
Specify Celery application path with address and port for Flower: ::
$ celery -A proj flower —address=127.0.0.6 —port=5566
Launch using docker: ::
$ docker run -p 5555:5555 mher/flower
Launch with unix socket file: ::
$ celery flower —unix-socket=/tmp/flower.sock
Broker URL and other configuration options can be passed through the standard Celery options (notice that they are after Celery command and before Flower sub-command): ::
API
Flower API enables to manage the cluster via REST API.
For example you can restart worker’s pool by: ::
$ curl -X POST http://localhost:5555/api/worker/pool/restart/myworker
Or call a task by: ::
$ curl -X POST -d » http://localhost:5555/api/task/async-apply/tasks.add
Or terminate executing task by: ::
$ curl -X POST -d ‘terminate=True’ http://localhost:5555/api/task/revoke/8a4da87b-e12b-4547-b89a-e92e4d1f8efd
For more info checkout API Reference _ and examples _.
Documentation
Documentation is available at Read the Docs _ and IPython Notebook Viewer _
License
Flower is licensed under BSD 3-Clause License. See the LICENSE file in the top distribution directory for the full license text.
Источник: pypi.org
Xiaomi Smart Flower Monitor — анализатор почвы и освещенности
Flower что это за программа
Целью данного руководства является установка и настройка Flower на боевом сервере. Предпогалается, что некоторые функции вашего Django-приложения обслуживает полностью функционирующий экземпляр Celery и вам необходима возможность отслеживать его работу. Мы установим Flower на тот же сервер, где расположено наше веб-приложение, настроим аутентификацию двумя разными способами, сконфигурируем Nginx для нашей задачи, обеспечим автоматический перезапуск при перегагрузке при помощи демона Supervisor. Статья имеет следующую структуру:
- Введение
- Установка Flower
- Настройка Supervisor для автоматического запуска Flower
- Конфигурация Nginx (1 способ)
- Конфигурация Nginx (2 способ).
- Настрока Google OAuth 2.0 аутентификации.
Введение
Flower — real-time система мониторинга для Celery. C его помощью можно контролировать все составляющие звенья экосистемы Celery. В соответсвии с официальной документацией Flower приведем неполный список его возможностей:
- Мониторинг в реальном времени с помощью Celery Events
- Ход выполнения и история
- Возможность показать детали задачи (аргументы, время запуска, время выполнения и многое другое)
- Графики и статистика
- Завершение работы и перезапуск рабочих экземпляров
- Управление размером рабочего пула и настройками автомасштабирования
- Применять ограничения по времени и скорости
- Отмена или прекращение задач
Как вы уже поняли — Flower прекрасный вспомагательный инструмент при работе с Celery. Больше информации вы можете найти тут.
Установка Flower
Все довольно просто. Установим Flower c помощью pip:
$ pip install flower
Запустим сервер с целью удостоверится, что все работает. Запуск призводим из того же каталога, где расположен файл manage.py:
$ flower -A project —port = 5555
Теперь, при условии что порт 5555 открыт, вы можете проверить что Flower успешно запустился по адресу http:// :5555. После этого вы можете остановить сервер комбинацией клавиш Сtrl-C.
Настройка Supervisor для автоматического запуска Flower
Несколько слов о Supervisor. Supervisor — это клиент-серверная система, которая позволяет контролировать ряд процессов в Unix-подобных системах. Состоит из сервера supervisord — он принимает и обрабатывает запросы от клиентов, supervisorctl — инструмент командной стоки(CLI), также имеется возможность запуска HTTP сервера для доступа через веб-интерфейс к supervisord. Supervisor запускается как основной процесс, а управляемые им процессы других приложений и служб выступают в роли субпроцессов.
Разобравшись, что из себя представляет Supervisor, настроим конфигурационный файл для запуска Flower под его управлением:
$ sudo touch /etc/supervisor/conf.d/my_flower.conf
$ sudo nano /etc/supervisor/conf.d/my_flower.conf
Заранее оговорюсь, что строку c директивой command мы несколько раз изменим в ходе статьи в зависимости от выбранного способа расположения или вида аутентификации. И так добавим в файл my_flower.conf следующее содержимое:
[program:my_flower]
command= /home/ my_user / my_project /venv/bin/flower -A project —port=5555 —basic_auth= user1 : password1
directory= /home/my_user/my_project/project
user= my_user
numprocs=1
Отмечу, что при данной конфигурации, файлы stdout и stderr должны быть созданы вручную, также мы указали базовый способ аутентификации. Чтобы узнать подробнее о настройках конфигурации Supervisor посетите эту страницу. Сохраните и закройте файл.
Теперь мы можем запустить сервер supervisord:
$ sudo supervisord
После выполнения этой команды мы имеем запущенный сервер supervisord и считанные файлы конфигурации. Запустим Flower с помощью CLI supervisorctl:
$ sudo supervisorctl start my_flower
Проверьте функционирует ли Flower. Так же вам становятся доступны такие команды как:
supervisorctl stop , supervisorctl restart , supervisorctl status .
Конфигурация Nginx (1 способ)
Первый вариант конфигурации Nginx описан в официальной документации Flower. В данном случае Flower будет располагаться на поддомене, т.е. адрес расположения будет выглядеть так http://flower. my_domain . На мои скромный взляд это отличный вариант, при этом есть одно исключение — если у вас настроен SSL сертификат, но при этом он не распространяется на поддомены. Если это не ваш случай, то стоит остановится на такой конфигурации.Обратите внимание — вам необходимо будет добавить дополнительную А-запись в личном кабинете поставщика услуг. Откроем файл конфигурации Nginx для вашего веб-приложения:
$ sudo nano /etc/nginx/sites-available/ my_project
Добавим дополнительную секцию server :
server <
listen 80;
server_name flower.example.com;
charset utf-8;
location / proxy_pass http://localhost:5555;
proxy_set_header Host $host;
proxy_redirect off;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection «upgrade»;
>
>
Закроем и сохраним файл конфигурации. Не забудьте перезапустить Nginx:
$ sudo systemctl restart nginx
Конфигурация Nginx (2 способ)
Второй способ подразумевает размещение Flower по адресу http:// my_domain /flower. Для этого изменим директиву command в файле конфигурации Supervisor. Откроем этот файл:
$ sudo nano /etc/supervisor/conf.d/my_flower.conf
Добавим в конец существующей строки параметр —url-prefix my-class-code»> command= /home/ my_user / my_project /venv/bin/flower -A project —port=5555 —basic_auth= user1 : password1 —url-prefix=»flower»
Следующим шагом будет изменение файла конфигурации Nginx:
$ sudo nano /etc/nginx/sites-available/my_project
Вместо добавления новой секции server, в уже имеющуюся секцию добавим дополнительльную location такого вида:
location ~ ^/flower/? <
proxy_pass http://localhost:5555;
rewrite ^/flower/?(.*)$ /$1 break;
>
Закроем и сохраним файл конфигурации. Не забудьте перезапустить Nginx:
$ sudo systemctl restart nginx
Настрока Google OAuth 2.0 аутентификации
Первым шагом будет создание нового проекта в консоле гугл разработчика и получение ключей:
- Перейдем в консоль разработчика Google
- Создадим новый проект
- Перейдем во вкладку «Учетные данные» и нажмем кпопку «Создать учетные данные». На этом шаге вам придется настроить «Окно запроса доступа». Обратите внимание на пункт с логотипом — не загружайте его сейчас (это требует длительной проверки на стороне Google).
- Завершите создание идентификатора клиента OAuth.
- Скопируйте ваши ключи,
На следующем этапе вам нужно изменить значение переменной command в конфигурационном файле Supervisor. Откроем и скоректируем запись:
$ sudo nano /etc/supervisor/conf.d/my_flower.conf
Для первого варианта конфигурации Nginx переменная command будет выглядить так:
Для второго варианта в том же виде, но с добавление —url-prefix:
Не забываем перезапускать процесс после изменения конфигурации:
$ sudo supervisorctl restart my_flower
Резюме
В этом руководстве мы пошагово рассмотрели процесс установки системы мониторига Flower для Django-проекта, используещего Сelery (довольно частое явление), а также настроили аутентификацию для Flower,используя Google Oauth. Я, в свою очередь надеюсь, что данный материал окажется полезным для вас и поможет сэкономить немного времени.
Источник: tiny-blog.ru
Асинхронные задания в Django с Celery
Если в вашем приложении есть какой-то длительный процесс, вы можете обрабатывать его не в стандартном потоке запросов/ответов, а в фоновом режиме.
К примеру, в вашем приложении пользователь должен отправить картинку-миниатюру (которую, скорее всего, нужно будет отредактировать) и подтвердить адрес электронной почты. Если ваше приложение обрабатывает изображение, а потом отправляет письмо для подтверждения в обработчике запросов, то конечному пользователю придется зачем-то ждать завершения выполнения обеих задач перед тем, как перезагрузить или закрыть страницу. Вместо этого, вы можете передать эти операции в очередь задач и оставить на обработку отдельному процессу, чтобы немедленно отправить пользователю ответ. В таком случае, конечный пользователь сможет заниматься другими делами на стороне клиента во время выполнения обработки в фоновом режиме. Ваше приложение в таком случае также сможет свободно отвечать на запросы других пользователей и клиентов.
Сегодня мы поговорим о процессе настройки и конфигурирования Celery и Redis для обработки длительных процессов в приложении на Django, чтобы решать такие задачи. Также мы воспользуемся Docker и Docker Compose, чтобы связать все части вместе, и рассмотрим, как тестировать задания Celery с помощью модульных и интеграционных тестов.
К концу этого руководства мы научимся:
- Интегрировать Celery в Django, чтобы создавать фоновые задания.
- Упаковывать Django, Celery и Redis с помощью Docker.
- Запускать процессы в фоновом режиме с помощью отдельного рабочего процесса.
- Сохранять логи Celery в файл.
- Настраивать Flower для мониторинга и администрирования заданий и воркеров (worker) Celery.
- Тестировать задания Celery с помощью модульных и интеграционных тестов.
Фоновые задачи
Для улучшения пользовательского опыта, продолжительные процессы должны выполняться в фоновом режиме вне обычного потока HTTP-запросов/ответов.
- Отправка писем для подтверждения;
- Веб-скейпинг и краулинг;
- Анализ данных;
- Обработка изображений;
- Генерация отчетов.
Рабочий процесс
Наша цель – разработать приложение на Django, которое для обработки продолжительных процессов вне цикла запрос/ответ использует Celery.
- Конечный пользователь генерирует новое задание, отправляя POST-запрос на сервер.
- В этом представлении задание добавляется в очередь, а id задания отправляется обратно на клиент.
- С помощью AJAX клиент продолжает опрашивать сервер, чтобы проверить состояние задания, в том время как само задание выполняется в фоновом режиме.
Создание проекта
Клонируйте проект из репозитория django-celery и выполните checkout по тегу v1 в ветке master:
$ git clone https://github.com/testdrivenio/django-celery —branch v1 —single-branch $ cd django-celery $ git checkout v1 -b master
Поскольку в общей сложности нам нужно работать с тремя процессами (Django, Redis, воркер), мы используем Docker для упрощения работы, соединив их так, чтобы мы могли все запустить одной командой в одном окне терминала.
Из корня проекта создайте образы и запустите Docker-контейнеры:
$ docker-compose up -d —build
Когда сборка завершится, перейдите на localhost:1337:
Убедитесь в том, что тесты проходят успешно:
$ docker-compose exec web python -m pytest ======================================== test session starts ======================================== platform linux — Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1 django: settings: core.settings (from ini) rootdir: /usr/src/app, inifile: pytest.ini plugins: django-3.8.0 collected 1 item tests/test_tasks.py . [100%] ========================================= 1 passed in 0.47s =========================================
Давайте взглянем на структуру проекта перед тем, как двигаться дальше:
├── .gitignore ├── LICENSE ├── README.md ├── docker-compose.yml └── project ├── Dockerfile ├── core │ ├── __init__.py │ ├── asgi.py │ ├── settings.py │ ├── urls.py │ └── wsgi.py ├── entrypoint.sh ├── manage.py ├── pytest.ini ├── requirements.txt ├── static │ ├── bulma.min.css │ ├── jquery-3.4.1.min.js │ ├── main.css │ └── main.js ├── tasks │ ├── __init__.py │ ├── apps.py │ ├── migrations │ │ └── __init__.py │ ├── templates │ │ └── home.html │ └── views.py └── tests ├── __init__.py └── test_tasks.py
Запуск задания
Обработчик событий в project/static/main.js подписан на нажатие на кнопку. По клику на сервер отправляет AJAX POST-запрос с соответствующим типом задания: 1 , 2 или 3 .
$(‘.button’).on(‘click’, function() < $.ajax(< url: ‘/tasks/’, data: < type: $(this).data(‘type’) >, method: ‘POST’, >) .done((res) => < getStatus(res.task_id); >) .fail((err) => < console.log(err); >); >);
На стороне сервера уже настроено представление для обработки запроса в project/tasks/views.py :
def run_task(request): if request.POST: task_type = request.POST.get(«type») return JsonResponse(, status=202)
А теперь начинается самое интересное: привязываем Celery!
Настройка Celery
Начнем с того, что добавим Celery и Redis в файл project/requirements.txt :
celery==4.4.1 Django==3.0.4 redis==3.4.1 pytest==5.4.1 pytest-django==3.8.0
Celery использует брокер сообщений — RabbitMQ, Redis или AWS Simple Queue Service (SQS) – чтобы упростить коммуникацию между воркером Celery и веб-приложением. Сообщения направляются к брокеру, а после обрабатываются воркером. После этого результаты отправляются на бэкенд.
Redis будет одновременно и брокером и бэкендом. Добавьте Redis и воркера Celery в файл docker-compose.yml следующим образом:
Обратите внимание на celery worker —app=core —loglevel=info :
- celery worker используется для запуска воркера Celery;
- —app=core используется для запуска core приложения Celery (которое мы коротко определим);
- —loglevel=info определяет уровень логирования информации.
CELERY_BROKER_URL = os.environ.get(«CELERY_BROKER», «redis://redis:6379/0») CELERY_RESULT_BACKEND = os.environ.get(«CELERY_BROKER», «redis://redis:6379/0»)
Затем создайте файл sample_tasks.py в project/tasks :
from celery import shared_task
Здесь, с помощью декоратора shared_task мы определили новую функцию-задание Celery, которая называется create_task .
Помните о том, что само задание не будет выполняться из процесса Django, оно будет выполнено воркером Celery.
А теперь добавьте файл celery.py в «project/core» :
import os from celery import Celery os.environ.setdefault(«DJANGO_SETTINGS_MODULE», «core.settings») app = Celery(«core») app.config_from_object(«django.conf:settings», namespace=»CELERY») app.autodiscover_tasks()
Что тут происходит?
- Для начала нужно установить значение по умолчанию для среды DJANGO_SETTINGS_MODULE , чтобы Celery знала, как найти проект Django.
- Затем мы создали экземпляр Celery с именем core и поместили в переменную app .
- Затем мы загрузили значения конфигурации Celery из объекта настроек из django.conf . Мы использовали namespace=«CELERY» для предотвращения коллизий с другими настройками Django. Таким образом, все настройки конфигурации для Celery должны начинаться с префикса CELERY_ .
- Наконец, app.autodiscover_tasks() говорит Celery искать задания из приложений, определенных в settings.INSTALLED_APPS .
from .celery import app as celery_app __all__ = («celery_app»,)
Запуск задания
Обновите представление, чтобы начать выполнение задания и отправить id:
Не забудьте импортировать задание:
from tasks.sample_tasks import create_task
Соберите образы и разверните новые контейнеры:
$ docker-compose up -d —build
Для запуска нового задания, выполните:
$ curl -F type=0 http://localhost:1337/tasks/
Вы увидите что-то вроде этого:
Статус задания
Вернитесь к обработчику событий на стороне клиента:
$(‘.button’).on(‘click’, function() < $.ajax(< url: ‘/tasks/’, data: < type: $(this).data(‘type’) >, method: ‘POST’, >) .done((res) => < getStatus(res.task_id); >) .fail((err) => < console.log(err); >); >);
Когда от AJAX-запроса вернется ответ, мы будем слать getStatus() с id задания каждую секунду:
function getStatus(taskID) < $.ajax(< url: `/tasks/$/`, method: ‘GET’ >) .done((res) => < const html = `
` $(‘#tasks’).prepend(html); const taskStatus = res.task_status; if (taskStatus === ‘SUCCESS’ || taskStatus === ‘FAILURE’) return false; setTimeout(function() < getStatus(res.task_id); >, 1000); >) .fail((err) => < console.log(err) >); >
Если ответ положительный, то новая строка будет добавлена к таблице DOM. Обновите представление get_status , чтобы вернуть статус:
from celery.result import AsyncResult
$ docker-compose up -d —build
Запустите новое задание:
$ curl -F type=1 http://localhost:1337/tasks/
Затем извлеките task_id из ответа и вызовите обновленный get_status , чтобы увидеть статус:
$ curl http://localhost:1337/tasks/25278457-0957-4b0b-b1da-2600525f812f/
Ту же информацию вы можете посмотреть в браузере:
Логи Celery
Обновите сервис celery в docker-compose.yml так, чтобы логи Celery отправились в отдельный файл:
Добавьте новую директорию в “project” и назовите ее “logs”. Затем добавьте в этот новый каталог положите файл celery.log .
$ docker-compose up -d —build
Вы должны видеть, как файл с логами локально заполняется после настройки volume:
Панель мониторинга Flower
Flower – это легкий веб-инструмент для мониторинга Celery в режиме реального времени. Вы можете отслеживать запущенные задания, увеличивать или уменьшать пул воркеров, отображать графики и статистику, например.
Добавьте его в requirements.txt :
celery==4.4.1 Django==3.0.4 flower==0.9.3 redis==3.4.1 pytest==5.4.1 pytest-django==3.8.0
Затем добавьте новый сервис в docker-compose.yml :
$ docker-compose up -d —build
Перейдите на localhost:5555 для просмотра панели мониторинга. Вы должны увидеть одного воркера:
Запустите еще несколько заданий, чтобы протестировать панель мониторинга:
Попробуйте добавить больше воркеров и посмотреть, как это скажется на производительности:
$ docker-compose up -d —build —scale celery=3
Тесты
Давайте начнем с самого простого теста:
def test_task(): assert sample_tasks.create_task.run(1) assert sample_tasks.create_task.run(2) assert sample_tasks.create_task.run(3)
Добавьте тест-кейс выше в project/tests/test_tasks.py и допишите следующий импорт:
from tasks import sample_tasks
Запустите этот тест:
$ docker-compose exec web python -m pytest -k «test_task and not test_home»
Выполнение данного теста займет около минуты:
======================================== test session starts ======================================== platform linux — Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1 django: settings: core.settings (from ini) rootdir: /usr/src/app, inifile: pytest.ini plugins: django-3.8.0, celery-4.4.1 collected 2 items / 1 deselected / 1 selected tests/test_tasks.py . [100%] ============================ 1 passed, 1 deselected in 62.43s (0:01:02) =============================
Стоит отметить, что в assert’ах выше мы использовали метод .run вместо .delay для непосредственного запуска задач, без использования воркера Celery.
Хотите использовать заглушки(mock), чтобы ускорить процесс?
from unittest.mock import patch, call
$ docker-compose exec web python -m pytest -k «test_mock_task» ======================================== test session starts ======================================== platform linux — Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1 django: settings: core.settings (from ini) rootdir: /usr/src/app, inifile: pytest.ini plugins: django-3.8.0, celery-4.4.1 collected 3 items / 2 deselected / 1 selected tests/test_tasks.py . [100%] ================================== 1 passed, 2 deselected in 1.13s ==================================
Видите? Теперь гораздо быстрее!
Как насчет полноценного интеграционного тестирования?
def test_task_status(client): response = client.post(reverse(«run_task»), ) content = json.loads(response.content) task_id = content[«task_id»] assert response.status_code == 202 assert task_id response = client.get(reverse(«get_status», args=[task_id])) content = json.loads(response.content) assert content == assert response.status_code == 200 while content[«task_status»] == «PENDING»: response = client.get(reverse(«get_status», args=[task_id])) content = json.loads(response.content) assert content ==
Помните, что этот тест использует того же брокера и бэкенд, что и в разработке. Вы можете создать новый экземпляр приложения Celery для тестирования:
app = celery.Celery(‘tests’, broker=CELERY_TEST_BROKER, backend=CELERY_TEST_BACKEND)
import json
И убедитесь в том, что тесты прошли успешно.
Сегодня мы познакомились с базовой настройкой Celery для выполнения долгосрочных заданий в приложении на Django. Вы должны отправлять в очередь обработки любые процессы, которые могут замедлить работу кода на стороне пользователя.
Также Celery можно использовать для выполнения повторяющихся задач и декомпозиции сложных ресурсоемких задач, чтобы распределить вычислительную нагрузку на несколько машин и уменьшить время выполнения и нагрузку на машину, которая обрабатывает запросы клиента.
Весь код вы можете найти в этом репозитории.
- Блог компании OTUS
- Python
- Программирование
- Django
Источник: habr.com