From 1a9613f22ebc48b97f808108912c1b76110ac42c Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 02:19:13 +0200 Subject: [PATCH 01/10] Hash profile files in larger chunks The asyncio port added an await between every 2048 byte read, which roughly doubled the cost of hashing. imgutil runs entire packed images through this, and the server pays it on rebase and media import. Read a megabyte per iteration instead. That still yields hundreds of times per gigabyte, so the event loop stays responsive, and sha512 is independent of the read size, so existing manifests remain valid. --- confluent_server/confluent/osimage.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/confluent_server/confluent/osimage.py b/confluent_server/confluent/osimage.py index 17d53c3a..33062708 100644 --- a/confluent_server/confluent/osimage.py +++ b/confluent_server/confluent/osimage.py @@ -1041,11 +1041,11 @@ def copy_file(src, dst): async def get_hash(fname): currhash = hashlib.sha512() with open(fname, 'rb') as currf: - currd = currf.read(2048) + currd = currf.read(1048576) await asyncio.sleep(0) while currd: currhash.update(currd) - currd = currf.read(2048) + currd = currf.read(1048576) await asyncio.sleep(0) return currhash.hexdigest() From aba564914f790ba8b4a84be141205703d61e588e Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 02:19:23 +0200 Subject: [PATCH 02/10] Limit imgutil manifest hashes to the profile source capture and pack hash the whole profile directory, which by that point holds rootimg.sfs, the kernel and the distribution initramfs. rebase only ever looks up entries that came from the profile source directory, so the image blobs cost gigabytes of hashing for nothing. Pass the source directory as the filter, as generate_stock_profiles already does. Older manifests keep working, since rebase reads their entries with a default. --- imgutil/imgutil | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/imgutil/imgutil b/imgutil/imgutil index ff8d68b7..1fb62b96 100644 --- a/imgutil/imgutil +++ b/imgutil/imgutil @@ -267,7 +267,7 @@ async def capture_remote(args): indir = '{}/profiles/default'.format(confdir) if os.path.exists(indir): copy_tree(indir, outdir) - hmap = await osimage.get_hashes(outdir) + hmap = await osimage.get_hashes(outdir, indir) with open('{0}/manifest.yaml'.format(outdir), 'w') as yout: yout.write('# This manifest enables rebase to know original source of profile data and if any customizations have been done\n') manifestdata = {'distdir': indir, 'disthashes': hmap} @@ -1663,7 +1663,7 @@ async def pack_image(args): indir = '{}/profiles/default'.format(confdir) if os.path.exists(indir): copy_tree(indir, outdir) - hmap = await osimage.get_hashes(outdir) + hmap = await osimage.get_hashes(outdir, indir) with open('{0}/manifest.yaml'.format(outdir), 'w') as yout: yout.write('# This manifest enables rebase to know original source of profile data and if any customizations have been done\n') manifestdata = {'distdir': indir, 'disthashes': hmap} From 38be080bec1f3f92d6e56cd665cbf7917d2be057 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 02:19:23 +0200 Subject: [PATCH 03/10] Accept a command list in check_call check_output unwraps a single list argument, check_call never did, so callers passing a list hit a TypeError out of create_subprocess_exec. Two callers do: the genisoimage run behind Windows profile imports, where an except Exception swallows the failure and the boot.iso is silently missing, and the nodeconfig run in discovery, which takes out automatic node configuration on discovery outright. --- confluent_server/confluent/util.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/confluent_server/confluent/util.py b/confluent_server/confluent/util.py index 339db356..a1a1be07 100644 --- a/confluent_server/confluent/util.py +++ b/confluent_server/confluent/util.py @@ -52,6 +52,8 @@ def mkdirp(path, mode=0o777): async def check_call(*cmd, **kwargs): + if len(cmd) == 1 and isinstance(cmd[0], (list, tuple)): + cmd = cmd[0] subproc = await asyncio.create_subprocess_exec(*cmd, **kwargs) rc = await subproc.wait() if rc != 0: From 4e9052012f033da164152ddf7a3818e98fbd9e7e Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 02:58:52 +0200 Subject: [PATCH 04/10] Keep imgutil pack and capture synchronous Both functions became coroutines solely to await one get_hashes call, but their bodies are long stretches of blocking work: mksquashfs, the encrypt_image copy loop, rsync, ssh and osdeploy. From Python 3.11 on, asyncio.run installs a SIGINT handler that cancels the main task and returns rather than raising, so an interrupt is only noticed at the next await. Interrupting a pack during mksquashfs surfaced as a CalledProcessError from the dying child instead of a KeyboardInterrupt, and with a base profile, where nothing is ever awaited, pack carried on and published the profile before exiting. Run the loop only around the call that needs it. --- imgutil/imgutil | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/imgutil/imgutil b/imgutil/imgutil index 1fb62b96..00500bbc 100644 --- a/imgutil/imgutil +++ b/imgutil/imgutil @@ -197,7 +197,7 @@ def build_el_boot_tree(targpath): gather_bootloader(targpath) -async def capture_remote(args): +def capture_remote(args): targ = args.node outdir = args.profilename os.umask(0o022) @@ -267,7 +267,7 @@ async def capture_remote(args): indir = '{}/profiles/default'.format(confdir) if os.path.exists(indir): copy_tree(indir, outdir) - hmap = await osimage.get_hashes(outdir, indir) + hmap = asyncio.run(osimage.get_hashes(outdir, indir)) with open('{0}/manifest.yaml'.format(outdir), 'w') as yout: yout.write('# This manifest enables rebase to know original source of profile data and if any customizations have been done\n') manifestdata = {'distdir': indir, 'disthashes': hmap} @@ -1015,13 +1015,13 @@ def main(): if args.subcommand == 'build': build_root(args) elif args.subcommand == 'capture': - asyncio.run(capture_remote(args)) + capture_remote(args) elif args.subcommand == 'unpack': unpack_image(args) elif args.subcommand == 'exec': exec_root(args) elif args.subcommand == 'pack': - asyncio.run(pack_image(args)) + pack_image(args) else: parser.print_usage() @@ -1544,7 +1544,7 @@ def recursecp(source, targ): shutil.copy2(source, targ) -async def pack_image(args): +def pack_image(args): outdir = args.profilename if '/' in outdir: raise Exception('Full path not supported, supply only the profile name\n') @@ -1663,7 +1663,7 @@ async def pack_image(args): indir = '{}/profiles/default'.format(confdir) if os.path.exists(indir): copy_tree(indir, outdir) - hmap = await osimage.get_hashes(outdir, indir) + hmap = asyncio.run(osimage.get_hashes(outdir, indir)) with open('{0}/manifest.yaml'.format(outdir), 'w') as yout: yout.write('# This manifest enables rebase to know original source of profile data and if any customizations have been done\n') manifestdata = {'distdir': indir, 'disthashes': hmap} From 547ecf16d4b34ffb1f8ad204d0e0cdf55aaba3e8 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 03:10:31 +0200 Subject: [PATCH 05/10] Release the crypt device if encrypt_image is interrupted Nothing unwound the loop device and dm-crypt mapping when the copy loop raised, so interrupting a pack stranded both, still holding the profile's rootimg.sfs. Tear them down from a finally. The retry loop moves with them, so also honour its tries counter, as unpack_image already does; spinning forever inside a finally would hang the interrupt it is meant to clean up after. A bounded retry loop can also give up, and the detach that follows would then fail with EBUSY and, raising from a finally, replace the exception that brought us here. Warn and leave both in place instead. --- imgutil/imgutil | 51 +++++++++++++++++++++++++++++-------------------- 1 file changed, 30 insertions(+), 21 deletions(-) diff --git a/imgutil/imgutil b/imgutil/imgutil index 00500bbc..96d01083 100644 --- a/imgutil/imgutil +++ b/imgutil/imgutil @@ -385,28 +385,37 @@ def encrypt_image(plainfile, cryptfile, keyfile): neededblocks += 1 loopdev = subprocess.check_output(['losetup', '-f']).decode('utf8').strip() subprocess.check_call(['losetup', loopdev, cryptfile]) - subprocess.check_call(['dmsetup', 'create', dmname, '--table', '0 {} crypt aes-xts-plain64 {} 0 {} 8'.format(neededblocks, key, loopdev)]) - subprocess.check_call(['dmsetup', 'mknodes', dmname]) - with open('/dev/mapper/{}'.format(dmname), 'wb') as cryptout: - with open(plainfile, 'rb+') as plainin: - lastoffset = 0 - chunk = plainin.read(2097152) - while chunk: - fallocate(plainin.fileno(), FALLOC_FL_KEEP_SIZE|FALLOC_FL_PUNCH_HOLE, lastoffset, len(chunk)) - lastoffset = plainin.tell() - cryptout.write(chunk) + try: + subprocess.check_call(['dmsetup', 'create', dmname, '--table', '0 {} crypt aes-xts-plain64 {} 0 {} 8'.format(neededblocks, key, loopdev)]) + subprocess.check_call(['dmsetup', 'mknodes', dmname]) + with open('/dev/mapper/{}'.format(dmname), 'wb') as cryptout: + with open(plainfile, 'rb+') as plainin: + lastoffset = 0 chunk = plainin.read(2097152) - mounted = True - tries = 30 - time.sleep(0.1) - while mounted: - tries -= 1 - try: - subprocess.check_call(['dmsetup', 'remove', dmname]) - mounted = False - except subprocess.CalledProcessError: - time.sleep(0.1) - subprocess.check_call(['losetup', '-d', loopdev]) + while chunk: + fallocate(plainin.fileno(), FALLOC_FL_KEEP_SIZE|FALLOC_FL_PUNCH_HOLE, lastoffset, len(chunk)) + lastoffset = plainin.tell() + cryptout.write(chunk) + chunk = plainin.read(2097152) + finally: + mounted = True + tries = 30 + time.sleep(0.1) + while mounted and tries: + tries -= 1 + try: + subprocess.check_call(['dmsetup', 'remove', dmname]) + mounted = False + except subprocess.CalledProcessError: + time.sleep(0.1) + if mounted: + sys.stderr.write( + 'Warning: unable to remove {0}, it and {1} are left behind\n'.format(dmname, loopdev)) + else: + try: + subprocess.check_call(['losetup', '-d', loopdev]) + except subprocess.CalledProcessError: + sys.stderr.write('Warning: unable to detach {0}\n'.format(loopdev)) oum = os.umask(0o077) with open(keyfile, 'w') as keyout: keyout.write('aes-xts-plain64\n{}\n'.format(key)) From f2c74b0be37004788215efa9465164182d737caf Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 06:52:37 +0200 Subject: [PATCH 06/10] Fingerprint installation media off the event loop scan_iso walks an entire ISO with blocking libarchive reads, yielding only once per entry, and the header-sum branch of fingerprint reads the whole file with no yield at all. Both run in the daemon, reached from MediaImporter.init on every fingerprint and importing request. The scan costs about 8us per entry and is indifferent to media size, since libarchive seeks past file data rather than reading it: measured at 80ms for 10k entries whether the image is 0.2 GB or 8.8 GB, and at 310ms for 40k. The header-sum branch is the one that scales with size, reading a multi-gigabyte image end to end. Make the pair plain functions and hand them to a thread instead. --- confluent_server/confluent/osimage.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/confluent_server/confluent/osimage.py b/confluent_server/confluent/osimage.py index 33062708..fda01d82 100644 --- a/confluent_server/confluent/osimage.py +++ b/confluent_server/confluent/osimage.py @@ -834,7 +834,7 @@ def fingerprint_initramfs(archive): return None -async def scan_iso(archive): +def scan_iso(archive): scanudf = False filesizes = {} filecontents = {} @@ -845,7 +845,6 @@ async def scan_iso(archive): for ent in reader: if str(ent).endswith('TRANS.TBL'): continue - await asyncio.sleep(0) filesizes[str(ent)] = ent.size if str(ent) == 'README.TXT': readmecontents = b'' @@ -909,13 +908,13 @@ def parse_bfb(archive): archive.seek(currsize, os.SEEK_CUR) return None -async def fingerprint(archive): +def fingerprint(archive): archive.seek(0) header = archive.read(32768) archive.seek(32769) if archive.read(6) == b'CD001\x01': # ISO image - isoinfo = await scan_iso(archive) + isoinfo = scan_iso(archive) name = None for fun in globals(): if fun.startswith('check_'): @@ -947,7 +946,7 @@ async def import_image(filename, callback, backend=False, mfd=None, custtargpath archive = os.fdopen(int(mfd), 'rb') else: archive = open(filename, 'rb') - identity = await fingerprint(archive) + identity = await asyncio.to_thread(fingerprint, archive) if not identity: return -1 identity, imginfo, funname = identity @@ -1207,7 +1206,7 @@ class MediaImporter(object): else: medfile = open(media, 'rb') try: - identity = await fingerprint(medfile) + identity = await asyncio.to_thread(fingerprint, medfile) finally: if not self.medfile: medfile.close() From b081c17b5582bb911e3a7f99d6b2771f58273d34 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 06:52:55 +0200 Subject: [PATCH 07/10] Let the import drain loop accumulate a line The loop that drains the importer's remaining output reads a byte at a time but clears currline on every iteration, one level out from where the earlier loop clears it. currline is therefore never longer than a single byte, so the percentage and ERROR: branches can never match and the tail of an import is silently discarded. Clear it only once a line has been consumed, as the earlier loop does. --- confluent_server/confluent/osimage.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/confluent_server/confluent/osimage.py b/confluent_server/confluent/osimage.py index fda01d82..8f743537 100644 --- a/confluent_server/confluent/osimage.py +++ b/confluent_server/confluent/osimage.py @@ -1301,7 +1301,7 @@ class MediaImporter(object): self.error = self.error.decode('utf8') self.phase = 'error' return - currline = b'' + currline = b'' a = await wkr.stdout.read(1) if self.oscategory: defprofile = '/opt/confluent/lib/osdeploy/{0}'.format( From 3e7da14a9a5bf8a6dd739d424b964eb7efe6cabd Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 17:02:17 +0200 Subject: [PATCH 08/10] Test the import drain loops for an error before a percentage Both loops that read the importer's output test for a percentage first, so an ERROR: line whose text carries a % takes the percentage branch and float() raises instead of the error being reported. The import target name can carry one too, and that one is user supplied. importmedia runs as a bare task, so the exception is swallowed and the client polls a phase that never advances. Test for ERROR: first and treat an unparsable percentage as no percentage. Set percent on the error path of the second loop as well, as the first already does. --- confluent_server/confluent/osimage.py | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/confluent_server/confluent/osimage.py b/confluent_server/confluent/osimage.py index 8f743537..d1f0ae61 100644 --- a/confluent_server/confluent/osimage.py +++ b/confluent_server/confluent/osimage.py @@ -1275,32 +1275,37 @@ class MediaImporter(object): nb = await wkr.stdout.read(128) currline += nb if b'\r' in currline: - if b'%' in currline: - val = currline.split(b'%')[0].strip() - if val: - self.percent = float(val) - elif b'ERROR:' in currline: + if b'ERROR:' in currline: self.error = currline.replace(b'ERROR:', b'') if not isinstance(self.error, str): self.error = self.error.decode('utf8') self.phase = 'error' self.percent = 100.0 return + elif b'%' in currline: + val = currline.split(b'%')[0].strip() + try: + self.percent = float(val) + except ValueError: + pass currline = b'' a = await wkr.stdout.read(1) while a: currline += a if b'\r' in currline: - if b'%' in currline: - val = currline.split(b'%')[0].strip() - if val: - self.percent = float(val) - elif b'ERROR:' in currline: + if b'ERROR:' in currline: self.error = currline.replace(b'ERROR:', b'') if not isinstance(self.error, str): self.error = self.error.decode('utf8') self.phase = 'error' + self.percent = 100.0 return + elif b'%' in currline: + val = currline.split(b'%')[0].strip() + try: + self.percent = float(val) + except ValueError: + pass currline = b'' a = await wkr.stdout.read(1) if self.oscategory: From 9956845009dcf51248607d3cd9bac448929a46f9 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 06:52:55 +0200 Subject: [PATCH 09/10] Do not block the import poll loop with time.sleep osimport polls import progress from a coroutine, so a blocking sleep between reads stalls the whole client loop. It was the only use of time in the script, so the import goes with it. --- confluent_server/bin/osdeploy | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/confluent_server/bin/osdeploy b/confluent_server/bin/osdeploy index 9c5af002..c2ad0323 100644 --- a/confluent_server/bin/osdeploy +++ b/confluent_server/bin/osdeploy @@ -10,7 +10,6 @@ import os.path import pwd import shutil import sys -import time path = os.path.dirname(os.path.realpath(__file__)) path = os.path.realpath(os.path.join(path, '..', 'lib', 'python')) @@ -597,7 +596,7 @@ async def osimport(imagefile, checkonly=False, custname=None): sys.stdout.flush() else: print(repr(rsp)) - time.sleep(0.5) + await asyncio.sleep(0.5) finally: if shortname: async for x in c.delete('/deployment/importing/{0}'.format(shortname)): From 4f112fb78d80cd3f0224e065a5e673be990e1c90 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 06:52:55 +0200 Subject: [PATCH 10/10] Skip the profile manifest when the server libraries are absent confluent_imgutil does not depend on confluent_server, and the yaml import is optional too, yet both capture and pack dereference osimage and yaml unconditionally when writing manifest.yaml. With confluent_osdeploy present but the server absent that raises rather than producing a profile. Guard the manifest on both being importable and say so, since rebase is what the manifest exists for. The yaml fallback now binds None instead of leaving the name undefined. The two call sites carried the manifest write verbatim in both, so fold them into one function rather than duplicate the guard as well. --- imgutil/imgutil | 26 +++++++++++++++----------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/imgutil/imgutil b/imgutil/imgutil index 96d01083..5f0ccfb2 100644 --- a/imgutil/imgutil +++ b/imgutil/imgutil @@ -26,7 +26,7 @@ import time try: import yaml except ImportError: - pass + yaml = None path = os.path.dirname(os.path.realpath(__file__)) path = os.path.realpath(os.path.join(path, '..', 'lib', 'python')) if path.startswith('/opt'): @@ -197,6 +197,18 @@ def build_el_boot_tree(targpath): gather_bootloader(targpath) +def write_manifest(outdir, indir): + if not (osimage and yaml): + sys.stderr.write('Warning: confluent server libraries unavailable, skipping manifest.yaml, ' + 'osdeploy rebase will not work for this profile\n') + return + hmap = asyncio.run(osimage.get_hashes(outdir, indir)) + with open('{0}/manifest.yaml'.format(outdir), 'w') as yout: + yout.write('# This manifest enables rebase to know original source of profile data and if any customizations have been done\n') + manifestdata = {'distdir': indir, 'disthashes': hmap} + yout.write(yaml.dump(manifestdata, default_flow_style=False)) + + def capture_remote(args): targ = args.node outdir = args.profilename @@ -267,11 +279,7 @@ def capture_remote(args): indir = '{}/profiles/default'.format(confdir) if os.path.exists(indir): copy_tree(indir, outdir) - hmap = asyncio.run(osimage.get_hashes(outdir, indir)) - with open('{0}/manifest.yaml'.format(outdir), 'w') as yout: - yout.write('# This manifest enables rebase to know original source of profile data and if any customizations have been done\n') - manifestdata = {'distdir': indir, 'disthashes': hmap} - yout.write(yaml.dump(manifestdata, default_flow_style=False)) + write_manifest(outdir, indir) label = '{0} {1} ({2})'.format(finfo['name'], finfo['version'], profname) with open(os.path.join(outdir, 'profile.yaml'), 'w') as profileout: profileout.write('label: {}\n'.format(label)) @@ -1672,11 +1680,7 @@ def pack_image(args): indir = '{}/profiles/default'.format(confdir) if os.path.exists(indir): copy_tree(indir, outdir) - hmap = asyncio.run(osimage.get_hashes(outdir, indir)) - with open('{0}/manifest.yaml'.format(outdir), 'w') as yout: - yout.write('# This manifest enables rebase to know original source of profile data and if any customizations have been done\n') - manifestdata = {'distdir': indir, 'disthashes': hmap} - yout.write(yaml.dump(manifestdata, default_flow_style=False)) + write_manifest(outdir, indir) tryupdate = True try: pwd.getpwnam('confluent')