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)

• Fix scheduler listener on buster's celery version (4.1.0-4)
• celery_backend.listener: Access directly existing queue_arguments

• Rebase to latest master

• celery_backend.listener: Use debugging logging statements

Rebase

Thanks for looking into this!

After tracking down the two upstream changes that made the code break, the changes need to be adapted.

Merge request was returned for changes

mentioned in commit 339028cb

mentioned in commit d5473d6e

As per inline comments:

• listener: Drop unused queue_arguments settings
• listener: Deal with messages received as list

Thanks for looking into this! After tracking down the two upstream changes that made the code break, the changes need to be adapted.

Awesome. Thanks. Adapted accordingly.

Cheers,

Merge request was accepted

approved this merge request

Some references in the commit message have been migrated:

• D312 is now !5 (closed)

Some references in the commit message have been migrated:

• D312 is now !5 (closed)

Rebase

Merge request was merged

closed