Celery-Django uygulamasında çalışanlardan olaylar nasıl izlenir?

Question

real-time monitoring of celery workers ile ilgili kereviz öğreticisine göre, çalışanlar tarafından üretilen olayları programatik olarak yakalayabilir ve buna göre harekete geçebilirsiniz.

Sorum şu: Bir monitörü, bir Celery-Django uygulamasında this örneğindeki gibi nasıl entegre edebilirim?

DÜZENLEME:

from celery import Celery 

def my_monitor(app): 
    state = app.events.State() 

    def announce_failed_tasks(event): 
     state.event(event) 
     task_id = event['uuid'] 

     print('TASK FAILED: %s[%s] %s' % (
      event['name'], task_id, state[task_id].info(),)) 
    with app.connection() as connection: 
     recv = app.events.Receiver(connection, handlers={ 
       'task-failed': announce_failed_tasks, 
       'worker-heartbeat': announce_dead_workers, 
     }) 
     recv.capture(limit=None, timeout=None, wakeup=True) 

if __name__ == '__main__': 
    celery = Celery(broker='amqp://guest@localhost//') 
    my_monitor(celery) 

Yani için sonuç almak için, işçi tarafından gönderilen task_failed olayı yakalamak için, ve öğretici gösterileri gibi onun task_id almak istiyorum: eğitimde kod örneği benziyor Bu görev, uygulamam için yapılandırılmış ve daha fazla işleyen sonuç arka ucundan. Benim problemim, bir django-kereviz projesinde olduğu gibi, uygulamanın nasıl elde edilebileceğinin açık olmadığıdır.

Ayrıca, bir çalışan bir görevi yürütmeyi bitirdiğinde sonuçların nasıl işleneceği konusunda başka herhangi bir fikre de açığım.

Answer 1

Bunu yapmanın bir yolunu buldum, ancak bunun çözüm olduğundan emin değilim, ama benim için çalışıyor. Monitör işlevi temel olarak doğrudan aracıya bağlanır ve farklı olay türlerini dinler. Kodum şunun gibi:

from celery.events import EventReceiver 
from kombu import Connection as BrokerConnection 

def my_monitor: 
    connection = BrokerConnection('amqp://guest:guest@localhost:5672//') 

    def on_event(event): 
     print "EVENT HAPPENED: ", event 

    def on_task_failed(event): 
     exception = event['exception'] 
     print "TASK FAILED!", event, " EXCEPTION: ", exception 

    while True: 
     try: 
      with connection as conn: 
       recv = EventReceiver(conn, 
           handlers={'task-failed' : on_task_failed, 
              'task-succeeded' : on_event, 
              'task-sent' : on_event, 
              'task-received' : on_event, 
              'task-revoked' : on_event, 
              'task-started' : on_event, 
              # OR: '*' : on_event 
              }) 
      recv.capture(limit=None, timeout=None) 
    except (KeyboardInterrupt, SystemExit): 
     print "EXCEPTION KEYBOARD INTERRUPT" 
     sys.exit() 

Hepsi bu kadar. Ve bunu normal uygulamadan farklı bir süreçte çalıştırıyorum, yani kereviz uygulamasının sadece bu işlevi yerine getiren bir çocuk süreci oluşturduğunu. HTH

Answer 2

Sorunlar Eğer kereviz config CELERY_SEND_EVENTS bayrağı olarak gerçek ayarlamanız gerekir

1. bir çift sakının.
2. Ayrıca, etkinlik izleyicisini çalışanınızdan yeni bir iş parçacığına da ayarlayabilirsiniz. İşte

benim uygulamasıdır: Ben biraz daha spesifik olmak zorundasınız düşünüyorum

class MonitorThread(object): 
    def __init__(self, celery_app, interval=1): 
     self.celery_app = celery_app 
     self.interval = interval 

     self.state = self.celery_app.events.State() 

     self.thread = threading.Thread(target=self.run, args=()) 
     self.thread.daemon = True 
     self.thread.start() 

    def catchall(self, event): 
     if event['type'] != 'worker-heartbeat': 
      self.state.event(event) 

     # logic here 

    def run(self): 
     while True: 
      try: 
       with self.celery_app.connection() as connection: 
        recv = self.celery_app.events.Receiver(connection, handlers={ 
         '*': self.catchall 
        }) 
        recv.capture(limit=None, timeout=None, wakeup=True) 

      except (KeyboardInterrupt, SystemExit): 
       raise 

      except Exception: 
       # unable to capture 
       pass 

      time.sleep(self.interval) 

if __name__ == '__main__': 
    app = get_celery_app() # returns app 
    MonitorThread(app) 
    app.start()