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

Posted 2023-02-24

【中文标题】芹菜工人过早退出不会调用 on_failure

【英文标题】：Celery worker exited prematurely doesn't invoke on_failure

【发布时间】：2018-01-06 17:09:15

【问题描述】：

我有以下代码：

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)

当我为我的结构提供错误的 ip 地址以 ssh 到 (1.2.3.4) 时，Celery 工作人员过早退出但不执行 on_failure 处理程序。

查看它在我的 celery worker 窗口中生成的日志：

[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 设置为 LAUNCHED 或 ERROR，以便我的最终用户可以手动重新启动此任务。

【参考方案1】：

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

首先是避免celery.state 和您自己的应用状态RemoteStatus.LAUNCHED 的重复（和同步！）。您必须从apply_async() 或至少id of the task 存储AsyncResult。

第二个是将可能导致WorkerLostError的操作包装到try/except中：

  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)

【讨论】：

嗨，你从哪里得到assert_execute？我在 Fabric 中根本找不到那种方法。 execute 单独不会引发 Worker Timeout 异常

是的，不应该有这样的方法，因为我猜你会写。您必须弄清楚为什么要杀死您的工人（请参阅***.com/questions/22805079/…），然后为这种不良情况编写一种断言