Qpid Transport Support

[bmbouter]

This PR adds support for Qpid as a valid transport type for Kombu, and includes code, unit tests, docstrings, doc changes, and dependency updates. This is very functional transport; it has been heavily tested with Celery, supports Celery events, and supports SSL. This is being used in the upcoming release of Pulp.

Issues related to this code can be routed to me for bugfix and troubleshooting.

The code

The bulk of the code itself is contained in the kombu.transport.qpid module.

NOTE: The code currently still does monkey patching of qpid.messaging. Current versions of qpid.messaging that are out in the wild (pypi, rhel6, fedora, etc) are missing two important patches, QPID-5637 and QPID-5557. Those patches have been accepted upstream, but not all places that provide Qpid bits contain those patches. This codebase introduces those patches at runtime, and eventually will be removed. It is safe to patch any version of Qpid 0.18+, and is safe to apply to an already patched installation.

The monkey patching has been tested currently to work with clients using 0.22, 0.24, and 0.26 versions of qpid.messaging, so it should produce the expected behavior no matter what version of tools it is patching. The patching has been tested against the versions currently on Pypi.

The code is flake8 compliant. You should be able to run everything as described on this celery page and it should look clean.

Unit Tests

The unit tests are all located in the kombu.tests.transport.test_qpid module. You can run all just these new tests using:

nosetests --with-coverage --cover-html -v kombu.tests.transport.test_qpid

To get any continuous integration environments to run these unit tests, you'll need to install the dependencies. Please install the dependencies in the CI environment by running:

pip install qpid-tools qpid-python

The Docs

The code itself contains a huge number of docstrings. They follow the syntax expectations of Kombu/Celery (line width, strict PEP8, etc).

The more general documentation has been updated in two important ways.

1. The transport has its own section which is partially from the docstrings. This contains important things like instructions on how to download dependencies.
2. Updates to things that are no longer true. For instance the docs use to say things like "only rabbitMQ supports SSL" which is no longer true.

You can build the docs using these instructions.

You also should build the API reference since most of the docs live in there. You can follow these instructions

Dependencies

This library relies on the pure-python packages named 'qpidtoollibs' and 'qpid.messaging'. Kombu expects all pure python dependencies to be put onto pypi, and they can be "bundled" into Kombu as a group of optional dependencies.

The files introduced have all imports guarded against ImportError Exceptions so that you can import any file without installing the dependencies, and you'll encounter no Exceptions.

qpidtoollibs is currently published on pypi at version 0.26 here.
qpid.messaging is currently published on pypi at version 0.26 here.

A bundle in Kombu has been created called 'qpid' which should allow someone to install the bundle using kombu[qpid]. The docs also explain how to do this.

[coveralls]

Coverage increased (+13.27%) when pulling 38cd23a on pulp:qpid-transport into 41791db on celery:master.

[coveralls]

Coverage increased (+13.27%) when pulling 38cd23a on pulp:qpid-transport into 41791db on celery:master.

[coveralls]

Coverage increased (+13.27%) when pulling 65fa159 on pulp:qpid-transport into 41791db on celery:master.

[coveralls]

Coverage increased (+13.27%) when pulling 65fa159 on pulp:qpid-transport into 41791db on celery:master.

[coveralls]

Coverage increased (+13.27%) when pulling 7b546d9 on pulp:qpid-transport into 146758c on celery:master.

[coveralls]

Coverage increased (+13.27%) when pulling 7b546d9 on pulp:qpid-transport into 146758c on celery:master.

[coveralls]

Coverage increased (+1.87%) when pulling 3054b74 on pulp:qpid-transport into e44ec99 on celery:master.

[coveralls]

Coverage increased (+1.87%) when pulling 3054b74 on pulp:qpid-transport into e44ec99 on celery:master.

[bmbouter]

As we talked about, the tests now pass on Travis without any 3rd part libraries needing to be installed!

It's almost 4K lines, but it's almost entirely in the three files that this PR introduces. Given your time availability, a deep review doesn't seem feasible. It's been tested heavily by two different QE teams, over 20 developers, and is included in a product being released very soon. We feel it is high quality. Every line of code introduced has also been reviewed by other RedHat developers as part of our internal process. As an aside, I'm committed to fixing defects that are discovered and adding features over time.

Could this please be merged? If you have any specific feedback or questions please let me know. After merging I can open a second PR to backport the changes to the earlier (Python 2.6x) Kombu that has already been released.

As always, thanks for your consideration and for creating such a great project. It's powering a RedHat product!

[bmbouter]

As we talked about, the tests now pass on Travis without any 3rd part libraries needing to be installed!

It's almost 4K lines, but it's almost entirely in the three files that this PR introduces. Given your time availability, a deep review doesn't seem feasible. It's been tested heavily by two different QE teams, over 20 developers, and is included in a product being released very soon. We feel it is high quality. Every line of code introduced has also been reviewed by other RedHat developers as part of our internal process. As an aside, I'm committed to fixing defects that are discovered and adding features over time.

Could this please be merged? If you have any specific feedback or questions please let me know. After merging I can open a second PR to backport the changes to the earlier (Python 2.6x) Kombu that has already been released.

As always, thanks for your consideration and for creating such a great project. It's powering a RedHat product!

[ask]

Isn't read blocking here? I don't see anything doing setblocking()?

It should be set non-blocking, and since this is the first read call it needs to catch if getattr(exc, errno) == errno.EAGAIN: raise ConnectionError.

[ask]

I guess qpid does not support async use, if using a non-blocking socket is difficult then you can keep the blocking behavior. Not optimal though, since it can cause pauses in the worker for big messages especially.

[bmbouter]

This read call is non blocking. The blocking behavior of the file descriptor is set at line 1398 using fnctl.

qpid.messaging is the library that is used, and it does support asyncrhonous message fetching. My use of it is consistent with that and expects qpid.messaging to be fetching messages behind the scenes asynchronously.

Given that the os.read() is not reading from qpid.messaging, do you expect it to raise errno.EAGAIN? If so in what situations? I couldn't think of any, but I could catch that exception like you described if there is a possibility it could be raised.

[ask]

It's better to have the worker block on self._session.next_receiver() than to start a thread. Eventually a process will be forked at the wrong time and the worker deadlocks.

[bmbouter]

If I eliminated the thread altogether, where would the self._session.next_receiver() be moved to? I can only think of two options, and neither one works.

1. Move it to the AIO callback function. In that case nothing would ever write to the file descriptor that celery is actually monitoring, so the blocking read would occur correctly except that on_readable would never be called. Qpid.messaging does not provide a file descriptor to monitor by so it's up to my code to create that construct for celery.
2. Move it to the drain_events method and make it a synchronous transport. That won't work because our code experiences an actual task cancellation bug #2070, and synchronous transports are unsupported and slow anyway.