如何檢查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

回答

相關問題