一句話結論：與其替每種錯誤寫關鍵字規則，不如讓 Drain3 把 log 歸納成「樣板」，只在出現沒看過的樣板時通知一次。已知的問題不吵你，新的問題才會冒出來。

遇到的問題

最常見的 log 告警做法是：log 裡出現 error，就通知。

這種規則在第一週還算好用，很快就會變成噪音。一個依賴服務重啟，幾百行 connection refused 一起噴出來，通知響不停。看完才發現是已知問題，下次又響。

接著開始加例外：這種錯誤不要報，那種超過幾次才報。規則越寫越多，卻還是會漏掉真正重要的事，也就是第一次出現的新錯誤。

其實真正想問的是：「這行 log 跟我以前看過的，是不是同一種？」

這就是 log 樣板（template）要解決的事。connection refused addr=10.0.0.8:5432 attempt=2 和 connection refused addr=10.0.0.41:5432 attempt=4 其實是同一種，只是變數不同。把變數抽掉，剩下的骨架就是樣板。每個樣板記下來，之後只要出現沒見過的，才值得通知。

Drain3 是一個做這件事的 Python 套件，演算法是 Drain，本身不用訓練資料，一行一行餵進去就會即時歸納。

手把手步驟

準備環境

只需要裝一個套件：

pip install drain3==0.9.11

下面的結果都是在 drain3==0.9.11、Python 3.12 實測的。

先準備一些假的 log

為了讓結果可以重現，先用一支小腳本產生假 log。第一批有三種問題，第二批是同樣的三種，外加一種全新的 disk quota exceeded：

# gen.py import random import sys random.seed(7) batch = sys.argv[1] def ip(): return f"10.0.{random.randint(0, 9)}.{random.randint(1, 250)}" def ts(): return f"2026-01-0{random.randint(1, 9)}T0{random.randint(0, 9)}:{random.randint(10, 59)}:{random.randint(10, 59)}Z" out = [] for _ in range(400): out.append(f'{ts()} level=error msg="connection refused" addr={ip()}:5432 attempt={random.randint(1, 5)}') for _ in range(300): out.append(f'{ts()} level=error msg="request failed" path=/api/items/{random.randint(1, 99999)} status=500 took={random.randint(5, 900)}ms') for _ in range(60): out.append(f'{ts()} level=warn msg="retry failed" job=sync-{random.randint(1, 40)} reason=timeout after {random.randint(1, 30)}s') if batch == "2": for _ in range(25): out.append(f'{ts()} level=error msg="disk quota exceeded" volume=pvc-{random.randint(1, 9)} used={random.randint(90, 99)}%') random.shuffle(out) print("\n".join(out)) 
python3 gen.py 1 > batch1.log # 760 行 python3 gen.py 2 > batch2.log # 785 行

核心腳本：只印出新樣板 

# novelty.py import re import sys from drain3 import TemplateMiner from drain3.file_persistence import FilePersistence from drain3.masking import MaskingInstruction from drain3.template_miner_config import TemplateMinerConfig STATE = sys.argv[1] config = TemplateMinerConfig() config.masking_instructions = [ MaskingInstruction(r"\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:\.\d+)?Z?", "TS"), MaskingInstruction(r"(?:\d{1,3}\.){3}\d{1,3}(?::\d+)?", "IP"), MaskingInstruction(r"\b0x[0-9a-fA-F]+\b|\b[0-9a-f]{12,}\b", "HEX"), MaskingInstruction(r"(?<![\w.])[-+]?\d+(?:\.\d+)?(?:ms|s|m|%)?(?![\w.])", "NUM"), ] config.drain_sim_th = 0.5 config.drain_depth = 4 config.snapshot_interval_minutes = 10 ** 6 # 不要中途存檔，結尾才存 miner = TemplateMiner(FilePersistence(STATE), config) new = {} lines = 0 for raw in sys.stdin: line = re.sub(r"\s+", " ", raw).strip() if not line: continue lines += 1 result = miner.add_log_message(line) if result["change_type"] == "cluster_created": new[result["cluster_id"]] = line miner.save_state("run-end") templates = {c.cluster_id: c.get_template() for c in miner.drain.clusters} print(f"lines={lines} total_templates={len(templates)} new_templates={len(new)}") for cid in new: print(f" NEW #{cid}: {templates[cid]}")

重點只有三個。

add_log_message 的回傳值。 它會告訴你這行 log 是 cluster_created（建了新樣板）、cluster_template_changed（既有樣板被泛化）還是 none。只看 cluster_created，就是「新樣板」。

FilePersistence。 樣板庫存在檔案裡，下一次執行會接著用。這是「以前看過」的記憶來源，沒有它每次執行都會把所有樣板當成新的。

masking（遮蔽）。 先用正規表示式把時間戳、IP、數字換成 <TS>、<IP>、<NUM>，再交給 Drain。後面的坑會看到它為什麼重要。

跑起來看結果

先餵第一批，建立基準：

python3 novelty.py state.bin < batch1.log 
lines=760 total_templates=3 new_templates=3 NEW #1: <TS> level=error msg="connection refused" addr=<IP> attempt=<NUM> NEW #2: <TS> level=error msg="request failed" path=/api/items/<NUM> status=<NUM> took=<NUM> NEW #3: <TS> level=warn msg="retry failed" job=sync-<NUM> reason=timeout after <NUM>

760 行 log 被歸納成 3 個樣板，變數都被抽掉了。

同一批再跑一次：

lines=760 total_templates=3 new_templates=0

沒有任何新樣板，所以不通知。

換第二批，裡面多了一種新錯誤：

python3 novelty.py state.bin < batch2.log 
lines=785 total_templates=4 new_templates=1 NEW #4: <TS> level=error msg="disk quota exceeded" volume=pvc-<NUM> used=<NUM>

785 行裡，前面三種問題的幾百行完全沒有動靜，只有 25 行的 disk quota exceeded 被抓出來。實際要接通知，就是把這幾行輸出送去聊天工具或 webhook 即可。

把 Loki 接進來

實際使用時，log 不是檔案，而是從 Loki 撈。排程（例如 Kubernetes CronJob）每隔幾小時跑一次，用 Loki 的 query_range API 撈上一輪到現在的錯誤類 log（LogQL 先用 |~ "(?i)(error|exception|fail|fatal|panic)" 篩一次），再餵進上面的流程。

這一段沒有包含在這篇的範例裡。樣板庫的檔案放在 PVC 上，Pod 重啟也不會遺失。

踩到的坑

坑一：變數沒遮蔽，樣板數量爆炸

這是最容易中的坑。Drain 預設會把「含數字的 token」當成變數，所以前面的假 log 就算不寫 masking，也只會得到 3 個樣板（實測結果也一樣）。

但如果變數是純字母，又剛好出現在行首附近，情況就不同。例如行首是 8 個隨機小寫字母的主機名：

hsreltpu worker sctapi failed to fetch rhgwp from upstream rrpmuehu worker eqmxav failed to fetch ycfys from upstream

這其實是同一種錯誤，但餵 500 行進去，不遮蔽的話會得到：

lines=500 total_templates=100 new_templates=100

100 個樣板，每一個都被當成「新樣板」，通知會被洗版。數字剛好是 Drain 預設單一節點最多 100 個分支，超過後才開始合併。

加一條遮蔽規則把行首那段換成 <ID>：

MaskingInstruction(r"^[a-z]{8}(?= worker)", "ID")

同樣 500 行：

lines=500 total_templates=1 new_templates=1 NEW #1: <ID> worker <*> failed to fetch <*> from upstream

100 個變成 1 個。要訣是：觀察輸出的樣板數，如果遠比預期的問題種類多，八成是有一種變數沒被遮蔽。檢查幾個看起來很像、只差一個詞的樣板，就能知道要補什麼規則。

坑二：第一次執行，所有樣板都是「新」的

樣板庫是空的，第一輪跑完，每一個樣板都是新樣板，你會一次收到幾十則通知。

做法是：第一輪只學不通知。可以在狀態裡記一個 bootstrapped 旗標，第一輪結束才設成 true，之後才開始通知。

坑三：偵測器自己的輸出也被當成 log

如果偵測器把通知內容印到 stdout，而 Loki 又收集了偵測器所在的 namespace，下一輪就會撈到它自己上一輪的輸出，把通知文字學成新樣板，然後又通知一次，無限循環。

查詢時要把偵測器自己的 container 排除（container!="detector"）。Loki 自己也會把查詢語句寫進它的 log，裡面剛好包含 error 這類關鍵字，同樣要排除。

這一點沒有在範例裡重現，是實際跑偵測器時遇到的。

為什麼這樣選

這個做法的限制要先講清楚。

「新樣板」只代表「以前沒出現過」，不代表「很嚴重」。一個無害的警告第一次出現，一樣會通知。它回答的是「有沒有新的事發生」，不回答「該不該緊張」。

另外，如果某個問題已經持續很久，因為早就學過它的樣板，之後不會再通知你。要補這個缺口，可以另外加一條「同一個樣板的出現次數突然暴增」的判斷，但那是另一個主題，這篇不展開。

換來的好處是維護成本很低：沒有關鍵字清單要更新，新服務上線不用先寫規則，而且通知量跟「新問題的種類數」成正比，而不是跟 log 量成正比。

如果 log 格式很固定、種類也有限，直接寫關鍵字規則可能更簡單。Drain3 比較適合服務很多、格式各異、不想一個個盯的情況。

延伸閱讀