Jak monitorować zdarzenia od pracowników w aplikacji Celery-Django?

Question

Zgodnie z samouczkiem selerem dotyczącym real-time monitoring of celery workers można również programowo uchwycić zdarzenia wygenerowane przez pracowników i podjąć odpowiednie działania.

Moje pytanie brzmi: jak mogę zintegrować monitor z przykładem w przykładzie this w aplikacji Celery-Django?

EDIT: Przykładowy kod w samouczku wygląda następująco:

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) 

Więc chcę uchwycić task_failed zdarzenia wysłane przez pracownika, a także uzyskać jego task_id jak poradnik pokazuje, aby uzyskać wynik dla to zadanie z zaplecza, które zostało skonfigurowane dla mojej aplikacji i dalej ją przetwarzać. Moim problemem jest to, że nie jest dla mnie oczywiste, jak uzyskać aplikację, ponieważ w projekcie django-selekcji nie jest dla mnie przejrzyste tworzenie biblioteki Selera.

Jestem również otwarty na wszelkie inne pomysły dotyczące przetwarzania wyników, gdy pracownik zakończył wykonywanie zadania.

Answer 1

Ok, znalazłem sposób na zrobienie tego, choć nie jestem pewien, czy to jest rozwiązanie, ale działa dla mnie. Funkcja monitora w zasadzie łączy się bezpośrednio z brokerem i nasłuchuje różnych typów zdarzeń. Mój kod wygląda następująco:

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

To wszystko. I uruchamiam to w innym procesie niż normalna aplikacja, co oznacza, że ​​tworzę proces potomny mojej aplikacji do selera, która tylko uruchamia tę funkcję. HTH

Answer 2

Strzeż się kilka pułapek

1. Musisz ustawić CELERY_SEND_EVENTS flagę jako prawdziwy w swojej selera config.
2. Możesz także ustawić monitor zdarzeń w nowym wątku od swojego pracownika.

Oto moja realizacja:

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