Issue 1:
Healthcheck fails after a few tries about in 5 minutes, and when I ping all the workers using

celery insect ping -A project

I don't find the worker, I find the other ones by the way.

I have three dockerized workers, using Amazon ECS two of which have light work to do and are not dependent on a VPN so I'm using fargate and the one that gives me problems is on EC2 using host networking mode in a server with a VPN. The other two workers don't have the same start up problem.

This worker has a beat that makes the worker do some work to update some sensors statuses every minute. It uses group calls to do parallel work.

REPORT with part of django settings

software -> celery:4.2.1 (windowlicker) kombu:4.2.1 py:3.6.8
billiard:3.5.0.4 redis:2.10.6
platform -> system:Linux arch:64bit imp:CPython
loader -> celery.loaders.app.AppLoader
settings -> transport:redis results:disabled

CELERY_ACCEPT_CONTENT: ['application/json']
CELERY_BEAT_SCHEDULE: {
'curator_indexes_migration': { 'schedule': <crontab: 0 4 * * * (m/h/d/dM/MY)>,
'task': 'sensor.tasks.statuses.curator_indexes_migration'},
'fetch_all_configurations_backup': { 'schedule': <crontab: 0 9 * * * (m/h/d/dM/MY)>,
'task': 'sensor.tasks.statuses.fetch_all_configurations_backup'},
'update_images': { 'schedule': <crontab: 0 * * * * (m/h/d/dM/MY)>,
'task': 'sensor.tasks.statuses.update_images'},
'update_statuses': { 'schedule': <crontab: * * * * * (m/h/d/dM/MY)>,
'task': 'sensor.tasks.statuses.update_statuses'}}
CELERY_BROKER_URL: 'redis://safety-staging.h1q2ld.0001.euw1.cache.amazonaws.com:6379/1'
CELERY_RESULT_BACKEND: None
CELERY_RESULT_SERIALIZER: 'json'
CELERY_TASK_ROUTES: {
'notification.tasks.*': {'queue': 'notification'},
'sensor.tasks.*': {'queue': 'sensor'}}
CELERY_TASK_SERIALIZER: 'json'
CELERY_TIMEZONE: 'Europe/Rome'

part of requirements.txt concerning celery and redis

aioredis==1.2.0
celery==4.2.1
Django==1.11
hiredis==0.2.0
kombu==4.2.1
redis==2.10.6

start script

#!/usr/bin/env bash
CELERY_DOMAIN=$(cat /dev/urandom | tr -dc 'a-zA-Z0-9' | fold -w 32 | head -n 1)
echo ${CELERY_DOMAIN} > /home/user/celery_domain.txt
printf "Celery domain is %s\n" "$CELERY_DOMAIN"
celery worker -A centroservizi -l INFO --autoscale=4,2 -Q sensor -n sensor@${CELERY_DOMAIN}

health check script

#!/usr/bin/env bash
CELERY_DOMAIN=$(cat /home/user/celery_domain.txt)
printf "Celery domain is %s\n" "$CELERY_DOMAIN"
celery inspect ping -A centroservizi -d sensor@${CELERY_DOMAIN}

Start log

Celery domain is G9ehGHbxIlsYFPb1JeW6Pdlar0gMIUue
[2019-01-08 07:14:57,184: INFO/MainProcess] Connected to redis://<elastic cache>.0001.euw1.cache.amazonaws.com:6379/1
[2019-01-08 07:14:57,200: INFO/MainProcess] mingle: searching for neighbors
[2019-01-08 07:14:58,306: INFO/MainProcess] mingle: sync with 2 nodes
[2019-01-08 07:14:58,307: INFO/MainProcess] mingle: sync complete
[2019-01-08 07:14:58,332: INFO/MainProcess] sensor@G9ehGHbxIlsYFPb1JeW6Pdlar0gMIUue ready.
[2019-01-08 07:14:59,026: INFO/MainProcess] Received task: sensor.tasks.statuses.update_sensor_status[2fa6e725-0a24-42c9-9171-572d7db3d31e]
[2019-01-08 07:14:59,033: INFO/MainProcess] Received task: sensor.tasks.statuses.update_sensor_status[296aaeb7-926a-4d1b-b62d-cec1783dbc70]
[2019-01-08 07:14:59,036: INFO/MainProcess] Received task: sensor.tasks.statuses.update_sensor_status[43cd23b1-5304-41df-ac82-1168cf53da60]
[2019-01-08 07:14:59,036: INFO/MainProcess] Scaling up 1 processes.
[2019-01-08 07:14:59,204: WARNING/MainProcess] consumer: Connection to broker lost. Trying to re-establish the connection...
Traceback (most recent call last):
File "/usr/local/lib/python3.6/site-packages/redis/connection.py", line 590, in send_packed_command
self._sock.sendall(item)
BrokenPipeError: [Errno 32] Broken pipe
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/usr/local/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 317, in start
blueprint.start(self)
File "/usr/local/lib/python3.6/site-packages/celery/bootsteps.py", line 119, in start
step.start(parent)
File "/usr/local/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 593, in start
c.loop(*c.loop_args())
File "/usr/local/lib/python3.6/site-packages/celery/worker/loops.py", line 91, in asynloop
next(loop)
File "/usr/local/lib/python3.6/site-packages/kombu/asynchronous/hub.py", line 276, in create_loop
tick_callback()
File "/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py", line 1033, in on_poll_start
cycle_poll_start()
File "/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py", line 315, in on_poll_start
self._register_BRPOP(channel)
File "/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py", line 301, in _register_BRPOP
channel._brpop_start()
File "/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py", line 707, in _brpop_start
self.client.connection.send_command('BRPOP', *keys)
File "/usr/local/lib/python3.6/site-packages/redis/connection.py", line 610, in send_command
self.send_packed_command(self.pack_command(*args))
File "/usr/local/lib/python3.6/site-packages/redis/connection.py", line 603, in send_packed_command
(errno, errmsg))
redis.exceptions.ConnectionError: Error 32 while writing to socket. Broken pipe.

If I can give you other info on the issue just ask

Issue 2:
Versions

Kombu: 4.0.2

Celery: 4.0.2

redis-py: 2.10.5

Redis: 3.2.6

Python: 2.7.12

Steps to reproduce

Start a redis server with timeout 1, I use the following config:

daemonize yes
pidfile ./redis.pid
port 0
unixsocket /tmp/celery.redis.test.sock
unixsocketperm 755
timeout 1
loglevel notice
logfile ./redis.log
databases 1

(I shove the unix socket into /tmp/ because there's a limit to how long a unix socket path can be.)

I put the Celery configuration in celeryconfig.py:

CELERY_BROKER_URL = 'redis+socket:///tmp/celery.redis.test.sock'
CELERY_RESULT_BACKEND = CELERY_BROKER_URL

The Celery app is in tasks.py:

import sys
import time

from celery import Celery, Task

app = Celery('tasks')
app.config_from_object('celeryconfig', namespace='CELERY')

@app.task
def add(x, y):
return x + y

I have test.py:

from celery.bin.multi import MultiTool

from tasks import app, add

workers = [
"A",
"B"
]

while True:
for worker in workers:
retcode = MultiTool().execute_from_commandline(["multi", "start", "-A",
"tasks", worker])
print "STARTED {0} WITH {1}".format(worker, retcode)

print "PING AFTER START", app.control.inspect().ping()
print add.delay(1, 2).get()
print "PING AFTER TASK", app.control.inspect().ping()
for worker in workers:

retcode = MultiTool().execute_from_commandline(["multi", "stopwait", "-A",
"tasks", worker])
print "STOPPED {0} WITH {1}".format(worker, retcode)
print "PING AFTER STOP", app.control.inspect().ping()

Run python test.py.

Expected Results

Should run forever without a crash.

Actual Results

Crashes at the end of the first iteration:

PING AFTER STOP
Traceback (most recent call last):
File "test.py", line 24, in <module>
print "PING AFTER STOP", app.control.inspect().ping()
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/celery/app/control.py", line 113, in ping
return self._request('ping')
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/celery/app/control.py", line 81, in _request
timeout=self.timeout, reply=True,
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/celery/app/control.py", line 436, in broadcast
limit, callback, channel=channel,
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/pidbox.py", line 321, in _broadcast
channel=chan)
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/pidbox.py", line 360, in _collect
self.connection.drain_events(timeout=timeout)
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/connection.py", line 301, in drain_events
return self.transport.drain_events(self.connection, **kwargs)
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/virtual/base.py", line 961, in drain_events
get(self._deliver, timeout=timeout)
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/redis.py", line 352, in get
self._register_BRPOP(channel)
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/redis.py", line 301, in _register_BRPOP
channel._brpop_start()
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/redis.py", line 707, in _brpop_start
self.client.connection.send_command('BRPOP', *keys)
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/redis/connection.py", line 563, in send_command
self.send_packed_command(self.pack_command(*args))
File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/redis/connection.py", line 556, in send_packed_command
(errno, errmsg))
redis.exceptions.ConnectionError: Error 32 while writing to socket. Broken pipe.

Observations

The code above in test.py imitates a sequence of operations that happen while testing an actual Django application of mine. The test suite starts and stops workers and executes some tasks on them.

Though the problem surfaced after I upgraded to Celery 4.x and thus started uising Kombu 4.x, a summary inspection of Kombu's source code in the 3.x series suggests the problem is present there too. It is unclear to me why I did not run into it in when running Kombu 3.x.

In my actual Redis setup I do not use a timeout 1 setting. Using timeout 1 is an easy way to cause Redis to close a connection. Other situations may be because the tcp-keepalive timeout deemed a connection "dead". Or because a client violated an output buffer limit, or for some other reason. What is clear is that Redis clients should be resilient in the face of connections that got closed by the server.

Indeed, the code of redis-py will generally retry sending commands that fail. This can be seen in execute_command:

def execute_command(self, *args, **options):
"Execute a command and return a parsed response"
pool = self.connection_pool
command_name = args[0]
connection = pool.get_connection(command_name, **options)
try:
connection.send_command(*args)
return self.parse_response(connection, command_name, **options)
except (ConnectionError, TimeoutError) as e:
connection.disconnect()
if not connection.retry_on_timeout and isinstance(e, TimeoutError):
raise
connection.send_command(*args)
return self.parse_response(connection, command_name, **options)
finally:
pool.release(connection)

Kombu's Redis Channel will generally benefit from redis-py's automatic retrying because most of the time the methods it calls on its client ultimately run execute_command. However, _brpop_start calls self.client.connection.send_command('BRPOP', *keys). If this call fails, the failure is automatically sent up the stack without a retry.

I've been able to work around the issue by changing the code of _brpop_start to:

def _brpop_start(self, timeout=1):
queues = self._queue_cycle.consume(len(self.active_queues))
if not queues:
return
keys = [self._q_for_pri(queue, pri) for pri in self.priority_steps
for queue in queues] + [timeout or 0]
self._in_poll = self.client.connection
from redis.exceptions import ConnectionError
try:
self.client.connection.send_command('BRPOP', *keys)
except ConnectionError:
self.client.connection.send_command('BRPOP', *keys)