Merge pull request #529 from jcastro/fix/oci-watchdog-instance-identity

Prevent OCI watchdog from restarting recycled VMIDs

.github/scripts/tests/test_oci_complete_errors_sk.py

@@ -145,13 +145,16 @@ class TechnicalMessages(unittest.TestCase):
 
     def test_watchdog_manual_stop_stays_stopped_without_start_or_notice(self):
         self.remote_language()
-        self.patch(oci_watchdog, 'watched', return_value={777: 'Fiction'})
+        identity = '11111111-1111-4111-8111-111111111111'
+        self.patch(oci_watchdog, 'watched', return_value={777: ('Fiction', identity)})
+        self.patch(oci_watchdog.instances, 'locked', side_effect=lambda *a: nullcontext())
+        self.patch(oci_watchdog, 'same_installation', return_value=True)
         self.patch(oci_watchdog, 'is_running', return_value=False)
         self.patch(oci_watchdog, 'is_busy', return_value=False)
         self.patch(oci_watchdog, 'read_tasks', return_value=[('vzshutdown', '777', 75)])
         start = self.patch(oci_watchdog, 'start', side_effect=AssertionError('manual stop was undone'))
         notice = self.patch(oci_watchdog.oci_operation_notice, 'notify')
-        state = oci_watchdog.new_state(); state['seen'] = 50
+        state = oci_watchdog.new_state(identity); state['seen'] = 50
         output = io.StringIO()
         with redirect_stdout(output): oci_watchdog.look({777: state}, 100)
         self.assertTrue(state['settled'])
@@ -159,7 +162,10 @@ class TechnicalMessages(unittest.TestCase):
         start.assert_not_called(); notice.assert_not_called()
 
     def test_watchdog_localization_does_not_change_retry_state_or_notification_payload(self):
-        self.patch(oci_watchdog, 'watched', return_value={777: 'Fiction'})
+        identity = '11111111-1111-4111-8111-111111111111'
+        self.patch(oci_watchdog, 'watched', return_value={777: ('Fiction', identity)})
+        self.patch(oci_watchdog.instances, 'locked', side_effect=lambda *a: nullcontext())
+        self.patch(oci_watchdog, 'same_installation', return_value=True)
         self.patch(oci_watchdog, 'is_running', return_value=False)
         self.patch(oci_watchdog, 'is_busy', return_value=False)
         self.patch(oci_watchdog, 'read_tasks', return_value=[])
@@ -168,7 +174,7 @@ class TechnicalMessages(unittest.TestCase):
         outcomes = []
         for locale in ('en', 'sk'):
             with patch.object(oci_ui, '_language', locale), patch.object(oci_ui, '_cache', self.sk):
-                state = oci_watchdog.new_state(); state['seen'] = 50
+                state = oci_watchdog.new_state(identity); state['seen'] = 50
                 text = io.StringIO()
                 with redirect_stdout(text): oci_watchdog.look({777: state}, 100)
                 outcomes.append((copy.deepcopy(state), notice.call_args, start.call_args))

.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'

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
 

oci/remote/oci_watchdog.py

@@ -46,8 +46,9 @@ STOP_TASKS = {'vzstop', 'vzshutdown', 'vzsuspend', 'vzreboot', 'vzdestroy', 'vzm
 HOST_TASKS = {'stopall', 'migrateall'}
 
 
-def new_state():
+def new_state(installation_id=None):
-    return {'seen': None, 'since': None, 'attempts': 0, 'retry_at': 0.0, 'settled': False, 'noticed': 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(path.parent.name)
-            vmid = int(record['vmid'])
+            record = instances.read(root, vmid)
-        except (OSError, ValueError, KeyError, TypeError):
+            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:
+        found[vmid] = (name, record['installation_id'])
-            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}'
     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():
+    for vmid, (name, installation_id) in apps.items():
-        state = states.setdefault(vmid, new_state())
+        state = states.get(vmid)
-        action = decide(state, is_running(vmid), now, lambda: is_busy(vmid, now),
+        if state is None or state.get('installation_id') != installation_id:
-                        lambda seen: stop_requested(read_tasks(), vmid, seen))
+            state = states[vmid] = new_state(installation_id)
-        data = {'app_name': name, 'vmid': vmid, 'containers': f'CT {vmid}'}
+        running = is_running(vmid)
-        if action == 'start':
+        if running or state['seen'] is None or state['settled']:
-            print(translate('CT {vmid} ({name}) stopped on its own; starting it again (attempt {attempt})').format(
+            decide(state, running, now, lambda: is_busy(vmid, now),
-                vmid=vmid, name=name, attempt=state['attempts']), flush=True)
+                   lambda seen: stop_requested(read_tasks(), vmid, seen))
-            try:
+            continue
-                started = start(vmid)
+        notice = None
-            except (OSError, subprocess.SubprocessError):
+        try:
-                started = False
+            # Updates/removals hold this lock too. Recheck the record and the
-            if started and notice_due(state, now):
+            # live identity before deciding, without charging busy operations a retry.
-                oci_operation_notice.notify('oci_watchdog_restarted', data)
+            with instances.locked(instances.ROOT):
-        elif action == 'give-up':
+                current = watched(instances.ROOT).get(vmid)
-            print(translate('CT {vmid} ({name}) could not be kept running; it is left stopped').format(
+                if current is None or current[1] != installation_id:
-                vmid=vmid, name=name), flush=True)
+                    states.pop(vmid, None)
-            oci_operation_notice.notify('oci_watchdog_failed', data)
+                    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():

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):