diff --git a/agents.md b/agents.md index 96391ac..bda5c2e 100644 --- a/agents.md +++ b/agents.md @@ -24,8 +24,10 @@ shard_core/ management/ Management API auth (hosted shards only) service/ → Business logic app_installation/ App install/uninstall/reinstall + background worker queue - app_lifecycle.py Auto-start/stop containers based on idle time - app_tools.py Docker CLI wrapper functions + app_lifecycle.py Two-tier idle control: RUNNING -> PAUSED -> STOPPED, PSI-driven LRU demotion, wake-on-request + app_tools.py Docker CLI wrapper functions (start/stop/pause/unpause/down) + memory_pressure.py PSI parsing (/host/pressure/memory), cgroup v2 memory.reclaim page-out + pause_metrics.py In-memory pause-tier telemetry accumulators (transitions, latencies, PSI snapshots) pairing.py Terminal pairing (JWT creation, code generation) backup.py Azure Blob Storage backup via rclone peer.py Peer shard management @@ -77,7 +79,7 @@ Tables: `identities`, `terminals`, `installed_apps`, `peers`, `backups`, `tours` ### Background Tasks Started at app lifespan startup, stopped at shutdown: - `InstallationWorker` — async task queue for app install/uninstall -- `PeriodicTask(control_apps, 30s)` — auto-start/stop app containers +- `PeriodicTask(control_apps, 30s)` — app idle lifecycle. With `apps.lifecycle.pause_enabled` (default off): RUNNING pauses after `idle_for_pause` (cgroup freeze + page-out to swap), PAUSED stops after `idle_for_stop`, and high memory PSI demotes the LRU app one tier per cycle. Flag off: legacy stop-only - `PeriodicTask(update_disk_space, 30s)` — disk monitoring - `CronTask(start_backup, "0 3 * * *")` — daily backup with random delay - `CronTask(docker_prune_images, daily)` — image cleanup diff --git a/config.toml b/config.toml index 7afa7aa..58a37a8 100644 --- a/config.toml +++ b/config.toml @@ -34,6 +34,10 @@ password = "bk11Ws.8OGakl-~t.7bKh5l9EYXTbHw.0T" [apps.lifecycle] refresh_interval = 10 +default_idle_for_pause = 60 +default_idle_for_stop = 10800 +psi_threshold = 10.0 +pause_enabled = false [apps] initial_apps = ["filebrowser", "immich", "paperless-ngx"] diff --git a/shard_core/data_model/app_meta.py b/shard_core/data_model/app_meta.py index 1475e90..44cbf5e 100644 --- a/shard_core/data_model/app_meta.py +++ b/shard_core/data_model/app_meta.py @@ -4,13 +4,13 @@ from pathlib import Path as FilePath from typing import Optional, List, Dict, Union -from pydantic import BaseModel, model_validator, field_validator +from pydantic import BaseModel, model_validator from shard_core.data_model import app_meta_migration from shard_core.settings import settings from shard_core.util import signals -CURRENT_VERSION = "1.2" +CURRENT_VERSION = "1.3" class InstallationReason(str, Enum): @@ -32,6 +32,7 @@ class Status(str, Enum): INSTALLING = "installing" STOPPED = "stopped" RUNNING = "running" + PAUSED = "paused" UNINSTALLATION_QUEUED = "uninstallation_queued" UNINSTALLING = "uninstalling" REINSTALLATION_QUEUED = "reinstallation_queued" @@ -89,25 +90,36 @@ class Entrypoint(BaseModel): class Lifecycle(BaseModel): always_on: bool = False - idle_time_for_shutdown: Optional[int] = None - - @field_validator("idle_time_for_shutdown") - @classmethod - def validate_idle_time_for_shutdown(cls, v): - if v and v < 5: - raise ValueError(f"idle_time_for_shutdown must be at least 5, was {v}") - return v + skip_pause: bool = False + idle_for_pause: int | None = None # None = use global default + idle_for_stop: int | None = None # None = use global default @model_validator(mode="after") - def validate_exclusivity(self): - if self.always_on and self.idle_time_for_shutdown is not None: + def validate_combinations(self): + if self.always_on and ( + self.skip_pause + or self.idle_for_pause is not None + or self.idle_for_stop is not None + ): + raise ValueError( + "if always_on is true, no other lifecycle field may be set" + ) + if self.skip_pause and self.idle_for_pause is not None: + raise ValueError("skip_pause and idle_for_pause are mutually exclusive") + if self.idle_for_pause is not None and self.idle_for_pause < 5: raise ValueError( - "if always_on is true, idle_time_for_shutdown must not be set" + f"idle_for_pause must be at least 5, was {self.idle_for_pause}" ) - if not self.always_on and self.idle_time_for_shutdown is None: + if self.idle_for_stop is not None and self.idle_for_stop < 5: raise ValueError( - "if always_on is false or not set, idle_time_for_shutdown must be set" + f"idle_for_stop must be at least 5, was {self.idle_for_stop}" ) + if ( + self.idle_for_pause is not None + and self.idle_for_stop is not None + and self.idle_for_pause >= self.idle_for_stop + ): + raise ValueError("idle_for_pause must be less than idle_for_stop") return self @@ -121,7 +133,7 @@ class AppMeta(BaseModel): icon: str entrypoints: List[Entrypoint] paths: Dict[str, Path] - lifecycle: Lifecycle = Lifecycle(idle_time_for_shutdown=60) + lifecycle: Lifecycle = Lifecycle() minimum_portal_size: VMSize = VMSize.XS store_info: Optional[StoreInfo] = None diff --git a/shard_core/data_model/app_meta_migration.py b/shard_core/data_model/app_meta_migration.py index 5816393..f13fc59 100644 --- a/shard_core/data_model/app_meta_migration.py +++ b/shard_core/data_model/app_meta_migration.py @@ -9,7 +9,23 @@ def migrate_1_1_to_1_2(values): return values +def migrate_1_2_to_1_3(values): + # Lifecycle rework for the PAUSED+PAGED tier: idle_time_for_shutdown + # becomes idle_for_stop; idle_for_pause stays unset and falls back to the + # global default. App-repository files are not regenerated — this runs on + # every read of a v1.2 app_meta.json. + lifecycle = values.get("lifecycle") or {} + if not lifecycle.get("always_on"): + idle_time_for_shutdown = lifecycle.pop("idle_time_for_shutdown", None) + if idle_time_for_shutdown is not None: + lifecycle["idle_for_stop"] = idle_time_for_shutdown + values["lifecycle"] = lifecycle + values["v"] = "1.3" + return values + + migrations = { "1.0": migrate_1_0_to_1_1, "1.1": migrate_1_1_to_1_2, + "1.2": migrate_1_2_to_1_3, } diff --git a/shard_core/data_model/backend/telemetry_model.py b/shard_core/data_model/backend/telemetry_model.py index fa2ef74..c30311e 100644 --- a/shard_core/data_model/backend/telemetry_model.py +++ b/shard_core/data_model/backend/telemetry_model.py @@ -1,11 +1,32 @@ # DO NOT MODIFY - copied from freeshard-controller import datetime +from typing import Dict, List, Optional from pydantic import BaseModel +class PauseTierTelemetry(BaseModel): + """Metrics from the shard_core PAUSED+PAGED app tier (FreeshardBase/freeshard#81). + + Sent by shards with the pause tier enabled; absent (None) from older shards + and from shards with the feature flag off. + """ + + # per-app transition counts, e.g. {"immich": {"running_to_paused": 3, "paused_to_running": 2}} + transitions: Dict[str, Dict[str, int]] = {} + pause_latency_ms_p50: Optional[float] = None + pause_latency_ms_p95: Optional[float] = None + unpause_latency_ms_p50: Optional[float] = None + unpause_latency_ms_p95: Optional[float] = None + # PSI `some avg10` sampled at each control cycle within the reporting interval + psi_some_avg10_snapshots: List[float] = [] + swap_total_kib: Optional[int] = None + swap_free_kib: Optional[int] = None + + class Telemetry(BaseModel): start_time: datetime.datetime end_time: datetime.datetime no_of_requests: int + pause_tier: Optional[PauseTierTelemetry] = None diff --git a/shard_core/service/app_installation/worker.py b/shard_core/service/app_installation/worker.py index e807ffc..3657260 100644 --- a/shard_core/service/app_installation/worker.py +++ b/shard_core/service/app_installation/worker.py @@ -17,6 +17,7 @@ docker_create_app_containers, docker_stop_app, docker_shutdown_app, + docker_unpause_app, ) from shard_core.settings import settings from shard_core.util import signals @@ -123,6 +124,12 @@ async def _install_app_from_existing_zip(app_name: str): async def _uninstall_app(app_name: str): try: + installed_app = await get_app_from_db(app_name) + if installed_app.status == Status.PAUSED: + # unfreeze while the status still says PAUSED — once it flips to + # UNINSTALLING nothing knows the containers are frozen, and a + # frozen stack can be neither stopped nor removed + await docker_unpause_app(app_name) await update_app_status(app_name, Status.UNINSTALLING) except KeyError: log.warning(f"during uninstallation of {app_name}: app not found in database") diff --git a/shard_core/service/app_lifecycle.py b/shard_core/service/app_lifecycle.py index 5991197..79e690e 100644 --- a/shard_core/service/app_lifecycle.py +++ b/shard_core/service/app_lifecycle.py @@ -1,18 +1,21 @@ import asyncio import logging import time -from typing import Dict +from typing import Dict, List from shard_core.database.connection import db_conn from shard_core.database import installed_apps as db_installed_apps from shard_core.data_model.app_meta import InstalledApp, Status -from shard_core.service import disk +from shard_core.service import disk, memory_pressure, pause_metrics from shard_core.service.app_tools import ( docker_start_app, docker_stop_app, + docker_pause_app, + docker_unpause_app, get_app_metadata, size_is_compatible, ) +from shard_core.settings import settings from shard_core.util import signals log = logging.getLogger(__name__) @@ -20,6 +23,9 @@ last_access_dict: Dict[str, float] = dict() background_tasks = set() +# Pressure demotion never touches an app accessed within this window (seconds). +RECENT_ACCESS_GRACE = 5 + @signals.on_request_to_app.connect async def ensure_app_is_running(app: InstalledApp): @@ -29,14 +35,28 @@ async def ensure_app_is_running(app: InstalledApp): if await size_is_compatible(app_meta.minimum_portal_size): global last_access_dict last_access_dict[app.name] = time.time() - task = asyncio.create_task( - docker_start_app(app.name), name=f"ensure {app.name} is running" - ) + # PAUSED wakes via unpause (ms to ~2s), everything else via the legacy + # start path. Calling docker_unpause_app directly also dodges the + # global 5s throttle on docker_start_app, which would silently drop + # the wake. This dispatch stays active even with pause_enabled=false, + # so already-paused apps still wake after a backout. + if app.status == Status.PAUSED: + coro = docker_unpause_app(app.name) + else: + coro = docker_start_app(app.name) + task = asyncio.create_task(coro, name=f"ensure {app.name} is running") background_tasks.add(task) task.add_done_callback(background_tasks.discard) async def control_apps(): + lifecycle_settings = settings().apps.lifecycle + pause_enabled = lifecycle_settings.pause_enabled + psi = memory_pressure.read_memory_pressure() if pause_enabled else 0.0 + pressure_high = pause_enabled and psi > lifecycle_settings.psi_threshold + if pause_enabled: + pause_metrics.record_psi_snapshot(psi) + async with db_conn() as conn: all_apps = await db_installed_apps.get_all(conn) installed_apps = [ @@ -44,23 +64,84 @@ async def control_apps(): for a in all_apps if a["status"] not in (Status.INSTALLATION_QUEUED, Status.INSTALLING) ] - tasks = [_control_app(app.name) for app in installed_apps] + tasks = [_control_app_time(app, pause_enabled) for app in installed_apps] await asyncio.gather(*tasks) + if pressure_high: + log.info(f"memory pressure high (PSI some avg10 = {psi}), demoting LRU app") + await _demote_lru(installed_apps) -async def _control_app(name: str): - global last_access_dict - app_meta = get_app_metadata(name) +async def _control_app_time(app: InstalledApp, pause_enabled: bool): + app_meta = get_app_metadata(app.name) + # Low disk stops every app, always_on included — a paused or running app + # can otherwise fill the disk completely. if disk.current_disk_usage.disk_space_low: - await docker_stop_app(name) + await docker_stop_app(app.name) return if app_meta.lifecycle.always_on: - if await size_is_compatible(app_meta.minimum_portal_size): - await docker_start_app(app_meta.name) - else: - last_access = last_access_dict.get(app_meta.name, 0.0) - idle_time_for_shutdown = app_meta.lifecycle.idle_time_for_shutdown - if last_access < time.time() - idle_time_for_shutdown: - await docker_stop_app(app_meta.name) + if app.status != Status.RUNNING and await size_is_compatible( + app_meta.minimum_portal_size + ): + await docker_start_app(app.name) + return + + idle = time.time() - last_access_dict.get(app.name, 0.0) + t2 = ( + app_meta.lifecycle.idle_for_stop + or settings().apps.lifecycle.default_idle_for_stop + ) + + # Feature flag off, or app opts out of the pause tier: legacy stop-only. + if not pause_enabled or app_meta.lifecycle.skip_pause: + if app.status == Status.RUNNING and idle >= t2: + await docker_stop_app(app.name) + return + + t1 = ( + app_meta.lifecycle.idle_for_pause + or settings().apps.lifecycle.default_idle_for_pause + ) + if app.status == Status.RUNNING and idle >= t1: + await docker_pause_app(app.name) + elif app.status == Status.PAUSED and idle >= t2: + await docker_stop_app(app.name) + + +async def _demote_lru(apps: List[InstalledApp]): + """Demote the least-recently-used demotable app one tier. + + RUNNING apps are considered first: pausing has less user impact than + stopping, and paused apps mostly have older last-access times — a single + LRU sort across both tiers would systematically stop paused apps instead. + Only when nothing is left to pause does the LRU paused app get stopped. + + One demotion per control cycle: the next cycle re-reads PSI and demotes + again only if pressure is still high, so a spike frees memory gradually + instead of stopping everything at once. + """ + running = [] + paused = [] + for app in apps: + app_meta = get_app_metadata(app.name) + if app_meta.lifecycle.always_on: + continue + if time.time() - last_access_dict.get(app.name, 0.0) <= RECENT_ACCESS_GRACE: + continue + if app.status == Status.RUNNING: + running.append(app) + elif app.status == Status.PAUSED: + paused.append(app) + + def lru(candidates: List[InstalledApp]) -> InstalledApp: + return min(candidates, key=lambda app: last_access_dict.get(app.name, 0.0)) + + if running: + victim = lru(running) + if get_app_metadata(victim.name).lifecycle.skip_pause: + await docker_stop_app(victim.name) + else: + await docker_pause_app(victim.name) + elif paused: + await docker_stop_app(lru(paused).name) diff --git a/shard_core/service/app_tools.py b/shard_core/service/app_tools.py index 92b39bf..bdd414e 100644 --- a/shard_core/service/app_tools.py +++ b/shard_core/service/app_tools.py @@ -1,11 +1,13 @@ import asyncio import json import logging +import time from pathlib import Path import shard_core.data_model.profile from shard_core.database.connection import db_conn from shard_core.database import installed_apps as db_installed_apps +from shard_core.service import memory_pressure, pause_metrics from shard_core.data_model.app_meta import ( Status, AppMeta, @@ -33,6 +35,11 @@ async def docker_start_app(name: str): app = await db_installed_apps.get_by_name(conn, name) app_status = app["status"] if app else None + if app_status == Status.PAUSED: + # a paused app wakes by unfreezing, not by compose up + await docker_unpause_app(name) + return + if app_status in [Status.STOPPED, Status.RUNNING, Status.DOWN]: log.debug(f"starting app {name=}") try: @@ -69,15 +76,65 @@ async def docker_start_app(name: str): log.debug(f"app {name=} has status {app_status}, skipping start") +async def docker_pause_app(name: str): + """Freeze a RUNNING app's containers and page their memory out to swap.""" + async with db_conn() as conn: + app = await db_installed_apps.get_by_name(conn, name) + app_status = app["status"] if app else None + if app_status == Status.RUNNING: + log.debug(f"pausing app {name=}") + pause_started = time.monotonic() + await subprocess( + *compose_command(), "pause", cwd=get_installed_apps_path() / name + ) + await memory_pressure.reclaim_compose_stack(name) + pause_metrics.record_pause_latency((time.monotonic() - pause_started) * 1000) + pause_metrics.record_app_transition(name, Status.RUNNING, Status.PAUSED) + async with db_conn() as conn: + await db_installed_apps.update_status(conn, name, Status.PAUSED) + await signals.on_apps_update.send_async() + else: + log.debug(f"app {name=} has {app_status=}, skipping pause") + + +async def docker_unpause_app(name: str): + async with db_conn() as conn: + app = await db_installed_apps.get_by_name(conn, name) + app_status = app["status"] if app else None + if app_status == Status.PAUSED: + log.debug(f"unpausing app {name=}") + unpause_started = time.monotonic() + await subprocess( + *compose_command(), "unpause", cwd=get_installed_apps_path() / name + ) + pause_metrics.record_unpause_latency( + (time.monotonic() - unpause_started) * 1000 + ) + pause_metrics.record_app_transition(name, Status.PAUSED, Status.RUNNING) + async with db_conn() as conn: + await db_installed_apps.update_status(conn, name, Status.RUNNING) + await signals.on_apps_update.send_async() + else: + log.debug(f"app {name=} has {app_status=}, skipping unpause") + + async def docker_stop_app(name: str, set_status: bool = True): async with db_conn() as conn: app = await db_installed_apps.get_by_name(conn, name) app_status = app["status"] if app else None - if app_status in [Status.RUNNING, Status.UNINSTALLING]: + if app_status in [Status.RUNNING, Status.PAUSED, Status.UNINSTALLING]: + if app_status == Status.PAUSED: + # a frozen container cannot be stopped — unfreeze first + await subprocess( + *compose_command(), "unpause", cwd=get_installed_apps_path() / name + ) await subprocess( *compose_command(), "stop", cwd=get_installed_apps_path() / name ) if set_status: + pause_metrics.record_app_transition( + name, Status(app_status), Status.STOPPED + ) async with db_conn() as conn: await db_installed_apps.update_status(conn, name, Status.STOPPED) await signals.on_apps_update.send_async() @@ -90,6 +147,12 @@ async def docker_shutdown_app(name: str, set_status: bool = True, force: bool = app = await db_installed_apps.get_by_name(conn, name) app_status = app["status"] if app else None if force or app_status in [Status.STOPPED, Status.UNINSTALLING]: + if app_status == Status.PAUSED: + # only reachable with force=True (process shutdown) — unfreeze so + # compose down can stop and remove the containers + await subprocess( + *compose_command(), "unpause", cwd=get_installed_apps_path() / name + ) await subprocess( *compose_command(), "down", cwd=get_installed_apps_path() / name ) diff --git a/shard_core/service/memory_pressure.py b/shard_core/service/memory_pressure.py new file mode 100644 index 0000000..20ddf12 --- /dev/null +++ b/shard_core/service/memory_pressure.py @@ -0,0 +1,63 @@ +import asyncio +import logging +import re +from pathlib import Path + +from shard_core.settings import settings +from shard_core.util.subprocess import subprocess, compose_command + +log = logging.getLogger(__name__) + +# PSI is bind-mounted by the core-version compose file: +# /proc/pressure:/host/pressure:ro (freeshard-controller#246) +PSI_PATH = Path("/host/pressure/memory") +# The host cgroup v2 hierarchy, bind-mounted read-write for memory.reclaim writes. +CGROUP_ROOT = Path("/sys/fs/cgroup") + +_PSI_SOME_AVG10_RE = re.compile(r"some.*?avg10=([\d.]+)") + + +def read_memory_pressure() -> float: + """Return the `some avg10` value from PSI, 0.0 if unavailable.""" + try: + text = PSI_PATH.read_text() + except OSError: + return 0.0 + match = _PSI_SOME_AVG10_RE.search(text) + return float(match.group(1)) if match else 0.0 + + +async def reclaim_compose_stack(app_name: str): + """Write each container's current RSS to its cgroup memory.reclaim, + proactively paging the frozen processes' anonymous pages out to swap.""" + app_path = Path(settings().path_root) / "core" / "installed_apps" / app_name + stdout = await subprocess(*compose_command(), "ps", "-q", cwd=app_path) + container_ids = [line.strip() for line in stdout.splitlines() if line.strip()] + for container_id in container_ids: + # The memory.reclaim write blocks while the kernel pages out — run it + # off the event loop. + await asyncio.to_thread(_reclaim_container, container_id) + + +def _reclaim_container(container_id: str): + cgroup = _find_cgroup(container_id) + if cgroup is None: + log.warning(f"no cgroup found for container {container_id}, skipping reclaim") + return + current = int((cgroup / "memory.current").read_text().strip()) + if current <= 0: + return + try: + (cgroup / "memory.reclaim").write_text(f"{current}\n") + except OSError as e: + # Partial reclaim surfaces as EAGAIN once nothing more can be paged + # out — the pages that could move are in swap already. + log.warning(f"memory.reclaim incomplete for container {container_id}: {e}") + + +def _find_cgroup(container_id: str) -> Path | None: + candidates = [ + CGROUP_ROOT / "system.slice" / f"docker-{container_id}.scope", # systemd driver + CGROUP_ROOT / "docker" / container_id, # cgroupfs driver + ] + return next((path for path in candidates if path.is_dir()), None) diff --git a/shard_core/service/pause_metrics.py b/shard_core/service/pause_metrics.py new file mode 100644 index 0000000..cf3dd06 --- /dev/null +++ b/shard_core/service/pause_metrics.py @@ -0,0 +1,58 @@ +"""In-memory accumulators for PAUSED+PAGED tier metrics. + +Kept separate from service.telemetry (which packages and sends them) so that +app_tools and app_lifecycle can record without importing telemetry's +controller-client dependency chain — importing telemetry from app_tools +trips the pre-existing signed_call/portal_controller import cycle. + +Reset after each successful telemetry send. Lost on restart, like every +other telemetry counter. +""" + +from typing import Dict, List + +from shard_core.settings import settings + +app_transitions: Dict[str, Dict[str, int]] = {} +pause_latencies_ms: List[float] = [] +unpause_latencies_ms: List[float] = [] +psi_snapshots: List[float] = [] + +_MAX_LATENCY_SAMPLES = 1000 +_MAX_PSI_SNAPSHOTS = 120 + + +def record_app_transition(app_name: str, from_status, to_status): + if not settings().telemetry.enabled: + return + key = f"{from_status.value}_to_{to_status.value}" + counters = app_transitions.setdefault(app_name, {}) + counters[key] = counters.get(key, 0) + 1 + + +def record_pause_latency(milliseconds: float): + if not settings().telemetry.enabled: + return + if len(pause_latencies_ms) < _MAX_LATENCY_SAMPLES: + pause_latencies_ms.append(milliseconds) + + +def record_unpause_latency(milliseconds: float): + if not settings().telemetry.enabled: + return + if len(unpause_latencies_ms) < _MAX_LATENCY_SAMPLES: + unpause_latencies_ms.append(milliseconds) + + +def record_psi_snapshot(psi_some_avg10: float): + if not settings().telemetry.enabled: + return + if len(psi_snapshots) < _MAX_PSI_SNAPSHOTS: + psi_snapshots.append(psi_some_avg10) + + +def reset(): + app_transitions.clear() + pause_latencies_ms.clear() + unpause_latencies_ms.clear() + psi_snapshots.clear() diff --git a/shard_core/service/telemetry.py b/shard_core/service/telemetry.py index 07f0591..10b2026 100644 --- a/shard_core/service/telemetry.py +++ b/shard_core/service/telemetry.py @@ -1,7 +1,11 @@ import datetime import logging +import re +from pathlib import Path +from typing import List, Optional, Tuple -from shard_core.data_model.backend.telemetry_model import Telemetry +from shard_core.data_model.backend.telemetry_model import Telemetry, PauseTierTelemetry +from shard_core.service import pause_metrics from shard_core.service.freeshard_controller import call_freeshard_controller from shard_core.settings import settings from shard_core.util.signals import on_terminal_auth, on_request_to_app @@ -12,6 +16,8 @@ no_of_requests: int = 0 last_send: datetime.datetime = datetime.datetime.now(datetime.timezone.utc) +MEMINFO_PATH = Path("/proc/meminfo") + @on_terminal_auth.connect @on_request_to_app.connect @@ -23,6 +29,60 @@ async def record_request(_): no_of_requests += 1 +def _percentile(samples: List[float], fraction: float) -> float: + if not 0.0 <= fraction <= 1.0: + raise ValueError(f"fraction must be between 0 and 1, was {fraction}") + ordered = sorted(samples) + return ordered[round(fraction * (len(ordered) - 1))] + + +def _read_swap_kib() -> Tuple[Optional[int], Optional[int]]: + try: + text = MEMINFO_PATH.read_text() + except OSError: + return None, None + total = re.search(r"^SwapTotal:\s+(\d+) kB", text, re.MULTILINE) + free = re.search(r"^SwapFree:\s+(\d+) kB", text, re.MULTILINE) + return ( + int(total.group(1)) if total else None, + int(free.group(1)) if free else None, + ) + + +def _build_pause_tier() -> Optional[PauseTierTelemetry]: + m = pause_metrics + if not ( + m.app_transitions + or m.pause_latencies_ms + or m.unpause_latencies_ms + or m.psi_snapshots + ): + return None + swap_total_kib, swap_free_kib = _read_swap_kib() + return PauseTierTelemetry( + transitions={ + app: dict(counters) for app, counters in m.app_transitions.items() + }, + pause_latency_ms_p50=( + _percentile(m.pause_latencies_ms, 0.5) if m.pause_latencies_ms else None + ), + pause_latency_ms_p95=( + _percentile(m.pause_latencies_ms, 0.95) if m.pause_latencies_ms else None + ), + unpause_latency_ms_p50=( + _percentile(m.unpause_latencies_ms, 0.5) if m.unpause_latencies_ms else None + ), + unpause_latency_ms_p95=( + _percentile(m.unpause_latencies_ms, 0.95) + if m.unpause_latencies_ms + else None + ), + psi_some_avg10_snapshots=list(m.psi_snapshots), + swap_total_kib=swap_total_kib, + swap_free_kib=swap_free_kib, + ) + + async def send_telemetry(): if not settings().telemetry.enabled: return @@ -32,7 +92,10 @@ async def send_telemetry(): now = datetime.datetime.now(datetime.timezone.utc) telemetry = Telemetry( - start_time=last_send, end_time=now, no_of_requests=no_of_requests + start_time=last_send, + end_time=now, + no_of_requests=no_of_requests, + pause_tier=_build_pause_tier(), ) try: await call_freeshard_controller( @@ -51,3 +114,4 @@ async def send_telemetry(): last_send = now no_of_requests = 0 + pause_metrics.reset() diff --git a/shard_core/settings.py b/shard_core/settings.py index 2b0acc5..e831ff2 100644 --- a/shard_core/settings.py +++ b/shard_core/settings.py @@ -45,6 +45,12 @@ class RegistrySettings(BaseModel): class AppLifecycleSettings(BaseModel): refresh_interval: int = 30 + default_idle_for_pause: int = 60 # T1: RUNNING -> PAUSED, seconds + default_idle_for_stop: int = ( + 10800 # T2: PAUSED -> STOPPED, seconds (3h — pressure demotion stops earlier when RAM is needed) + ) + psi_threshold: float = 10.0 # /proc/pressure/memory `some avg10` demotion trigger + pause_enabled: bool = False # rollout kill-switch for the PAUSED+PAGED tier class AppLastAccessSettings(BaseModel): diff --git a/tests/config.toml b/tests/config.toml index 1df24ea..d479402 100644 --- a/tests/config.toml +++ b/tests/config.toml @@ -1,5 +1,7 @@ [apps.lifecycle] refresh_interval = 2 +default_idle_for_pause = 5 +default_idle_for_stop = 12 [apps.last_access] max_update_frequency = 3 diff --git a/tests/mock_app_store/pause_cycle/app_meta.json b/tests/mock_app_store/pause_cycle/app_meta.json new file mode 100644 index 0000000..7408e3c --- /dev/null +++ b/tests/mock_app_store/pause_cycle/app_meta.json @@ -0,0 +1,26 @@ +{ + "v": "1.3", + "app_version": "0.1.0", + "name": "pause_cycle", + "pretty_name": "Pause Cycle", + "icon": "icon.svg", + "entrypoints": [ + { + "container_name": "pause_cycle", + "container_port": 80, + "entrypoint_port": "http" + } + ], + "paths": { + "": { + "access": "private" + }, + "/pub": { + "access": "public" + } + }, + "lifecycle": {}, + "store_info": { + "description_short": "an app that pauses, then stops, on the global default tiers" + } +} diff --git a/tests/mock_app_store/pause_cycle/docker-compose.yml.template b/tests/mock_app_store/pause_cycle/docker-compose.yml.template new file mode 100644 index 0000000..5d432b0 --- /dev/null +++ b/tests/mock_app_store/pause_cycle/docker-compose.yml.template @@ -0,0 +1,15 @@ +version: '3.5' + +networks: + portal: + external: true + +services: + pause_cycle: + restart: always + image: nginx:alpine + container_name: pause_cycle + volumes: + - "{{ fs.app_data }}:/data" + networks: + - portal diff --git a/tests/mock_app_store/pause_cycle/icon.svg b/tests/mock_app_store/pause_cycle/icon.svg new file mode 100644 index 0000000..5e78ecc --- /dev/null +++ b/tests/mock_app_store/pause_cycle/icon.svg @@ -0,0 +1,147 @@ + +image/svg+xml + + + + + \ No newline at end of file diff --git a/tests/mock_app_store/pause_cycle/pause_cycle.zip b/tests/mock_app_store/pause_cycle/pause_cycle.zip new file mode 100644 index 0000000..5b77c86 Binary files /dev/null and b/tests/mock_app_store/pause_cycle/pause_cycle.zip differ diff --git a/tests/test_app_lifecycle_pause.py b/tests/test_app_lifecycle_pause.py new file mode 100644 index 0000000..23692fa --- /dev/null +++ b/tests/test_app_lifecycle_pause.py @@ -0,0 +1,279 @@ +"""Unit tests for the two-tier idle logic and PSI-driven LRU demotion. + +The docker primitives and metadata lookups are patched — these tests pin the +decision logic of _control_app_time and _demote_lru, not Docker behavior +(that lives in the integration tests). +""" + +import time +from unittest.mock import AsyncMock, patch + +import pytest + +from shard_core.data_model.app_meta import ( + AppMeta, + InstalledApp, + Lifecycle, + Status, +) +from shard_core.service import app_lifecycle +from tests.conftest import settings_override + + +def _meta(lifecycle: Lifecycle) -> AppMeta: + return AppMeta( + v="1.3", + app_version="1.0.0", + name="test", + pretty_name="Test", + icon="icon.svg", + entrypoints=[], + paths={}, + lifecycle=lifecycle, + ) + + +@pytest.fixture +def docker_mocks(): + with ( + patch.object(app_lifecycle, "docker_start_app", new=AsyncMock()) as start, + patch.object(app_lifecycle, "docker_stop_app", new=AsyncMock()) as stop, + patch.object(app_lifecycle, "docker_pause_app", new=AsyncMock()) as pause, + patch.object(app_lifecycle, "docker_unpause_app", new=AsyncMock()) as unpause, + patch.object( + app_lifecycle, "size_is_compatible", new=AsyncMock(return_value=True) + ), + ): + yield {"start": start, "stop": stop, "pause": pause, "unpause": unpause} + + +@pytest.fixture(autouse=True) +def clean_last_access(): + app_lifecycle.last_access_dict.clear() + yield + app_lifecycle.last_access_dict.clear() + + +def _app(name: str, status: Status, idle: float) -> InstalledApp: + app_lifecycle.last_access_dict[name] = time.time() - idle + return InstalledApp(name=name, status=status) + + +PAUSE_ON = {"apps": {"lifecycle": {"pause_enabled": True}}} +# tests/config.toml: default_idle_for_pause=5, default_idle_for_stop=12 + + +async def test_running_app_pauses_after_t1(docker_mocks): + app = _app("a", Status.RUNNING, idle=7) + with ( + settings_override(PAUSE_ON), + patch.object( + app_lifecycle, "get_app_metadata", return_value=_meta(Lifecycle()) + ), + ): + await app_lifecycle._control_app_time(app, pause_enabled=True) + docker_mocks["pause"].assert_awaited_once_with("a") + docker_mocks["stop"].assert_not_awaited() + + +async def test_running_app_below_t1_is_left_alone(docker_mocks): + app = _app("a", Status.RUNNING, idle=2) + with ( + settings_override(PAUSE_ON), + patch.object( + app_lifecycle, "get_app_metadata", return_value=_meta(Lifecycle()) + ), + ): + await app_lifecycle._control_app_time(app, pause_enabled=True) + docker_mocks["pause"].assert_not_awaited() + docker_mocks["stop"].assert_not_awaited() + + +async def test_paused_app_stops_after_t2(docker_mocks): + app = _app("a", Status.PAUSED, idle=15) + with ( + settings_override(PAUSE_ON), + patch.object( + app_lifecycle, "get_app_metadata", return_value=_meta(Lifecycle()) + ), + ): + await app_lifecycle._control_app_time(app, pause_enabled=True) + docker_mocks["stop"].assert_awaited_once_with("a") + docker_mocks["pause"].assert_not_awaited() + + +async def test_paused_app_below_t2_stays_paused(docker_mocks): + app = _app("a", Status.PAUSED, idle=7) + with ( + settings_override(PAUSE_ON), + patch.object( + app_lifecycle, "get_app_metadata", return_value=_meta(Lifecycle()) + ), + ): + await app_lifecycle._control_app_time(app, pause_enabled=True) + docker_mocks["stop"].assert_not_awaited() + docker_mocks["pause"].assert_not_awaited() + + +async def test_flag_off_keeps_legacy_stop_only(docker_mocks): + app = _app("a", Status.RUNNING, idle=15) + with patch.object( + app_lifecycle, "get_app_metadata", return_value=_meta(Lifecycle()) + ): + await app_lifecycle._control_app_time(app, pause_enabled=False) + docker_mocks["stop"].assert_awaited_once_with("a") + docker_mocks["pause"].assert_not_awaited() + + +async def test_skip_pause_app_never_pauses(docker_mocks): + app = _app("a", Status.RUNNING, idle=15) + with ( + settings_override(PAUSE_ON), + patch.object( + app_lifecycle, + "get_app_metadata", + return_value=_meta(Lifecycle(skip_pause=True)), + ), + ): + await app_lifecycle._control_app_time(app, pause_enabled=True) + docker_mocks["stop"].assert_awaited_once_with("a") + docker_mocks["pause"].assert_not_awaited() + + +async def test_per_app_idle_overrides_win(docker_mocks): + # per-app t1=60 keeps the app running where the global default (5) would pause + app = _app("a", Status.RUNNING, idle=10) + with ( + settings_override(PAUSE_ON), + patch.object( + app_lifecycle, + "get_app_metadata", + return_value=_meta(Lifecycle(idle_for_pause=60, idle_for_stop=120)), + ), + ): + await app_lifecycle._control_app_time(app, pause_enabled=True) + docker_mocks["pause"].assert_not_awaited() + docker_mocks["stop"].assert_not_awaited() + + +async def test_always_on_app_is_started_not_paused(docker_mocks): + app = _app("a", Status.STOPPED, idle=9999) + with ( + settings_override(PAUSE_ON), + patch.object( + app_lifecycle, + "get_app_metadata", + return_value=_meta(Lifecycle(always_on=True)), + ), + ): + await app_lifecycle._control_app_time(app, pause_enabled=True) + docker_mocks["start"].assert_awaited_once_with("a") + docker_mocks["pause"].assert_not_awaited() + docker_mocks["stop"].assert_not_awaited() + + +async def test_low_disk_stops_even_always_on_apps(docker_mocks): + app = _app("a", Status.RUNNING, idle=0) + with ( + settings_override(PAUSE_ON), + patch.object( + app_lifecycle, + "get_app_metadata", + return_value=_meta(Lifecycle(always_on=True)), + ), + patch.object( + app_lifecycle.disk, + "current_disk_usage", + app_lifecycle.disk.DiskUsage(total_gb=10, free_gb=0.1, disk_space_low=True), + ), + ): + await app_lifecycle._control_app_time(app, pause_enabled=True) + docker_mocks["stop"].assert_awaited_once_with("a") + docker_mocks["start"].assert_not_awaited() + docker_mocks["pause"].assert_not_awaited() + + +async def test_demote_lru_pauses_least_recently_used_running_app(docker_mocks): + apps = [ + _app("newer", Status.RUNNING, idle=100), + _app("older", Status.RUNNING, idle=200), + ] + with patch.object( + app_lifecycle, "get_app_metadata", return_value=_meta(Lifecycle()) + ): + await app_lifecycle._demote_lru(apps) + docker_mocks["pause"].assert_awaited_once_with("older") + docker_mocks["stop"].assert_not_awaited() + + +async def test_demote_lru_prefers_pausing_running_over_stopping_paused(docker_mocks): + # the paused app is older, but pausing a running app has less user impact + # than stopping a paused one — running apps are demoted first + apps = [ + _app("running", Status.RUNNING, idle=100), + _app("paused", Status.PAUSED, idle=200), + ] + with patch.object( + app_lifecycle, "get_app_metadata", return_value=_meta(Lifecycle()) + ): + await app_lifecycle._demote_lru(apps) + docker_mocks["pause"].assert_awaited_once_with("running") + docker_mocks["stop"].assert_not_awaited() + + +async def test_demote_lru_stops_lru_paused_when_nothing_left_to_pause(docker_mocks): + apps = [ + _app("paused_newer", Status.PAUSED, idle=100), + _app("paused_older", Status.PAUSED, idle=200), + ] + with patch.object( + app_lifecycle, "get_app_metadata", return_value=_meta(Lifecycle()) + ): + await app_lifecycle._demote_lru(apps) + docker_mocks["stop"].assert_awaited_once_with("paused_older") + docker_mocks["pause"].assert_not_awaited() + + +async def test_demote_lru_stops_running_skip_pause_victim(docker_mocks): + apps = [_app("a", Status.RUNNING, idle=200)] + with patch.object( + app_lifecycle, + "get_app_metadata", + return_value=_meta(Lifecycle(skip_pause=True)), + ): + await app_lifecycle._demote_lru(apps) + docker_mocks["stop"].assert_awaited_once_with("a") + docker_mocks["pause"].assert_not_awaited() + + +async def test_demote_lru_excludes_always_on_recent_and_stopped(docker_mocks): + metas = { + "always_on": _meta(Lifecycle(always_on=True)), + "recent": _meta(Lifecycle()), + "stopped": _meta(Lifecycle()), + } + apps = [ + _app("always_on", Status.RUNNING, idle=500), + _app("recent", Status.RUNNING, idle=1), # within RECENT_ACCESS_GRACE + _app("stopped", Status.STOPPED, idle=500), + ] + with patch.object( + app_lifecycle, "get_app_metadata", side_effect=lambda name: metas[name] + ): + await app_lifecycle._demote_lru(apps) + docker_mocks["pause"].assert_not_awaited() + docker_mocks["stop"].assert_not_awaited() + + +async def test_demote_lru_demotes_exactly_one_app_per_cycle(docker_mocks): + apps = [ + _app("a", Status.RUNNING, idle=300), + _app("b", Status.RUNNING, idle=200), + _app("c", Status.RUNNING, idle=100), + ] + with patch.object( + app_lifecycle, "get_app_metadata", return_value=_meta(Lifecycle()) + ): + await app_lifecycle._demote_lru(apps) + assert docker_mocks["pause"].await_count == 1 + docker_mocks["pause"].assert_awaited_once_with("a") diff --git a/tests/test_app_meta_migration.py b/tests/test_app_meta_migration.py index 7a2e467..69dc2e7 100644 --- a/tests/test_app_meta_migration.py +++ b/tests/test_app_meta_migration.py @@ -1,24 +1,93 @@ -from shard_core.data_model.app_meta import AppMeta, Lifecycle, VMSize - - -def test_migrate_1_0_to_1_2(): - app_meta_in = AppMeta( - v="1.0", - app_version="1.0.0", - name="test", - pretty_name="not set", - icon="test", - entrypoints=[], - paths={}, - lifecycle=Lifecycle(always_on=False, idle_time_for_shutdown=60), - minimum_portal_size=VMSize.XS, - store_info=None, +import json +import zipfile +from pathlib import Path + +from shard_core.data_model.app_meta import AppMeta + +MOCK_APP_STORE = Path(__file__).parent / "mock_app_store" + + +def _base_app_meta_json(v: str, lifecycle: dict | None) -> dict: + values = { + "v": v, + "app_version": "1.0.0", + "name": "test", + "icon": "test", + "entrypoints": [], + "paths": {}, + "minimum_portal_size": "xs", + "store_info": None, + } + if lifecycle is not None: + values["lifecycle"] = lifecycle + return values + + +def test_migrate_1_0_to_current(): + app_meta_in_json = _base_app_meta_json( + "1.0", {"always_on": False, "idle_time_for_shutdown": 60} ) - app_meta_in_json = app_meta_in.model_dump(exclude={"pretty_name"}) - app_meta_in_json["v"] = "1.0" app_meta_out = AppMeta.model_validate(app_meta_in_json) - assert app_meta_out.v == "1.2" + assert app_meta_out.v == "1.3" # done by migration to 1.1 assert app_meta_out.pretty_name == "Test" + # done by migration to 1.3 + assert app_meta_out.lifecycle.idle_for_stop == 60 + assert app_meta_out.lifecycle.idle_for_pause is None + assert app_meta_out.lifecycle.skip_pause is False + + +def test_migrate_1_2_to_1_3_idle_time_becomes_idle_for_stop(): + values = _base_app_meta_json( + "1.2", {"always_on": False, "idle_time_for_shutdown": 3600} + ) + values["pretty_name"] = "Test" + + app_meta = AppMeta.model_validate(values) + assert app_meta.v == "1.3" + assert app_meta.lifecycle.idle_for_stop == 3600 + assert app_meta.lifecycle.idle_for_pause is None + + +def test_migrate_1_2_to_1_3_always_on_stays_always_on(): + values = _base_app_meta_json("1.2", {"always_on": True}) + values["pretty_name"] = "Test" + + app_meta = AppMeta.model_validate(values) + assert app_meta.v == "1.3" + assert app_meta.lifecycle.always_on is True + assert app_meta.lifecycle.idle_for_stop is None + + +def test_migrate_1_2_to_1_3_missing_lifecycle_gets_defaults(): + values = _base_app_meta_json("1.2", None) + values["pretty_name"] = "Test" + + app_meta = AppMeta.model_validate(values) + assert app_meta.v == "1.3" + assert app_meta.lifecycle.always_on is False + assert app_meta.lifecycle.skip_pause is False + assert app_meta.lifecycle.idle_for_pause is None + assert app_meta.lifecycle.idle_for_stop is None + + +def test_all_mock_app_store_metas_migrate(): + """Every app_meta.json fixture (the app-repository stand-ins) must load + through the migration chain without regeneration.""" + zips = sorted(MOCK_APP_STORE.glob("*/*.zip")) + assert zips, "expected mock app store zips" + for zip_path in zips: + with zipfile.ZipFile(zip_path) as zf: + values = json.loads(zf.read("app_meta.json")) + old_lifecycle = values.get("lifecycle", {}) + app_meta = AppMeta.model_validate(values) + assert app_meta.v == "1.3", zip_path + if old_lifecycle.get("always_on"): + assert app_meta.lifecycle.always_on is True, zip_path + elif "idle_time_for_shutdown" in old_lifecycle: + assert ( + app_meta.lifecycle.idle_for_stop + == old_lifecycle["idle_time_for_shutdown"] + ), zip_path diff --git a/tests/test_app_pause_lifecycle.py b/tests/test_app_pause_lifecycle.py new file mode 100644 index 0000000..23c6252 --- /dev/null +++ b/tests/test_app_pause_lifecycle.py @@ -0,0 +1,129 @@ +"""Integration tests for the PAUSED+PAGED tier against real containers. + +The pause_cycle mock app carries an empty lifecycle, so both tiers come from +the test config's global defaults (T1=5s pause, T2=12s stop) with the 2s +control-loop refresh. + +Not covered here: the memory.reclaim RSS-drop assertion needs root access to +the host cgroup filesystem, which CI and dev runs don't have — the write +logic is unit-tested against a fake cgroup tree in test_memory_pressure.py, +and the actual paging effect is verified in the manual pre-rollout benchmark +(see the design spec's testing strategy). +""" + +import time + +import docker +from fastapi import status + +from shard_core.data_model.app_meta import InstalledApp, Status +from shard_core.service import pause_metrics +from tests.conftest import settings_override +from tests.util import install_app, retry_async, wait_until_app_uninstalled + +PAUSE_ON = {"apps": {"lifecycle": {"pause_enabled": True}}} +APP_NAME = "pause_cycle" + + +def _assert_state(docker_client, api_client): + async def _get_status(): + response = await api_client.get(f"protected/apps/{APP_NAME}") + response.raise_for_status() + return InstalledApp.model_validate(response.json()).status + + def container_status(): + return docker_client.containers.get(APP_NAME).status + + return _get_status, container_status + + +async def _wake_via_request(api_client): + response = await api_client.get( + "internal/auth", + headers={ + "X-Forwarded-Host": f"{APP_NAME}.myshard.org", + "X-Forwarded-Uri": "/pub", + }, + ) + response.raise_for_status() + + +async def test_full_tier_cycle_and_fast_wake(requests_mock, api_client): + docker_client = docker.from_env() + get_status, container_status = _assert_state(docker_client, api_client) + + with settings_override(PAUSE_ON): + await install_app(api_client, APP_NAME) + pause_metrics.reset() + + await _wake_via_request(api_client) + + async def assert_running(): + assert container_status() == "running" + assert await get_status() == Status.RUNNING + + await retry_async(assert_running, timeout=15, retry_errors=[AssertionError]) + + # T1: idle >= 5s pauses the containers, status follows + async def assert_paused(): + assert container_status() == "paused" + assert await get_status() == Status.PAUSED + + await retry_async(assert_paused, timeout=20, retry_errors=[AssertionError]) + + # wake-on-request from PAUSED + wake_started = time.monotonic() + await _wake_via_request(api_client) + await retry_async(assert_running, timeout=10, retry_errors=[AssertionError]) + wake_wall_time = time.monotonic() - wake_started + # generous ceiling for loaded CI runners; the precise target is pinned + # on the recorded unpause latency below + assert wake_wall_time < 10 + + # the compose unpause itself must be fast (design target: <=2s p95) + assert pause_metrics.unpause_latencies_ms + assert pause_metrics.unpause_latencies_ms[-1] < 2000 + + # T2: after another T1 the app re-pauses, after idle >= 12s it stops + await retry_async(assert_paused, timeout=20, retry_errors=[AssertionError]) + + async def assert_stopped(): + assert container_status() == "exited" + assert await get_status() == Status.STOPPED + + await retry_async(assert_stopped, timeout=25, retry_errors=[AssertionError]) + + # transitions were recorded for telemetry + transitions = pause_metrics.app_transitions[APP_NAME] + assert transitions["running_to_paused"] >= 2 + assert transitions["paused_to_running"] >= 1 + assert transitions["paused_to_stopped"] >= 1 + + +async def test_uninstall_while_paused(requests_mock, api_client): + docker_client = docker.from_env() + get_status, container_status = _assert_state(docker_client, api_client) + + with settings_override(PAUSE_ON): + await install_app(api_client, APP_NAME) + await _wake_via_request(api_client) + + async def assert_paused(): + assert container_status() == "paused" + assert await get_status() == Status.PAUSED + + await retry_async(assert_paused, timeout=25, retry_errors=[AssertionError]) + + # uninstalling a frozen stack must unfreeze before stop/down + response = await api_client.delete(f"protected/apps/{APP_NAME}") + assert response.status_code == status.HTTP_204_NO_CONTENT + await wait_until_app_uninstalled(api_client, APP_NAME) + + async def assert_gone(): + try: + docker_client.containers.get(APP_NAME) + except docker.errors.NotFound: + return + raise AssertionError(f"container {APP_NAME} still exists") + + await retry_async(assert_gone, timeout=15, retry_errors=[AssertionError]) diff --git a/tests/test_memory_pressure.py b/tests/test_memory_pressure.py new file mode 100644 index 0000000..5812504 --- /dev/null +++ b/tests/test_memory_pressure.py @@ -0,0 +1,78 @@ +from unittest.mock import AsyncMock, patch + +import pytest + +from shard_core.service import memory_pressure + +PSI_SAMPLE = """some avg10=12.34 avg60=5.67 avg300=1.23 total=123456 +full avg10=3.21 avg60=1.11 avg300=0.42 total=65432 +""" + + +def test_read_memory_pressure_parses_some_avg10(tmp_path, monkeypatch): + psi_file = tmp_path / "memory" + psi_file.write_text(PSI_SAMPLE) + monkeypatch.setattr(memory_pressure, "PSI_PATH", psi_file) + assert memory_pressure.read_memory_pressure() == 12.34 + + +def test_read_memory_pressure_missing_file_returns_zero(tmp_path, monkeypatch): + monkeypatch.setattr(memory_pressure, "PSI_PATH", tmp_path / "does-not-exist") + assert memory_pressure.read_memory_pressure() == 0.0 + + +def test_read_memory_pressure_malformed_returns_zero(tmp_path, monkeypatch): + psi_file = tmp_path / "memory" + psi_file.write_text("garbage that is not PSI\n") + monkeypatch.setattr(memory_pressure, "PSI_PATH", psi_file) + assert memory_pressure.read_memory_pressure() == 0.0 + + +@pytest.fixture +def fake_cgroup_root(tmp_path, monkeypatch): + monkeypatch.setattr(memory_pressure, "CGROUP_ROOT", tmp_path) + return tmp_path + + +def _make_cgroup(root, relative: str, memory_current: int): + cgroup = root / relative + cgroup.mkdir(parents=True) + (cgroup / "memory.current").write_text(f"{memory_current}\n") + (cgroup / "memory.reclaim").write_text("") + return cgroup + + +def test_reclaim_container_writes_rss_systemd_driver(fake_cgroup_root): + cgroup = _make_cgroup( + fake_cgroup_root, "system.slice/docker-abc123.scope", 104857600 + ) + memory_pressure._reclaim_container("abc123") + assert (cgroup / "memory.reclaim").read_text() == "104857600\n" + + +def test_reclaim_container_writes_rss_cgroupfs_driver(fake_cgroup_root): + cgroup = _make_cgroup(fake_cgroup_root, "docker/def456", 2048) + memory_pressure._reclaim_container("def456") + assert (cgroup / "memory.reclaim").read_text() == "2048\n" + + +def test_reclaim_container_missing_cgroup_is_skipped(fake_cgroup_root): + # must not raise + memory_pressure._reclaim_container("unknown") + + +def test_reclaim_container_zero_usage_writes_nothing(fake_cgroup_root): + cgroup = _make_cgroup(fake_cgroup_root, "docker/ghi789", 0) + memory_pressure._reclaim_container("ghi789") + assert (cgroup / "memory.reclaim").read_text() == "" + + +async def test_reclaim_compose_stack_reclaims_each_container(fake_cgroup_root): + cgroup_a = _make_cgroup(fake_cgroup_root, "docker/aaa", 100) + cgroup_b = _make_cgroup(fake_cgroup_root, "docker/bbb", 200) + with patch.object( + memory_pressure, "subprocess", new=AsyncMock(return_value="aaa\nbbb\n\n") + ): + await memory_pressure.reclaim_compose_stack("someapp") + assert (cgroup_a / "memory.reclaim").read_text() == "100\n" + assert (cgroup_b / "memory.reclaim").read_text() == "200\n" diff --git a/tests/test_model.py b/tests/test_model.py index bccd6f9..7bae692 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -21,14 +21,38 @@ def prop_field_included(self) -> str: def test_lifecycle(): + # valid combinations + Lifecycle() Lifecycle(always_on=True) - Lifecycle(idle_time_for_shutdown=5) + Lifecycle(skip_pause=True) + Lifecycle(idle_for_pause=5) + Lifecycle(idle_for_stop=5) + Lifecycle(idle_for_pause=120, idle_for_stop=7200) + Lifecycle(skip_pause=True, idle_for_stop=1800) + + # always_on excludes everything else + with pytest.raises(ValueError): + Lifecycle(always_on=True, skip_pause=True) + with pytest.raises(ValueError): + Lifecycle(always_on=True, idle_for_pause=10) + with pytest.raises(ValueError): + Lifecycle(always_on=True, idle_for_stop=10) + + # skip_pause excludes idle_for_pause with pytest.raises(ValueError): - Lifecycle(idle_time_for_shutdown=4) + Lifecycle(skip_pause=True, idle_for_pause=10) + + # minimums + with pytest.raises(ValueError): + Lifecycle(idle_for_pause=4) + with pytest.raises(ValueError): + Lifecycle(idle_for_stop=4) + + # ordering when both set with pytest.raises(ValueError): - Lifecycle() + Lifecycle(idle_for_pause=60, idle_for_stop=60) with pytest.raises(ValueError): - Lifecycle(always_on=True, idle_time_for_shutdown=10) + Lifecycle(idle_for_pause=120, idle_for_stop=60) def test_vm_size(): diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index ea0a025..5e761dc 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -49,3 +49,62 @@ async def test_telemetry_disabled( assert telemetry.no_of_requests == 0 assert not mock_call_freeshard_controller.called + + +@patch("shard_core.service.telemetry.call_freeshard_controller", new_callable=AsyncMock) +async def test_telemetry_pause_tier_roundtrip( + mock_call_freeshard_controller: AsyncMock, app_client +): + from shard_core.data_model.app_meta import Status + from shard_core.service import pause_metrics + + pause_metrics.reset() + pause_metrics.record_app_transition("immich", Status.RUNNING, Status.PAUSED) + pause_metrics.record_app_transition("immich", Status.PAUSED, Status.RUNNING) + pause_metrics.record_app_transition("immich", Status.RUNNING, Status.PAUSED) + pause_metrics.record_pause_latency(100.0) + pause_metrics.record_pause_latency(300.0) + pause_metrics.record_unpause_latency(50.0) + pause_metrics.record_psi_snapshot(2.5) + pause_metrics.record_psi_snapshot(11.0) + + await telemetry.send_telemetry() + + body = mock_call_freeshard_controller.await_args_list[0].kwargs["body"] + tel = Telemetry.model_validate(json.loads(body.decode())) + assert tel.pause_tier is not None + assert tel.pause_tier.transitions["immich"]["running_to_paused"] == 2 + assert tel.pause_tier.transitions["immich"]["paused_to_running"] == 1 + assert tel.pause_tier.pause_latency_ms_p50 in (100.0, 300.0) + assert tel.pause_tier.unpause_latency_ms_p50 == 50.0 + assert tel.pause_tier.psi_some_avg10_snapshots == [2.5, 11.0] + + # accumulators reset after a successful send + assert pause_metrics.app_transitions == {} + assert pause_metrics.pause_latencies_ms == [] + assert pause_metrics.psi_snapshots == [] + + +@patch("shard_core.service.telemetry.call_freeshard_controller", new_callable=AsyncMock) +async def test_telemetry_without_pause_activity_omits_pause_tier( + mock_call_freeshard_controller: AsyncMock, app_client +): + from shard_core.service import pause_metrics + + pause_metrics.reset() + await telemetry.record_request("arg") + await telemetry.send_telemetry() + + body = mock_call_freeshard_controller.await_args_list[0].kwargs["body"] + tel = Telemetry.model_validate(json.loads(body.decode())) + assert tel.pause_tier is None + + +def test_percentile_rejects_out_of_range_fraction(): + import pytest + + with pytest.raises(ValueError): + telemetry._percentile([1.0, 2.0], 95) + with pytest.raises(ValueError): + telemetry._percentile([1.0, 2.0], -0.1) + assert telemetry._percentile([1.0, 2.0, 3.0], 0.5) == 2.0