Merge pull request #493 from jcastro/perf/oci-manage-record-sync

Speed up OCI Manage by caching unchanged Proxmox config reads

.github/workflows/test-oci-registries.yml

@@ -28,6 +28,12 @@ jobs:
           python-version: '3.13'
       - name: Install OCI test dependencies
         run: python3 -m pip install -r oci/requirements.txt
+      - name: Run carried record, recovery and configuration cache regressions
+        env:
+          PYTHONDONTWRITEBYTECODE: '1'
+        run: |
+          python3 -m unittest discover -s oci/tests -p 'test_carried*.py' -v
+          python3 -m unittest discover -s oci/tests -p test_restore_recovery.py -v
       - name: Run read-only volume preparation regressions
         env:
           PYTHONDONTWRITEBYTECODE: '1'

oci/remote/oci_carried_record.py

@@ -38,6 +38,8 @@ DIRECTORY = '.proxmenux'
 NAME = 'oci-record.json'
 KIND = 'proxmenux.oci-carried-record'
 STAMP = 'carried.sha256'
+CONFIG_CACHE = 'carried-config.json'
+LOCAL_CONFIGS = Path('/etc/pve/lxc')
 LIMIT = 16 * 1024 * 1024
 STACK_CONTRACT = '/etc/pve/priv/proxmenux-stack-{}.json'
 NVIDIA_HOOK = re.compile(r'/usr/local/lib/proxmenux/oci/nvidia-mount-([a-f0-9]{64})\.sh')
@@ -279,6 +281,47 @@ def read_stamp(path):
     return value if isinstance(value, dict) else {}
 
 
+def configuration(root, vmid, verify=False):
+    """Reuse pct's formatted output only while its local source is unchanged.
+
+    Opening Manage checks every carried copy. Starting a fresh pct process
+    for each unchanged guest dominates that check. The private host cache is
+    keyed by the complete native configuration, including notes, pending
+    changes and snapshot sections. Lifecycle verification always calls pct.
+    An unavailable source, invalid cache or concurrent edit falls back to pct.
+    """
+    source = LOCAL_CONFIGS / f'{int(vmid)}.conf'
+    cache = instances.location(root, vmid).parent / CONFIG_CACHE
+
+    def fingerprint():
+        try:
+            return sha(source.read_bytes())
+        except OSError:
+            return None
+
+    before = fingerprint()
+    if before is not None and not verify:
+        try:
+            if not cache.is_symlink() and cache.stat().st_size <= LIMIT:
+                value = json.loads(cache.read_text())
+                if (isinstance(value, dict) and value.get('schema_version') == 1
+                        and value.get('source_sha256') == before
+                        and isinstance(value.get('config'), str) and value['config']
+                        and fingerprint() == before):
+                    return value['config']
+        except (OSError, ValueError):
+            pass
+    result = _run('pct', 'config', str(vmid))
+    if result.returncode != 0:
+        return None
+    # Do not associate output from one revision with another revision's hash.
+    if before is not None and fingerprint() == before:
+        with contextlib.suppress(OSError, ValueError):
+            instances.write(cache, {'schema_version': 1, 'source_sha256': before,
+                                    'config': result.stdout})
+    return result.stdout
+
+
 def superseded(record, existing, stamp):
     """Whether the container carries a copy this host did not write, with a
     record that differs from the one of this host. That is a container that
@@ -308,10 +351,10 @@ def carry(root, vmid, mount_stopped=True, verify=False, adopted=False):
     if record.get('status') != 'installed' or record.get('pending_transaction') \
             or record.get('pending_stack_transaction'):
         return 'skipped'
-    result = _run('pct', 'config', str(vmid))
-    if result.returncode != 0 or instances.identity(result.stdout.encode()) != record['installation_id']:
+    config = configuration(root, vmid, verify)
+    if config is None or instances.identity(config.encode()) != record['installation_id']:
         return 'skipped'
-    value = bundle(record, result.stdout)
+    value = bundle(record, config)
     expected = digest(value)
     record_path = instances.location(root, vmid)
     path = record_path.parent / STAMP
@@ -340,7 +383,7 @@ def carry(root, vmid, mount_stopped=True, verify=False, adopted=False):
             else:
                 value['saved_at'] = datetime.datetime.now(datetime.timezone.utc).isoformat()
                 value['generation'] = uuid.uuid4().hex
-                write_copy(rootfs, value, mapped_root(result.stdout))
+                write_copy(rootfs, value, mapped_root(config))
                 stamp = {'generation': value['generation']}
                 outcome = 'carried'
         if outcome == 'stale':

oci/tests/test_carried_config_cache.py

@@ -0,0 +1,134 @@
+"""A fast Manage entry still detects config edits and verifies lifecycle work."""
+import contextlib
+import copy
+import json
+import os
+from pathlib import Path
+import sys
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / 'remote'))
+import oci_carried_record as carried
+
+
+class CarriedConfigCacheTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name) / 'instances'
+        self.source = Path(self.tmp.name) / 'lxc'
+        self.source.mkdir()
+        self.config = self.source / '120.conf'
+        self.config.write_text('# proxmenux-instance=example\nunprivileged: 1\n')
+        (self.root / '120').mkdir(parents=True)
+        self.cache = self.root / '120' / carried.CONFIG_CACHE
+        self.patch = patch.object(carried, 'LOCAL_CONFIGS', self.source)
+        self.patch.start()
+        self.addCleanup(self.patch.stop)
+
+    def response(self, text='description: proxmenux-instance=example\nunprivileged: 1\n', status=0):
+        return SimpleNamespace(returncode=status, stdout=text)
+
+    def test_warm_reads_avoid_pct_and_cache_is_private(self):
+        with patch.object(carried, '_run', return_value=self.response()) as run:
+            first = carried.configuration(self.root, 120)
+            self.assertEqual(carried.configuration(self.root, 120), first)
+        run.assert_called_once_with('pct', 'config', '120')
+        self.assertEqual(self.cache.stat().st_mode & 0o777, 0o600)
+
+    def test_native_edits_invalidate_even_when_size_and_mtime_are_restored(self):
+        with patch.object(carried, '_run', side_effect=[self.response(), self.response('new')]) as run:
+            carried.configuration(self.root, 120)
+            info = self.config.stat()
+            self.config.write_text(self.config.read_text().replace('example', 'changed'))
+            os.utime(self.config, ns=(info.st_atime_ns, info.st_mtime_ns))
+            self.assertEqual(carried.configuration(self.root, 120), 'new')
+        self.assertEqual(run.call_count, 2)
+
+    def test_verify_forces_a_fresh_pct_read_with_a_warm_cache(self):
+        with patch.object(carried, '_run', side_effect=[self.response(), self.response('verified')]) as run:
+            carried.configuration(self.root, 120)
+            self.assertEqual(carried.configuration(self.root, 120, verify=True), 'verified')
+        self.assertEqual(run.call_count, 2)
+
+    def test_missing_native_source_never_uses_or_populates_a_cache(self):
+        self.config.unlink()
+        with patch.object(carried, '_run', return_value=self.response()) as run:
+            carried.configuration(self.root, 120)
+            carried.configuration(self.root, 120)
+        self.assertEqual(run.call_count, 2)
+        self.assertFalse(self.cache.exists())
+
+    def test_concurrent_edit_does_not_cache_output_for_the_wrong_revision(self):
+        def command(*args):
+            self.config.write_text('changed while pct was reading\n')
+            return self.response()
+        with patch.object(carried, '_run', side_effect=command):
+            carried.configuration(self.root, 120)
+        self.assertFalse(self.cache.exists())
+
+    def test_edit_during_cache_lookup_falls_back_to_pct(self):
+        with patch.object(carried, '_run', return_value=self.response()):
+            carried.configuration(self.root, 120)
+        read_text = Path.read_text
+        def edit_during_read(path, *args, **kwargs):
+            text = read_text(path, *args, **kwargs)
+            if path == self.cache:
+                self.config.write_text('changed while reading the cache\n')
+            return text
+        with patch.object(Path, 'read_text', edit_during_read), \
+                patch.object(carried, '_run', return_value=self.response('fresh')) as run:
+            self.assertEqual(carried.configuration(self.root, 120), 'fresh')
+        run.assert_called_once_with('pct', 'config', '120')
+
+    def test_corrupt_cache_and_pct_failure_are_not_accepted(self):
+        for text in ('not json', '[]', '{"schema_version": 999}'):
+            with self.subTest(cache=text):
+                self.cache.write_text(text)
+                with patch.object(carried, '_run', return_value=self.response(status=1)) as run:
+                    self.assertIsNone(carried.configuration(self.root, 120))
+                run.assert_called_once()
+                self.assertEqual(self.cache.read_text(), text)
+
+    def test_symlink_cache_does_not_read_or_replace_its_target(self):
+        target = Path(self.tmp.name) / 'outside'
+        target.write_text('private original')
+        self.cache.symlink_to(target)
+        with patch.object(carried, '_run', return_value=self.response()) as run:
+            self.assertEqual(carried.configuration(self.root, 120), self.response().stdout)
+        run.assert_called_once()
+        self.assertTrue(self.cache.is_symlink())
+        self.assertEqual(target.read_text(), 'private original')
+
+    def test_cached_config_still_detects_a_restored_container_record(self):
+        installation = '488ed3cb-1145-477a-b90f-c46777d69fb2'
+        record = {'schema_version': 1, 'vmid': 120, 'installation_id': installation,
+                  'status': 'installed', 'deployment': {'hostname': 'current'}}
+        carried.instances.write(carried.instances.location(self.root, 120), record)
+        native = f'description: proxmenux-instance={installation}\n'
+        with patch.object(carried, '_run', return_value=self.response(native)):
+            carried.configuration(self.root, 120)
+        rootfs = Path(self.tmp.name) / 'rootfs'
+        rootfs.mkdir()
+        previous = copy.deepcopy(record)
+        previous['deployment']['hostname'] = 'restored-snapshot'
+        carried.write_copy(rootfs, {'kind': carried.KIND, 'schema_version': 1,
+                                   'record': previous, 'generation': 'restored-generation'},
+                           (os.getuid(), os.getgid()))
+        stamp = self.root / '120' / carried.STAMP
+        stamp.write_text(json.dumps({'generation': 'host-generation'}))
+        with patch.object(carried, '_run', side_effect=AssertionError('cached config expected')), \
+                patch.object(carried, 'read_cluster', return_value=None), \
+                patch.object(carried, 'running_pid', return_value=123), \
+                patch.object(carried, 'container_root', return_value=contextlib.nullcontext(rootfs)):
+            self.assertEqual(carried.carry(self.root, 120, mount_stopped=False), 'stale')
+        self.assertFalse(carried.instances.location(self.root, 120).exists())
+        self.assertTrue((self.root / '120' / f'retired-{installation}.json').exists())
+
+
+if __name__ == '__main__':
+    unittest.main()