请教: 如何使用 nb_log 打印task_id

[anjia0532]

通过文档 https://funboost.readthedocs.io/zh/latest/articles/c4.html#funboost-current-task 已经知道通过 fct.function_result_status.task_id 获取当前任务id。
能否通过添加一个handler或者别的方式，把task_id加到logger的上下文中，进而实现任务中不用每个print/info都添加 task_id 参数，但是打印结果里有 task_id。

作用是，把某次task的日志可以通过 task_id 串起来。

import random
import time
from nb_log import get_logger

from funboost import boost, FunctionResultStatusPersistanceConfig,BoosterParams
from funboost.core.current_task import funboost_current_task
logger = get_logger(__name__)

@boost(BoosterParams(queue_name='queue_test_fct', qps=2,concurrent_num=5,))
def f(a, b):
    fct = funboost_current_task()
    task_id = fct.function_result_status.task_id
    # 添加一个handler或者别的方式，把 task_id 放到logger的上下文中

    logger.info(f"a={a}, b={b}")
    # 打印结果类似 包含 task id 以及代码中输出的 a=1,b=2

    logger.info(f"a+b={a}+{b}={a+b}")
    # 打印结果类似 包含 task id 以及代码中输出的 a+b=1+2=3
    return a + b


if __name__ == '__main__':

    for i in range(0, 200):
        f.push(i, b=i * 2)

    f.consume()

[anjia0532]

类似 https://cloud.tencent.com/developer/article/2318115
或者 open-telemetry/opentelemetry-python#3615

[anjia0532]

目前可以用 [p%(process)d_t%(thread)d] 感觉基本够用

[ydf0509]

主意不错，我可以加上这个，等我消息。

[anjia0532]

[抱拳] 多谢

[ydf0509]

试过了,自定义继承 Formatter类,是最方便的,也可以在handler里面实现,但是handler种类太多了实现起来改formattor更方便.已经测试可以了.

但是有个问题,nb_log的控制台屏幕流写入和文件写入和es mongo handlers等都可以设置异步批量写入以提高日志性能,在异步批量写入的情况下,真正写入和格式化消息时候所在的线程和协程 和用户当前logger.info(msgxx) 不在同一个线程/协程.

所以准备专门加个AdapterLogger类,自动在extra入参 里面机上taskid,

[ydf0509]

nb_log为了实现超高性能, 可以异步批量写入. 异步批量写入,就是当前线程只是把要记录的日志put到queue里面,另外有单独的线程从queue里面取出很多消息,然后一次性批量写入到文件或者屏幕.

[anjia0532]

期待

[ydf0509]

已经发布了.
修改了设置超时时间后,可以获取上下文的信息.
新增了 日志模板中自动显示task_id.

安装43.0版本

[ydf0509]

注意现有的配置文件的 NB_LOG_FORMATER_INDEX_FOR_CONSUMER_AND_PUBLISHER ,要改成新的 带task_id的模板

class FunboostCommonConfig(DataClassBase):
    # nb_log包的第几个日志模板，内置了7个模板，可以在你当前项目根目录下的nb_log_config.py文件扩展模板。
    # NB_LOG_FORMATER_INDEX_FOR_CONSUMER_AND_PUBLISHER = 11  # 7是简短的不可跳转，5是可点击跳转的，11是可显示ip 进程 线程的模板。
    NB_LOG_FORMATER_INDEX_FOR_CONSUMER_AND_PUBLISHER = logging.Formatter(
        f'%(asctime)s-({nb_log_config_default.computer_ip},{nb_log_config_default.computer_name})-[p%(process)d_t%(thread)d] - %(name)s - "%(filename)s:%(lineno)d" - %(funcName)s - %(levelname)s - %(task_id)s - %(message)s',
        "%Y-%m-%d %H:%M:%S",)   # 这个是带task_id的日子模板,日子可以显示task_id,方便用户窗帘排查某一个任务的所有日志.

[ydf0509]

测试例子如下

https://github.com/ydf0509/funboost/blob/9cb8d6940477bfb93ec86b6d4af13f0cb1975345/test_frame/test_funboost_current_task/test_current_task.py

[ydf0509]

[anjia0532]

嗯，可以了。
nb_log_config.py

# 忽略其他
FORMATTER_DICT = {
    # 忽略其他
    11: logging.Formatter(
        f'({computer_ip},{computer_name})-[p%(process)d_t%(thread)d] %(asctime)s - %(name)s - "%(filename)s:%(lineno)d" - %(funcName)s - %(levelname)s - %(message)s', "%Y-%m-%d %H:%M:%S"),  # 对7改进，带进程和线程显示的日志模板以及ip和主机名。
    12: logging.Formatter(
        f'%(asctime)s-({computer_ip},{computer_name})-[p%(process)d_t%(thread)d] - %(name)s - "%(filename)s:%(lineno)d" - %(funcName)s - %(levelname)s - %(task_id)s - %(message)s',
        "%Y-%m-%d %H:%M:%S",)   # 这个是带task_id的日子模板,日子可以显示task_id,方便用户窗帘排查某一个任务的所有日志.
}

FORMATTER_KIND = 11 
# 忽略其他

这里的 FORMATTER_KIND 不能用 带 %(task_id)s 的模板 ,否则会报 ValueError: Formatting field not found in record: 'task_id' ，因为有些非任务函数地方不包含 fct。

funboost_config.py

# 忽略其他
class FunboostCommonConfig(DataClassBase):
   # 这里可以用带 %(task_id)s 的
    NB_LOG_FORMATER_INDEX_FOR_CONSUMER_AND_PUBLISHER = 12
   # 忽略其他

test_current_task.py

import random
import time

from funboost.core.current_task import funboost_current_task
from funboost.core.task_id_logger import TaskIdLogger
from nb_log import LogManager
from nb_log_config import FORMATTER_DICT

LOG_FILENAME_QUEUE_FCT = 'funboost.log'
# 使用TaskIdLogger创建的日志配合带task_id的日志模板，每条日志会自动带上task_id，方便用户搜索日志，定位某一个任务id的所有日志。

logger = LogManager(__name__, logger_cls=TaskIdLogger).get_logger_and_add_handlers(
    formatter_template=FORMATTER_DICT[12], )


@boost(BoosterParams(queue_name='queue_test_fct', qps=2,concurrent_num=5,log_filename=LOG_FILENAME_QUEUE_FCT))
def f(a, b):
    fct = funboost_current_task() # 线程/协程隔离级别的上下文

    fct.logger.warning('如果不想亲自创建logger对象，可以使用fct.logger来记录日志，fct.logger是当前队列的消费者logger对象')

    time.sleep(2)
    if random.random() > 0.99:
        raise Exception(f'{a} {b} 模拟出错啦')
    logger.debug(f'哈哈 a: {a}')
    logger.debug(f'哈哈 b: {b}')
    logger.info(a+b)

    return a + b

if __name__ == '__main__':
    # f(5, 6)  # 可以直接调用

    for i in range(0, 200):
        f.push(i, b=i * 2)

    f.consume()
    f.multi_process_consume(2)

这样打印的会带 task_id 而且不用再没处打印都塞一个 extra={'task_id': xxx}

不过如果不自己定义 logger， 用 fct.logger 并且 funboost_config.py NB_LOG_FORMATER_INDEX_FOR_CONSUMER_AND_PUBLISHER=11 和 nb_log_config.py FORMATTER_KIND = 11 那么打印的也不含 task_id

[ydf0509]

是的,就没打算所有logger都用带taskid的日志模板啊.
就是日志模板和TaskIdLogger需要对应起来.

nb_log_config.py的FORMATTER_KIND 是全局默认日志模板, NB_LOG_FORMATER_INDEX_FOR_CONSUMER_AND_PUBLISHER 是专门为funboost 发布者消费者制定的日志模板,因为一般情况希望打印文件全路径,funboost发布者消费者希望只打印短文件路径.

[ydf0509]

如果所有logger都使用 FORMATTER_KIND 的模板,那就没必要 NB_LOG_FORMATER_INDEX_FOR_CONSUMER_AND_PUBLISHER 这个配置了.

[anjia0532]

嗯，是的，我只是说明下这个情况，防止有些人遇到后一脸懵

[ydf0509]

TaskIdLogger是在_log方法里面对extra自动添加了task_id ,如果获取不到task_id是 使用no_task_id替代.

普通的logging.Logger是没有自动添加 task_id到extra.

我给的例子还特意说了 logger = LogManager(name, logger_cls=TaskIdLogger).get_logger_and_add_handlers(....) 需要指定logger_cls

[ydf0509]

有说明

[ydf0509]

例如这个日志不在运行消费的上下文时候,会使用no_task_id替代.

所以主要是模板和logger_cls要对应起来, 用户可以在没有funboost_current_task上下文的地方使用日志打印,只是用no_task_id替代.

[anjia0532]

明白了