xCAT/confluent
2
0
Fork 0
mirror of https://github.com/xcat2/confluent.git synced 2026-08-03 07:57:02 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #258 from Obihoernchen/imgutil-async

Browse Source 
imgutil: fix async-port fallout in the image pack/capture path
This commit is contained in:
Jarrod Johnson
2026-07-28 08:45:18 -04:00
committed by GitHub
parent 46296d4751 4f112fb78d
commit 86e4603b82
4 changed files with 75 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
confluent_server/bin/osdeploy
+1 -2
View File
@@ -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)):
confluent_server/confluent/osimage.py
+23 -19
View File
@@ -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(
confluent_server/confluent/util.py
+2
View File
@@ -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:
imgutil/imgutil
+49 -36
View File
@@ -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')