Rabbit-logger
rabbit-logger
— это библиотека для интеграции APM (Application Performance Monitoring) и логирования с RabbitMQ. Она предоставляет интерфейс для отслеживания метрик функций и удобное логирование через синхронные и асинхронные обработчики. Логи и метрики записываются в RabbitMq и в дальнейшем вы можете развернуть у себя образ из Dockerhub [клик] (тут будет ссылка), где данные логи будут записываться в БД clickhouse, и с помощью интерфейса Metabase вы сможете просматривать логи и информацию о своих функциях из своих сервисов.
Установка
Установите библиотеку с помощью pip:
pip install rabbit-logger
Настройка логирования
Добавьте обработчики в конфигурацию LOGGING
. Вы можете выбрать синхронный или асинхронный обработчик.
Пример настройки:
from logging.config import dictConfig
import logging
LOGGING = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"verbose": {"format": "%(levelname)s %(asctime)s %(module)s %(message)s"},
"simple": {"format": "%(levelname)s %(message)s"},
},
"handlers": {
"rabbit-logger": {
"level": "INFO",
"class": "rabbit_logger.logger.udp_handler.UDPSyncLoggerHandler",
"udp_host": "0.0.0.0",
"udp_port": 9999,
"server_name": "test",
},
"console": {
"level": "DEBUG",
"class": "logging.StreamHandler",
"formatter": "simple",
},
},
"loggers": {
"my_app": {
"handlers": ["rabbit-logger", "console"],
"level": "INFO",
"propagate": True,
},
},
}
dictConfig(LOGGING)
logger = logging.getLogger("my_app")
logger.setLevel(logging.DEBUG)
logger.info("Это информационное сообщение!")
logger.error("Ошибка!")
Использование APM
Синхронный APM
Для синхронного мониторинга создайте экземпляр SyncAPMHandler
:
from rabbit_logger.apm.udp_apm import UdpAPMHandler
UDPHandler = UdpAPMHandler(
udp_host="localhost",
udp_port=9999,
server_name="test"
)
@UDPHandler.apm
def my_sync_function():
print("Выполнение функции...")
time.sleep(2)
return "Готово!"
result = my_sync_function()
print(result)
Асинхронный APM
Для асинхронного мониторинга используйте AsyncAPMHandler
:
import asyncio
from rabbit_logger.apm.async_apm import AsyncAPMHandler
APMHandler = AsyncAPMHandler(
rabbit_host="localhost",
rabbit_port=5672,
rabbit_user="user",
rabbit_password="password",
server_name="my_server"
)
async def main():
await APMHandler.connect()
@APMHandler.apm
async def my_async_function():
print("Выполнение асинхронной функции...")
await asyncio.sleep(2)
return "Готово!"
result = await my_async_function()
print(result)
await APMHandler.close()
asyncio.run(main())
Обязательные параметры
Лицензия
Этот проект лицензирован под MIT License.
Обратная связь
Если у вас есть вопросы, предложения или вы нашли ошибку, создайте issue в GitHub репозитории.