如何检查Celery/Supervisor是否正在运行使用Python

如何在Python中编写一个脚本，该脚本在计算机上运行时输出（Ubuntu）？如何检查Celery/Supervisor是否正在运行使用Python

我的用例。我有一个简单的Python文件，有一些任务。我没有使用Django或Flask。我使用主管来运行任务队列。例如，

tasks.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

这一切工作，我现在想，如果有芹菜/管理进程正在运行，检查页面。即类似这样的事情可能使用Flask允许我托管给出200状态的页面，以允许我进行负载平衡。

例如...

check_status.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()

来源

2015-10-30 Prometheus

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

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...')

来源

2015-11-05 13:30:19 Rotten194

我没有任何留言给今天，所以我通过评论upvoting这种方法。 – sobolevn

我不认为'app = celery.Celery（'tasks'，broker ='redis：//'）'是必须的。 'app'变量在其他地方不使用。 – djromero

@djromero已经有一段时间了，因为我检查了它是如何工作的，但是IIRC需要在库内部实例化单例应用程序。如果您在其他地方使用芹菜并且创建芹菜实例，则不需要。 – Rotten194

如何使用子进程，如果不知道这是一个好主意：

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

来源

2015-10-31 05:14:10

-2

以我的经验，我会设置一条消息来跟踪它是否完整，以便队列负责重试任务。

来源

2015-11-06 21:38:09 hematag

你可以从supervisorctl status输出

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'

来源

2015-11-08 18:49:23

稀疏的Web用户界面，配有主管解析process state。也许你可以使用它。它可以在管理员配置中启用。寻找的关键是[inet_http_server]

你甚至可以看看那件作品的源代码，以获得想法来实现你自己的想法。

来源

2015-11-10 12:51:24 gipsy

看来，这条线在Rotten194 's answer：

status.app = status.get_app()

应该

status.app = status.get_app(app)

来源

2016-10-26 04:31:57

这并不适用于芹菜，但任何人都在这里结束了，看看supervisord运行，检查您的supervisord.conf配置文件中是否存在为supervisord定义的pid文件。如果是这样，它正在运行;如果没有，它不是。默认的pidfile是/tmp/supervisord.pid，这是我在下面使用的。

import os 
import sys 

if os.path.isfile("/tmp/supervisord.pid"): 
    print "supervisord is running." 
    sys.exit()

来源

2017-11-28 00:16:39

如何检查Celery/Supervisor是否正在运行使用Python

回答

相关问题