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)): diff --git a/confluent_server/confluent/osimage.py b/confluent_server/confluent/osimage.py index 17d53c3a..d1f0ae61 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 @@ -1041,11 +1040,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() @@ -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() @@ -1276,33 +1275,38 @@ 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 - currline = b'' + 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: defprofile = '/opt/confluent/lib/osdeploy/{0}'.format( 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: diff --git a/imgutil/imgutil b/imgutil/imgutil index ff8d68b7..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,7 +197,19 @@ def build_el_boot_tree(targpath): gather_bootloader(targpath) -async def capture_remote(args): +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 os.umask(0o022) @@ -267,11 +279,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) - 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)) @@ -385,28 +393,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)) @@ -1015,13 +1032,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 +1561,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,11 +1680,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) - 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')