Reuse OCI images and apply resource edits without rebuilding

This commit is contained in:
Jonatan
2026-10-17 17:11:26 +02:00
parent ce1a3de719
commit 784ce13c14
11 changed files with 481 additions and 14 deletions
+112
View File
@@ -0,0 +1,112 @@
"""Apply CPU/RAM-only proposals to the existing guest, without an image archive."""
import copy
import hashlib
from pathlib import Path
import uuid
import oci_instances as instances
import oci_instance_transaction as transaction
from oci_installation_state import parse_config, sha
from oci_ui import translate, msg_info, msg_ok, msg_warn
CONFIG_DIR = Path('/etc/pve/lxc')
def changes(record, candidate, config):
"""Return a narrow native edit, or None when the normal installer is needed."""
previous = transaction.with_default_healthcheck(copy.deepcopy(record))
candidate = transaction.with_default_healthcheck(copy.deepcopy(candidate))
old, new = previous['deployment'], candidate['deployment']
if previous['template'] != candidate['template'] or old.get('host_monitor'):
return None
if {k: v for k, v in old.items() if k != 'resources'} != {
k: v for k, v in new.items() if k != 'resources'}:
return None
original, desired = old.get('resources', {}), new.get('resources', {})
changed = {key for key in original.keys() | desired.keys() if original.get(key) != desired.get(key)}
if not changed or changed - {'cores', 'memory_mb'}:
return None
cfg = parse_config(config)
cpu = 'cpulimit' if 'cpulimit' in cfg and 'cores' not in cfg else 'cores'
parameters = {}
for key in sorted(changed):
value = desired.get(key)
if isinstance(value, bool) or not isinstance(value, int) or value < (128 if key == 'memory_mb' else 1):
return None
native = 'memory' if key == 'memory_mb' else cpu
if native not in cfg:
return None
parameters[native] = str(value)
return parameters
def guarded_set(vmid, installation_id, expected, parameters):
# pct uses this SHA1 under its own configuration lock. Sample the native
# file before checking pct's view, so a concurrent edit cannot be accepted
# accidentally by taking a fresh digest after validation.
native = (CONFIG_DIR / f'{vmid}.conf').read_bytes()
current = instances.command('pct', 'config', str(vmid))
if current != expected or instances.identity(current) != installation_id:
raise ValueError(translate('The instance changed while it was being edited; open Modify again'))
command = ['pct', 'set', str(vmid), '--digest', hashlib.sha1(native).hexdigest()]
for key, value in sorted(parameters.items()):
command += ['--' + key, value]
transaction.run(*command)
def expected_config(before, after, parameters):
def unrelated(config):
return b''.join(line for line in config.splitlines(keepends=True)
if line.split(b':', 1)[0].decode() not in parameters)
new = parse_config(after)
return unrelated(before) == unrelated(after) and all(new.get(key) == value for key, value in parameters.items())
def apply(root, record, candidate, config, parameters):
"""The caller holds the instance lock and has checked image identity/preflight."""
if changes(record, candidate, config) != parameters:
raise ValueError(translate('The instance changed while it was being edited; open Modify again'))
vmid = record['vmid']
directory = instances.location(root, vmid).parent / 'modifications' / uuid.uuid4().hex
instances.write(directory / 'before.json', {'record': record, 'config': config.decode()})
transaction.open_log(directory)
msg_info(translate('Applying CPU and RAM changes without rebuilding the container...'))
applied = False
after = None
try:
guarded_set(vmid, record['installation_id'], config, parameters)
applied = True
after = instances.command('pct', 'config', str(vmid))
if not expected_config(config, after, parameters):
raise ValueError(translate('The instance changed while it was being edited; open Modify again'))
observed = record['observed']
fresh = instances.observe(vmid, record['installation_id'], observed['archive_path'],
observed['resolved_registry_digest'], observed['image'])
if fresh['config_sha256'] != sha(after):
raise ValueError(translate('The instance changed while it was being edited; open Modify again'))
updated = copy.deepcopy(record)
updated.update(deployment=copy.deepcopy(candidate['deployment']), observed=fresh)
instances.write(instances.location(root, vmid), updated)
except Exception:
if applied:
try:
current = instances.command('pct', 'config', str(vmid))
if not expected_config(config, current, parameters):
raise ValueError('Concurrent configuration change; resource rollback refused')
previous = parse_config(config)
guarded_set(vmid, record['installation_id'], current,
{key: previous[key] for key in parameters})
instances.write(instances.location(root, vmid), record)
except Exception as error:
transaction.log(f'resource rollback: {type(error).__name__}')
msg_warn(translate('The resource edit could not be fully rolled back. Open Manage again to review and import the current Proxmox settings.'))
raise
# Updating the carried record is best effort here, as it is after other
# management operations. The normal menu also refreshes it on return.
try:
import oci_carried_record
oci_carried_record.carry(root, vmid, mount_stopped=False)
except (OSError, ValueError, RuntimeError):
transaction.log('carried record will be refreshed on the next management entry')
msg_ok(translate('Changes applied. Data kept.'))
return directory
+39 -14
View File
@@ -8,6 +8,7 @@ import re
import shlex
import subprocess
import sys
import tarfile
import tempfile
import oci_instances as instances
@@ -56,7 +57,7 @@ def run_quiet(args, error, capture=False):
process.wait()
def resolve_archive(desired, config, current=None, check=None, required_digest=None):
def resolve_archive(desired, config, current=None, check=None, required_digest=None, local_archives=()):
# Shared by individual and coordinated operations; no guest mutation here.
# When the registry still serves the current digest nothing is downloaded.
reference = desired['template']['container_contract']['image']['reference']
@@ -83,6 +84,27 @@ def resolve_archive(desired, config, current=None, check=None, required_digest=N
return None, digest
if check:
check()
verifier = Path(__file__).with_name('verify_oci_archive.py')
def intact(path):
try:
run_quiet([sys.executable, str(verifier), str(path)],
translate('The image did not pass the integrity check'))
return same_image(candidate, image_from_archive(str(path)))
except (RuntimeError, OSError, ValueError, KeyError, tarfile.TarError):
return False
# Initial installations and updates use different archive names. The
# instance record knows the actual archive; validate it before looking
# only for the deterministic update-cache name. Never delete a rejected
# recorded path: another instance may still refer to it.
for saved in local_archives:
path = Path(saved) if saved else None
if path and not path.is_symlink() and path.is_file():
msg_info(translate('Verifying the image integrity...'))
if intact(path):
msg_ok(translate('Using the verified image from the cache'))
return path, digest
storage = desired['deployment']['template_storage']
if not re.fullmatch(r'[A-Za-z0-9_-]+', storage):
raise ValueError(translate('Invalid template storage'))
@@ -91,16 +113,6 @@ def resolve_archive(desired, config, current=None, check=None, required_digest=N
archive.parent.mkdir(parents=True, exist_ok=True)
if archive.is_symlink():
raise ValueError(translate('Unsafe OCI archive path'))
verifier = Path(__file__).with_name('verify_oci_archive.py')
def intact(path):
try:
run_quiet([sys.executable, str(verifier), str(path)],
translate('The image did not pass the integrity check'))
except RuntimeError:
return False
return same_image(candidate, image_from_archive(str(path)))
if archive.exists():
msg_info(translate('Verifying the image integrity...'))
if intact(archive):
@@ -181,10 +193,23 @@ def update(vmid, acknowledge_external_data=False, proposal=None, keep_backup=Non
changes = transaction.external_changes(record, config)
transaction.preflight(record, desired, config)
msg_ok(translate('Container checked'))
current = record['observed']['image'] if operation == 'update' else None
archive, digest = resolve_archive(desired, config, current, lambda: work_backup.locate(
instances.location(instances.ROOT, vmid).parent / 'transactions' / 'check', [vmid], record, vmid))
resource_changes = None
if operation == 'recreate' and not keep_backup:
import oci_resource_modify
resource_changes = oci_resource_modify.changes(record, desired, config)
current = record['observed']['image'] if operation == 'update' or resource_changes else None
check = lambda: work_backup.locate(
instances.location(instances.ROOT, vmid).parent / 'transactions' / 'check', [vmid], record, vmid)
if operation == 'recreate':
archive, digest = resolve_archive(desired, config, current, check,
local_archives=[record['observed'].get('archive_path')])
else:
archive, digest = resolve_archive(desired, config, current, check)
if archive is None:
if resource_changes:
import oci_operation_notice
with oci_operation_notice.operation([vmid], 'modify', application_name(record, vmid)):
return oci_resource_modify.apply(instances.ROOT, record, desired, config, resource_changes)
msg_ok(translate('The image is already up to date; nothing was changed.'))
return
file_storage = None
+312
View File
@@ -0,0 +1,312 @@
"""Unchanged OCI images permit native resource edits or verified local reuse."""
import copy
import hashlib
import json
from pathlib import Path
import re
import stat
import sys
import tempfile
import unittest
from unittest.mock import Mock, patch
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT/'remote'))
import oci_carried_record
import oci_instances as instances
import oci_resource_modify as resources
import oci_update_current as update
DIGEST = 'sha256:'+'a'*64
NEW = 'sha256:'+'b'*64
IDENTITY = '11111111-1111-4111-8111-111111111111'
CONFIG = (f'arch: amd64\ncores: 2\ndescription: proxmenux-instance={IDENTITY}\n'
'memory: 512\nrootfs: local-zfs:subvol-115-disk-0,size=8G\n'
'lxc.environment.runtime: A=one\nlxc.environment.runtime: B=two\n').encode()
def record():
return {'schema_version': 1, 'vmid': 115, 'installation_id': IDENTITY, 'status': 'installed',
'template': {'id': 'example-app', 'catalog_ui': {'title': 'Example'},
'container_contract': {'image': {'reference': 'example/app:latest'}},
'proxmox': {'installer_profile': {'startup_healthcheck': {'type': 'running'}}}},
'deployment': {'vmid': 115, 'hostname': 'example', 'template_storage': 'local',
'resources': {'cores': 2, 'memory_mb': 512, 'swap_mb': 512},
'network': {'bridge': 'vmbr0', 'ipv4': 'dhcp'}, 'mounts': [], 'devices': [],
'environment': [{'name': 'PASSWORD', 'value': 'private-test-value', 'sensitive': True}]},
'observed': {'config': CONFIG.decode(), 'config_sha256': resources.sha(CONFIG),
'image': {'manifest_digest': DIGEST, 'architecture': 'amd64', 'os': 'linux'},
'resolved_registry_digest': DIGEST, 'archive_path': '/missing/deleted-image.tar'}}
class ResourceModifyTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)/'instances'
self.configs = Path(self.temp.name)/'configs'
self.configs.mkdir()
self.file = self.configs/'115.conf'
self.file.write_bytes(CONFIG)
self.saved = record()
self.candidate = copy.deepcopy(self.saved)
self.candidate['deployment']['resources'].update(cores=3, memory_mb=768)
self.params = {'cores': '3', 'memory': '768'}
instances.write(instances.location(self.root, 115), self.saved)
self.calls = []
self.native_failure = None
def run(*args):
self.calls.append(args)
self.assertEqual(args[:3], ('pct', 'set', '115'))
expected = args[args.index('--digest')+1]
self.assertEqual(expected, hashlib.sha1(self.file.read_bytes()).hexdigest())
if self.native_failure:
raise self.native_failure
text = self.file.read_text()
for index in range(5, len(args), 2):
text = re.sub(r'^'+args[index][2:]+r': .*$', args[index][2:]+': '+args[index+1], text, flags=re.M)
self.file.write_text(text)
return b''
def observe(*args):
result = copy.deepcopy(self.saved['observed'])
result.update(config=self.file.read_text(), config_sha256=resources.sha(self.file.read_bytes()))
return result
patches = [patch.object(resources, 'CONFIG_DIR', self.configs),
patch.object(instances, 'command', side_effect=lambda *args: self.file.read_bytes()),
patch.object(resources.transaction, 'run', side_effect=run),
patch.object(resources.transaction, 'open_log'), patch.object(resources.transaction, 'log'),
patch.object(instances, 'observe', side_effect=observe),
patch.object(oci_carried_record, 'carry', return_value='carried')]
for p in patches:
p.start()
self.addCleanup(p.stop)
def apply(self):
return resources.apply(self.root, self.saved, self.candidate, CONFIG, self.params)
def test_resource_only_change_needs_no_archive_and_preserves_every_other_field(self):
directory = self.apply()
after = instances.read(self.root, 115)
self.assertEqual(after['deployment']['resources'], {'cores': 3, 'memory_mb': 768, 'swap_mb': 512})
for key in ('network', 'mounts', 'devices', 'environment'):
self.assertEqual(after['deployment'][key], self.saved['deployment'][key])
self.assertEqual(after['observed']['image'], self.saved['observed']['image'])
self.assertEqual(after['observed']['archive_path'], '/missing/deleted-image.tar')
self.assertEqual(after['observed']['config_sha256'], resources.sha(self.file.read_bytes()))
self.assertNotIn('pending_transaction', after)
self.assertEqual(len(self.calls), 1)
self.assertEqual(stat.S_IMODE((directory/'before.json').stat().st_mode), 0o600)
self.assertEqual(json.loads((directory/'before.json').read_text())['record'], self.saved)
self.assertNotIn('private-test-value', str(self.calls))
def test_network_environment_template_and_extra_device_edits_use_the_normal_installer(self):
for mutate in (
lambda c: c['deployment']['network'].update(ipv4='192.0.2.2/24'),
lambda c: c['deployment']['environment'][0].update(value='changed'),
lambda c: c['template']['container_contract']['image'].update(reference='example/app:other'),
lambda c: c['deployment']['devices'].append({'host_path': '/dev/ttyACM0'}),
lambda c: c['deployment']['resources'].update(swap_mb=1024),
):
candidate = copy.deepcopy(self.candidate)
mutate(candidate)
self.assertIsNone(resources.changes(self.saved, candidate, CONFIG))
def test_invalid_values_and_unchanged_proposal_do_not_use_the_native_path(self):
for value in (True, 0, -1, 64, 128.5, '768'):
candidate = copy.deepcopy(self.saved)
candidate['deployment']['resources']['memory_mb'] = value
self.assertIsNone(resources.changes(self.saved, candidate, CONFIG))
self.assertIsNone(resources.changes(self.saved, copy.deepcopy(self.saved), CONFIG))
def test_cpu_limit_profile_stays_a_cpu_limit(self):
config = CONFIG.replace(b'cores: 2', b'cpulimit: 2')
self.assertEqual(resources.changes(self.saved, self.candidate, config), {'cpulimit': '3', 'memory': '768'})
def test_native_failure_leaves_saved_record_and_config_unchanged(self):
self.native_failure = RuntimeError('native API rejected edit')
with self.assertRaisesRegex(RuntimeError, 'rejected'):
self.apply()
self.assertEqual(self.file.read_bytes(), CONFIG)
self.assertEqual(instances.read(self.root, 115), self.saved)
def test_observation_failure_rolls_back_resource_values(self):
with patch.object(instances, 'observe', side_effect=RuntimeError('observation failed')):
with self.assertRaisesRegex(RuntimeError, 'observation failed'):
self.apply()
self.assertEqual(self.file.read_bytes(), CONFIG)
self.assertEqual(instances.read(self.root, 115), self.saved)
self.assertEqual(len(self.calls), 2)
def test_record_write_failure_rolls_back_resources(self):
real_write = instances.write
failed = False
def write(path, value):
nonlocal failed
if Path(path) == instances.location(self.root, 115) and not failed:
failed = True
raise OSError('record could not be saved')
real_write(path, value)
with patch.object(instances, 'write', side_effect=write):
with self.assertRaisesRegex(OSError, 'could not be saved'):
self.apply()
self.assertEqual(self.file.read_bytes(), CONFIG)
self.assertEqual(instances.read(self.root, 115), self.saved)
def test_configuration_drift_is_rejected_before_native_set(self):
self.file.write_bytes(CONFIG.replace(b'memory: 512', b'memory: 1024'))
with self.assertRaisesRegex(ValueError, 'changed'):
self.apply()
self.assertEqual(self.calls, [])
def test_repeated_environment_entries_are_all_preserved_by_the_guard(self):
changed = CONFIG.replace(b'cores: 2', b'cores: 3').replace(b'memory: 512', b'memory: 768')
self.assertTrue(resources.expected_config(CONFIG, changed, self.params))
self.assertFalse(resources.expected_config(CONFIG, changed.replace(b'A=one', b'A=foreign'), self.params))
def test_a_concurrent_unrelated_change_is_not_overwritten_during_rollback(self):
def observe(*args):
self.file.write_bytes(self.file.read_bytes().replace(b'A=one', b'A=foreign'))
raise RuntimeError('concurrent edit')
with patch.object(instances, 'observe', side_effect=observe), patch.object(resources, 'msg_warn') as warn:
with self.assertRaisesRegex(RuntimeError, 'concurrent edit'):
self.apply()
self.assertIn(b'A=foreign', self.file.read_bytes())
self.assertEqual(len(self.calls), 1)
warn.assert_called_once()
self.assertEqual(instances.read(self.root, 115), self.saved)
def test_a_carried_record_error_does_not_undo_the_successful_edit(self):
with patch.object(oci_carried_record, 'carry', side_effect=RuntimeError('copy unavailable')):
self.apply()
self.assertEqual(instances.read(self.root, 115)['deployment']['resources']['memory_mb'], 768)
self.assertEqual(len(self.calls), 1)
class CachedArchiveTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.path = Path(self.temp.name)/'original-installation.tar'
self.path.write_bytes(b'fixture verified by stub')
self.desired = record()
self.image = {'manifest_digest': DIGEST}
self.candidate = dict(self.image, registry_digest=DIGEST)
def resolve(self, paths, image=None):
def run(args, error, capture=False):
self.assertNotEqual(args[0], 'skopeo', 'No layers should be downloaded')
return json.dumps(self.candidate) if '-c' in args else b''
with patch.object(update, 'run_quiet', side_effect=run), \
patch.object(update, 'image_from_archive', return_value=image or self.image), \
patch.object(instances, 'command') as command:
result = update.resolve_archive(self.desired, b'arch: amd64\n', local_archives=paths)
command.assert_not_called()
return result
def test_original_installation_archive_is_reused_without_update_cache_lookup(self):
self.assertEqual(self.resolve([self.path]), (self.path, DIGEST))
def test_invalid_original_cache_is_kept_and_a_second_matching_archive_is_reused(self):
other = self.path.with_name('other.tar')
other.write_bytes(b'second fixture')
def image(path):
return {'manifest_digest': NEW if str(path) == str(self.path) else DIGEST}
with patch.object(update, 'run_quiet', side_effect=lambda args, *a, **kw: json.dumps(self.candidate) if '-c' in args else b''), \
patch.object(update, 'image_from_archive', side_effect=image), \
patch.object(instances, 'command') as command:
result = update.resolve_archive(self.desired, b'arch: amd64\n', local_archives=[self.path, other])
self.assertEqual(result, (other, DIGEST))
self.assertTrue(self.path.exists())
command.assert_not_called()
def test_a_new_digest_downloads_once_instead_of_using_an_old_archive(self):
candidate = {'manifest_digest': NEW, 'registry_digest': NEW}
cached = self.path.with_name('update.tar')
copies = []
def run(args, *a, **kwargs):
if '-c' in args:
return json.dumps(candidate)
if args[0] == 'skopeo':
copies.append(args)
Path(args[-1].removeprefix('oci-archive:')).write_bytes(b'new verified fixture')
return b''
def image(path):
return {'manifest_digest': DIGEST if str(path) == str(self.path) else NEW}
with patch.object(update, 'run_quiet', side_effect=run), \
patch.object(update, 'image_from_archive', side_effect=image), \
patch.object(instances, 'command', return_value=str(cached).encode()), \
patch.object(update.work_backup, 'available', return_value=1024**4), \
patch.object(update.transaction, 'log'):
result = update.resolve_archive(self.desired, b'arch: amd64\n', local_archives=[self.path])
self.assertEqual(result, (cached, NEW))
self.assertEqual(len(copies), 1)
self.assertIn('docker://example/app@'+NEW, copies[0])
self.assertEqual(self.path.read_bytes(), b'fixture verified by stub')
def test_a_broken_or_symlink_original_cache_does_not_hide_a_valid_update_cache(self):
cached = self.path.with_name('update.tar')
cached.write_bytes(b'valid update fixture')
link = self.path.with_name('link.tar')
link.symlink_to(self.path)
verified = []
def run(args, *a, **kwargs):
if '-c' in args:
return json.dumps(self.candidate)
self.assertNotEqual(args[0], 'skopeo')
verified.append(Path(args[-1]))
if Path(args[-1]) == self.path:
raise RuntimeError('corrupt archive')
return b''
with patch.object(update, 'run_quiet', side_effect=run), \
patch.object(update, 'image_from_archive', return_value=self.image), \
patch.object(instances, 'command', return_value=str(cached).encode()):
result = update.resolve_archive(self.desired, b'arch: amd64\n', local_archives=[link, self.path])
self.assertEqual(result, (cached, DIGEST))
self.assertNotIn(link, verified)
self.assertTrue(link.is_symlink())
self.assertTrue(self.path.exists())
class UpdateRoutingTests(unittest.TestCase):
def run_update(self, new_image=False, resource_edit=True):
saved = record()
candidate = copy.deepcopy(saved)
candidate['deployment']['resources']['memory_mb'] = 768
if not resource_edit:
candidate['deployment']['environment'][0]['value'] = 'changed'
proposal = {'operation': 'recreate', 'candidate': candidate}
import contextlib, oci_operation_notice
with patch.object(instances, 'locked', return_value=contextlib.nullcontext()), \
patch.object(instances, 'read', return_value=saved), \
patch.object(instances, 'command', return_value=CONFIG), \
patch.object(update.transaction, 'candidate_contract', return_value=candidate), \
patch.object(update.transaction, 'external_changes', return_value={}), \
patch.object(update.transaction, 'preflight'), \
patch.object(update, 'resolve_archive', return_value=(Path('/new.tar') if new_image or not resource_edit else None, NEW if new_image else DIGEST)) as resolve, \
patch.object(resources, 'apply') as apply_resources, \
patch.object(update.transaction, 'apply') as apply_image, \
patch.object(oci_operation_notice, 'operation', return_value=contextlib.nullcontext()):
update.update(115, proposal=proposal)
return resolve, apply_resources, apply_image
def test_unchanged_image_routes_cpu_ram_edits_to_the_existing_guest(self):
resolve, native, image = self.run_update()
self.assertEqual(resolve.call_args.args[2]['manifest_digest'], DIGEST)
native.assert_called_once()
image.assert_not_called()
def test_a_new_image_still_uses_the_normal_verified_recreation(self):
_resolve, native, image = self.run_update(new_image=True)
native.assert_not_called()
image.assert_called_once()
def test_other_modifications_offer_the_recorded_archive_to_the_resolver(self):
resolve, native, image = self.run_update(resource_edit=False)
self.assertIsNone(resolve.call_args.args[2])
self.assertEqual(resolve.call_args.kwargs['local_archives'], ['/missing/deleted-image.tar'])
native.assert_not_called()
image.assert_called_once()
if __name__ == '__main__':
unittest.main()