如何捕获Ray Actor/Task的标准输出（stdout）日志？

由于Ray Actor/Task运行在独立进程中，当前进程的StreamHandler无法跨进程捕获其日志，以下是几种可行的解决方法：

1. 在Actor内部配置日志处理器 #

每个Actor进程独立配置日志捕获逻辑，将自身日志存入内存流后通过远程调用返回给驱动进程。

示例代码：

import logging
import ray
from io import StringIO

@ray.remote
class LoggingActor:
    def __init__(self):
        # 初始化Actor进程内的日志工具
        self.logger = logging.getLogger(__name__)
        self.log_stream = StringIO()
        handler = logging.StreamHandler(self.log_stream)
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.WARNING)  # 设置日志级别

    def log_message(self):
        self.logger.warning('这是来自Ray Actor的警告消息')
        # 返回捕获到的日志内容
        return self.log_stream.getvalue()

ray.init()
actor = LoggingActor.remote()
captured_output = ray.get(actor.log_message.remote())
print("捕获到的日志 -> ", captured_output, "<- 结束")

2. 利用Ray原生日志配置转发到驱动进程 #

通过Ray初始化参数和日志工具，将Actor/Task的日志直接输出到驱动进程的标准输出，再由驱动进程统一捕获。

示例代码：

import logging
import ray
from ray.util.logging import setup_logging

# 初始化Ray时开启日志转发到驱动进程
ray.init(log_to_driver=True)

# 统一配置全局日志格式与级别
setup_logging(
    logging_level=logging.WARNING,
    log_format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

@ray.remote
class LoggingActor:
    def log_message(self):
        logger = logging.getLogger(__name__)
        logger.warning('这是来自Ray Actor的警告消息')

actor = LoggingActor.remote()
ray.get(actor.log_message.remote())

此时Actor的日志会直接打印到驱动进程的终端，若需要捕获到内存流，可以在驱动进程中重定向sys.stdout或sys.stderr。

3. 自定义日志收集Actor实现集中管理 #

创建一个专门的日志收集Actor，让所有业务Actor将日志发送到该收集Actor，实现集群内所有日志的统一存储与查询。

示例代码：

import logging
import ray

@ray.remote
class LogCollectorActor:
    def __init__(self):
        self.log_records = []

    def add_log(self, log_content):
        self.log_records.append(log_content)

    def get_all_logs(self):
        return self.log_records

@ray.remote
class LoggingActor:
    def __init__(self, collector_actor):
        self.collector = collector_actor
        # 自定义日志处理器，将日志发送到收集Actor
        self.logger = logging.getLogger(__name__)
        custom_handler = logging.Handler()
        custom_handler.emit = self._send_to_collector
        self.logger.addHandler(custom_handler)
        self.logger.setLevel(logging.WARNING)

    def _send_to_collector(self, record):
        # 格式化日志并发送
        formatted_log = self.logger.format(record)
        ray.get(self.collector.add_log.remote(formatted_log))

    def log_message(self):
        self.logger.warning('这是来自Ray Actor的警告消息')

# 初始化日志收集Actor
log_collector = LogCollectorActor.remote()
# 初始化业务Actor并绑定收集器
actor = LoggingActor.remote(log_collector)
ray.get(actor.log_message.remote())
# 获取所有收集到的日志
all_logs = ray.get(log_collector.get_all_logs.remote())
print("捕获到的所有日志：", all_logs)

内容的提问来源于stack exchange，提问作者tz01x