Làm thế nào để theo dõi các sự kiện từ công nhân trong ứng dụng Celery-Django?

Question

Theo hướng dẫn cần tây liên quan đến real-time monitoring of celery workers, người ta cũng có thể nắm bắt theo chương trình các sự kiện do công nhân sản xuất và thực hiện hành động tương ứng.

Câu hỏi của tôi là làm cách nào tôi có thể tích hợp màn hình dưới dạng màn hình trong ví dụ this, trong ứng dụng Celery-Django?

EDIT: Các mã ví dụ trong hướng dẫn trông giống như:

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) 

Vì vậy, tôi muốn chụp sự kiện task_failed gửi bởi người lao động, và để có được task_id của nó giống như các chương trình hướng dẫn, để có được những kết quả cho nhiệm vụ này từ backend kết quả đã được cấu hình cho ứng dụng của tôi và xử lý nó thêm nữa. Vấn đề của tôi là nó không phải là rõ ràng với tôi làm thế nào để có được các ứng dụng, như trong một dự án django-cần tây nó không phải là minh bạch cho tôi sự khởi tạo của thư viện Celery.

Tôi cũng mở cho bất kỳ ý tưởng nào khác về cách xử lý kết quả khi một công nhân đã hoàn thành việc thực hiện một tác vụ.

Answer 1

Ok, tôi đã tìm ra cách để làm điều này, mặc dù tôi không chắc chắn đây là giải pháp, nhưng nó hoạt động cho tôi. Chức năng giám sát về cơ bản kết nối trực tiếp với người môi giới và lắng nghe các loại sự kiện khác nhau. Mã của tôi trông giống như sau:

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

Đây là tất cả. Và tôi chạy điều này trong một quy trình khác với ứng dụng bình thường, có nghĩa là tôi tạo ra một tiến trình con của ứng dụng cần tây của tôi chỉ chạy chức năng này. HTH

Answer 2

Cẩn thận với một vài gotchas

1. Bạn cần phải thiết lập CELERY_SEND_EVENTS cờ như đúng trong cấu hình cần tây bạn.
2. Bạn cũng có thể đặt trình giám sát sự kiện trong một chủ đề mới từ nhân viên của mình.

Đây là triển khai thực hiện của tôi:

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