tegwick
35dbe4dd42
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa
232 lines
10 KiB
Python
Executable file
232 lines
10 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Capture non-secret Forgejo resource and recovery evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
CPU_FACTORS = {"n": 0.000001, "u": 0.001, "m": 1.0, "": 1000.0}
|
|
BYTE_FACTORS = {"": 1, "Ki": 1024, "Mi": 1024**2, "Gi": 1024**3, "Ti": 1024**4, "K": 1000, "M": 1000**2, "G": 1000**3}
|
|
|
|
|
|
def cpu_millicores(value: str | None) -> int:
|
|
match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)(n|u|m)?", str(value or "0"))
|
|
if not match:
|
|
raise ValueError(f"unsupported CPU quantity: {value}")
|
|
return round(float(match.group(1)) * CPU_FACTORS[match.group(2) or ""])
|
|
|
|
|
|
def bytes_value(value: str | None) -> int:
|
|
match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)(Ki|Mi|Gi|Ti|K|M|G)?", str(value or "0"))
|
|
if not match:
|
|
raise ValueError(f"unsupported byte quantity: {value}")
|
|
return round(float(match.group(1)) * BYTE_FACTORS[match.group(2) or ""])
|
|
|
|
|
|
def resource_totals(deployment: dict[str, Any]) -> dict[str, Any]:
|
|
spec = (((deployment.get("spec") or {}).get("template") or {}).get("spec") or {})
|
|
result = {
|
|
"requests": {"cpu_millicores": 0, "memory_bytes": 0},
|
|
"limits": {"cpu_millicores": 0, "memory_bytes": 0},
|
|
}
|
|
replicas = int((deployment.get("spec") or {}).get("replicas") or 0)
|
|
for container in spec.get("containers") or []:
|
|
resources = container.get("resources") or {}
|
|
for kind in ("requests", "limits"):
|
|
values = resources.get(kind) or {}
|
|
result[kind]["cpu_millicores"] += cpu_millicores(values.get("cpu")) * replicas
|
|
result[kind]["memory_bytes"] += bytes_value(values.get("memory")) * replicas
|
|
return result
|
|
|
|
|
|
def parse_top(text: str | None) -> dict[str, dict[str, int]]:
|
|
result: dict[str, dict[str, int]] = {}
|
|
for line in (text or "").splitlines():
|
|
fields = line.split()
|
|
if len(fields) >= 3:
|
|
result[fields[0]] = {"cpu_millicores": cpu_millicores(fields[1]), "memory_bytes": bytes_value(fields[2])}
|
|
return result
|
|
|
|
|
|
def build_observation(snapshot: dict[str, Any], captured_at: str) -> dict[str, Any]:
|
|
gaps: list[str] = []
|
|
top = parse_top(snapshot.get("top"))
|
|
if snapshot.get("top") is None:
|
|
gaps.append("forge pod metrics unavailable")
|
|
deployments = snapshot["deployments"].get("items") or []
|
|
deployment_rows = []
|
|
for deployment in deployments:
|
|
meta = deployment.get("metadata") or {}
|
|
status = deployment.get("status") or {}
|
|
name = str(meta.get("name"))
|
|
matching = [metric for pod, metric in top.items() if pod.startswith(f"{name}-")]
|
|
observed = None
|
|
if snapshot.get("top") is not None:
|
|
observed = {
|
|
"cpu_millicores": sum(item["cpu_millicores"] for item in matching),
|
|
"memory_bytes": sum(item["memory_bytes"] for item in matching),
|
|
}
|
|
deployment_rows.append({
|
|
"name": name,
|
|
"desired_replicas": int((deployment.get("spec") or {}).get("replicas") or 0),
|
|
"available_replicas": int(status.get("availableReplicas") or 0),
|
|
**resource_totals(deployment),
|
|
"observed": observed,
|
|
})
|
|
|
|
pvcs = []
|
|
total_pvc = 0
|
|
for item in snapshot["pvcs"].get("items") or []:
|
|
meta = item.get("metadata") or {}
|
|
spec = item.get("spec") or {}
|
|
requested = bytes_value(((spec.get("resources") or {}).get("requests") or {}).get("storage"))
|
|
total_pvc += requested
|
|
pvcs.append({
|
|
"name": meta.get("name"),
|
|
"storage_class": spec.get("storageClassName"),
|
|
"requested_bytes": requested,
|
|
"phase": (item.get("status") or {}).get("phase"),
|
|
})
|
|
|
|
database = snapshot["database"]
|
|
db_spec = database.get("spec") or {}
|
|
db_status = database.get("status") or {}
|
|
backups = [
|
|
item for item in snapshot["backups"].get("items") or []
|
|
if ((item.get("spec") or {}).get("cluster") or {}).get("name") == "forgejo-db"
|
|
]
|
|
completed = [item for item in backups if (item.get("status") or {}).get("phase") == "completed"]
|
|
if not backups:
|
|
gaps.append("forgejo-db has no Backup or ScheduledBackup evidence")
|
|
if snapshot.get("forge_data_kib") is None:
|
|
gaps.append("forge shared-storage used bytes unavailable")
|
|
if snapshot.get("runner_data_kib") is None:
|
|
gaps.append("runner storage used bytes unavailable")
|
|
gaps.extend([
|
|
"repository and package usage are not yet attributable to named consumers",
|
|
"runner job counts and durations are unavailable from the current read-only surface",
|
|
"material network traffic is unavailable",
|
|
])
|
|
|
|
return {
|
|
"schema_version": "railiance.forge-resource-observation.v1",
|
|
"record_type": "usage_observation",
|
|
"resource_id": "resource:railiance:forgejo",
|
|
"source": "railiance-forge",
|
|
"workplan_id": "RFORGE-WP-0002",
|
|
"reef": "reef-railiance",
|
|
"captured_at": captured_at,
|
|
"capacity": {
|
|
"deployments": deployment_rows,
|
|
"persistent_volumes": pvcs,
|
|
"pvc_requested_bytes": total_pvc,
|
|
"forge_data_used_bytes": int(snapshot["forge_data_kib"]) * 1024 if snapshot.get("forge_data_kib") is not None else None,
|
|
"runner_data_used_bytes": int(snapshot["runner_data_kib"]) * 1024 if snapshot.get("runner_data_kib") is not None else None,
|
|
"database": {
|
|
"instances": db_spec.get("instances"),
|
|
"ready_instances": db_status.get("readyInstances"),
|
|
"storage_requested_bytes": bytes_value((db_spec.get("storage") or {}).get("size")),
|
|
"phase": db_status.get("phase"),
|
|
},
|
|
},
|
|
"recovery": {
|
|
"database_continuous_archiving": any(c.get("type") == "ContinuousArchiving" and c.get("status") == "True" for c in db_status.get("conditions") or []),
|
|
"database_backups_seen": len(backups),
|
|
"database_backups_completed": len(completed),
|
|
"shared_storage_restore_drill": False,
|
|
"runner_restore_drill": False,
|
|
},
|
|
"allocation": {
|
|
"method_version": "forge-raw-drivers-v1",
|
|
"candidate_drivers": ["repository_count", "package_bytes", "runner_job_minutes"],
|
|
"consumer_groups": ["helix-forge", "coulomb-social", "railiance"],
|
|
"attributed": [],
|
|
"residual": {"forge_data_used_bytes": int(snapshot["forge_data_kib"]) * 1024 if snapshot.get("forge_data_kib") is not None else None, "reason": "consumer attribution source is not yet available"},
|
|
"selection_owner": "resource-control",
|
|
},
|
|
"measurement_gaps": sorted(set(gaps)),
|
|
"provenance": {
|
|
"commands": [
|
|
"kubectl get deploy -n forgejo -o json",
|
|
"kubectl get pvc -n forgejo -o json",
|
|
"kubectl top pods -n forgejo --no-headers",
|
|
"kubectl get cluster forgejo-db -n databases -o json",
|
|
"kubectl get backup,scheduledbackup -n databases -o json",
|
|
"kubectl exec deploy/forgejo-gitea -- du -sk /data",
|
|
"kubectl exec deploy/forgejo-runner -- du -sk /data",
|
|
],
|
|
"repository_contents_read": False,
|
|
"secret_surfaces_read": False,
|
|
},
|
|
}
|
|
|
|
|
|
def run(prefix: list[str], args: list[str], *, optional: bool = False) -> str | None:
|
|
completed = subprocess.run(prefix + args, text=True, capture_output=True, check=False)
|
|
if completed.returncode != 0:
|
|
if optional:
|
|
return None
|
|
raise subprocess.CalledProcessError(completed.returncode, prefix + args, completed.stdout, completed.stderr)
|
|
return completed.stdout
|
|
|
|
|
|
def collect(remote: str | None, kubectl: str) -> dict[str, Any]:
|
|
prefix = ["ssh", "-o", "BatchMode=yes", remote, kubectl] if remote else [kubectl]
|
|
load = lambda args: json.loads(run(prefix, args) or "{}")
|
|
data = {
|
|
"deployments": load(["get", "deploy", "-n", "forgejo", "-o", "json"]),
|
|
"pvcs": load(["get", "pvc", "-n", "forgejo", "-o", "json"]),
|
|
"database": load(["get", "cluster", "forgejo-db", "-n", "databases", "-o", "json"]),
|
|
"backups": load(["get", "backup,scheduledbackup", "-n", "databases", "-o", "json"]),
|
|
"top": run(prefix, ["top", "pods", "-n", "forgejo", "--no-headers"], optional=True),
|
|
}
|
|
# Avoid an inner shell here: when kubectl is executed through SSH, shell
|
|
# quoting can otherwise turn ``sh -c 'du ...'`` into ``sh -c du`` and
|
|
# silently measure the container working directory instead of /data.
|
|
forge_du = run(prefix, ["exec", "-n", "forgejo", "deploy/forgejo-gitea", "-c", "gitea", "--", "du", "-sk", "/data"], optional=True)
|
|
runner_du = run(prefix, ["exec", "-n", "forgejo", "deploy/forgejo-runner", "-c", "runner", "--", "du", "-sk", "/data"], optional=True)
|
|
|
|
def du_kib(output: str | None) -> int | None:
|
|
fields = (output or "").split()
|
|
return int(fields[0]) if fields and fields[0].isdigit() else None
|
|
|
|
data["forge_data_kib"] = du_kib(forge_du)
|
|
data["runner_data_kib"] = du_kib(runner_du)
|
|
captured_at = datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
|
return build_observation(data, captured_at)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--remote")
|
|
parser.add_argument("--kubectl", default="kubectl")
|
|
parser.add_argument("--output-dir", type=Path)
|
|
args = parser.parse_args()
|
|
try:
|
|
observation = collect(args.remote, args.kubectl)
|
|
except (OSError, subprocess.CalledProcessError, json.JSONDecodeError, ValueError) as exc:
|
|
print(f"forge observation failed: {exc}", file=sys.stderr)
|
|
return 1
|
|
rendered = json.dumps(observation, indent=2) + "\n"
|
|
if args.output_dir:
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
|
stamp = observation["captured_at"].replace(":", "").replace("-", "")
|
|
destination = args.output_dir / f"{stamp}.json"
|
|
destination.write_text(rendered, encoding="utf-8")
|
|
(args.output_dir / "latest.json").write_text(rendered, encoding="utf-8")
|
|
print(destination)
|
|
else:
|
|
sys.stdout.write(rendered)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|