loki-service

Python logging handler for Grafana Loki，支持自定义 HTTP Headers、Basic 认证、异步队列、结构化日志。

特性

• 开箱即用 — LokiLogger 一行代码完成异步队列 + 控制台 + Loki 推送
• 生产级健壮性 — Loki 异常静默隔离、HTTP 请求超时保护、队列满时非阻塞丢弃、进程退出自动 flush
• 结构化日志 — 自动注入 path（调用者文件:行号）、ip（主机名）、trace_err（异常堆栈）等上下文字段
• 控制台着色 — 内置 ColorFormatter 按日志级别彩色输出，也支持自定义 logging.Formatter
• 多租户支持 — 通过 headers 传入 X-Scope-OrgID 实现多租户隔离
• 灵活组合 — 底层 LokiHandler / LokiQueueHandler 可自由搭配标准 logging 模块
• Loki API V0 & V1 — 支持 Loki < 0.4.0（V0）和 >= 0.4.0（V1）两种协议

安装

pip install loki-service

快速开始

LokiLogger（推荐）

开箱即用的生产级Logger，内置异步队列 + 控制台输出：

from loki_service import LokiLogger

# 完整模式：控制台 + Loki 推送
logger = LokiLogger(
    app_name="my-app",
    loki_url="http://loki:3100/loki/api/v1/push",
    level="INFO",
    env="production",
    headers={"X-Scope-OrgID": "my-tenant"},
    auth=("username", "password"),
)

# 本地模式：仅控制台输出，不启动 Loki 推送（loki_url 默认为 None）
logger = LokiLogger(app_name="my-app")

# 基本用法
logger.info("用户登录成功")
logger.error("请求失败", status_code=500, request_id="abc123")

# 传入 dict 作为消息
logger.warning({"event": "timeout", "cost": 3200})

# 在 except 块中调用，自动捕获异常堆栈到 trace_err 字段
try:
    1 / 0
except ZeroDivisionError:
    logger.error("除零异常")
    # 输出 JSON 中将包含 "trace_err": "Traceback (most recent call last)..."

# 如果不需要异常堆栈，显式关闭
logger.info("普通消息", is_trace=False)

# 额外关键字段直接传入
logger.error("订单失败", order_id="ORD-001", cost=3200)
# 输出: {"msg": "订单失败", "order_id": "ORD-001", "cost": 3200, "path": "...", "ip": "...", "trace_err": "..."}

# 位置参数 + msg 关键字同时传入时，关键字值存为 msg_extra
logger.debug("主消息", msg="补充信息")
# 输出: {"msg": "主消息", "msg_extra": "补充信息", ...}

# 启用 asctime 注入
logger.info("带时间戳", is_asctime=True)
# 输出: {"msg": "带时间戳", "asctime": "2026-07-08 15:30:00.123", ...}

自动注入字段

LokiLogger 的 info / error / warning / debug / critical / success 方法会将消息序列化为 JSON，并自动注入以下上下文字段：

额外的 **kwargs 会直接作为 JSON 字段输出：

logger.error("订单失败", order_id="ORD-001", cost=3200)
# 输出: {"msg": "订单失败", "order_id": "ORD-001", "cost": 3200, "path": "...", "ip": "..."}

LokiLogger 参数说明：

文件日志（转轮）

通过 log_file 参数启用文件日志，内置 RotatingFileHandler 自动转轮：

from loki_service import LokiLogger

logger = LokiLogger(
    app_name="my-app",
    log_file="/var/log/my-app/app.log",  # 目录不存在时自动创建
    max_bytes=10 * 1024 * 1024,          # 单文件最大 10MB
    backup_count=5,                        # 保留 5 个旧文件（共 6 个文件）
)

logger.info("这条日志同时输出到控制台和文件")

转轮规则：当 app.log 达到 max_bytes 时，自动重命名为 app.log.1，新日志写入新的 app.log。最多保留 backup_count 个旧文件，最旧的被删除。

LokiLogger 生命周期管理

LokiLogger 在进程退出时（atexit）会自动 flush 队列中剩余日志。也可以手动关闭：

logger = LokiLogger(app_name="my-app")
# ... 使用 logger ...
logger.close()  # 停止队列监听，等待剩余日志发送完毕

如果需要获取底层 logging.Logger 实例（例如添加自定义 Handler）：

raw_logger = logger.get_logger()

自定义格式化

各输出未指定 formatter 时使用各自默认格式。通过 formatter 参数可统一自定义：

from loki_service import LokiLogger, ColorFormatter
import logging

# 启用彩色输出（按日志级别着色：DEBUG灰/INFO绿/WARNING黄/ERROR红/CRITICAL紫）
logger = LokiLogger(app_name="my-app", formatter=ColorFormatter())

# 完全自定义格式（同时应用于控制台、文件、Loki）
logger = LokiLogger(
    app_name="my-app",
    formatter=logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"),
)

重复初始化警告

如果同一个 app_name 多次创建 LokiLogger，由于 logging.getLogger() 返回同一实例，新的配置参数不会生效，并在 DEBUG 级别输出警告。请确保每个 app_name 只初始化一次。

LokiHandler（底层 Handler）

适合需要自行组合 Handler 的场景：

import logging
import loki_service

handler = loki_service.LokiHandler(
    url="http://loki:3100/loki/api/v1/push",
    tags={"application": "my-app"},
    auth=("username", "password"),
    version="1",                              # Loki >= 0.4.0 使用 "1"，< 0.4.0 使用 "0"
    headers={"X-Scope-OrgID": "my-tenant"},
)

logger = logging.getLogger("my-logger")
logger.addHandler(handler)
logger.error("Something happened", extra={"tags": {"service": "my-service"}})

LokiHandler 参数说明：

SafeLokiHandler

SafeLokiHandler 继承自 LokiHandler，在 Loki 推送异常时完全静默（吞掉所有异常），绝不影响业务逻辑。LokiLogger 内部默认使用的就是此 Handler。

from loki_service import SafeLokiHandler

handler = SafeLokiHandler(
    url="http://loki:3100/loki/api/v1/push",
    tags={"app": "my-app"},
    version="1",
)

LokiQueueHandler（异步队列模式）

使用 LokiQueueHandler 在独立线程中发送日志，不阻塞主线程：

import logging
import loki_service
from multiprocessing import Queue

handler = loki_service.LokiQueueHandler(
    Queue(-1),
    url="http://loki:3100/loki/api/v1/push",
    tags={"application": "my-app"},
    version="1",
)

logger = logging.getLogger("my-logger")
logger.addHandler(handler)

Loki 标签体系

Loki 使用标签（Labels）对日志流进行索引和查询。以下是各组件的标签行为：

标签名只允许字母、数字和下划线。包含 . - 空格等的标签名会被自动替换为 _。

多租户（X-Scope-OrgID）

Loki 多租户模式通过 X-Scope-OrgID 请求头区分租户，使用 headers 参数传入：

from loki_service import LokiLogger

logger = LokiLogger(
    app_name="order-service",
    loki_url="http://loki:3100/loki/api/v1/push",
    headers={"X-Scope-OrgID": "tenant-abc"},
)

logger.info("订单创建成功", order_id="ORD-001")

等效 curl 请求：

curl -X POST http://loki:3100/loki/api/v1/push \
  -H "Content-Type: application/json" \
  -H "X-Scope-OrgID: tenant-abc" \
  -d '{"streams": [{"stream": {"app": "order-service", "severity": "info"}, "values": [["1234567890000000000", "订单创建成功"]]}]}'

生产环境注意事项

• 队列溢出保护：当异步队列满时，新日志会被静默丢弃而非阻塞业务线程，防止 Loki 故障拖垮整个服务。
• 异常隔离：SafeLokiHandler 会吞掉所有 Loki 推送异常，logging.raiseExceptions 也被设为 False。
• HTTP 超时保护：每次 Loki 推送请求默认 10 秒超时（可通过 timeout 参数调整），防止 Loki 不可达时 listener 线程无限阻塞导致程序无法退出。
• 自动 flush：进程退出时通过 atexit 自动停止所有 QueueListener，确保队列中剩余日志发送完毕。
• 重复初始化：相同 app_name 重复创建 LokiLogger 时，新参数不生效，请注意避免。