手动返回 celery 任务的错误结果和状态失败？

Question

我创建了 celery 任务来运行一些通过 nodejs 用 javascript 编写的各种作业。该任务基本上是一个调用nodejs的subprocess.popen。

nodejs 作业在退出时将返回非零状态，以及写入 stderr 的错误信息。

发生这种情况时，我想获取标准错误，并将这些作为“结果”返回给 celery，以及 FAILURE 状态，这样我的作业监视器可以反映作业失败。

我该怎么做？

这是我的任务

@app.task
def badcommand():
    try:
       output = subprocess.check_output('ls foobar',stderr=subprocess.STDOUT,shell=True)
       return output
    except subprocess.CalledProcessError as er:
       #What do I do here to return er.output, and set the status to fail?

如果我没有捕捉到子进程异常，作业会正常失败，但结果是空的，我会得到一个回溯堆栈跟踪。

如果我捕捉到异常，并返回er.output，则作业成功完成。

Answer 1

您可以使用celery.app.task.Task.update_state 方法更新当前任务状态。

@app.task(bind=True)
def badcommand(self):
    try:
       output = subprocess.check_output('ls foobar',stderr=subprocess.STDOUT,shell=True)
       return output
    except subprocess.CalledProcessError as er:
        self.update_state(state='FAILURE', meta={'exc': er})

请注意，app.task 装饰器的 bind 参数是在 Celery 3.1 中引入的。如果您仍在使用旧版本，我认为您可以这样调用update_state 任务方法：

@app.task
def badcommand():
    ...
    except subprocess.CalledProcessError as er:
        badcommand.update_state(state='FAILURE', meta={'exc': er})    

Answer 2

您可以使用具有指定功能的基础来说明失败时的处理方式。

class YourBase(Task):
    def on_success(self, retval, task_id, args, kwargs):
        print "Failure"

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print "Success"

@app.task(base=YourBase)
def badcommand():
   output = subprocess.check_output('ls foobar', stderr=subprocess.STDOUT, shell=True)
   return output

这些是您的基类可以使用的处理程序：http://celery.readthedocs.org/en/latest/userguide/tasks.html#handlers