如何通过Python检查Celery/Supervisor是否正在运行

Question

如果芹菜在机器上运行（Ubuntu），如何用Python编写输出脚本？

我的用例。我有一个简单的python文件和一些任务。我没有用姜黄或烧瓶。我使用主管来运行任务队列。例如

任务.py

from celery import Celery, task
app = Celery('tasks')
@app.task()
def add_together(a, b):
    return a + b

主管：

[program:celery_worker]
directory = /var/app/
command=celery -A tasks worker info

这一切工作，我现在想有网页，检查芹菜/主管进程是否正在运行。i、 像这样的东西可能使用烧瓶允许我主持一个200状态的页面，允许我负载平衡。

例如。。。

检查状态.py

from flask import Flask

app = Flask(__name__)

@app.route('/')
def status_check():

    #check supervisor is running
    if supervisor:
         return render_template('up.html')
    else:
        return render_template('down.html')

if __name__ == '__main__':
    app.run()

Answer 1

不如使用子流程，不确定它是否是一个好主意：

>>> import subprocess
>>> output = subprocess.check_output('ps aux'.split())
>>> 'supervisord' in output
True

Answer 2

通过导入celery.bin.celery包，可以通过代码运行celery status命令：

import celery
import celery.bin.base
import celery.bin.celery
import celery.platforms

app = celery.Celery('tasks', broker='redis://')

status = celery.bin.celery.CeleryCommand.commands['status']()
status.app = status.get_app()

def celery_is_up():
    try:
        status.run()
        return True
    except celery.bin.base.Error as e:
        if e.status == celery.platforms.EX_UNAVAILABLE:
            return False
        raise e

if __name__ == '__main__':
    if celery_is_up():
        print('Celery up!')
    else:
        print('Celery not responding...')

Answer 3

可以从supervisorctl status输出解析process state

import subprocess

def is_celery_worker_running():
    ctl_output = subprocess.check_output('supervisorctl status celery_worker'.split()).strip()
    if ctl_output == 'unix:///var/run/supervisor.sock no such file':
        # supervisord not running
        return False
    elif ctl_output == 'No such process celery_worker':
        return False
    else:
        state = ctl_output.split()[1]
        return state == 'RUNNING'