流导致芹菜

Question

我正在尝试使用芹菜在一组服务器上安排和运行任务.每个任务都运行一段时间(几个小时),并涉及使用子进程使用给定的输入调用某个程序.该程序在stdout和stderr中产生大量输出.

是否有某种方法可以近乎实时地向客户端显示程序生成的输出？流输出,以便客户端可以观察服务器上运行的任务所产生的输出,而无需登录服务器？

Answer 1

您没有指定许多要求和约束.我假设你已经在某个地方有一个redis实例.

你可以做的是逐行读取其他进程的输出并通过redis发布:

这是一个可以将echo数据导入文件/tmp/foo以进行测试的示例:

import redis
redis_instance = redis.Redis()
p = subprocess.Popen(shlex.split("tail -f /tmp/foo"), stdout=subprocess.PIPE)

while True:
    line = p.stdout.readline()
    if line:
        redis_instance.publish('process log', line)
    else:
        break

在一个单独的过程中:

import redis

redis_instance = redis.Redis()
pubsub = redis_instance.pubsub()
pubsub.subscribe('process log')

while True:
    for message in pubsub.listen():
        print message  #  or use websockets to comunicate with a browser

如果您希望该过程结束,您可以例如在芹菜任务完成后发送"退出".

您可以使用不同的通道(字符串输入subscribe)来分隔不同进程的输出.

如果需要,您还可以将日志输出存储在redis中,

redis_instance.rpush('process log', message)

然后完全检索它.

Answer 2

我看到如何做的一种方法是编写将用于stderr和stdout的自定义Logger(参见文档:

from celery.app.log import Logger
Logger.redirect_stdouts_to_logger(MyLogger())

您的记录器可以将数据保存到数据库,Memcached,Redis或用于获取数据的任何共享存储.

我不确定记录器的结构,但我猜这样的东西会起作用:

from logging import Logger

class MyLogger(Logger):
    def log(lvl, msg):
        # Do something with the message