aioluent 1.2.9

FluentSender接口

sender.FluentSender是Fluentd的结构化事件记录器。

默认情况下，记录器假定本地启动了fluentd守护进程。您也可以通过传递选项来指定远程记录器。

from aiofluent import sender

# for local fluent
logger = sender.FluentSender('app')

# for remote fluent
logger = sender.FluentSender('app', host='host', port=24224)

要发送事件，请使用您的事件调用emit方法。以下示例将事件发送到fluentd，带有标签‘app.follow’和属性‘from’和‘to’。

# Use current time
logger.emit('follow', {'from': 'userA', 'to': 'userB'})

# Specify optional time
cur_time = int(time.time())
logger.emit_with_time('follow', cur_time, {'from': 'userA', 'to':'userB'})

您可以通过emit的返回值来检测错误。如果在emit过程中发生错误，emit将返回False，并通过last_error方法获取错误对象。

if not logger.emit('follow', {'from': 'userA', 'to': 'userB'}):
    print(logger.last_error)
    logger.clear_last_error() # clear stored error after handled errors

如果您想关闭客户端，请调用close()方法。

logger.close()

基于事件的接口

此API是sender.FluentSender的包装器。

首先，您需要调用sender.setup()来创建全局的sender.FluentSender日志实例。此调用只需在应用程序开始时调用一次。

以下是基于事件的API的初始化代码

from aiofluent import sender

# for local fluent
sender.setup('app')

# for remote fluent
sender.setup('app', host='host', port=24224)

然后，请创建以下事件。这将带有标签‘app.follow’和属性‘from’和‘to’的事件发送到fluentd。

from aiofluent import event

# send event to fluentd, with 'app.follow' tag
event.Event('follow', {
  'from': 'userA',
  'to':   'userB'
})

event.Event有一个限制，不能返回成功/失败结果。

其他基于事件的接口方法。

sender.get_global_sender # get instance of global sender
sender.close # Call FluentSender#close

缓冲区溢出处理器

您可以在连接失败的情况下注入自己的自定义进程来处理缓冲区溢出。这将减少数据丢失，而不是简单地丢弃数据。

import msgpack
from io import BytesIO

def handler(pendings):
    unpacker = msgpack.Unpacker(BytesIO(pendings))
    for unpacked in unpacker:
        print(unpacked)

logger = sender.FluentSender('app', host='host', port=24224, buffer_overflow_handler=handler)

您应该在处理程序中处理任何异常。aiofluent忽略来自buffer_overflow_handler的异常。

在调用close()时存在挂起事件时，也将调用此处理程序。

Python日志处理程序接口

此客户端库还具有用于Python日志模块的FluentHandler类。

import logging
from aiofluent import handler

custom_format = {
  'host': '%(hostname)s',
  'where': '%(module)s.%(funcName)s',
  'type': '%(levelname)s',
  'stack_trace': '%(exc_text)s'
}

logging.basicConfig(level=logging.INFO)
l = logging.getLogger('fluent.test')
h = handler.FluentHandler('app.follow', host='host', port=24224)
formatter = handler.FluentRecordFormatter(custom_format)
h.setFormatter(formatter)
l.addHandler(h)
l.info({
  'from': 'userA',
  'to': 'userB'
})
l.info('{"from": "userC", "to": "userD"}')
l.info("This log entry will be logged with the additional key: 'message'.")

您还可以通过logging.config.dictConfig自定义格式化程序。

import logging.config
import yaml

with open('logging.yaml') as fd:
    conf = yaml.load(fd)

logging.config.dictConfig(conf['logging'])

以下是一个示例配置logging.yaml：

logging:
    version: 1

    formatters:
      brief:
        format: '%(message)s'
      default:
        format: '%(asctime)s %(levelname)-8s %(name)-15s %(message)s'
        datefmt: '%Y-%m-%d %H:%M:%S'
      fluent_fmt:
        '()': fluent.handler.FluentRecordFormatter
        format:
          level: '%(levelname)s'
          hostname: '%(hostname)s'
          where: '%(module)s.%(funcName)s'

    handlers:
        console:
            class : logging.StreamHandler
            level: DEBUG
            formatter: default
            stream: ext://sys.stdout
        fluent:
            class: fluent.handler.FluentHandler
            host: localhost
            port: 24224
            tag: test.logging
            formatter: fluent_fmt
            level: DEBUG
        none:
            class: logging.NullHandler

    loggers:
        amqp:
            handlers: [none]
            propagate: False
        conf:
            handlers: [none]
            propagate: False
        '': # root logger
            handlers: [console, fluent]
            level: DEBUG
            propagate: False