E 会計/ファイナンス・コスト効率 — Cloud Logging の取り込み費を「severity 一括除外で $0」にする提案を、内訳を測って書き直す（DEBUG 戻し忘れ 312 GiB × 拒否理由の残存 0% → リクエスト単位バッファ × 差額 $18 を人件費に換算）

概要

問題

MOps 販促システムチームの後輩が、coupon-api（GKE、月 1.82 億リクエスト）の Cloud Logging 費を下げる提案を上げてきた。あなたはレビューを頼まれた。

実態（合成ログ）

制約・前提条件

• 行サイズ 仮定: access 1,126 B／INFO 358 B／DEBUG 307 B／probe 512 B／canonical 1,638 B／audit 700 B。月 1.82 億リクエスト、Pod 24 台、LOG_LEVEL=DEBUG は 8 月から戻し忘れ
• 単価 公式（参照日 2026-10-02、cloud.google.com/stackdriver/pricing）: 取り込み $0.50/GiB（30 日分の保管込み）／プロジェクトあたり月 50 GiB 無料／30 日超の保管 $0.01/GiB/月
• 他サービスが同プロジェクトで月 30 GiB 仮定。150 円/$、技術者 6,000 円/h、問い合わせ窓 90 日 仮定

期待する回答形式: 問題点 7 点／内訳と方針比較のコード／実行例／提案書の書き直し／導入後の確認と止め方／迷った案と前提が変わったときの判断／扱わなかった範囲

悪い提案 (Before)

ヒント（段階的開示）

費用 = max(取り込み量 + 他サービス − 50GiB, 0) × $0.50/GiB
損益分岐の時間 = (good と bad の月額差 × 150円) ÷ 6,000円/h
good: DEBUG はバッファ／失敗・>800ms は文脈ごと／拒否は audit 100%／成功は canonical 10%／probe はソースで止める

問題点分析（7点）

内訳・残存率・パターン（SVG）

棒の長さは GiB に比例（実測:合成ログ）。Bad は 3.6 GiB まで減るが、調べられる率はエラー・拒否ともに 0%。

Bad → Good（コード差分）

📋 コピー# Bad: 後輩案（取り込み後に severity で一括除外）
LOG_LEVEL=DEBUG                          # 8月から戻していない
logger.info("rule_check %s", rule)       # リクエストごとに INFO 2〜4 行
logger.debug("db_read ms=%d", ms)        # DEBUG 4〜8 行（312 GiB/月）

# _Default シンクの除外フィルタ
severity < ERROR                         # 捨てるのは取り込みの後。アプリは出し続ける

# 結果: 文脈の残存率 エラー0% / 拒否0%

📋 コピー# Good: リクエスト単位のバッファ + canonical line
log = RequestLog(emit, trace_id)         # リクエストの入口で作る
log.note("rule_check", rule=rule)        # 即出力せず溜めるだけ
log.note("db_read", ms=ms)

log.finish(outcome, coupon=cid)          # 出口で1回だけ判断する
# error / >800ms : 文脈ごと1行で残す（100%）
# denied         : audit 行を100%（DB に無い）
# granted        : canonical line を10%（sample_rate 付き）

# 結果: 文脈の残存率 エラー100% / 拒否100%、費用 $337 → $17.66

適用パターン: Request-scoped buffer + Canonical Log Line（Stripe の canonical log lines、wide events の考え方）。書籍の章対応は無し（C/E の日のため）。

模範解答

費用の内訳と方針比較（logcost.py）

📋 コピー"""Cloud Logging 取り込み費の方針比較（合成ログ）。Python 3.12+。

出所ラベル: 単価=公式(2026-10-02参照) / 台数・件数=仮定 / バイト数・削減率=合成ログの実測。
"""
import random
from dataclasses import dataclass

GIB = 1024**3
PRICE_PER_GIB = 0.50          # 公式: ストレージへの取り込み $0.50/GiB（30日保管込み）
FREE_GIB = 50                 # 公式: プロジェクトあたり月 50GiB 無料
RETAIN_PER_GIB_MONTH = 0.01   # 公式: 30日超の保管 $0.01/GiB/月
YEN_PER_USD = 150             # 仮定
ENG_YEN_PER_HOUR = 6000       # 仮定: 技術者の時間単価（会社負担込み）
REQ_PER_MONTH = 182_000_000   # 仮定: 6M/日 × 30.4日
PODS = 24                     # 仮定
PROBES_PER_POD_MONTH = 2 * 6 * 60 * 24 * 30.4   # readiness+liveness 各10秒おき
SAMPLE_N = 200_000            # 合成するリクエスト数
SUCCESS_KEEP = 0.10           # 成功の canonical line を残す率（sample_rate をフィールドに持つ）
OTHER_GIB = 30                # 仮定: 同じ GCP プロジェクトの他サービスが使う量（無料枠 50GiB は共有）
AUDIT_DAYS = 90               # 仮定: 問い合わせ対応の窓。拒否理由を90日残す
SLOW_MS = 800                 # 仮定: p99 付近。これ超は全件残す

B = dict(access=1126, info=358, debug=307, probe=512, stack=5120, canon=1638, audit=700)

@dataclass
class Req:
    outcome: str      # granted / denied / error
    ms: int
    n_info: int
    n_debug: int

def synth(rng: random.Random) -> Req:
    r = rng.random()
    outcome = "error" if r < 0.004 else "denied" if r < 0.164 else "granted"
    ms = int(rng.lognormvariate(4.6, 0.8))     # 中央値約100ms
    return Req(outcome, ms, rng.choice([2, 3, 3, 4]), rng.choice([4, 6, 6, 8]))

def bytes_baseline(q: Req) -> int:       # 現状: DEBUG のまま・全行を出す
    b = B["access"] + q.n_info * B["info"] + q.n_debug * B["debug"]
    return b + (B["stack"] if q.outcome == "error" else 0)

def bytes_bad(q: Req) -> int:            # 後輩案: severity<ERROR を除外フィルタで捨てる
    return B["stack"] if q.outcome == "error" else 0

def bytes_good(q: Req, rng: random.Random) -> int:
    audit = B["audit"] if q.outcome == "denied" else 0  # 拒否は DB に残らないのでログが唯一の記録。付与は coupon_grants 行が正本
    if q.outcome == "error" or q.ms > SLOW_MS:          # 失敗・遅いものは文脈ごと全部（バッファを吐く）
        return B["canon"] + q.n_debug * B["debug"] + B["stack"] * (q.outcome == "error") + audit
    return (B["canon"] if rng.random() < SUCCESS_KEEP else 0) + audit

def lines_kept(policy: str, q: Req, rng: random.Random, audit: bool = False) -> int:
    """そのリクエストの文脈行（INFO+DEBUG）が何行残るか。audit=True は拒否理由の行が残るか。"""
    ctx = q.n_info + q.n_debug
    if policy == "base":
        return ctx
    if policy == "bad":
        return 0                      # severity<ERROR は全部捨てる。ERROR 行（スタック）自体は残るが文脈ではない
    if audit:
        return 1                      # good: 拒否の判断は100%残す
    if q.outcome == "error" or q.ms > SLOW_MS:
        return ctx                    # バッファを吐く
    return ctx if rng.random() < SUCCESS_KEEP else 0

def monthly_usd(gib: float) -> float:
    return max(gib + OTHER_GIB - FREE_GIB, 0) * PRICE_PER_GIB

def run(seed: int = 7) -> dict:
    rng = random.Random(seed)
    reqs = [synth(rng) for _ in range(SAMPLE_N)]
    scale = REQ_PER_MONTH / SAMPLE_N
    probe_gib = PODS * PROBES_PER_POD_MONTH * B["probe"] / GIB
    out = {"probe_gib": probe_gib}
    out["base_gib"] = sum(bytes_baseline(q) for q in reqs) * scale / GIB + probe_gib
    out["bad_gib"] = sum(bytes_bad(q) for q in reqs) * scale / GIB      # probe も info なので消える
    out["good_gib"] = sum(bytes_good(q, rng) for q in reqs) * scale / GIB   # probe はソースで停止
    for k in ("base", "bad", "good"):
        out[f"{k}_usd"] = monthly_usd(out[f"{k}_gib"])
    err = [q for q in reqs if q.outcome == "error"]
    den = [q for q in reqs if q.outcome == "denied"]
    out["n_err"] = len(err); out["n_den"] = len(den)
    # 調べられる率: エラー1件の前後の文脈（INFO/DEBUG）が残っているか／拒否の理由が残っているか
    # 調べられる率を「残った行」から数える（固定値を置かない）
    out["ctx_base"] = sum(lines_kept("base", q, rng) >= q.n_info + q.n_debug for q in err) / len(err)
    out["ctx_bad"] = sum(lines_kept("bad", q, rng) >= q.n_info + q.n_debug for q in err) / len(err)
    out["ctx_good"] = sum(lines_kept("good", q, rng) >= q.n_info + q.n_debug for q in err) / len(err)
    out["deny_bad"] = sum(lines_kept("bad", q, rng, audit=True) > 0 for q in den) / len(den)
    out["deny_good"] = sum(lines_kept("good", q, rng, audit=True) > 0 for q in den) / len(den)
    out["ok_ctx_bad"] = 0.0 if False else sum(lines_kept("bad", q, rng) > 0 for q in reqs if q.outcome == "granted") / sum(q.outcome == "granted" for q in reqs)
    out["ok_ctx_good"] = sum(lines_kept("good", q, rng) >= q.n_info + q.n_debug for q in reqs if q.outcome == "granted") / sum(q.outcome == "granted" for q in reqs)
    audit_gib = sum(B["audit"] for q in reqs if q.outcome == "denied") * scale / GIB
    out["audit_gib"] = audit_gib
    out["audit_retain_usd"] = audit_gib * RETAIN_PER_GIB_MONTH * (AUDIT_DAYS - 30) / 30.4
    out["saving_good_usd"] = out["base_usd"] - out["good_usd"] - out["audit_retain_usd"]
    out["extra_bad_usd"] = out["good_usd"] + out["audit_retain_usd"] - out["bad_usd"]
    out["breakeven_hours"] = out["extra_bad_usd"] * YEN_PER_USD / ENG_YEN_PER_HOUR
    out["slow_share"] = sum(q.ms > SLOW_MS for q in reqs) / len(reqs)
    return out

def sensitivity() -> None:
    global SUCCESS_KEEP
    for keep in (0.01, 0.10, 1.0):
        SUCCESS_KEEP = keep
        o = run()
        print(f"keep={keep:>4}: good {o['good_gib']:7.1f} GiB  ${o['good_usd']:7.2f}/月  成功の文脈が残る率 {o['ok_ctx_good']:.1%}")
    SUCCESS_KEEP = 0.10

if __name__ == "__main__":
    o = run()
    for k, v in o.items():
        print(f"{k:12s} {v:,.3f}")
    sensitivity()

リクエスト単位のバッファ（reqlog.py）

📋 コピー"""リクエスト単位のログバッファ + canonical line（Python 3.12+、標準ライブラリのみ）。"""
import json
import random
import time
from collections.abc import Callable
from dataclasses import dataclass, field

SUCCESS_KEEP = 0.10   # 成功リクエストの canonical line を残す率
SLOW_MS = 800         # これを超えたら文脈ごと全部残す
MAX_BUFFERED = 50     # 暴走リクエストでメモリを食わない上限

@dataclass
class RequestLog:
    """1リクエストの DEBUG/INFO を溜め、失敗・遅延・拒否のときだけ吐く。"""
    emit: Callable[[dict], None]            # 実際の出力先（stdout JSON → Cloud Logging）
    trace_id: str
    rng: random.Random = field(default_factory=random.Random)
    _buf: list[dict] = field(default_factory=list)
    _t0: float = field(default_factory=time.monotonic)

    def note(self, msg: str, **kv) -> None:
        if len(self._buf) < MAX_BUFFERED:   # DEBUG/INFO は即出力せず溜めるだけ
            self._buf.append({"msg": msg, **kv})

    def finish(self, outcome: str, **fields) -> None:
        ms = int((time.monotonic() - self._t0) * 1000) if "ms" not in fields else fields.pop("ms")
        keep_ctx = outcome == "error" or ms > SLOW_MS
        sampled = self.rng.random() < SUCCESS_KEEP
        if outcome == "denied":             # 拒否は DB に残らない。理由を必ず残す（audit）
            self.emit({"severity": "NOTICE", "kind": "audit", "trace": self.trace_id,
                       "outcome": outcome, "ms": ms, **fields})
        if keep_ctx or sampled:
            line = {"severity": "ERROR" if outcome == "error" else "INFO", "kind": "canonical",
                    "trace": self.trace_id, "outcome": outcome, "ms": ms,
                    "sample_rate": 1.0 if keep_ctx else SUCCESS_KEEP, **fields}
            if keep_ctx:
                line["context"] = self._buf  # 溜めた文脈を1行にまとめて添える
            self.emit(line)

if __name__ == "__main__":
    out: list[dict] = []
    for i, (outcome, ms) in enumerate([("granted", 90), ("denied", 120), ("error", 300), ("granted", 950)]):
        log = RequestLog(out.append, f"t{i}", random.Random(i))
        log.note("rule_check", rule="birthday_month")
        log.note("db_read", ms=12)
        log.finish(outcome, ms=ms, coupon="BDAY10")
    for line in out:
        print(json.dumps(line, ensure_ascii=False))

テスト（7件）

📋 コピーimport random
import logcost as L
import reqlog as R

def test_breakdown_sums_to_baseline():
    o = L.run()
    assert 690 < o["base_gib"] < 700

def test_bad_loses_all_context_good_keeps_errors():
    o = L.run()
    assert o["ctx_bad"] == 0.0 and o["deny_bad"] == 0.0
    assert o["ctx_good"] == 1.0 and o["deny_good"] == 1.0

def test_good_cheaper_than_baseline_by_over_90pct():
    o = L.run()
    assert o["good_usd"] < 0.1 * o["base_usd"]

def test_free_tier_is_project_wide():
    assert L.monthly_usd(20) == 0.0 and L.monthly_usd(21) == 0.5

def test_error_always_emitted_with_context():
    out = []
    for i in range(200):
        r = R.RequestLog(out.append, f"t{i}", random.Random(i)); r.note("a"); r.finish("error", ms=10)
    assert len(out) == 200 and all(l["context"] for l in out)

def test_sampling_rate_close_to_keep():
    out = []
    for i in range(20000):
        r = R.RequestLog(out.append, f"t{i}", random.Random(i)); r.finish("granted", ms=10)
    assert 0.09 < len(out) / 20000 < 0.11

def test_denied_always_audited():
    out = []
    for i in range(300):
        r = R.RequestLog(out.append, f"t{i}", random.Random(i)); r.finish("denied", ms=10)
    assert sum(l["kind"] == "audit" for l in out) == 300

実行例

📋 コピーprobe_gib    6.012
base_gib     694.422
bad_gib      3.649
good_gib     55.323
base_usd     337.211
bad_usd      0.000
good_usd     17.661
n_err        841.000
n_den        31,923.000
ctx_base     1.000
ctx_bad      0.000
ctx_good     1.000
deny_bad     0.000
deny_good    1.000
ok_ctx_bad   0.000
ok_ctx_good  0.104
audit_gib    18.938
audit_retain_usd 0.374
saving_good_usd 319.176
extra_bad_usd 18.035
breakeven_hours 0.451
slow_share   0.004
keep=0.01: good    30.4 GiB  $   5.20/月  成功の文脈が残る率 1.4%
keep= 0.1: good    55.3 GiB  $  17.66/月  成功の文脈が残る率 10.4%
keep= 1.0: good   302.9 GiB  $ 141.46/月  成功の文脈が残る率 100.0%
{"severity": "NOTICE", "kind": "audit", "trace": "t1", "outcome": "denied", "ms": 120, "coupon": "BDAY10"}
{"severity": "ERROR", "kind": "canonical", "trace": "t2", "outcome": "error", "ms": 300, "sample_rate": 1.0, "coupon": "BDAY10", "context": [{"msg": "rule_check", "rule": "birthday_month"}, {"msg": "db_read", "ms": 12}]}
{"severity": "INFO", "kind": "canonical", "trace": "t3", "outcome": "granted", "ms": 950, "sample_rate": 1.0, "coupon": "BDAY10", "context": [{"msg": "rule_check", "rule": "birthday_month"}, {"msg": "db_read", "ms": 12}]}

推奨案と提案書の書き直し

件名: coupon-api のログ費を月 $337 → $18 にする（95% 減）。失敗・遅延・拒否の文脈は 100% 残す

現状: 月 694 GiB。DEBUG 312／access 191／INFO 182 GiB で 99%。DEBUG は 8 月の戻し忘れ。

提案: (1) DEBUG を OFF にしバッファへ。失敗・800ms 超・拒否のときだけ文脈ごと 1 行 (2) 成功は canonical line を 10%（sample_rate 付き）、件数・率はメトリクス (3) probe はソースで除外

効果: 月 $337 → $17.66（−$319、年約 57 万円）。実装 12 時間（仮定 約 7.2 万円）で 1.5 か月回収。

残るリスク: 成功した普通のリクエストの文脈は 10%。付与は coupon_grants が正本。

確認と止め方: 1 サービス 1 日のカナリア（1.8 GiB/日前後）→全体。予算アラート月 $50。LOG_LEVEL=DEBUG で元に戻る。

見直す条件: 他サービスが無料枠を使い切る／文脈不足が月 1 回を超える／監査要件が変わる。

迷った案と採らなかった理由

ポイント解説

1. 費用は中身の内訳で決める: probe は 0.9%。「ヘルスチェックのせい」は外れ、最大は DEBUG の戻し忘れ。
2. 捨てる単位をリクエストに: 95% 減らしても調査に必要な 100% が残る。
3. 正本がある記録は二重に持たない: 付与は DB。拒否だけ audit で 100%（18.9 GiB）。
4. 最後の数ドルは人の時間と比べる: 差 月 $18 は 0.45 時間。
5. 無料枠は共有物: 他サービス次第で月 $0 が $27 に。

実務への応用

メール配信 Cloud Run・Argo Workflows のバッチでも、まず内訳と LOG_LEVEL の戻し忘れを確認。件数・率は OpenTelemetry のメトリクスから DataDog へ。拒否の audit は「クーポンが付かない」問い合わせの一次回答に使える。

扱わなかった範囲（止め方）

• 本番の実ログの分布は測っていない。行サイズ・結果の割合は仮定、遅延は対数正規の仮定。導入前に 1 日分を実測すること
• セール日のピーク、ログ書き込みの CPU・レイテンシへの影響（未測定）、log-based metrics（別課金）、vended network logs、DataDog・外部 SIEM
• 世界の実物との対応: canonical log lines・wide events・tail-based sampling のうち、リクエスト内バッファ＋canonical line まで。複数サービスをまたぐ tail-based sampling は扱っていない

発展問題: 複数サービスをまたぐトレースで、OpenTelemetry Collector の tail-based sampling をどう設定し、メモリと待ち時間をいくら払うか。

今日のまとめ

ログ費は severity で一括除外せず、内訳を測り、失敗・遅い・拒否の文脈だけ残して 95% 減らす。残りの 5% は 0 にするより、人の時間と比べて止める。

自己評価（あとで記入）