取得処理が失敗したとき、誰かが気づく必要があります。仕組み自体は簡単で、例外を捕まえてメッセージを送るだけです。数行で書けます。
難しいのはその先です。送った通知が読まれ続けるかどうか。最初の1週間はみんな見ますが、毎日1通ずつ届くようになると、1か月後には誰も開かなくなります。そして本当に困る失敗も、同じ場所で見過ごされます。
この回では、通知が形骸化しないための設計を扱います。技術ではなく、人がどう振る舞うかを前提にした話になります。
毎日鳴るものは通知ではない
最も避けたいのは、正常時にも鳴る通知です。「取得が完了しました」を毎日送ると、それが日常になります。日常になったものの中から異常を見つけるのは、人間には向いていません。
同じ理由で、対処しようのない警告も送るべきではありません。相手のサーバが一時的に落ちていて、で自動的に復旧したなら、それは人に知らせる必要がありません。記録に残せば十分です。
| 状況 | 通知 | 理由 |
|---|---|---|
| 自動で復旧した一時的な失敗 | しない | 対処が要らない |
| 再試行しても失敗し続けた | する | 人が判断する必要がある |
| 取得件数が想定より大きく違う | する | 静かに壊れている可能性 |
| 正常に完了した | しない | 毎日鳴ると見なくなる |
| 何日も実行されていない | する | 気づきようがない種類の異常 |
処理が失敗すれば通知できますが、そもそも起動しなかった場合は何も起きません。定期実行の設定が消えた、サーバが止まった。これらは失敗の通知では捕まえられないので、別の場所から「動いた形跡があるか」を見る必要があります。
が一時的に不安定なのはよくあることで、の 5xx が数回出ても自動で復旧します。これを毎回通知していると、本当に止まった日の通知が同じ見た目で埋もれます。自動で直るものは記録に留めてください。
逆に、通知すべきなのに見落とされがちなのが件数の異常です。取得は成功していて、例外も出ていない。ただ件数が普段の1割しかない。EP.13 で扱った検算を通知と繋いでおくと、この種の静かな失敗を捕まえられます。
動いた形跡を別の場所から見る
対策は単純で、取得が成功するたびに最後に成功した時刻を記録し、それを別の処理が見に行きます。想定した間隔を超えて更新されていなければ、何かが起きています。
この確認だけは、取得と同じ場所で動かさないでください。同じ場所が止まれば、確認する側も止まります。別の仕組みから見ることに意味があります。
import jsonfrom datetime import datetime, timedelta, timezonefrom pathlib import Path
HEARTBEAT = Path("data/heartbeat.json")
def mark_success(job: str, path: Path = HEARTBEAT) -> None: """成功したときだけ更新する。失敗時に更新すると意味が無くなる。""" data = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {} data[job] = datetime.now(timezone.utc).isoformat() path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def stale_jobs(expected: dict[str, timedelta], path: Path = HEARTBEAT) -> list[str]: """想定した間隔を超えて成功していない処理を返す。""" if not path.exists(): return list(expected) # 一度も動いていない
data = json.loads(path.read_text(encoding="utf-8")) now = datetime.now(timezone.utc) late = [] for job, interval in expected.items(): last = data.get(job) if last is None or now - datetime.fromisoformat(last) > interval: late.append(job) return lateこの確認は でも でも構いませんが、取得を動かしている場所とは別にするのが条件です。同居させると、その場所ごと止まったときに沈黙します。
通知に「次に何をするか」を書く
「取得に失敗しました」だけの通知は、受け取った人を困らせます。何が起きて、どう対処すればよいかが書かれていないと、まず状況を調べるところから始まります。深夜に受け取ったなら、なおさらです。
含めるべきは、どの処理か、いつ、何回目の失敗か、直前の例外、そして再実行の方法です。EP.36 で手動起動の口を用意しておけば、通知にそのコマンドを書けます。
def build_alert(job: str, attempt: int, error: Exception, target_date: str) -> str: """調べ直さなくても対処できる情報を入れる。""" return "\n".join( [ f"[取得失敗] {job}", f"対象日: {target_date}", f"{attempt} 回目の失敗（自動再試行は打ち切り）", f"例外: {type(error).__name__}: {error}", "", "再実行:", f" python -m fetch.run --job {job} --date {target_date}", ] )通知の宛先も決めておいてください。個人宛にすると、その人が休んだ日に止まります。複数人が見る場所へ送るのが基本です。誰が対処するかは、受け取ってから決めれば構いません。
通知は、作った直後がいちばん機能します。時間とともに形骸化するので、たまに「最近この通知を見たか」を確かめるのが実際上の維持方法になります。次回は、取得そのものの全体像を把握する話をします。
通知は仕組みではなく習慣です。作った直後がいちばん機能し、あとは劣化していきます。劣化を前提に、たまに見直す時間を取るほうが現実的です。
通知先を決めるとき、受け取った人が何をすればよいかまで一緒に決めておいてください。通知だけが飛んで、誰も動かない状態がいちばん無駄です。
この記事の感想を教えてください
あなたの 1 クリックで、本当にこの記事は更新されます。「もっと詳しく」「続編希望」が一定数集まった記事は、 ふくふくが 実際に内容を拡充したり続編記事を公開 します。 送信したリアクションはお使いのブラウザに記録され、再カウントされません。