Loguru + Redis 集中日誌（現代 Python Logging 最佳實踐）

利用 Loguru 的 Callback Sink 機制，將微服務中分散的日誌自動結構化後寫入 Redis List，實現跨容器集中監控，無需 ELK 等重量級基礎設施。

概述

在微服務或多 Replica 架構下，日誌分散在各個 Container 中。Debug 時需要逐一 shell 進入各容器查看，效率極低。將所有日誌集中到一個可查詢的中央存儲，是微服務可觀測性的基本需求。

Loguru 是現代 Python 的首選 logging 套件，其 logger.add() 接受任意 Callable 作為 Sink，讓「寫日誌」的動作完全可自訂。相較於 Python 標準 logging 模組需要繁瑣地配置 Formatter、Handler、Logger，Loguru 開箱即用且結構化資料（時間、檔案、行號）自動附帶在 message.record 中。

本方案架構：Application → Loguru → redis_sink() → Redis List 以 logs:{service}:{date} 為 Key Pattern 按服務與日期分桶，每筆日誌為 JSON 物件，保留一週後自動過期。

核心內容

為什麼選擇 Loguru

優點	說明
開箱即用	無需繁瑣配置 `Formatter` 與 `Handler`，一行即可輸出有顏色的結構化日誌
強大的 Sink 機制	除檔案與終端機外，接受任意 Callable、async coroutine、`logging.Handler`
結構化 Context	`message.record` 包含完整的 time、level、file、line、message 等資訊

Redis Sink 架構設計

日誌流向：

Application → Loguru → redis_sink() → Redis List (logs:{service}:{date})

Redis Key Pattern：

logs:{service_name}:{YYYY-MM-DD}

設計考量：

按日期分桶：方便按天查詢，過期清理粒度明確
LPUSH 作為 Queue：新日誌插入 List 頭部，最新的在前面
TTL 7 天：expired=86400 * 7 自動清理舊日誌，不需要手動維護
結構化 JSON：每筆日誌包含 service、host、file:line，跨服務查詢時可快速定位來源

Sink 機制詳解

logger.add() 的 sink 參數接受以下類型：

類型	範例	說明
File-like object	`sys.stderr`、`open("file.log", "w")`	任何具有 `.write()` 方法的物件，可選實作 `flush()`、`stop()`、`complete()`
檔案路徑	`"app.log"`、`pathlib.Path("app.log")`	字串或 Path，支援 rotation/retention 額外參數
Callable（函數）	`lambda msg: print(msg)`	完全自訂的日誌處理邏輯
Async coroutine	`async def my_sink(msg): ...`	Loguru 自動透過 `loop.create_task()` 加入事件循環，需用 `complete()` 等待完成
`logging.Handler`	`logging.StreamHandler()`	相容標準 logging 生態系，Loguru 自動轉換 record 格式

重要限制：Logging 函數不具備可重入性（not reentrant）。避免在 Sink 內部或 signal handler 中再次呼叫 logger，否則可能造成 deadlock。若需規避，請明確停用（disable）該 sink。

關鍵要點

Loguru 的 message.record 帶有完整 context，無需手動格式化
Redis Sink 以 logs:{service}:{date} 分桶，按服務與日期獨立查詢
TTL 設 7 天可自動清理，避免 Redis 記憶體無限增長
Sink 函數內的 exception 必須 catch 並寫 stderr，避免 logging 系統本身 crash 導致主程式中斷
is_json 參數依 Redis client 實作不同，使用前確認所用 library 的 API

實際應用

適合以下場景：

Kubernetes 多 Replica 服務（無法預先知道日誌在哪個 Pod）
微服務架構（多個服務共用一個 Redis，按 service_name 區分）
需要即時查詢最新日誌但不想導入完整 ELK Stack 的場景

消費端只需 LRANGE logs:{service}:{date} 0 -1 即可取得全天日誌，或用 BRPOP 做即時串流。

部署設定參考

以下為實際部署時使用的完整設定，供日後查詢與複製使用。

Redis Sink 完整實作（Python）

python

import os
import sys
import json
import datetime
import loguru

def redis_sink(message):
    """
    Loguru Callback Sink：將日誌結構化後寫入 Redis List
    """
    try:
        record = message.record  # 包含完整 Context (time, file, line...)

        log_data = {
            "timestamp": record["time"].isoformat(),
            "level": record["level"].name,
            "service": os.environ.get("SERVICE_NAME", "unknown"),
            "host": os.environ.get("COMPUTERNAME", "localhost"),
            "file": f"{record['file'].name}:{record['line']}",
            "message": record["message"],
        }

        today_str = datetime.date.today().strftime("%Y-%m-%d")
        redis_key = f"logs:{log_data['service']}:{today_str}"

        conn = get_redis_connection()
        conn.lpush(
            redis_key,
            json.dumps(log_data),
        )
        conn.expire(redis_key, 86400 * 7)   # 保留 7 天

    except Exception as e:
        # 避免 Log 系統本身 crash 導致主程式中斷
        sys.stderr.write(f"Redis Sink Error: {e}\n")


# 向 Loguru 註冊 Sink
loguru.logger.add(redis_sink, level="INFO")

環境變數

變數	說明
`SERVICE_NAME`	服務名稱，用於 Redis Key 分桶（如 `api-server`、`worker`）
`COMPUTERNAME`	主機名稱，Pod 環境通常為 Pod Name

Redis Key 查詢範例

bash

# 查詢今日全部日誌
LRANGE logs:api-server:2026-05-28 0 -1

# 查詢最新 100 筆
LRANGE logs:api-server:2026-05-28 0 99

# 即時串流（阻塞等待新日誌）
BRPOP logs:api-server:2026-05-28 0

來源

Loguru 官方文件 — Sink 機制完整說明
原始素材：2026-05-25 現代化 Python Logging 最佳實踐：Loguru + Redis.md

Loguru + Redis 集中日誌（現代 Python Logging 最佳實踐） ​

概述 ​

核心內容 ​

為什麼選擇 Loguru ​

Redis Sink 架構設計 ​

Sink 機制詳解 ​

關鍵要點 ​

實際應用 ​

部署設定參考 ​

Redis Sink 完整實作（Python） ​

環境變數 ​

Redis Key 查詢範例 ​

相關概念 ​

來源 ​