过早退出的芹菜工人不会调用on_failure

Question

我有以下代码：

class StatusTask(automata_celery.Task):
def on_success(self, retval, task_id, args, kwargs):
with app.app_context():
cloaker = Cloaker.query.get(args[0])
cloaker.status = RemoteStatus.LAUNCHED
db.session.commit()
def on_failure(self, exc, task_id, args, kwargs, einfo):
with app.app_context():
cloaker = Cloaker.query.get(args[0])
cloaker.status = RemoteStatus.ERROR
db.session.commit()

@celery.task(base=StatusTask)
def deploy_cloaker(cloaker_id):
"""To prevent launching while we are launching, we will
disable launching until the cloaker's status is LAUNCHED
"""
cloaker = Cloaker.query.get(cloaker_id)
if not cloaker.can_launch():
return
cloaker.status = RemoteStatus.LAUNCHING
db.session.commit()
host = cloaker.server.ssh_user + '@' + cloaker.server.ip
execute(fabric_deploy_cloaker, cloaker, hosts=host)

def fabric_deploy_cloaker(cloaker):
domain = cloaker.domain
sudo('rm -rf /var/www/%s/html' % domain)          # Restartable process
sudo('mkdir -p /var/www/%s/html' % domain)

当我为要 ssh 的结构提供错误的 IP 地址时 (1.2.3.4(，Celery worker 会过早退出，但不执行on_failure处理程序。

查看它在我的芹菜工人窗口上生成的日志：

[2017-07-31 01:04:45,231: WARNING/PoolWorker-8] [root@1.2.3.45] Executing task 'fabric_deploy_cloaker'
[2017-07-31 01:04:45,231: WARNING/PoolWorker-8] [root@1.2.3.45] sudo: rm -rf /var/www/google.com/html
[2017-07-31 01:04:55,328: WARNING/PoolWorker-8] Fatal error: Timed out trying to connect to 1.2.3.45 (tried 1 time)
Underlying exception:
timed out
[2017-07-31 01:04:55,328: WARNING/PoolWorker-8] Aborting.
[2017-07-31 01:04:59,126: ERROR/MainProcess] Task handler raised error: WorkerLostError('Worker exited prematurely: exitcode 0.',)
Traceback (most recent call last):
File "/Users/vng/.virtualenvs/AutomataHeroku/lib/python2.7/site-packages/billiard/pool.py", line 1224, in mark_as_worker_lost
human_status(exitcode)),
WorkerLostError: Worker exited prematurely: exitcode 0.

但是，当我检查此任务的状态时，我看到以下内容：state=FAILURE status=FAILURE message=Worker exited prematurely: exitcode 0.

如何优雅地处理此错误？

我的应用程序需要将 cloaker.status 设置为 LAUNCH 或 ERROR，以便我的最终用户可以手动重新启动此任务。

Answer 1

我在项目中遇到了同样的问题，并找到了两种可能的解决方法：

首先是避免重复(和同步！(celery.state和您自己的应用程序状态RemoteStatus.LAUNCHED。您必须存储任务的apply_async()或至少 id 中的AsyncResult。

其次是将导致WorkerLostError的操作包装为 try/例外：

host = cloaker.server.ssh_user + '@' + cloaker.server.ip
try:
assert_execute(fabric_deploy_cloaker, cloaker, hosts=host)
except Exception:
raise FabricDeployError("Something went wrong")
else:
execute(fabric_deploy_cloaker, cloaker, hosts=host)