5.0.7 SSL connection problem

[pavelion]

After upgrade to the 5.0.7 celery workers are unable to connect to RabbitMQ (3.9.11) using SSL. Worker dies immediately after start and following exceptions are logged. Reverting to 5.0.6 solves the connection problem.

2021-12-15 07:31:14 [11] CRITICAL Unrecoverable error: OperationalError('[Errno 32] Broken pipe') (/usr/local/lib/python3.9/site-packages/celery/worker/worker.py:start)
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/kombu/connection.py", line 524, in _ensured
    return fun(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/kombu/messaging.py", line 193, in _publish
    [maybe_declare(entity) for entity in declare]
  File "/usr/local/lib/python3.9/site-packages/kombu/messaging.py", line 193, in 
    [maybe_declare(entity) for entity in declare]
  File "/usr/local/lib/python3.9/site-packages/kombu/messaging.py", line 99, in maybe_declare
    return maybe_declare(entity, self.channel, retry, **retry_policy)
  File "/usr/local/lib/python3.9/site-packages/kombu/common.py", line 110, in maybe_declare
    return _maybe_declare(entity, channel)
  File "/usr/local/lib/python3.9/site-packages/kombu/common.py", line 150, in _maybe_declare
    entity.declare(channel=channel)
  File "/usr/local/lib/python3.9/site-packages/kombu/entity.py", line 180, in declare
    return (channel or self.channel).exchange_declare(
  File "/usr/local/lib/python3.9/site-packages/amqp/channel.py", line 608, in exchange_declare
    self.send_method(
  File "/usr/local/lib/python3.9/site-packages/amqp/abstract_channel.py", line 57, in send_method
    conn.frame_writer(1, self.channel_id, sig, args, content)
  File "/usr/local/lib/python3.9/site-packages/amqp/method_framing.py", line 183, in write_frame
    write(view[:offset])
  File "/usr/local/lib/python3.9/site-packages/amqp/transport.py", line 355, in write
    self._write(s)
  File "/usr/local/lib/python3.9/site-packages/amqp/transport.py", line 599, in _write
    n = write(s)
  File "/usr/local/lib/python3.9/ssl.py", line 1118, in write
    return self._sslobj.write(data)
BrokenPipeError: [Errno 32] Broken pipe

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/kombu/connection.py", line 447, in _reraise_as_library_errors
    yield
  File "/usr/local/lib/python3.9/site-packages/kombu/connection.py", line 536, in _ensured
    self.collect()
  File "/usr/local/lib/python3.9/site-packages/kombu/connection.py", line 366, in collect
    gc_transport(self._connection)
  File "/usr/local/lib/python3.9/site-packages/kombu/transport/pyamqp.py", line 173, in _collect
    connection.collect()
  File "/usr/local/lib/python3.9/site-packages/amqp/connection.py", line 470, in collect
    self._transport.close()
  File "/usr/local/lib/python3.9/site-packages/amqp/transport.py", line 275, in close
    self._shutdown_transport()
  File "/usr/local/lib/python3.9/site-packages/amqp/transport.py", line 564, in _shutdown_transport
    self.sock = self.sock.unwrap()
  File "/usr/local/lib/python3.9/ssl.py", line 1285, in unwrap
    s = self._sslobj.shutdown()
BrokenPipeError: [Errno 32] Broken pipe

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/celery/worker/worker.py", line 203, in start
    self.blueprint.start(self)
  File "/usr/local/lib/python3.9/site-packages/celery/bootsteps.py", line 116, in start
    step.start(parent)
  File "/usr/local/lib/python3.9/site-packages/celery/bootsteps.py", line 365, in start
    return self.obj.start()
  File "/usr/local/lib/python3.9/site-packages/celery/worker/consumer/consumer.py", line 326, in start
    blueprint.start(self)
  File "/usr/local/lib/python3.9/site-packages/celery/bootsteps.py", line 116, in start
    step.start(parent)
  File "/usr/local/lib/python3.9/site-packages/celery/worker/consumer/mingle.py", line 37, in start
    self.sync(c)
  File "/usr/local/lib/python3.9/site-packages/celery/worker/consumer/mingle.py", line 41, in sync
    replies = self.send_hello(c)
  File "/usr/local/lib/python3.9/site-packages/celery/worker/consumer/mingle.py", line 54, in send_hello
    replies = inspect.hello(c.hostname, our_revoked._data) or {}
  File "/usr/local/lib/python3.9/site-packages/celery/app/control.py", line 389, in hello
    return self._request('hello', from_node=from_node, revoked=revoked)
  File "/usr/local/lib/python3.9/site-packages/celery/app/control.py", line 106, in _request
    return self._prepare(self.app.control.broadcast(
  File "/usr/local/lib/python3.9/site-packages/celery/app/control.py", line 741, in broadcast
    return self.mailbox(conn)._broadcast(
  File "/usr/local/lib/python3.9/site-packages/kombu/pidbox.py", line 335, in _broadcast
    self._publish(command, arguments, destination=destination,
  File "/usr/local/lib/python3.9/site-packages/kombu/pidbox.py", line 303, in _publish
    producer.publish(
  File "/usr/local/lib/python3.9/site-packages/kombu/messaging.py", line 177, in publish
    return _publish(
  File "/usr/local/lib/python3.9/site-packages/kombu/connection.py", line 557, in _ensured
    errback and errback(exc, 0)
  File "/usr/local/lib/python3.9/contextlib.py", line 137, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/usr/local/lib/python3.9/site-packages/kombu/connection.py", line 451, in _reraise_as_library_errors
    raise ConnectionError(str(exc)) from exc
kombu.exceptions.OperationalError: [Errno 32] Broken pipe

[auvipy]

@pawl @michael-lazar can you verify this?

[pb-dod]

I just tried reproducing this issue by enabling SSL for rabbitmq on this branch and bumping py-amqp to 5.0.7: https://github.com/pb-dod/celery_pyamqp_memory_leak/tree/fix_ssl_2

I didn't see the immediate shutdown due to the BrokenPipeError.

It looks like we should put try/except OSError around this line though:

py-amqp/amqp/transport.py

Line 564 in d4c879f

self.sock = self.sock.unwrap()

Because it's also calling shutdown: https://github.com/python/cpython/blob/f4c03484da59049eb62a9bf7777b963e2267d187/Lib/ssl.py#L1322

[auvipy]

ow is it your another github ID? would appreciate another PR for that. along with that I want to release another version

[pb-dod]

I was able to reproduce the issue by restarting rabbitmq (with docker-compose restart rabbitmq) while running celery with SSL in my example.

Should have a PR up with a fix shortly. Oops, yeah this is @pawl's other github id.