ログが語ることを数字にする

目標

標準ライブラリだけで、アクセスログとデプロイのCSVから、運用メトリクス(件数・ステータスの分布・上位のパス・分位数・最悪の1分・サービスごとの失敗率)を取り出すツールを作ります。

なぜ重要なのか

本番サーバーには、pandasもインターネットもありませんが、csv・Counter・statistics・datetimeはあります。1行ずつ読むパーサー、分位数、タイムゾーン付きの時刻の比較。この3つを身につければ、ログが5千行でも50万行でも、同じツールで答えを出せます。用意されたデータは、/opt/fixtures/pyops/access.log(nginx combined + 最後の列は処理時間(秒)、2026-09-10 02:00から05:00 KSTまでで、03:12から03:17までが障害の区間)と、/opt/fixtures/pyops/deploys.csv(service,version,deployed_at,duration_s,status)です。

ステップ

/root/pyops/data/logstat.pyを作ってください。logstat.py count <로그>(プレースホルダーはログです)は、全体の行数と、正規表現でパースに成功した行数を、lines=<n> parsed=<m>の1行で出力します。パースでは、時刻・パス・ステータスコード・処理時間を取り出します。
logstat.py status <로그>は、ステータスコードのクラスごとの件数を、2xx=<n> 3xx=<n> 4xx=<n> 5xx=<n>の1行で出力します(Counter)。
logstat.py top <로그> --n 5は、リクエスト数が多いパスの上位n件を、<건수> <경로>(プレースホルダーは、件数とパスです)の形式で、1行に1つずつ、多い順に出力します(most_common)。
logstat.py latency <로그>は、処理時間のp50・p95・p99を、p50=<x> p95=<x> p99=<x>で出力します。statistics.quantiles(times, n=100)(既定のmethod)の[49]・[94]・[98]を、小数第3位に丸めます。
logstat.py worst <로그>は、5xxが最も多かった1分の区間を、worst_minute=<YYYY-MM-DDTHH:MM> count=<n>で出力します。時刻は、strptimeでaware datetimeを作り、秒を0に揃えてまとめます。ログのタイムゾーン(+0900)のまま書きます。
/root/pyops/data/deploys.pyを作ってください。deploys.py summary <csv>は、サービスごとにservice=<이름> total=<n> failed=<m> fail_rate=<0.000> median_s=<x>(プレースホルダーは名前です)を、サービス名の順に1行ずつ出力します(DictReader・statistics.median、fail_rateは小数第3位)。
logstat.py status <로그> --since <ISO> --until <ISO>で、時刻のフィルターを付けてください。2つの値は、2026-09-10T03:00:00+09:00のように、タイムゾーン付きのISO 8601で、since ≤ ts < untilの行だけを数えます。
logstat.py report <로그> --jsonは、上のメトリクスをすべて含むJSONオブジェクトを1つ、標準出力に出力します。キーは、lines・parsed・status(クラスごとのオブジェクト)・top([[경로, 건수], ...]の5件。プレースホルダーは、パスと件数です)・latency(p50・p95・p99)・worst_minute({"minute": ..., "count": ...})です。これを/root/pyops/data/report.jsonに保存します。

参考

正規表現の例: \[(?P<ts>[^\]]+)\] "(?P<method>\S+) (?P<path>\S+) [^"]*" (?P<status>\d{3}) \d+ "[^"]*" "[^"]*" (?P<rt>[\d.]+)$
時刻: datetime.strptime(ts, "%d/%b/%Y:%H:%M:%S %z")、フィルターの入力: datetime.fromisoformat(s)
よくあるミスは、平均で遅延を報告すること、naiveとawareの時刻を比較すること、パースに失敗した行を数えないことです。

1行ずつ読んで、パースの成功を数える

/root/pyops/data/logstat.pyを作ってください。logstat.py count <로그>(プレースホルダーはログです)が、lines=<n> parsed=<m>を出力します。時刻・パス・ステータス・処理時間を取り出す正規表現でパースします。

ファイルはfor line in fで1行ずつ読み、re.compileしたパターンのsearchがNoneならスキップしますが、linesは数えます。参考の項の正規表現を、そのまま使ってもかまいません。

ステータスコードのクラスごとの件数

logstat.py status <로그>(プレースホルダーはログです)が、2xx=<n> 3xx=<n> 4xx=<n> 5xx=<n>の1行を出力します。

Counterに、f"{status // 100}xx"をキーとして入れてください。ないクラスは、0として出力する必要があります(dict.get(k, 0))。

リクエストが多いパスの上位n件

logstat.py top <로그> --n 5(プレースホルダーはログです)が、<건수> <경로>(プレースホルダーは、件数とパスです)を、多い順に1行に1つずつ出力します。

Counter.most_common(n)が、(値と件数)のリストを、多い順に返します。出力の順序は、件数パスです。

平均ではなく分位数

logstat.py latency <로그>(プレースホルダーはログです)が、p50=<x> p95=<x> p99=<x>を出力します。statistics.quantiles(times, n=100)の[49]・[94]・[98]を、小数第3位に丸めます。

n=100なら、分割点が99個なので、インデックス49・94・98が、p50・p95・p99です。methodは、既定値(exclusive)のままにしてください。

5xxが最も多かった1分

logstat.py worst <로그>(プレースホルダーはログです)が、worst_minute=<YYYY-MM-DDTHH:MM> count=<n>を出力します。時刻は、ログのタイムゾーンのまま、秒を0に揃えて、1分単位でまとめます。

strptimeに%zを使えば、aware datetimeになります。ts.replace(second=0, microsecond=0)をCounterのキーにして、most_common(1)を見てください。出力は、strftime("%Y-%m-%dT%H:%M")です。

デプロイのCSVの、サービスごとの失敗率

/root/pyops/data/deploys.pyを作ってください。deploys.py summary <csv>が、サービス名の順に、service=<이름> total=<n> failed=<m> fail_rate=<0.000> median_s=<x>(プレースホルダーは名前です)を1行ずつ出力します。

csv.DictReaderは、行ごとに辞書を返し、値はすべて文字列です。defaultdict(list)にduration_sを集めて、statistics.medianを出力してください。fail_rateは、failed / totalを小数第3位までです。

タイムゾーン付きの時刻で絞り込む

logstat.py status <로그> --since <ISO> --until <ISO>(プレースホルダーはログです)が、since ≤ ts < untilの行だけを数えます。2つの値は、タイムゾーン付きのISO 8601(例: 2026-09-10T03:00:00+09:00)です。

argparseのtype=datetime.fromisoformatで受け取れば、aware datetimeになります。naiveの値と比較するとTypeErrorが出るので、入力にタイムゾーンがあるかを確認してください。

すべてをJSONレポートに

logstat.py report <로그> --json(プレースホルダーはログです)が、lines・parsed・status・top(5件、[パス、件数])・latency・worst_minute({minute, count})を含むJSONオブジェクトを1つ出力し、それを/root/pyops/data/report.jsonに保存します。

json.dumpsにdatetimeを入れると失敗します。文字列に変換してから入れてください。logstat.py report ... --json > /root/pyops/data/report.jsonで保存します。