diff --git a/.github/workflows/test-oci-registries.yml b/.github/workflows/test-oci-registries.yml index 5c17f077..9937206b 100644 --- a/.github/workflows/test-oci-registries.yml +++ b/.github/workflows/test-oci-registries.yml @@ -36,6 +36,10 @@ jobs: python-version: '3.13' - name: Install OCI test dependencies run: python3 -m pip install -r oci/requirements.txt + - name: Run watchdog identity and lifecycle regressions + env: + PYTHONDONTWRITEBYTECODE: '1' + run: python3 -m unittest discover -s oci/tests -p test_watchdog.py -v - name: Run OCI image reuse and resource edit regressions env: PYTHONDONTWRITEBYTECODE: '1' diff --git a/oci/README.md b/oci/README.md index cf735188..4e79e632 100644 --- a/oci/README.md +++ b/oci/README.md @@ -373,6 +373,11 @@ The variables the image itself defines are imported with it and are not written Where a path goes follows one rule. A path that holds content of the user, such as `/media`, `/downloads`, `/music` or `/backups`, is offered as a container disk or as a host directory, and the installation asks which. The list is in `catalog/volume-policy.json`. Every other path is state of the application and goes to a container disk without asking. Writing `"installation_choice": ["managed-volume", "host-bind"]` or `"default": "host-bind"` on a volume offers the host directory for a path that is not in that list. **`restart`**. With `unless-stopped`, `always` or `on-failure`, the watchdog is proposed by default: an application that stops on its own is started again. +Its running history belongs to the installation UUID, not just the VMID. Before +restarting a guest, it rechecks the saved record and the live ProxMenux identity +under the registry operation lock. A stale record therefore cannot start an +unrelated guest or a temporary `oci-data-holder` that reused the same VMID. +Updates, removals, intentional stops and guests first seen stopped remain left alone. ### `first_run`: what is shown when the installation ends diff --git a/oci/remote/oci_watchdog.py b/oci/remote/oci_watchdog.py index 50c157c1..1c9b0947 100644 --- a/oci/remote/oci_watchdog.py +++ b/oci/remote/oci_watchdog.py @@ -46,8 +46,9 @@ STOP_TASKS = {'vzstop', 'vzshutdown', 'vzsuspend', 'vzreboot', 'vzdestroy', 'vzm HOST_TASKS = {'stopall', 'migrateall'} -def new_state(): - return {'seen': None, 'since': None, 'attempts': 0, 'retry_at': 0.0, 'settled': False, 'noticed': None} +def new_state(installation_id=None): + return {'installation_id': installation_id, 'seen': None, 'since': None, + 'attempts': 0, 'retry_at': 0.0, 'settled': False, 'noticed': None} def notice_due(state, now): @@ -117,23 +118,37 @@ def decide(state, running, now, busy, requested): def watched(root): - """The installed applications that asked for the watchdog: {vmid: name}.""" + """The installed applications that asked for it: {vmid: (name, identity)}.""" found = {} for path in sorted(root.glob('*/oci-compose.json')): try: - record = json.loads(path.read_text()) - vmid = int(record['vmid']) - except (OSError, ValueError, KeyError, TypeError): + vmid = int(path.parent.name) + record = instances.read(root, vmid) + if record.get('status') != 'installed' or record.get('deployment', {}).get('watchdog') is not True: + continue + if record.get('pending_stack_transaction') or record.get('pending_transaction'): + continue + title = record.get('template', {}).get('catalog_ui', {}).get('title') + name = (title.get('en_US') if isinstance(title, dict) else title) or f'CT {vmid}' + except (OSError, ValueError, KeyError, TypeError, AttributeError): continue - if record.get('status') != 'installed' or record.get('deployment', {}).get('watchdog') is not True: - continue - if record.get('pending_stack_transaction') or record.get('pending_transaction'): - continue - title = record.get('template', {}).get('catalog_ui', {}).get('title') - found[vmid] = (title.get('en_US') if isinstance(title, dict) else title) or f'CT {vmid}' + found[vmid] = (name, record['installation_id']) return found +def same_installation(vmid, installation_id): + """Match the live identity; None means it could not be read this time.""" + if not (CONFIGS / f'{vmid}.conf').is_file(): + return False + try: + config = subprocess.run(['pct', 'config', str(vmid)], capture_output=True, check=False, timeout=30) + if config.returncode == 0: + return instances.identity(config.stdout) == installation_id + except (OSError, ValueError, subprocess.SubprocessError): + pass + return None + + def is_running(vmid): return (CGROUPS / str(vmid)).is_dir() @@ -201,24 +216,50 @@ def look(states, now): apps = watched(instances.ROOT) for vmid in set(states) - set(apps): del states[vmid] - for vmid, name in apps.items(): - state = states.setdefault(vmid, new_state()) - action = decide(state, is_running(vmid), now, lambda: is_busy(vmid, now), - lambda seen: stop_requested(read_tasks(), vmid, seen)) - data = {'app_name': name, 'vmid': vmid, 'containers': f'CT {vmid}'} - if action == 'start': - print(translate('CT {vmid} ({name}) stopped on its own; starting it again (attempt {attempt})').format( - vmid=vmid, name=name, attempt=state['attempts']), flush=True) - try: - started = start(vmid) - except (OSError, subprocess.SubprocessError): - started = False - if started and notice_due(state, now): - oci_operation_notice.notify('oci_watchdog_restarted', data) - elif action == 'give-up': - print(translate('CT {vmid} ({name}) could not be kept running; it is left stopped').format( - vmid=vmid, name=name), flush=True) - oci_operation_notice.notify('oci_watchdog_failed', data) + for vmid, (name, installation_id) in apps.items(): + state = states.get(vmid) + if state is None or state.get('installation_id') != installation_id: + state = states[vmid] = new_state(installation_id) + running = is_running(vmid) + if running or state['seen'] is None or state['settled']: + decide(state, running, now, lambda: is_busy(vmid, now), + lambda seen: stop_requested(read_tasks(), vmid, seen)) + continue + notice = None + try: + # Updates/removals hold this lock too. Recheck the record and the + # live identity before deciding, without charging busy operations a retry. + with instances.locked(instances.ROOT): + current = watched(instances.ROOT).get(vmid) + if current is None or current[1] != installation_id: + states.pop(vmid, None) + continue + matches = same_installation(vmid, installation_id) + if matches is None: + continue + if not matches: + states.pop(vmid, None) + continue + name = current[0] + action = decide(state, is_running(vmid), now, lambda: is_busy(vmid, now), + lambda seen: stop_requested(read_tasks(), vmid, seen)) + if action == 'start': + print(translate('CT {vmid} ({name}) stopped on its own; starting it again (attempt {attempt})').format( + vmid=vmid, name=name, attempt=state['attempts']), flush=True) + try: + started = start(vmid) + except (OSError, subprocess.SubprocessError): + started = False + if started and notice_due(state, now): + notice = 'oci_watchdog_restarted' + elif action == 'give-up': + print(translate('CT {vmid} ({name}) could not be kept running; it is left stopped').format( + vmid=vmid, name=name), flush=True) + notice = 'oci_watchdog_failed' + except BlockingIOError: + continue + if notice: + oci_operation_notice.notify(notice, {'app_name': name, 'vmid': vmid, 'containers': f'CT {vmid}'}) def run(): diff --git a/oci/tests/test_watchdog.py b/oci/tests/test_watchdog.py index 89962ee9..8fd70d84 100644 --- a/oci/tests/test_watchdog.py +++ b/oci/tests/test_watchdog.py @@ -2,9 +2,11 @@ import json from pathlib import Path +import subprocess import sys import tempfile import unittest +from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "remote")) @@ -148,10 +150,26 @@ class Registry(unittest.TestCase): } for vmid, record in records.items(): (root / str(vmid)).mkdir() - (root / str(vmid) / "oci-compose.json").write_text(json.dumps({"vmid": vmid, **record})) + (root / str(vmid) / "oci-compose.json").write_text(json.dumps({ + "schema_version": 1, "vmid": vmid, + "installation_id": "11111111-1111-4111-8111-111111111111", **record})) (root / "107").mkdir() (root / "107" / "oci-compose.json").write_text("{broken") - self.assertEqual(watchdog.watched(root), {101: "Jellyfin", 106: "CT 106"}) + self.assertEqual(watchdog.watched(root), { + 101: ("Jellyfin", "11111111-1111-4111-8111-111111111111"), + 106: ("CT 106", "11111111-1111-4111-8111-111111111111")}) + + def test_a_record_without_a_valid_installation_identity_is_not_watched(self): + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + path = root / "109" / "oci-compose.json" + path.parent.mkdir() + for identity in (None, "", "not-a-uuid", 123, []): + with self.subTest(identity=identity): + path.write_text(json.dumps({ + "schema_version": 1, "vmid": 109, "status": "installed", + "installation_id": identity, "deployment": {"watchdog": True}})) + self.assertEqual(watchdog.watched(root), {}) class Start(unittest.TestCase): @@ -165,6 +183,154 @@ class Start(unittest.TestCase): self.assertEqual(run.call_args.args[0], ["systemd-run", "--scope", "--quiet", "--collect", "pct", "start", "109"]) +class InstallationIdentity(unittest.TestCase): + """A recycled VMID is not the application the watchdog saw running.""" + VMID = 109 + ORIGINAL = "11111111-1111-4111-8111-111111111111" + REPLACEMENT = "22222222-2222-4222-8222-222222222222" + + def setUp(self): + folder = tempfile.TemporaryDirectory() + self.addCleanup(folder.cleanup) + self.base = Path(folder.name) + self.registry = self.base / "instances" + self.configs = self.base / "lxc" + self.cgroups = self.base / "cgroups" + for path in (self.registry, self.configs, self.cgroups): + path.mkdir() + for target, value in (("CONFIGS", self.configs), ("CGROUPS", self.cgroups)): + patcher = patch.object(watchdog, target, value) + patcher.start() + self.addCleanup(patcher.stop) + patcher = patch.object(watchdog.instances, "ROOT", self.registry) + patcher.start() + self.addCleanup(patcher.stop) + for name, target, kwargs in ( + ("start", watchdog, {"return_value": True}), + ("is_busy", watchdog, {"return_value": False}), + ("read_tasks", watchdog, {"return_value": []}), + ("notify", watchdog.oci_operation_notice, {}), + ("run", watchdog.subprocess, {"side_effect": self.command})): + patcher = patch.object(target, name, **kwargs) + setattr(self, name, patcher.start()) + self.addCleanup(patcher.stop) + self.states = {} + self.record() + self.config(self.ORIGINAL) + + def record(self, identity=None, **extra): + path = self.registry / str(self.VMID) / "oci-compose.json" + path.parent.mkdir(exist_ok=True) + path.write_text(json.dumps({ + "schema_version": 1, "vmid": self.VMID, + "installation_id": identity or self.ORIGINAL, + "status": "installed", "deployment": {"watchdog": True}, + "template": {"catalog_ui": {"title": {"en_US": "Example"}}}, **extra})) + + def config(self, identity): + self.config_text(f"description: proxmenux-instance={identity}\narch: amd64\n") + + def config_text(self, text): + (self.configs / f"{self.VMID}.conf").write_text(text) + + def command(self, argv, **kwargs): + self.assertEqual(argv, ["pct", "config", str(self.VMID)]) + return subprocess.CompletedProcess(argv, 0, (self.configs / f"{self.VMID}.conf").read_bytes(), b"") + + def seen_running(self): + (self.cgroups / str(self.VMID)).mkdir() + watchdog.look(self.states, 100) + (self.cgroups / str(self.VMID)).rmdir() + + def assert_left_alone(self): + watchdog.look(self.states, 110) + self.start.assert_not_called() + self.notify.assert_not_called() + + def test_transaction_holder_is_not_started_for_a_stale_application_record(self): + self.seen_running() + self.config_text("description: proxmenux-transaction=example-update\nhostname: oci-data-holder\n") + self.assert_left_alone() + + def test_another_installation_at_the_same_vmid_is_not_started(self): + self.seen_running() + self.config(self.REPLACEMENT) + self.assert_left_alone() + + def test_a_marker_in_old_snapshot_notes_does_not_own_the_live_guest(self): + self.seen_running() + self.config_text("description: unrelated container\n\n[old]\n" + f"description: proxmenux-instance={self.ORIGINAL}\n") + self.assert_left_alone() + + def test_an_unreadable_live_configuration_is_not_started(self): + self.seen_running() + self.run.side_effect = None + self.run.return_value = subprocess.CompletedProcess([], 2, b"", b"missing config") + self.assert_left_alone() + self.run.side_effect = self.command + watchdog.look(self.states, 111) + self.start.assert_called_once_with(self.VMID) + self.assertEqual(self.states[self.VMID]["attempts"], 1) + + def test_a_removed_guest_drops_its_running_history_without_a_pve_query(self): + self.seen_running() + (self.configs / f"{self.VMID}.conf").unlink() + self.assert_left_alone() + self.run.assert_not_called() + self.assertNotIn(self.VMID, self.states) + + def test_a_configuration_timeout_is_retried_without_consuming_an_attempt(self): + self.seen_running() + self.run.side_effect = subprocess.TimeoutExpired("pct", 30) + self.assert_left_alone() + self.run.side_effect = self.command + watchdog.look(self.states, 111) + self.start.assert_called_once_with(self.VMID) + self.assertEqual(self.states[self.VMID]["attempts"], 1) + + def test_a_new_installation_does_not_inherit_running_history(self): + self.seen_running() + self.record(self.REPLACEMENT) + self.config(self.REPLACEMENT) + self.assert_left_alone() + (self.cgroups / str(self.VMID)).mkdir() + watchdog.look(self.states, 120) + (self.cgroups / str(self.VMID)).rmdir() + watchdog.look(self.states, 130) + self.start.assert_called_once_with(self.VMID) + + def test_a_transaction_started_after_the_scan_is_respected(self): + self.seen_running() + + def updating(vmid): + self.record(status="updating", pending_transaction="/journal") + return False + + with patch.object(watchdog, "is_running", side_effect=updating): + self.assert_left_alone() + + def test_a_busy_registry_does_not_start_or_consume_a_restart_attempt(self): + self.seen_running() + with watchdog.instances.locked(self.registry): + self.assert_left_alone() + watchdog.look(self.states, 111) + self.start.assert_called_once_with(self.VMID) + self.assertEqual(self.states[self.VMID]["attempts"], 1) + + def test_the_same_installation_is_still_restarted_after_a_crash(self): + self.seen_running() + watchdog.look(self.states, 110) + self.start.assert_called_once_with(self.VMID) + self.notify.assert_called_once() + + def test_a_recycled_vmid_does_not_receive_a_crash_loop_alert(self): + self.seen_running() + self.states[self.VMID].update(attempts=len(watchdog.DELAYS), retry_at=0) + self.config(self.REPLACEMENT) + self.assert_left_alone() + + class Switch(unittest.TestCase): """Turning the watchdog on or off for a whole application.""" def registry(self, folder):