1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@taquangku

Cloud Build の履歴を CSV に出して、失敗したままのトリガーだけ再実行する

1
Last updated at Posted at 2026-10-07

マイクロサービス構成で Cloud Build のトリガーが数十個あると、「今日のビルドでまだ落ちたままのサービスはどれか」をコンソールで追うのがつらくなります。再実行したものも時系列に混ざって並ぶので、「1回落ちたが直った」と「再実行しても落ちている」が一目で区別できません。

標準ライブラリだけの Python スクリプト2本で解決しました。使うのは実質このコマンドだけです。

python3 rerun_failed_builds.py --project my-project        # 失敗中のトリガーを表示
python3 rerun_failed_builds.py --project my-project --run  # それを再実行

cloudbuild_20261007_develop.csv: 56 builds (SUCCESS 25 / OTHER 31)
実行中 2 件（対象外）: build-sample-service, build-other-service
対象 2 件（dry-run）
  gcloud builds triggers run build-sample-service --branch=develop --region=asia-northeast1 --project=my-project
  ...

考え方

トリガーごとに最新の1件だけを見る。 これだけで「まだ落ちているサービス」の一覧になります。

rows.sort(key=lambda r: r["create_time_jst"])
latest = {}
for r in rows:
    latest[r["trigger_name"]] = r      # 時系列順に上書き → 最後に残るのが最新

役割は2本に分けています。

スクリプト 役割
fetch_cloudbuild_history.py 履歴を CSV に出力する
rerun_failed_builds.py CSV を読んで、最新が失敗のトリガーを再実行する

rerun_failed_builds.py は起動時に fetch_cloudbuild_history.py を自動で呼ぶので、普段は後者を意識する必要はありません（--no-refresh でスキップ可）。CSV 単体は Excel で集計するときに使えます。

ハマったところ

1. --region を付けないと 0 件になる

Cloud Build の履歴はリージョンごとに分かれています。省略すると global を見にいきます。

gcloud builds list --project=my-project --limit=10                            # 0 件
gcloud builds list --project=my-project --region=asia-northeast1 --limit=10   # こちら

2. --page-size を付けないと静かに取りこぼす

これが一番ハマりました。--limit=1000 を付けているのに、返ってくる件数が頭打ちになります。

# 47 件しか返らない
gcloud builds list --project=my-project --region=asia-northeast1 \
  --limit=1000 --filter='createTime>="2026-10-07T00:00:00+09:00"' --format='value(id)' | wc -l

# 66 件（正しい）
gcloud builds list --project=my-project --region=asia-northeast1 \
  --limit=1000 --page-size=500 --filter='createTime>="2026-10-07T00:00:00+09:00"' --format='value(id)' | wc -l

--filter はクライアント側で評価されるため、--page-size 未指定だとページングが途中で打ち切られて古いビルドが欠落します。エラーにならず件数だけ減るので気づきにくい。実際、この取りこぼしのせいで「トリガー数が勝手に減る」という謎の挙動に悩まされました。

3. ブランチ名は substitutions にある

トップレベルの source には GitHub App 経由だとコミットSHAしか入らないことがあります。実用的な情報は substitutions 配下です。

{
  "substitutions": {
    "TRIGGER_NAME": "build-sample-service",
    "BRANCH_NAME": "develop",
    "REPO_NAME": "my-repo",
    "SHORT_SHA": "55679e5",
    "_PR_NUMBER": "123"
  }
}

4. タイムスタンプが RFC3339 として壊れていることがある

datetime.fromisoformat() に素で食わせると落ちます。実際に返ってきた値:

2026-10-06T23:27:0218577Z     ← 秒とマイクロ秒の間にドットが無い

正規表現で分解してからパースしています。

5. WORKING を失敗扱いしない

status != "SUCCESS" で判定すると、実行中のビルドまで再実行対象になります。終了ステータスだけを対象にしましょう。

FAILED_STATUSES  = {"FAILURE", "TIMEOUT", "INTERNAL_ERROR", "EXPIRED", "CANCELLED"}
RUNNING_STATUSES = {"WORKING", "QUEUED", "PENDING"}

コード

fetch_cloudbuild_history.py

fetch_cloudbuild_history.py
#!/usr/bin/env python3
"""Cloud Build の履歴を取得して CSV に保存する。

既定は当日(JST)分・全ブランチ。

使い方:
    python3 fetch_cloudbuild_history.py --project my-project
    python3 fetch_cloudbuild_history.py --project my-project --date 2026-10-06
    python3 fetch_cloudbuild_history.py --project my-project --branch develop --days 3
"""
import argparse
import csv
import datetime
import json
import re
import subprocess
import sys

JST = datetime.timezone(datetime.timedelta(hours=9))
FIELDS = [
    "build_id", "status", "trigger_name", "repo", "branch", "short_sha",
    "pr_number", "create_time_jst", "start_time_jst", "finish_time_jst",
    "duration_sec", "failure_detail", "log_url",
]
TS_RE = re.compile(r"(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d+))?(Z|[+-]\d{2}:\d{2})$")


def parse_ts(value):
    """RFC3339 文字列を datetime に変換（小数秒の桁数が不正な値にも対応）。"""
    if not value:
        return None
    m = TS_RE.match(value)
    if not m:
        return None
    base, frac, offset = m.group(1), m.group(2) or "0", m.group(3)
    offset = "+00:00" if offset == "Z" else offset
    return datetime.datetime.fromisoformat(base + offset) + datetime.timedelta(seconds=float("0." + frac))


def to_jst(value):
    ts = parse_ts(value)
    return ts.astimezone(JST).strftime("%Y-%m-%d %H:%M:%S") if ts else ""


def duration_sec(start, finish):
    s, f = parse_ts(start), parse_ts(finish)
    return str(round((f - s).total_seconds())) if s and f else ""


def fetch_builds(project, region, since, until, limit):
    filter_expr = f'createTime>="{since}" AND createTime<"{until}"'
    cmd = [
        "gcloud", "builds", "list",
        f"--project={project}", f"--region={region}",
        f"--filter={filter_expr}", f"--limit={limit}",
        # --page-size を省くと gcloud がページングを打ち切り、取りこぼす
        "--page-size=500", "--format=json",
    ]
    proc = subprocess.run(cmd, capture_output=True, text=True)
    if proc.returncode != 0:
        sys.exit(f"gcloud failed:\n{proc.stderr.strip()}")
    return json.loads(proc.stdout or "[]")


def to_row(build):
    sub = build.get("substitutions", {})
    return {
        "build_id": build.get("id", ""),
        "status": build.get("status", ""),
        "trigger_name": sub.get("TRIGGER_NAME", ""),
        "repo": sub.get("REPO_NAME", ""),
        "branch": sub.get("BRANCH_NAME", ""),
        "short_sha": sub.get("SHORT_SHA", ""),
        "pr_number": sub.get("_PR_NUMBER", ""),
        "create_time_jst": to_jst(build.get("createTime")),
        "start_time_jst": to_jst(build.get("startTime")),
        "finish_time_jst": to_jst(build.get("finishTime")),
        "duration_sec": duration_sec(build.get("startTime"), build.get("finishTime")),
        "failure_detail": (build.get("failureInfo") or {}).get("detail", "").replace("\n", " "),
        "log_url": build.get("logUrl", ""),
    }


def main():
    today = datetime.datetime.now(JST).date()
    p = argparse.ArgumentParser()
    p.add_argument("--project", required=True)
    p.add_argument("--region", default="asia-northeast1")
    p.add_argument("--branch", default="", help="空文字（既定）で全ブランチ")
    p.add_argument("--date", default=today.isoformat(), help="取得開始日(JST, YYYY-MM-DD)。既定は本日")
    p.add_argument("--days", type=int, default=1, help="--date から何日分か")
    p.add_argument("--limit", type=int, default=1000)
    p.add_argument("--out", default=None, help="既定: cloudbuild_<YYYYMMDD>_<branch>.csv")
    args = p.parse_args()

    start = datetime.date.fromisoformat(args.date)
    end = start + datetime.timedelta(days=args.days)
    since = f"{start.isoformat()}T00:00:00+09:00"
    until = f"{end.isoformat()}T00:00:00+09:00"

    builds = fetch_builds(args.project, args.region, since, until, args.limit)
    builds.sort(key=lambda b: b.get("createTime", ""))
    rows = [to_row(b) for b in builds]
    if args.branch:
        rows = [r for r in rows if r["branch"] == args.branch]

    out = args.out or f"cloudbuild_{start.strftime('%Y%m%d')}_{args.branch or 'all'}.csv"
    with open(out, "w", newline="", encoding="utf-8-sig") as f:  # Excel 用に BOM 付き
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        writer.writerows(rows)

    ok = sum(1 for r in rows if r["status"] == "SUCCESS")
    print(f"{out}: {len(rows)} builds (SUCCESS {ok} / OTHER {len(rows) - ok})")


if __name__ == "__main__":
    main()

CSV の主な列は status / trigger_name / branch / short_sha / create_time_jst / duration_sec / failure_detail / log_url。Excel で文字化けしないよう encoding="utf-8-sig"（BOM 付き）で書いています。

日付フィルタはオフセット付きで書けるので、JST の0時境界をそのまま渡せます。

createTime>="2026-10-07T00:00:00+09:00" AND createTime<"2026-10-08T00:00:00+09:00"

rerun_failed_builds.py

再実行は gcloud builds triggers run です。誤爆するとビルドが一斉に走るので、既定を dry-run にして --run を付けたときだけ実行します。

rerun_failed_builds.py
#!/usr/bin/env python3
"""本日の最新ビルドが失敗しているトリガーを再実行する。

fetch_cloudbuild_history.py が出力した CSV を読み、トリガーごとの最新行が
SUCCESS でないものを `gcloud builds triggers run` で再実行する。
既定は dry-run（コマンド表示のみ）。実行するには --run を付ける。

実行前に fetch_cloudbuild_history.py を自動実行して CSV を最新化する
（スキップするには --no-refresh）。

使い方:
    python3 rerun_failed_builds.py --project my-project               # 確認のみ
    python3 rerun_failed_builds.py --project my-project --run         # 再実行
    python3 rerun_failed_builds.py --project my-project --run --only api- batch-
    python3 rerun_failed_builds.py --project my-project --no-refresh  # 既存CSVのまま
"""
import argparse
import csv
import datetime
import os
import subprocess
import sys

JST = datetime.timezone(datetime.timedelta(hours=9))
# 再実行対象にする終了ステータス（WORKING/QUEUED は実行中なので対象外）
FAILED_STATUSES = {"FAILURE", "TIMEOUT", "INTERNAL_ERROR", "EXPIRED", "CANCELLED"}
RUNNING_STATUSES = {"WORKING", "QUEUED", "PENDING"}


def latest_builds(csv_path):
    """トリガーごとの最新ビルド行を返す。"""
    with open(csv_path, encoding="utf-8-sig") as f:
        rows = list(csv.DictReader(f))
    rows.sort(key=lambda r: r["create_time_jst"])
    latest = {}
    for r in rows:
        latest[r["trigger_name"]] = r
    return list(latest.values())


def main():
    today = datetime.datetime.now(JST).date()
    p = argparse.ArgumentParser()
    p.add_argument("--project", required=True)
    p.add_argument("--region", default="asia-northeast1")
    p.add_argument("--branch", default="", help="対象ブランチ")
    p.add_argument("--csv", default=None, help="既定: cloudbuild_<本日>_<branch>.csv")
    p.add_argument("--only", nargs="*", default=None, help="トリガー名の部分一致で絞り込み")
    p.add_argument("--run", action="store_true", help="実際に再実行する（省略時は dry-run）")
    p.add_argument("--no-refresh", action="store_true", help="CSV の自動再取得をスキップする")
    args = p.parse_args()

    csv_path = args.csv or f"cloudbuild_{today.strftime('%Y%m%d')}_{args.branch or 'all'}.csv"

    if not args.no_refresh:
        fetch = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fetch_cloudbuild_history.py")
        proc = subprocess.run(
            [sys.executable, fetch, f"--project={args.project}", f"--region={args.region}",
             f"--branch={args.branch}", f"--out={csv_path}"],
            capture_output=True, text=True)
        if proc.returncode != 0:
            sys.exit(f"履歴の再取得に失敗しました:\n{proc.stderr.strip()}")
        print(proc.stdout.strip())

    try:
        latest = latest_builds(csv_path)
    except FileNotFoundError:
        sys.exit(f"CSV が見つかりません: {csv_path}\n先に fetch_cloudbuild_history.py を実行してください。")

    running = [r for r in latest if r["status"] in RUNNING_STATUSES]
    if running:
        print(f"実行中 {len(running)} 件（対象外）: " + ", ".join(sorted(r["trigger_name"] for r in running)))
    targets = [r for r in latest if r["status"] in FAILED_STATUSES]

    if args.only:
        targets = [r for r in targets if any(k in r["trigger_name"] for k in args.only)]
    if not targets:
        print("再実行対象なし")
        return

    print(f"対象 {len(targets)} 件" + ("" if args.run else "（dry-run）"))
    for r in sorted(targets, key=lambda r: r["trigger_name"]):
        cmd = [
            "gcloud", "builds", "triggers", "run", r["trigger_name"],
            f"--branch={args.branch}", f"--region={args.region}", f"--project={args.project}",
        ]
        if not args.run:
            print("  " + " ".join(cmd))
            continue
        proc = subprocess.run(cmd, capture_output=True, text=True)
        status = "OK" if proc.returncode == 0 else "NG"
        detail = proc.stdout.strip() if proc.returncode == 0 else proc.stderr.strip().splitlines()[-1]
        print(f"  [{status}] {r['trigger_name']}: {detail[:160]}")


if __name__ == "__main__":
    main()

使い方

python3 rerun_failed_builds.py --project my-project --branch develop        # 確認（dry-run）
python3 rerun_failed_builds.py --project my-project --branch develop --run  # 再実行
python3 rerun_failed_builds.py --project my-project --run --only api- batch-  # 一部だけ

失敗の詳細ログは CSV の log_url を開くか、CLI でも追えます。

gcloud builds log <build_id> --region=asia-northeast1 --project=my-project

履歴 CSV だけ欲しいときは単体でも叩けます。

python3 fetch_cloudbuild_history.py --project my-project --date 2026-10-06  # 日付指定（JST）
python3 fetch_cloudbuild_history.py --project my-project --days 7           # 7日分

おわりに

コードを直さずに再実行しても、同じコミットでは同じ結果になります。 再実行が手軽になるぶん原因調査を飛ばしがちなので、dry-run → log_url でログ確認、を挟む運用にしています。

ポイントをまとめると:

  • Cloud Build の履歴はリージョンごと。--region 必須
  • --page-size を付けないと静かに取りこぼす
  • ブランチ名・トリガー名は substitutions から取る
  • createTime の小数秒が壊れていることがあるので fromisoformat() を素で使わない
  • 「トリガーごとの最新1件」を見るだけで、まだ落ちているサービスの一覧になる
1
0
1

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?