diff --git a/confluent_server/confluent/syncfiles.py b/confluent_server/confluent/syncfiles.py index 50ac8e52..a57c7d05 100644 --- a/confluent_server/confluent/syncfiles.py +++ b/confluent_server/confluent/syncfiles.py @@ -191,12 +191,33 @@ async def sync_list_to_node(sl, node, suffixes, peerip=None): for ent in sl.appendoncemap: stage_ent(sl.appendoncemap, ent, os.path.join(targdir, suffixes['appendonce']), True) + filelist = [] + for root, dirs, files in os.walk(targdir): + if not dirs and not files and root != targdir: + # otherwise empty directories have to be named to be synced + filelist.append(os.path.relpath(root, targdir)) + for filename in files: + filelist.append( + os.path.relpath(os.path.join(root, filename), targdir)) await sshutil.prep_ssh_key('/etc/confluent/ssh/automation') targip = node if peerip: targip = peerip + # Naming every file (--files-from implies --relative) together with + # --no-implied-dirs confines the preservation flags to the content + # actually being synchronized. Without it, the parent directories + # traversed on the way get the attributes of the staging copy, which + # is what forced preservation to be rolled back in the past. output, stderr = await util.check_output( - 'rsync', '-rvLD', targdir + '/', 'root@[{}]:/'.format(targip)) + 'rsync', '-vLpgotDA', '--no-implied-dirs', '--files-from=-', + # content out of confluent's own storage is owned by the account + # the daemon runs as, which normally does not exist on the node + # and would arrive as a meaningless numeric id; every other owner + # is still matched by name as usual + '--usermap={}:root'.format(os.getuid()), + '--groupmap={}:root'.format(os.getgid()), + targdir + '/', 'root@[{}]:/'.format(targip), + input=''.join([x + '\n' for x in filelist]).encode('utf8')) except subprocess.CalledProcessError as e: unreadablefiles = [] for root, dirnames, filenames in os.walk(targdir): @@ -251,6 +272,15 @@ def stage_ent(currmap, ent, targdir, appendexist=False): def mkpathorlink(source, destination, appendexist=False): if os.path.isdir(source): mkdirp(destination) + # Files are staged as symlinks, so rsync reads their attributes + # through to the real file, but a directory is staged as a directory + # and has to be given the source attributes to pass them along. + srcstat = os.stat(source) + shutil.copystat(source, destination) + try: + os.chown(destination, srcstat.st_uid, srcstat.st_gid) + except OSError: # lacking CAP_CHOWN, leave it owned by the daemon + pass for ent in os.listdir(source): currsrc = os.path.join(source, ent) currdst = os.path.join(destination, ent) diff --git a/confluent_server/confluent/util.py b/confluent_server/confluent/util.py index a1a1be07..0f3ae2c8 100644 --- a/confluent_server/confluent/util.py +++ b/confluent_server/confluent/util.py @@ -63,12 +63,13 @@ async def check_call(*cmd, **kwargs): -async def check_output(*cmd): +async def check_output(*cmd, input=None): if len(cmd) == 1 and isinstance(cmd[0], (list, tuple)): cmd = cmd[0] process = await asyncio.create_subprocess_exec( - *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) - stdout, stderr = await process.communicate() + *cmd, stdin=asyncio.subprocess.PIPE if input is not None else None, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) + stdout, stderr = await process.communicate(input) retcode = process.returncode if retcode: raise subprocess.CalledProcessError(