Implement SAND-WP-0008: host telemetry and self-canary
Add profile.sandbox-canary, HostSnapshot/inventory/stale schemas, SSH collectors, before/after provision deltas, telemetry export to State Hub and local JSON, default `sandboxer create` self-deploy, inspect/reap-stale CLI, runbook, and CoulombCore verification (26 tests pass).
This commit is contained in:
parent
582c1dd3c6
commit
c0a9261cdc
22 changed files with 1047 additions and 26 deletions
|
|
@ -9,13 +9,22 @@ import typer
|
|||
|
||||
from sandboxer import __version__
|
||||
from sandboxer.core.manager import SandboxManager
|
||||
from sandboxer.defaults import resolve_create_defaults
|
||||
from sandboxer.models import ActorType, Consumer, SandboxCreateRequest
|
||||
from sandboxer.placement import resolve_host
|
||||
from sandboxer.profiles.loader import load_profile
|
||||
from sandboxer.telemetry.export import export_telemetry
|
||||
from sandboxer.telemetry.introspection import build_introspection_report, collect_host_snapshot
|
||||
from sandboxer.telemetry.inventory import HostInventoryScanner
|
||||
from sandboxer.telemetry.reap import reap_stale
|
||||
|
||||
app = typer.Typer(
|
||||
name="sandboxer",
|
||||
help="Provision and manage isolated sandbox environments.",
|
||||
no_args_is_help=True,
|
||||
)
|
||||
inspect_app = typer.Typer(help="Host introspection without provisioning.")
|
||||
app.add_typer(inspect_app, name="inspect")
|
||||
|
||||
|
||||
@app.callback()
|
||||
|
|
@ -39,13 +48,36 @@ def _parse_inputs(values: list[str]) -> dict[str, str]:
|
|||
return inputs
|
||||
|
||||
|
||||
def _print_status(status: object) -> None:
|
||||
typer.echo(json.dumps(status, default=str, indent=2))
|
||||
def _print_json(data: object) -> None:
|
||||
typer.echo(json.dumps(data, default=str, indent=2))
|
||||
|
||||
|
||||
def _print_telemetry_summary(telemetry: dict | None) -> None:
|
||||
if not telemetry:
|
||||
return
|
||||
delta = telemetry.get("provision_delta") or telemetry.get("destroy_delta")
|
||||
stale = telemetry.get("stale_candidates", [])
|
||||
if delta:
|
||||
typer.echo(
|
||||
f"\nTelemetry: load Δ {delta.get('load_1m_delta', 0):+.3f}, "
|
||||
f"mem avail Δ {delta.get('mem_available_mb_delta', 0):+d} MB, "
|
||||
f"stale candidates: {len(stale)}",
|
||||
err=True,
|
||||
)
|
||||
after = delta.get("after") if delta else None
|
||||
if after:
|
||||
typer.echo(
|
||||
f" host load={after.get('load_1m')} mem_avail={after.get('mem_available_mb')} MB "
|
||||
f"disk_free={after.get('disk_root_avail_gb')} GB",
|
||||
err=True,
|
||||
)
|
||||
|
||||
|
||||
@app.command("create")
|
||||
def sandbox_create(
|
||||
profile: Annotated[str, typer.Option("--profile", help="Profile id")],
|
||||
profile: Annotated[
|
||||
str | None, typer.Option("--profile", help="Profile id (default: canary self-deploy)")
|
||||
] = None,
|
||||
input: Annotated[
|
||||
list[str] | None,
|
||||
typer.Option("--input", help="Input key=value (repeatable)"),
|
||||
|
|
@ -54,10 +86,12 @@ def sandbox_create(
|
|||
project: Annotated[str, typer.Option(help="Calling project id")] = "sand-boxer",
|
||||
host: Annotated[str | None, typer.Option(help="Override placement host")] = None,
|
||||
) -> None:
|
||||
"""Provision a sandbox from a profile."""
|
||||
"""Provision a sandbox. No args → canary self-deploy of sand-boxer."""
|
||||
parsed = _parse_inputs(input or [])
|
||||
resolved_profile, resolved_inputs = resolve_create_defaults(profile, parsed)
|
||||
request = SandboxCreateRequest(
|
||||
profile=profile,
|
||||
inputs=_parse_inputs(input or []),
|
||||
profile=resolved_profile,
|
||||
inputs=resolved_inputs,
|
||||
consumer=Consumer(actor=ActorType(actor), project=project),
|
||||
)
|
||||
manager = SandboxManager()
|
||||
|
|
@ -66,7 +100,9 @@ def sandbox_create(
|
|||
except Exception as exc:
|
||||
typer.echo(f"Error: {exc}", err=True)
|
||||
raise typer.Exit(code=1) from exc
|
||||
_print_status(status.model_dump(mode="json"))
|
||||
payload = status.model_dump(mode="json")
|
||||
_print_json(payload)
|
||||
_print_telemetry_summary(status.telemetry)
|
||||
|
||||
|
||||
@app.command("get")
|
||||
|
|
@ -76,7 +112,7 @@ def sandbox_get(sandbox_id: str) -> None:
|
|||
if not status:
|
||||
typer.echo(f"Sandbox not found: {sandbox_id}", err=True)
|
||||
raise typer.Exit(code=1)
|
||||
_print_status(status.model_dump(mode="json"))
|
||||
_print_json(status.model_dump(mode="json"))
|
||||
|
||||
|
||||
@app.command("list")
|
||||
|
|
@ -87,7 +123,7 @@ def sandbox_list(
|
|||
items = SandboxManager().list()
|
||||
if state:
|
||||
items = [s for s in items if s.state.value == state]
|
||||
_print_status([s.model_dump(mode="json") for s in items])
|
||||
_print_json([s.model_dump(mode="json") for s in items])
|
||||
|
||||
|
||||
@app.command("destroy")
|
||||
|
|
@ -99,7 +135,8 @@ def sandbox_destroy(sandbox_id: str) -> None:
|
|||
except KeyError as exc:
|
||||
typer.echo(str(exc), err=True)
|
||||
raise typer.Exit(code=1) from exc
|
||||
_print_status(status.model_dump(mode="json"))
|
||||
_print_json(status.model_dump(mode="json"))
|
||||
_print_telemetry_summary(status.telemetry)
|
||||
|
||||
|
||||
@app.command("recreate")
|
||||
|
|
@ -111,7 +148,72 @@ def sandbox_recreate(sandbox_id: str) -> None:
|
|||
except (KeyError, Exception) as exc:
|
||||
typer.echo(f"Error: {exc}", err=True)
|
||||
raise typer.Exit(code=1) from exc
|
||||
_print_status(status.model_dump(mode="json"))
|
||||
_print_json(status.model_dump(mode="json"))
|
||||
|
||||
|
||||
@inspect_app.command("host")
|
||||
def inspect_host(
|
||||
host: Annotated[str | None, typer.Option(help="Sandbox host")] = None,
|
||||
profile_id: Annotated[
|
||||
str, typer.Option(help="Profile for placement resolution")
|
||||
] = "profile.sandbox-canary",
|
||||
) -> None:
|
||||
"""Host snapshot and inventory (no sandbox create)."""
|
||||
profile = load_profile(profile_id)
|
||||
resolved = resolve_host(profile, override=host)
|
||||
snapshot = collect_host_snapshot(resolved)
|
||||
scanner = HostInventoryScanner(resolved)
|
||||
inventory = scanner.scan_inventory()
|
||||
stale = scanner.find_stale(SandboxManager().store)
|
||||
report = build_introspection_report(
|
||||
host=resolved,
|
||||
sandbox_id="inspect",
|
||||
profile=profile,
|
||||
provision_before=snapshot,
|
||||
provision_after=snapshot,
|
||||
store=SandboxManager().store,
|
||||
)
|
||||
export_telemetry(report)
|
||||
_print_json(
|
||||
{
|
||||
"host_snapshot": snapshot.model_dump(mode="json"),
|
||||
"inventory": inventory.model_dump(mode="json"),
|
||||
"stale_candidates": [s.model_dump(mode="json") for s in stale],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@inspect_app.command("stale")
|
||||
def inspect_stale(
|
||||
host: Annotated[str | None, typer.Option(help="Sandbox host")] = None,
|
||||
older_than: Annotated[float, typer.Option(help="Stale threshold hours")] = 24.0,
|
||||
) -> None:
|
||||
"""List stale sandbox candidates."""
|
||||
profile = load_profile("profile.sandbox-canary")
|
||||
resolved = resolve_host(profile, override=host)
|
||||
scanner = HostInventoryScanner(resolved, stale_hours=older_than)
|
||||
stale = scanner.find_stale(SandboxManager().store, stale_hours=older_than)
|
||||
_print_json([s.model_dump(mode="json") for s in stale])
|
||||
|
||||
|
||||
@app.command("reap-stale")
|
||||
def reap_stale_cmd(
|
||||
host: Annotated[str | None, typer.Option(help="Sandbox host")] = None,
|
||||
older_than: Annotated[float, typer.Option(help="Reap threshold hours")] = 24.0,
|
||||
apply: Annotated[bool, typer.Option("--apply", help="Actually remove stale resources")] = False,
|
||||
) -> None:
|
||||
"""Report or remove stale sandboxes on host (default: dry-run)."""
|
||||
profile = load_profile("profile.sandbox-canary")
|
||||
resolved = resolve_host(profile, override=host)
|
||||
results = reap_stale(
|
||||
resolved,
|
||||
SandboxManager().store,
|
||||
dry_run=not apply,
|
||||
stale_hours=older_than,
|
||||
)
|
||||
mode = "apply" if apply else "dry-run"
|
||||
typer.echo(f"reap-stale ({mode}): {len(results)} candidate(s)", err=True)
|
||||
_print_json([r.model_dump(mode="json") for r in results])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
|
|
@ -13,6 +13,12 @@ from sandboxer.models import (
|
|||
)
|
||||
from sandboxer.placement import resolve_host
|
||||
from sandboxer.profiles.loader import load_profile
|
||||
from sandboxer.telemetry.export import export_telemetry
|
||||
from sandboxer.telemetry.introspection import (
|
||||
build_introspection_report,
|
||||
collect_host_snapshot,
|
||||
profile_wants_telemetry,
|
||||
)
|
||||
|
||||
|
||||
class SandboxManager:
|
||||
|
|
@ -24,6 +30,8 @@ class SandboxManager:
|
|||
extension = load_extension(profile.extension)
|
||||
backend = resolve_backend(extension)
|
||||
resolved_host = resolve_host(profile, override=host)
|
||||
wants_telemetry = profile_wants_telemetry(profile)
|
||||
base_dir = extension.config.get("base_dir", "/tmp/sandboxer")
|
||||
|
||||
now = utcnow()
|
||||
status = SandboxStatus(
|
||||
|
|
@ -43,6 +51,10 @@ class SandboxManager:
|
|||
status.updated_at = utcnow()
|
||||
emit_lifecycle_event(status, event_type=event_type_for_state(status.state))
|
||||
|
||||
provision_before = None
|
||||
if wants_telemetry:
|
||||
provision_before = collect_host_snapshot(resolved_host)
|
||||
|
||||
try:
|
||||
handle = backend.provision(profile, request.inputs, resolved_host)
|
||||
status.sandbox_id = handle["sandbox_id"]
|
||||
|
|
@ -54,6 +66,21 @@ class SandboxManager:
|
|||
status.state = SandboxState.READY
|
||||
status.ready_at = utcnow()
|
||||
status.updated_at = status.ready_at
|
||||
|
||||
if wants_telemetry and provision_before:
|
||||
provision_after = collect_host_snapshot(resolved_host)
|
||||
report = build_introspection_report(
|
||||
host=resolved_host,
|
||||
sandbox_id=status.sandbox_id,
|
||||
profile=profile,
|
||||
provision_before=provision_before,
|
||||
provision_after=provision_after,
|
||||
store=self.store,
|
||||
base_dir=base_dir,
|
||||
)
|
||||
status.telemetry = report.model_dump(mode="json")
|
||||
export_telemetry(report)
|
||||
|
||||
self.store.save(status)
|
||||
emit_lifecycle_event(status, event_type=event_type_for_state(status.state))
|
||||
return status
|
||||
|
|
@ -86,6 +113,12 @@ class SandboxManager:
|
|||
profile = load_profile(status.profile_id)
|
||||
extension = load_extension(profile.extension)
|
||||
backend = resolve_backend(extension)
|
||||
wants_telemetry = profile_wants_telemetry(profile)
|
||||
base_dir = extension.config.get("base_dir", "/tmp/sandboxer")
|
||||
|
||||
destroy_before = None
|
||||
if wants_telemetry and status.host:
|
||||
destroy_before = collect_host_snapshot(status.host)
|
||||
|
||||
status.state = SandboxState.DESTROYING
|
||||
status.updated_at = utcnow()
|
||||
|
|
@ -106,6 +139,21 @@ class SandboxManager:
|
|||
status.state = SandboxState.DESTROYED
|
||||
status.destroyed_at = utcnow()
|
||||
status.updated_at = status.destroyed_at
|
||||
|
||||
if wants_telemetry and destroy_before and status.host:
|
||||
destroy_after = collect_host_snapshot(status.host)
|
||||
report = build_introspection_report(
|
||||
host=status.host,
|
||||
sandbox_id=status.sandbox_id,
|
||||
profile=profile,
|
||||
destroy_before=destroy_before,
|
||||
destroy_after=destroy_after,
|
||||
store=self.store,
|
||||
base_dir=base_dir,
|
||||
)
|
||||
status.telemetry = report.model_dump(mode="json")
|
||||
export_telemetry(report)
|
||||
|
||||
self.store.save(status)
|
||||
emit_lifecycle_event(status, event_type=event_type_for_state(status.state))
|
||||
return status
|
||||
|
|
|
|||
30
src/sandboxer/defaults.py
Normal file
30
src/sandboxer/defaults.py
Normal file
|
|
@ -0,0 +1,30 @@
|
|||
"""Default paths and profile resolution for CLI."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
DEFAULT_CANARY_PROFILE = "profile.sandbox-canary"
|
||||
DEFAULT_COMPOSE_PROFILE = "profile.compose-e2e"
|
||||
|
||||
|
||||
def repo_root() -> Path:
|
||||
override = os.environ.get("SANDBOXER_REPO_ROOT")
|
||||
if override:
|
||||
return Path(override).expanduser().resolve()
|
||||
return Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def resolve_create_defaults(
|
||||
profile: str | None,
|
||||
inputs: dict[str, str],
|
||||
) -> tuple[str, dict[str, str]]:
|
||||
"""Apply default profile and repo per SAND-WP-0008-T06."""
|
||||
resolved = dict(inputs)
|
||||
user_repo = "repo" in resolved
|
||||
if not user_repo:
|
||||
resolved["repo"] = str(repo_root())
|
||||
if profile is None:
|
||||
profile = DEFAULT_COMPOSE_PROFILE if user_repo else DEFAULT_CANARY_PROFILE
|
||||
return profile, resolved
|
||||
|
|
@ -39,6 +39,7 @@ def emit_lifecycle_event(
|
|||
"actor_type": status.consumer.actor.value,
|
||||
"state": status.state.value,
|
||||
"reachability": status.reachability.model_dump() if status.reachability else None,
|
||||
"telemetry": status.telemetry,
|
||||
"timestamps": {
|
||||
"created_at": status.created_at.isoformat(),
|
||||
"updated_at": status.updated_at.isoformat(),
|
||||
|
|
|
|||
|
|
@ -83,6 +83,7 @@ class ReachabilitySpec(BaseModel):
|
|||
class ProfileMetadata(BaseModel):
|
||||
cost_class: Literal["self-hosted", "saas-metered"] = "self-hosted"
|
||||
latency_class: str = "standard"
|
||||
observability: Literal["none", "canary"] = "none"
|
||||
|
||||
|
||||
class Profile(BaseModel):
|
||||
|
|
@ -141,6 +142,7 @@ class SandboxStatus(BaseModel):
|
|||
reachability: Reachability | None = None
|
||||
inputs: dict[str, str] = Field(default_factory=dict)
|
||||
error: str | None = None
|
||||
telemetry: dict | None = None # IntrospectionReport JSON when canary
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
ready_at: datetime | None = None
|
||||
|
|
|
|||
1
src/sandboxer/telemetry/__init__.py
Normal file
1
src/sandboxer/telemetry/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
"""Host telemetry and introspection."""
|
||||
64
src/sandboxer/telemetry/export.py
Normal file
64
src/sandboxer/telemetry/export.py
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
"""Telemetry export sinks."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Protocol
|
||||
|
||||
import httpx
|
||||
|
||||
from sandboxer.lifecycle.state_hub import hub_url
|
||||
from sandboxer.telemetry.models import IntrospectionReport
|
||||
|
||||
|
||||
class TelemetrySink(Protocol):
|
||||
"""Future export target (artifact-store, Prometheus, ClickHouse)."""
|
||||
|
||||
def publish(self, report: IntrospectionReport) -> None: ...
|
||||
|
||||
|
||||
def telemetry_dir() -> Path:
|
||||
base = Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share"))
|
||||
path = base / "sandboxer" / "telemetry"
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
return path
|
||||
|
||||
|
||||
def export_local_artifact(report: IntrospectionReport) -> Path:
|
||||
path = telemetry_dir() / f"{report.sandbox_id}.json"
|
||||
path.write_text(json.dumps(report.model_dump(mode="json"), indent=2, default=str))
|
||||
return path
|
||||
|
||||
|
||||
def export_state_hub(report: IntrospectionReport) -> dict | None:
|
||||
if os.environ.get("SANDBOXER_NO_STATE_HUB", "").lower() in ("1", "true", "yes"):
|
||||
return None
|
||||
payload = {
|
||||
"event_type": "note",
|
||||
"summary": (
|
||||
f"Telemetry {report.sandbox_id}: load Δ "
|
||||
f"{report.provision_delta.load_1m_delta if report.provision_delta else 0}, "
|
||||
f"stale={len(report.stale_candidates)}"
|
||||
),
|
||||
"author": "sandboxer",
|
||||
"detail": report.model_dump(mode="json"),
|
||||
}
|
||||
try:
|
||||
response = httpx.post(f"{hub_url()}/progress/", json=payload, timeout=10.0)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except httpx.HTTPError:
|
||||
return None
|
||||
|
||||
|
||||
def export_telemetry(report: IntrospectionReport) -> Path:
|
||||
path = export_local_artifact(report)
|
||||
export_state_hub(report)
|
||||
return path
|
||||
|
||||
|
||||
class NoopTelemetrySink:
|
||||
def publish(self, report: IntrospectionReport) -> None:
|
||||
export_telemetry(report)
|
||||
122
src/sandboxer/telemetry/host_snapshot.py
Normal file
122
src/sandboxer/telemetry/host_snapshot.py
Normal file
|
|
@ -0,0 +1,122 @@
|
|||
"""Collect HostSnapshot over SSH."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from sandboxer.extensions.ssh import SSHConfig
|
||||
from sandboxer.lifecycle.store import utcnow
|
||||
from sandboxer.telemetry.models import HostSnapshot
|
||||
|
||||
|
||||
def parse_loadavg(text: str) -> tuple[float, float, float]:
|
||||
parts = text.strip().split()
|
||||
return float(parts[0]), float(parts[1]), float(parts[2])
|
||||
|
||||
|
||||
def parse_meminfo(text: str) -> tuple[int, int]:
|
||||
total = avail = 0
|
||||
for line in text.splitlines():
|
||||
if line.startswith("MemTotal:"):
|
||||
total = int(line.split()[1]) // 1024
|
||||
elif line.startswith("MemAvailable:"):
|
||||
avail = int(line.split()[1]) // 1024
|
||||
return total, avail
|
||||
|
||||
|
||||
def parse_free_m(text: str) -> tuple[int, int]:
|
||||
for line in text.splitlines():
|
||||
if line.startswith("Mem:"):
|
||||
parts = line.split()
|
||||
return int(parts[1]), int(parts[6])
|
||||
return 0, 0
|
||||
|
||||
|
||||
def parse_df_root(text: str) -> tuple[float, float]:
|
||||
line = text.strip().splitlines()[-1]
|
||||
parts = line.split()
|
||||
used_pct = float(parts[4].rstrip("%"))
|
||||
avail = parts[3]
|
||||
mult = 1.0
|
||||
if avail[-1] in "KMGT":
|
||||
mult = {"K": 1 / 1e6, "M": 1 / 1e3, "G": 1.0, "T": 1000.0}[avail[-1]]
|
||||
avail = avail[:-1]
|
||||
return used_pct, float(avail) * mult
|
||||
|
||||
|
||||
def parse_container_count(text: str) -> int:
|
||||
lines = [ln for ln in text.strip().splitlines() if ln.strip()]
|
||||
return len(lines)
|
||||
|
||||
|
||||
class HostSnapshotCollector:
|
||||
def __init__(
|
||||
self, host: str, *, ssh_user: str | None = None, ssh_key: str | None = None
|
||||
) -> None:
|
||||
self.ssh = SSHConfig.from_env(host, user=ssh_user, key=ssh_key)
|
||||
self.host = host
|
||||
|
||||
def collect(self, *, collected_at: datetime | None = None) -> HostSnapshot:
|
||||
when = collected_at or utcnow()
|
||||
runtime = self._detect_runtime()
|
||||
load = self._run("cat /proc/loadavg")
|
||||
cpu = self._run("nproc")
|
||||
mem = self._run("free -m | awk '/^Mem:/{print $2\" \"$7}'")
|
||||
disk = self._run("df -h / | tail -1")
|
||||
running = self._run(f"{runtime} ps -q 2>/dev/null")
|
||||
sandbox = self._run(
|
||||
f"{runtime} ps -q --filter label=io.podman.compose.project=sbx 2>/dev/null"
|
||||
)
|
||||
if sandbox == "" and runtime == "docker":
|
||||
sandbox = self._run(
|
||||
"docker ps -q --filter label=com.docker.compose.project=sbx 2>/dev/null"
|
||||
)
|
||||
|
||||
load_vals = parse_loadavg(load) if load else (0.0, 0.0, 0.0)
|
||||
cpu_count = int(cpu.strip()) if cpu.strip().isdigit() else 0
|
||||
if mem and mem.strip():
|
||||
parts = mem.strip().split()
|
||||
mem_total, mem_avail = int(parts[0]), int(parts[1])
|
||||
else:
|
||||
mem_total, mem_avail = 0, 0
|
||||
disk_used, disk_avail = parse_df_root(disk) if disk else (0.0, 0.0)
|
||||
|
||||
return HostSnapshot(
|
||||
collected_at=when,
|
||||
host=self.host,
|
||||
load_1m=load_vals[0],
|
||||
load_5m=load_vals[1],
|
||||
load_15m=load_vals[2],
|
||||
cpu_count=cpu_count,
|
||||
mem_total_mb=mem_total,
|
||||
mem_available_mb=mem_avail,
|
||||
disk_root_used_pct=disk_used,
|
||||
disk_root_avail_gb=disk_avail,
|
||||
running_containers=parse_container_count(running),
|
||||
sandbox_containers=parse_container_count(sandbox),
|
||||
container_runtime=runtime,
|
||||
)
|
||||
|
||||
def _detect_runtime(self) -> str:
|
||||
rc, _ = self.ssh.run("command -v podman")
|
||||
if rc == 0:
|
||||
return "podman"
|
||||
rc, _ = self.ssh.run("command -v docker")
|
||||
if rc == 0:
|
||||
return "docker"
|
||||
return "unknown"
|
||||
|
||||
def _run(self, cmd: str) -> str:
|
||||
rc, out = self.ssh.run(cmd, timeout=10)
|
||||
if rc != 0:
|
||||
return ""
|
||||
return out
|
||||
|
||||
|
||||
def compute_delta(before: HostSnapshot, after: HostSnapshot) -> dict[str, float | int]:
|
||||
return {
|
||||
"load_1m_delta": round(after.load_1m - before.load_1m, 3),
|
||||
"mem_available_mb_delta": after.mem_available_mb - before.mem_available_mb,
|
||||
"running_containers_delta": after.running_containers - before.running_containers,
|
||||
"sandbox_containers_delta": after.sandbox_containers - before.sandbox_containers,
|
||||
}
|
||||
65
src/sandboxer/telemetry/introspection.py
Normal file
65
src/sandboxer/telemetry/introspection.py
Normal file
|
|
@ -0,0 +1,65 @@
|
|||
"""Assemble IntrospectionReport for canary profiles."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from sandboxer.lifecycle.store import SandboxStore, utcnow
|
||||
from sandboxer.models import Profile
|
||||
from sandboxer.telemetry.host_snapshot import HostSnapshot, compute_delta
|
||||
from sandboxer.telemetry.inventory import HostInventoryScanner
|
||||
from sandboxer.telemetry.models import IntrospectionReport, ProvisionDelta
|
||||
|
||||
|
||||
def profile_wants_telemetry(profile: Profile) -> bool:
|
||||
if profile.id == "profile.sandbox-canary":
|
||||
return True
|
||||
return profile.metadata.observability == "canary"
|
||||
|
||||
|
||||
def build_introspection_report(
|
||||
*,
|
||||
host: str,
|
||||
sandbox_id: str,
|
||||
profile: Profile,
|
||||
store: SandboxStore,
|
||||
base_dir: str = "/tmp/sandboxer",
|
||||
provision_before: HostSnapshot | None = None,
|
||||
provision_after: HostSnapshot | None = None,
|
||||
destroy_before: HostSnapshot | None = None,
|
||||
destroy_after: HostSnapshot | None = None,
|
||||
) -> IntrospectionReport:
|
||||
scanner = HostInventoryScanner(host, base_dir=base_dir)
|
||||
inventory = scanner.scan_inventory()
|
||||
stale = scanner.find_stale(store)
|
||||
|
||||
provision_delta = None
|
||||
if provision_before and provision_after:
|
||||
provision_delta = ProvisionDelta(
|
||||
before=provision_before,
|
||||
after=provision_after,
|
||||
**compute_delta(provision_before, provision_after),
|
||||
)
|
||||
|
||||
destroy_delta = None
|
||||
if destroy_before and destroy_after:
|
||||
destroy_delta = ProvisionDelta(
|
||||
before=destroy_before,
|
||||
after=destroy_after,
|
||||
**compute_delta(destroy_before, destroy_after),
|
||||
)
|
||||
|
||||
return IntrospectionReport(
|
||||
host=host,
|
||||
sandbox_id=sandbox_id,
|
||||
profile_id=profile.id,
|
||||
collected_at=utcnow(),
|
||||
provision_delta=provision_delta,
|
||||
destroy_delta=destroy_delta,
|
||||
inventory=inventory,
|
||||
stale_candidates=stale,
|
||||
)
|
||||
|
||||
|
||||
def collect_host_snapshot(host: str) -> HostSnapshot:
|
||||
from sandboxer.telemetry.host_snapshot import HostSnapshotCollector
|
||||
|
||||
return HostSnapshotCollector(host).collect()
|
||||
177
src/sandboxer/telemetry/inventory.py
Normal file
177
src/sandboxer/telemetry/inventory.py
Normal file
|
|
@ -0,0 +1,177 @@
|
|||
"""Sandbox inventory and stale candidate discovery."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from sandboxer.extensions.ssh import SSHConfig
|
||||
from sandboxer.lifecycle.store import SandboxStore, utcnow
|
||||
from sandboxer.models import SandboxState
|
||||
from sandboxer.telemetry.models import InventoryEntry, SandboxInventory, StaleCandidate
|
||||
|
||||
_PROJECT_RE = re.compile(r"^(sbx-.+|e2e-.+)$")
|
||||
|
||||
|
||||
def _age_hours(epoch_str: str) -> float | None:
|
||||
try:
|
||||
# stat format %Y
|
||||
ts = int(epoch_str.strip())
|
||||
return round((datetime.now(UTC).timestamp() - ts) / 3600, 2)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _profile_hint_from_project(project: str) -> str | None:
|
||||
if project.startswith("sbx-"):
|
||||
parts = project.split("-")
|
||||
if len(parts) >= 3:
|
||||
return f"profile.{parts[1]}"
|
||||
return None
|
||||
|
||||
|
||||
class HostInventoryScanner:
|
||||
def __init__(
|
||||
self,
|
||||
host: str,
|
||||
*,
|
||||
base_dir: str = "/tmp/sandboxer",
|
||||
ssh_user: str | None = None,
|
||||
stale_hours: float = 24.0,
|
||||
) -> None:
|
||||
self.host = host
|
||||
self.base_dir = base_dir
|
||||
self.ssh = SSHConfig.from_env(host, user=ssh_user)
|
||||
self.stale_hours = stale_hours
|
||||
|
||||
def scan_inventory(self) -> SandboxInventory:
|
||||
when = utcnow()
|
||||
entries: list[InventoryEntry] = []
|
||||
entries.extend(self._scan_directories())
|
||||
entries.extend(self._scan_compose_projects())
|
||||
return SandboxInventory(
|
||||
host=self.host,
|
||||
base_dir=self.base_dir,
|
||||
collected_at=when,
|
||||
entries=entries,
|
||||
)
|
||||
|
||||
def find_stale(
|
||||
self,
|
||||
store: SandboxStore,
|
||||
*,
|
||||
stale_hours: float | None = None,
|
||||
) -> list[StaleCandidate]:
|
||||
threshold = stale_hours if stale_hours is not None else self.stale_hours
|
||||
inventory = self.scan_inventory()
|
||||
on_host_ids = {e.id for e in inventory.entries}
|
||||
store_by_id = {
|
||||
s.sandbox_id: s
|
||||
for s in store.list_all()
|
||||
if s.state != SandboxState.DESTROYED
|
||||
}
|
||||
|
||||
candidates: list[StaleCandidate] = []
|
||||
|
||||
for entry in inventory.entries:
|
||||
in_store = entry.id in store_by_id
|
||||
if not in_store:
|
||||
kind = "orphan_dir" if entry.kind == "directory" else "orphan_compose"
|
||||
candidates.append(
|
||||
StaleCandidate(
|
||||
kind=kind,
|
||||
id=entry.id,
|
||||
path=entry.path,
|
||||
age_hours=entry.age_hours,
|
||||
action="reap",
|
||||
reason="present on host but absent from local store",
|
||||
)
|
||||
)
|
||||
elif entry.age_hours is not None and entry.age_hours >= threshold:
|
||||
candidates.append(
|
||||
StaleCandidate(
|
||||
kind="aged_dir" if entry.kind == "directory" else "orphan_compose",
|
||||
id=entry.id,
|
||||
path=entry.path,
|
||||
age_hours=entry.age_hours,
|
||||
action="reap",
|
||||
reason=f"older than {threshold}h threshold",
|
||||
)
|
||||
)
|
||||
|
||||
for sid, status in store_by_id.items():
|
||||
if sid not in on_host_ids and status.host == self.host:
|
||||
candidates.append(
|
||||
StaleCandidate(
|
||||
kind="zombie_record",
|
||||
id=sid,
|
||||
path=status.reachability.remote_dir if status.reachability else None,
|
||||
age_hours=None,
|
||||
action="inspect",
|
||||
reason="recorded in store but missing on host",
|
||||
)
|
||||
)
|
||||
|
||||
return candidates
|
||||
|
||||
def _scan_directories(self) -> list[InventoryEntry]:
|
||||
cmd = (
|
||||
f"if [ -d {self.base_dir} ]; then "
|
||||
f"find {self.base_dir} -mindepth 1 -maxdepth 1 -type d "
|
||||
f"-printf '%f %Y\\n'; fi"
|
||||
)
|
||||
rc, out = self.ssh.run(cmd, timeout=15)
|
||||
if rc != 0 or not out.strip():
|
||||
return []
|
||||
entries = []
|
||||
for line in out.strip().splitlines():
|
||||
parts = line.split()
|
||||
if len(parts) < 2:
|
||||
continue
|
||||
sid, mtime = parts[0], parts[1]
|
||||
entries.append(
|
||||
InventoryEntry(
|
||||
kind="directory",
|
||||
id=sid,
|
||||
path=f"{self.base_dir}/{sid}",
|
||||
age_hours=_age_hours(mtime),
|
||||
)
|
||||
)
|
||||
return entries
|
||||
|
||||
def _scan_compose_projects(self) -> list[InventoryEntry]:
|
||||
runtime = "podman" if self._has_podman() else "docker"
|
||||
if runtime == "podman":
|
||||
cmd = (
|
||||
"podman ps -a --format '{{.Labels}}' 2>/dev/null | "
|
||||
"grep -o 'io.podman.compose.project=sbx[^, ]*' | "
|
||||
"sed 's/io.podman.compose.project=//' | sort -u"
|
||||
)
|
||||
else:
|
||||
cmd = (
|
||||
"docker ps -a --format '{{.Label \"com.docker.compose.project\"}}' 2>/dev/null | "
|
||||
"grep '^sbx' | sort -u"
|
||||
)
|
||||
rc, out = self.ssh.run(cmd, timeout=15)
|
||||
if rc != 0 or not out.strip():
|
||||
return []
|
||||
entries = []
|
||||
for project in out.strip().splitlines():
|
||||
project = project.strip()
|
||||
if not _PROJECT_RE.match(project):
|
||||
continue
|
||||
sandbox_id = project.rsplit("-", 1)[-1]
|
||||
entries.append(
|
||||
InventoryEntry(
|
||||
kind="compose_project",
|
||||
id=sandbox_id,
|
||||
path=f"compose:{project}",
|
||||
age_hours=None,
|
||||
profile_hint=_profile_hint_from_project(project),
|
||||
)
|
||||
)
|
||||
return entries
|
||||
|
||||
def _has_podman(self) -> bool:
|
||||
rc, _ = self.ssh.run("command -v podman")
|
||||
return rc == 0
|
||||
70
src/sandboxer/telemetry/models.py
Normal file
70
src/sandboxer/telemetry/models.py
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
"""Telemetry and introspection schemas."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
SCHEMA_VERSION = "0.1"
|
||||
|
||||
|
||||
class HostSnapshot(BaseModel):
|
||||
collected_at: datetime
|
||||
host: str
|
||||
load_1m: float = 0.0
|
||||
load_5m: float = 0.0
|
||||
load_15m: float = 0.0
|
||||
cpu_count: int = 0
|
||||
mem_total_mb: int = 0
|
||||
mem_available_mb: int = 0
|
||||
disk_root_used_pct: float = 0.0
|
||||
disk_root_avail_gb: float = 0.0
|
||||
running_containers: int = 0
|
||||
sandbox_containers: int = 0
|
||||
container_runtime: str = "unknown"
|
||||
|
||||
|
||||
class InventoryEntry(BaseModel):
|
||||
kind: str # directory | compose_project
|
||||
id: str
|
||||
path: str | None = None
|
||||
age_hours: float | None = None
|
||||
profile_hint: str | None = None
|
||||
|
||||
|
||||
class SandboxInventory(BaseModel):
|
||||
host: str
|
||||
base_dir: str
|
||||
collected_at: datetime
|
||||
entries: list[InventoryEntry] = Field(default_factory=list)
|
||||
|
||||
|
||||
class StaleCandidate(BaseModel):
|
||||
kind: str # orphan_dir | orphan_compose | zombie_record | aged_dir
|
||||
id: str
|
||||
path: str | None = None
|
||||
age_hours: float | None = None
|
||||
action: str # reap | inspect | ignore
|
||||
reason: str
|
||||
|
||||
|
||||
class ProvisionDelta(BaseModel):
|
||||
before: HostSnapshot
|
||||
after: HostSnapshot
|
||||
load_1m_delta: float = 0.0
|
||||
mem_available_mb_delta: int = 0
|
||||
running_containers_delta: int = 0
|
||||
sandbox_containers_delta: int = 0
|
||||
|
||||
|
||||
class IntrospectionReport(BaseModel):
|
||||
schema_version: str = SCHEMA_VERSION
|
||||
host: str
|
||||
sandbox_id: str
|
||||
profile_id: str
|
||||
collected_at: datetime
|
||||
provision_delta: ProvisionDelta | None = None
|
||||
destroy_delta: ProvisionDelta | None = None
|
||||
inventory: SandboxInventory | None = None
|
||||
stale_candidates: list[StaleCandidate] = Field(default_factory=list)
|
||||
36
src/sandboxer/telemetry/reap.py
Normal file
36
src/sandboxer/telemetry/reap.py
Normal file
|
|
@ -0,0 +1,36 @@
|
|||
"""Stale sandbox reap (dry-run and apply)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from sandboxer.extensions.ssh import SSHConfig
|
||||
from sandboxer.lifecycle.store import SandboxStore
|
||||
from sandboxer.telemetry.inventory import HostInventoryScanner
|
||||
from sandboxer.telemetry.models import StaleCandidate
|
||||
|
||||
|
||||
def reap_stale(
|
||||
host: str,
|
||||
store: SandboxStore,
|
||||
*,
|
||||
dry_run: bool = True,
|
||||
stale_hours: float = 24.0,
|
||||
base_dir: str = "/tmp/sandboxer",
|
||||
) -> list[StaleCandidate]:
|
||||
scanner = HostInventoryScanner(host, base_dir=base_dir, stale_hours=stale_hours)
|
||||
candidates = [
|
||||
c for c in scanner.find_stale(store, stale_hours=stale_hours) if c.action == "reap"
|
||||
]
|
||||
if dry_run:
|
||||
return candidates
|
||||
|
||||
compose_cmd = os.environ.get("SANDBOXER_COMPOSE_CMD", "podman-compose")
|
||||
ssh = SSHConfig.from_env(host)
|
||||
for item in candidates:
|
||||
if item.kind in ("orphan_dir", "aged_dir") and item.path:
|
||||
ssh.run(f"rm -rf {item.path}", timeout=30)
|
||||
elif item.kind == "orphan_compose" and item.path and item.path.startswith("compose:"):
|
||||
project = item.path.split(":", 1)[1]
|
||||
ssh.run(f"{compose_cmd} -p {project} down -v 2>/dev/null || true", timeout=60)
|
||||
return candidates
|
||||
Loading…
Add table
Add a link
Reference in a new issue