Fix scheduler listener on buster's celery version (4.1.0-4)

Without this, the listener just fails as some internal function is no longer defined:

• Problem 1.

Traceback (most recent call last):
  File "/usr/lib/python3.6/runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "/usr/lib/python3.6/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/home/tony/work/inria/repo/swh/swh-environment/swh-scheduler/swh/scheduler/celery_backend/listener.py", line 178, in <module>
    main()
  File "/usr/lib/python3/dist-packages/click/core.py", line 722, in __call__
    return self.main(*args, **kwargs)
  File "/usr/lib/python3/dist-packages/click/core.py", line 697, in main
    rv = self.invoke(ctx)
  File "/usr/lib/python3/dist-packages/click/core.py", line 895, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/lib/python3/dist-packages/click/core.py", line 535, in invoke
    return callback(*args, **kwargs)
  File "/home/tony/work/inria/repo/swh/swh-environment/swh-scheduler/swh/scheduler/celery_backend/listener.py", line 174, in main
    event_monitor(main_app, backend=scheduler)
  File "/home/tony/work/inria/repo/swh/swh-environment/swh-scheduler/swh/scheduler/celery_backend/listener.py", line 149, in event_monitor
    node_id='listener-%s' % socket.gethostname(),
  File "/home/tony/work/inria/repo/swh/swh-environment/swh-scheduler/swh/scheduler/celery_backend/listener.py", line 30, in __init__
    queue_arguments=self._get_queue_arguments())
AttributeError: 'ReliableEventReceiver' object has no attribute '_get_queue_arguments'

• Problem 2.

Traceback (most recent call last):
  File "/usr/lib/python3.6/runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "/usr/lib/python3.6/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/home/tony/work/inria/repo/swh/swh-environment/swh-scheduler/swh/scheduler/celery_backend/listener.py", line 207, in <module>
    main()
  File "/usr/lib/python3/dist-packages/click/core.py", line 759, in __call__
    return self.main(*args, **kwargs)
  File "/usr/lib/python3/dist-packages/click/core.py", line 714, in main
    rv = self.invoke(ctx)
  File "/usr/lib/python3/dist-packages/click/core.py", line 951, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/lib/python3/dist-packages/click/core.py", line 552, in invoke
    return callback(*args, **kwargs)
  File "/home/tony/work/inria/repo/swh/swh-environment/swh-scheduler/swh/scheduler/celery_backend/listener.py", line 203, in main
    event_monitor(main_app, backend=scheduler)
  File "/home/tony/work/inria/repo/swh/swh-environment/swh-scheduler/swh/scheduler/celery_backend/listener.py", line 173, in event_monitor
    recv.capture(limit=None, timeout=None, wakeup=True)
  File "/usr/local/lib/python3.6/dist-packages/celery/events/receiver.py", line 92, in capture
    return list(self.consume(limit=limit, timeout=timeout, wakeup=wakeup))
  File "/usr/local/lib/python3.6/dist-packages/kombu/mixins.py", line 192, in consume
    conn.drain_events(timeout=safety_interval)
  File "/usr/local/lib/python3.6/dist-packages/kombu/connection.py", line 301, in drain_events
    return self.transport.drain_events(self.connection, **kwargs)
  File "/usr/local/lib/python3.6/dist-packages/kombu/transport/pyamqp.py", line 103, in drain_events
    return connection.drain_events(**kwargs)
  File "/usr/local/lib/python3.6/dist-packages/amqp/connection.py", line 491, in drain_events
    while not self.blocking_read(timeout):
  File "/usr/local/lib/python3.6/dist-packages/amqp/connection.py", line 497, in blocking_read
    return self.on_inbound_frame(frame)
  File "/usr/local/lib/python3.6/dist-packages/amqp/method_framing.py", line 77, in on_frame
    callback(channel, msg.frame_method, msg.frame_args, msg)
  File "/usr/local/lib/python3.6/dist-packages/amqp/connection.py", line 501, in on_inbound_method
    method_sig, payload, content,
  File "/usr/local/lib/python3.6/dist-packages/amqp/abstract_channel.py", line 128, in dispatch_method
    listener(*args)
  File "/usr/local/lib/python3.6/dist-packages/amqp/channel.py", line 1597, in _on_basic_deliver
    fun(msg)
  File "/usr/local/lib/python3.6/dist-packages/kombu/messaging.py", line 624, in _receive_callback
    return on_m(message) if on_m else self.receive(decoded, message)
  File "/usr/local/lib/python3.6/dist-packages/kombu/messaging.py", line 590, in receive
    [callback(body, message) for callback in callbacks]
  File "/usr/local/lib/python3.6/dist-packages/kombu/messaging.py", line 590, in <listcomp>
    [callback(body, message) for callback in callbacks]
  File "/home/tony/work/inria/repo/swh/swh-environment/swh-scheduler/swh/scheduler/celery_backend/listener.py", line 47, in _receive
    type, body = self.event_from_message(body)
  File "/usr/local/lib/python3.6/dist-packages/celery/events/receiver.py", line 103, in event_from_message
    type = body['type']
TypeError: list indices must be integers or slices, not str

version:

$ dpkg -l python3-celery
Desired=Unknown/Install/Remove/Purge/Hold
| Status=Not/Inst/Conf-files/Unpacked/halF-conf/Half-inst/trig-aWait/Trig-pend
|/ Err?=(none)/Reinst-required (Status,Err: uppercase=bad)
||/ Name                                                           Version                              Architecture                         Description
+++-==============================================================-====================================-====================================-=================================================================================================================================
ii  python3-celery                                                 4.1.0-4                              all                                  async task/job queue based on message passing (Python3 version)

---

Migrated from D312 (view on Phabricator)

swh/scheduler/celery_backend/listener.py

@@ -5,6 +5,7 @@
 
 import click
 import datetime
+import logging
 import socket
 
 from arrow import utcnow
@@ -26,22 +27,29 @@ class ReliableEventReceiver(EventReceiver):
                            exchange=self.exchange,
                            routing_key=self.routing_key,
                            auto_delete=False,
-                           durable=True,
-                           queue_arguments=self._get_queue_arguments())
+                           durable=True)
 
     def get_consumers(self, consumer, channel):
         return [consumer(queues=[self.queue],
                          callbacks=[self._receive], no_ack=False,
                          accept=self.accept)]
 
-    def _receive(self, body, message):
-        type, body = self.event_from_message(body)
-        self.process(type, body, message)
+    def _receive(self, bodies, message):
+        logging.debug('## event-receiver: bodies: %s' % bodies)
+        logging.debug('## event-receiver: message: %s' % message)
+        if not isinstance(bodies, list):  # celery<4 returned body as element
+            bodies = [bodies]
+        for body in bodies:
+            type, body = self.event_from_message(body)
+            self.process(type, body, message)
 
     def process(self, type, event, message):
         """Process the received event by dispatching it to the appropriate
         handler."""
         handler = self.handlers.get(type) or self.handlers.get('*')
+        logging.debug('## event-receiver: type: %s' % type)
+        logging.debug('## event-receiver: event: %s' % event)
+        logging.debug('## event-receiver: handler: %s' % handler)
         handler and handler(event, message)
 
 
@@ -93,6 +101,9 @@ def event_monitor(app, backend):
         try_perform_actions()
 
     def task_started(event, message):
+        logging.debug('#### task_started: event: %s' % event)
+        logging.debug('#### task_started: message: %s' % message)
+
         queue_action({
             'action': 'start_task_run',
             'args': [event['uuid']],
@@ -106,8 +117,11 @@ def event_monitor(app, backend):
         })
 
     def task_succeeded(event, message):
+        logging.debug('#### task_succeeded: event: %s' % event)
+        logging.debug('#### task_succeeded: message: %s' % message)
         result = event['result']
 
+        logging.debug('#### task_succeeded: result: %s' % result)
         try:
             status = result.get('status')
             if status == 'success':
@@ -127,6 +141,9 @@ def event_monitor(app, backend):
         })
 
     def task_failed(event, message):
+        logging.debug('#### task_failed: event: %s' % event)
+        logging.debug('#### task_failed: message: %s' % message)
+
         queue_action({
             'action': 'end_task_run',
             'args': [event['uuid']],
@@ -159,7 +176,12 @@ def event_monitor(app, backend):
     '--database', '-d', help='Scheduling database DSN')
 @click.option('--url', '-u',
               help="(Optional) Scheduler's url access")
-def main(cls, database, url):
+@click.option('--verbose', is_flag=True, default=False,
+              help='Default to be silent')
+def main(cls, database, url, verbose):
+    if verbose:
+        logging.basicConfig(level=logging.DEBUG if verbose else logging.INFO)
+
     scheduler = None
     override_config = {}
     if cls == 'local':