# retoor <retoor@molodetz.nl>
import json
from devplacepy import config
from devplacepy.services.base import BaseService, ConfigField
from devplacepy.services.containers import api, store
from devplacepy.services.containers.runtime import get_backend
from devplacepy.services.devii.tasks.schedule import next_run, now_utc, to_iso
from devplacepy.services.audit import record as audit
AUTO_RESTART = ("always", "on-failure", "unless-stopped")
def _audit_reconcile(inst, action, summary, metadata=None):
audit.record_system(
"container.reconcile.action",
actor_kind="service",
origin="scheduler",
target_type="instance",
target_uid=inst.get("uid"),
target_label=inst.get("name"),
metadata={**(metadata or {}), "reconcile_action": action},
summary=summary,
links=[audit.instance(inst.get("uid"), inst.get("name"))],
)
class ContainerService(BaseService):
default_enabled = False
min_interval = 1
title = "Containers"
description = (
"Reconciles desired container state against the docker daemon: launches, stops, restarts, "
"reaps orphans, fires schedules, and samples per-instance metrics. Requires access to the "
"docker socket."
)
config_fields = [
ConfigField(
"container_metrics_every",
"Metrics sample every (ticks)",
type="int",
default=1,
minimum=1,
maximum=60,
help="Sample docker stats once every N reconcile ticks.",
group="Containers",
),
]
def __init__(self):
super().__init__(name="containers", interval_seconds=5)
self._metric_tick = 0
async def run_once(self) -> None:
backend = get_backend()
try:
ps_rows = await backend.ps(label_filter=api.INSTANCE_LABEL)
except Exception as exc:
self.log(f"docker ps failed: {exc}")
return
by_uid = {}
for row in ps_rows:
uid = row.labels.get(api.INSTANCE_LABEL)
if uid:
by_uid[uid] = row
instances = store.all_instances()
known = {inst["uid"] for inst in instances}
for uid, row in by_uid.items():
if uid not in known:
try:
await backend.rm(row.container_id, force=True)
self.log(f"reaped orphan container {row.name}")
_audit_reconcile(
{"uid": uid, "name": row.name},
"orphan_reap",
f"reconciler reaped orphan container {row.name}",
)
except Exception as exc:
self.log(f"orphan rm failed for {row.name}: {exc}")
for inst in instances:
try:
await self._reconcile(backend, inst, by_uid.get(inst["uid"]))
except Exception as exc:
self.log(f"reconcile {inst.get('name')} failed: {exc}")
await self._fire_schedules()
self._metric_tick += 1
if (
self._metric_tick
% max(1, self.get_config().get("container_metrics_every", 1))
== 0
):
await self._sample_metrics(backend, by_uid)
async def _reconcile(self, backend, inst, ps) -> None:
uid = inst["uid"]
desired = inst["desired_state"]
status = inst["status"]
if status == store.ST_REMOVING:
if ps is not None:
await backend.rm(ps.container_id, force=True)
store.delete_instance(uid)
self.log(f"removed instance {inst['name']}")
return
if desired == store.DESIRED_RUNNING:
if ps is None:
await self._launch(backend, inst)
elif ps.state == "running":
changes = {}
if status != store.ST_RUNNING:
changes["status"] = store.ST_RUNNING
changes["container_id"] = ps.container_id
changes["started_at"] = inst.get("started_at") or store.now()
ip, gateway = await self._capture_net(backend, ps.container_id)
if ip and ip != (inst.get("container_ip") or ""):
changes["container_ip"] = ip
if gateway and gateway != (inst.get("container_gateway") or ""):
changes["container_gateway"] = gateway
if changes:
store.update_instance(uid, changes)
elif ps.state == "paused":
await backend.unpause(ps.container_id)
store.update_instance(uid, {"status": store.ST_RUNNING})
elif ps.state == "created":
await backend.start(ps.container_id)
store.update_instance(uid, {"status": store.ST_RUNNING})
else:
await self._handle_exit(backend, inst, ps)
elif desired == store.DESIRED_STOPPED:
if ps is not None and ps.state in ("running", "restarting", "paused"):
await backend.stop(ps.container_id)
if status != store.ST_STOPPED:
store.update_instance(
uid, {"status": store.ST_STOPPED, "stopped_at": store.now()}
)
elif desired == store.DESIRED_PAUSED:
if ps is not None and ps.state == "running":
await backend.pause(ps.container_id)
store.update_instance(uid, {"status": store.ST_PAUSED})
async def _launch(self, backend, inst) -> None:
spec = api.run_spec_for(inst, config.CONTAINER_IMAGE)
try:
cid = await backend.run(spec)
except Exception as exc:
store.update_instance(inst["uid"], {"status": store.ST_CRASHED})
store.record_event(
inst, "launch_failed", "reconciler", "", {"reason": str(exc)}
)
self.log(f"launch {inst['name']} failed: {exc}")
return
changes = {
"container_id": cid,
"status": store.ST_RUNNING,
"started_at": store.now(),
}
ip, gateway = await self._capture_net(backend, cid)
if ip:
changes["container_ip"] = ip
if gateway:
changes["container_gateway"] = gateway
store.update_instance(inst["uid"], changes)
store.record_event(inst, "start", "reconciler", "")
self.log(f"launched instance {inst['name']}")
_audit_reconcile(inst, "start", f"reconciler started instance {inst['name']}")
async def _capture_net(self, backend, cid: str) -> tuple:
try:
data = await backend.inspect(cid)
except Exception:
return "", ""
return (
api.container_ip_from_inspect(data),
api.container_gateway_from_inspect(data),
)
async def _exit_logs(self, backend, container_id: str) -> str:
lines: list = []
async def collect(line: str) -> None:
lines.append(line)
try:
await backend.logs(container_id, follow=False, tail=20, on_log=collect)
except Exception:
return ""
return "\n".join(lines)[-2000:]
async def _handle_exit(self, backend, inst, ps) -> None:
uid = inst["uid"]
exit_code = ps.exit_code or 0
logs = await self._exit_logs(backend, ps.container_id)
policy = inst.get("restart_policy", "never")
if policy in AUTO_RESTART and not (policy == "on-failure" and exit_code == 0):
try:
await backend.start(ps.container_id)
store.update_instance(
uid,
{
"status": store.ST_RUNNING,
"restart_count": int(inst.get("restart_count") or 0) + 1,
},
)
store.record_event(
inst,
"policy_restart",
"reconciler",
"",
{"exit_code": exit_code, "logs": logs},
)
_audit_reconcile(
inst,
"policy_restart",
f"reconciler restarted instance {inst['name']} per {policy} policy",
{"exit_code": exit_code},
)
return
except Exception as exc:
self.log(f"policy restart {inst['name']} failed: {exc}")
terminal = store.ST_CRASHED if exit_code != 0 else store.ST_STOPPED
store.update_instance(
uid,
{
"status": terminal,
"desired_state": store.DESIRED_STOPPED,
"exit_code": exit_code,
"stopped_at": store.now(),
},
)
if terminal == store.ST_CRASHED:
store.record_event(
inst, "crash", "reconciler", "", {"exit_code": exit_code, "logs": logs}
)
async def _fire_schedules(self) -> None:
now = now_utc()
now_iso = to_iso(now)
for sched in store.due_schedules(now_iso):
inst = store.get_instance(sched["instance_uid"])
if inst is None:
store.delete_schedule(sched["uid"])
continue
action = sched["action"]
desired = (
store.DESIRED_RUNNING if action == "start" else store.DESIRED_STOPPED
)
store.update_instance(inst["uid"], {"desired_state": desired})
store.record_event(inst, f"schedule_{action}", "scheduler", "")
_audit_reconcile(
inst,
f"schedule_{action}",
f"reconciler fired scheduled {action} on instance {inst['name']}",
{"schedule_uid": sched["uid"]},
)
cols = json.loads(sched.get("schedule_json") or "{}")
run_count = int(sched.get("run_count") or 0) + 1
upcoming = next_run(
cols.get("kind"), cols.get("every_seconds"), cols.get("cron"), now
)
changes = {"last_run_at": now_iso, "run_count": run_count}
max_runs = cols.get("max_runs")
if upcoming is None or (max_runs and run_count >= max_runs):
changes["enabled"] = 0
changes["next_run_at"] = ""
else:
changes["next_run_at"] = to_iso(upcoming)
store.update_schedule(sched["uid"], changes)
self.log(f"schedule fired: {action} {inst['name']}")
async def _sample_metrics(self, backend, by_uid) -> None:
running = {uid: row for uid, row in by_uid.items() if row.state == "running"}
if not running:
return
try:
samples = await backend.stats_once(
[row.container_id for row in running.values()]
)
except Exception as exc:
self.log(f"docker stats failed: {exc}")
return
name_to_uid = {row.name: uid for uid, row in running.items()}
for name, sample in samples.items():
uid = name_to_uid.get(name)
if uid:
store.insert_metric(uid, sample)
def collect_metrics(self) -> dict:
instances = store.all_instances()
counts = {}
for inst in instances:
counts[inst["status"]] = counts.get(inst["status"], 0) + 1
running = counts.get(store.ST_RUNNING, 0)
stats = [
{"label": "Instances", "value": len(instances)},
{"label": "Running", "value": running},
{"label": "Stopped", "value": counts.get(store.ST_STOPPED, 0)},
{"label": "Crashed", "value": counts.get(store.ST_CRASHED, 0)},
{"label": "Paused", "value": counts.get(store.ST_PAUSED, 0)},
]
rows = [
[
inst.get("name", "")[:32],
inst.get("status", ""),
inst.get("desired_state", ""),
inst.get("restart_policy", ""),
int(inst.get("restart_count") or 0),
]
for inst in instances[:15]
]
table = {
"columns": ["Instance", "Status", "Desired", "Policy", "Restarts"],
"rows": rows,
}
return {"stats": stats, "table": table}
Reference in New Issue View Git Blame Copy Permalink