diff --git a/confluent_osdeploy/common/profile/scripts/syncfileclient b/confluent_osdeploy/common/profile/scripts/syncfileclient index beaa3216..878dc36e 100644 --- a/confluent_osdeploy/common/profile/scripts/syncfileclient +++ b/confluent_osdeploy/common/profile/scripts/syncfileclient @@ -210,6 +210,24 @@ def appendonce(basepath, filename): with open(targname, 'ab') as targhdl: targhdl.write(thedata) +def syncerror(exc): + # grab_url_with_status raises with the raw response body. Return the + # server reported error text, or None for anything else, e.g. a dropped + # connection. + msg = exc.args[0] if exc.args else None + if isinstance(msg, bytes): + msg = msg.decode('utf8', 'replace') + if not isinstance(msg, str): + return None + try: + parsed = json.loads(msg) + except ValueError: + return None + if isinstance(parsed, dict) and 'error' in parsed: + return parsed['error'] + return None + + def synchronize(): tmpdir = tempfile.mkdtemp() appendoncedir = tempfile.mkdtemp() @@ -240,7 +258,17 @@ def synchronize(): lastrsp = '' while status != 204: time.sleep(1+(2*random.random())) - status, rsp = ac.grab_url_with_status('/confluent-api/self/remotesyncfiles') + try: + status, rsp = ac.grab_url_with_status('/confluent-api/self/remotesyncfiles') + except Exception as e: + errmsg = syncerror(e) + if errmsg is None: + # Not server reported, let the caller retry. + raise + sys.stderr.write( + 'Error performing syncfiles: {}\n'.format(errmsg)) + sys.stderr.flush() + return 500 if not isinstance(rsp, str): rsp = rsp.decode('utf8') if status == 200: @@ -268,6 +296,7 @@ def synchronize(): appendonce(appendoncedir, os.path.join(dirn[0], filen)) if lastrsp: lastrsp = json.loads(lastrsp) + if lastrsp: opts = lastrsp.get('options', {}) for fname in opts: uid = -1 @@ -306,5 +335,8 @@ if __name__ == '__main__': sys.stderr.write('\n') sys.stderr.flush() status = 300 + if status >= 400 and status != 503: + # Terminal failure; 503 (sync already running) stays retryable. + sys.exit(1) if status not in (204, 200): time.sleep((random.random()*3)+2) diff --git a/confluent_osdeploy/el10-diskless/profiles/default/scripts/syncfileclient b/confluent_osdeploy/el10-diskless/profiles/default/scripts/syncfileclient deleted file mode 120000 index 219244e6..00000000 --- a/confluent_osdeploy/el10-diskless/profiles/default/scripts/syncfileclient +++ /dev/null @@ -1 +0,0 @@ -../../../../el9-diskless/profiles/default/scripts/syncfileclient \ No newline at end of file diff --git a/confluent_osdeploy/el7-diskless/profiles/default/scripts/syncfileclient b/confluent_osdeploy/el7-diskless/profiles/default/scripts/syncfileclient index cca0f57d..37c7bed1 100644 --- a/confluent_osdeploy/el7-diskless/profiles/default/scripts/syncfileclient +++ b/confluent_osdeploy/el7-diskless/profiles/default/scripts/syncfileclient @@ -7,6 +7,7 @@ import os import shutil import pwd import grp +import sys try: from importlib.machinery import SourceFileLoader def load_source(mod, path): @@ -214,6 +215,24 @@ def appendonce(basepath, filename): with open(targname, 'ab') as targhdl: targhdl.write(thedata) +def syncerror(exc): + # grab_url_with_status raises with the raw response body. Return the + # server reported error text, or None for anything else, e.g. a dropped + # connection. + msg = exc.args[0] if exc.args else None + if isinstance(msg, bytes): + msg = msg.decode('utf8', 'replace') + if not isinstance(msg, str): + return None + try: + parsed = json.loads(msg) + except ValueError: + return None + if isinstance(parsed, dict) and 'error' in parsed: + return parsed['error'] + return None + + def synchronize(): tmpdir = tempfile.mkdtemp() appendoncedir = tempfile.mkdtemp() @@ -225,7 +244,17 @@ def synchronize(): lastrsp = '' while status != 204: time.sleep(2) - status, rsp = ac.grab_url_with_status('/confluent-api/self/remotesyncfiles') + try: + status, rsp = ac.grab_url_with_status('/confluent-api/self/remotesyncfiles') + except Exception as e: + errmsg = syncerror(e) + if errmsg is None: + # Not server reported, let the caller retry. + raise + sys.stderr.write( + 'Error performing syncfiles: {}\n'.format(errmsg)) + sys.stderr.flush() + return 500 if not isinstance(rsp, str): rsp = rsp.decode('utf8') if status == 200: @@ -253,6 +282,7 @@ def synchronize(): appendonce(appendoncedir, os.path.join(dirn[0], filen)) if lastrsp: lastrsp = json.loads(lastrsp) + if lastrsp: opts = lastrsp.get('options', {}) for fname in opts: uid = -1 diff --git a/confluent_osdeploy/el7/profiles/default/scripts/syncfileclient b/confluent_osdeploy/el7/profiles/default/scripts/syncfileclient index 02dbcc4d..17565570 100644 --- a/confluent_osdeploy/el7/profiles/default/scripts/syncfileclient +++ b/confluent_osdeploy/el7/profiles/default/scripts/syncfileclient @@ -7,6 +7,7 @@ import shutil import pwd import time import grp +import sys try: from importlib.machinery import SourceFileLoader def load_source(mod, path): @@ -214,6 +215,24 @@ def appendonce(basepath, filename): with open(targname, 'ab') as targhdl: targhdl.write(thedata) +def syncerror(exc): + # grab_url_with_status raises with the raw response body. Return the + # server reported error text, or None for anything else, e.g. a dropped + # connection. + msg = exc.args[0] if exc.args else None + if isinstance(msg, bytes): + msg = msg.decode('utf8', 'replace') + if not isinstance(msg, str): + return None + try: + parsed = json.loads(msg) + except ValueError: + return None + if isinstance(parsed, dict) and 'error' in parsed: + return parsed['error'] + return None + + def synchronize(): tmpdir = tempfile.mkdtemp() appendoncedir = tempfile.mkdtemp() @@ -225,7 +244,17 @@ def synchronize(): lastrsp = '' while status != 204: time.sleep(2) - status, rsp = ac.grab_url_with_status('/confluent-api/self/remotesyncfiles') + try: + status, rsp = ac.grab_url_with_status('/confluent-api/self/remotesyncfiles') + except Exception as e: + errmsg = syncerror(e) + if errmsg is None: + # Not server reported, let the caller retry. + raise + sys.stderr.write( + 'Error performing syncfiles: {}\n'.format(errmsg)) + sys.stderr.flush() + return 500 if not isinstance(rsp, str): rsp = rsp.decode('utf8') if status == 200: @@ -253,6 +282,7 @@ def synchronize(): appendonce(appendoncedir, os.path.join(dirn[0], filen)) if lastrsp: lastrsp = json.loads(lastrsp) + if lastrsp: opts = lastrsp.get('options', {}) for fname in opts: uid = -1 diff --git a/confluent_osdeploy/el8-diskless/profiles/default/scripts/syncfileclient b/confluent_osdeploy/el8-diskless/profiles/default/scripts/syncfileclient deleted file mode 100644 index beaa3216..00000000 --- a/confluent_osdeploy/el8-diskless/profiles/default/scripts/syncfileclient +++ /dev/null @@ -1,310 +0,0 @@ -#!/usr/bin/python3 -import random -import time -import subprocess -import importlib -import tempfile -import json -import os -import shutil -import pwd -import grp -import sys -from importlib.machinery import SourceFileLoader -try: - apiclient = SourceFileLoader('apiclient', '/opt/confluent/bin/apiclient').load_module() -except FileNotFoundError: - apiclient = SourceFileLoader('apiclient', '/etc/confluent/apiclient').load_module() - - -def partitionhostsline(line): - comment = '' - try: - cmdidx = line.index('#') - comment = line[cmdidx:] - line = line[:cmdidx].strip() - except ValueError: - pass - if not line: - return '', [], comment - ipaddr, names = line.split(maxsplit=1) - names = names.split() - return ipaddr, names, comment - -class HostMerger(object): - def __init__(self): - self.byip = {} - self.byname = {} - self.sourcelines = [] - self.targlines = [] - - def read_source(self, sourcefile): - with open(sourcefile, 'r') as hfile: - self.sourcelines = hfile.read().split('\n') - while not self.sourcelines[-1]: - self.sourcelines = self.sourcelines[:-1] - for x in range(len(self.sourcelines)): - line = self.sourcelines[x] - currip, names, comment = partitionhostsline(line) - if currip: - self.byip[currip] = x - for name in names: - self.byname[name] = x - - def read_target(self, targetfile): - with open(targetfile, 'r') as hfile: - lines = hfile.read().split('\n') - if not lines[-1]: - lines = lines[:-1] - for y in range(len(lines)): - line = lines[y] - currip, names, comment = partitionhostsline(line) - if currip in self.byip: - x = self.byip[currip] - if self.sourcelines[x] is None: - # have already consumed this enntry - continue - self.targlines.append(self.sourcelines[x]) - self.sourcelines[x] = None - continue - for name in names: - if name in self.byname: - x = self.byname[name] - if self.sourcelines[x] is None: - break - self.targlines.append(self.sourcelines[x]) - self.sourcelines[x] = None - break - else: - self.targlines.append(line) - - def write_out(self, targetfile): - while not self.targlines[-1]: - self.targlines = self.targlines[:-1] - if not self.targlines: - break - while not self.sourcelines[-1]: - self.sourcelines = self.sourcelines[:-1] - if not self.sourcelines: - break - with open(targetfile, 'w') as hosts: - for line in self.targlines: - hosts.write(line + '\n') - for line in self.sourcelines: - if line is not None: - hosts.write(line + '\n') - - -class CredMerger: - def __init__(self): - try: - with open('/etc/login.defs', 'r') as ldefs: - defs = ldefs.read().split('\n') - except FileNotFoundError: - defs = [] - lkup = {} - self.discardnames = {} - self.shadowednames = {} - for line in defs: - try: - line = line[:line.index('#')] - except ValueError: - pass - keyval = line.split() - if len(keyval) < 2: - continue - lkup[keyval[0]] = keyval[1] - self.uidmin = int(lkup.get('UID_MIN', 1000)) - self.uidmax = int(lkup.get('UID_MAX', 60000)) - self.gidmin = int(lkup.get('GID_MIN', 1000)) - self.gidmax = int(lkup.get('GID_MAX', 60000)) - self.shadowlines = None - - def read_passwd(self, source, targfile=False): - self.read_generic(source, self.uidmin, self.uidmax, targfile) - - def read_group(self, source, targfile=False): - self.read_generic(source, self.gidmin, self.gidmax, targfile) - - def read_generic(self, source, minid, maxid, targfile): - if targfile: - self.targdata = [] - else: - self.sourcedata = [] - with open(source, 'r') as inputfile: - for line in inputfile.read().split('\n'): - try: - name, _, uid, _ = line.split(':', 3) - uid = int(uid) - except ValueError: - continue - if targfile: - if uid < minid or uid > maxid: - self.targdata.append(line) - else: - self.discardnames[name] = 1 - else: - if name[0] in ('+', '#', '@'): - self.sourcedata.append(line) - elif uid >= minid and uid <= maxid: - self.sourcedata.append(line) - - def read_shadow(self, source): - self.shadowlines = [] - try: - with open(source, 'r') as inshadow: - for line in inshadow.read().split('\n'): - try: - name, _ = line.split(':' , 1) - except ValueError: - continue - if name in self.discardnames: - continue - self.shadowednames[name] = 1 - self.shadowlines.append(line) - except FileNotFoundError: - return - - def write_out(self, outfile): - with open(outfile, 'w') as targ: - for line in self.targdata: - targ.write(line + '\n') - for line in self.sourcedata: - targ.write(line + '\n') - if outfile == '/etc/passwd': - if self.shadowlines is None: - self.read_shadow('/etc/shadow') - with open('/etc/shadow', 'w') as shadout: - for line in self.shadowlines: - shadout.write(line + '\n') - for line in self.sourcedata: - name, _ = line.split(':', 1) - if name[0] in ('+', '#', '@'): - continue - if name in self.shadowednames: - continue - shadout.write(name + ':!:::::::\n') - if outfile == '/etc/group': - if self.shadowlines is None: - self.read_shadow('/etc/gshadow') - with open('/etc/gshadow', 'w') as shadout: - for line in self.shadowlines: - shadout.write(line + '\n') - for line in self.sourcedata: - name, _ = line.split(':' , 1) - if name in self.shadowednames: - continue - shadout.write(name + ':!::\n') - -def appendonce(basepath, filename): - with open(filename, 'rb') as filehdl: - thedata = filehdl.read() - targname = filename.replace(basepath, '') - try: - with open(targname, 'rb') as filehdl: - targdata = filehdl.read() - except IOError: - targdata = b'' - if thedata in targdata: - return - with open(targname, 'ab') as targhdl: - targhdl.write(thedata) - -def synchronize(): - tmpdir = tempfile.mkdtemp() - appendoncedir = tempfile.mkdtemp() - try: - ac = apiclient.HTTPSClient() - myips = [] - ipaddrs = subprocess.check_output(['ip', '-br', 'a']).split(b'\n') - for line in ipaddrs: - isa = line.split() - if len(isa) < 3 or isa[1] != b'UP': - continue - for addr in isa[2:]: - if addr.startswith(b'fe80::') or addr.startswith(b'169.254'): - continue - addr = addr.split(b'/')[0] - if not isinstance(addr, str): - addr = addr.decode('utf8') - myips.append(addr) - data = json.dumps({'merge': tmpdir, 'appendonce': appendoncedir, 'myips': myips}) - status, rsp = ac.grab_url_with_status('/confluent-api/self/remotesyncfiles', data) - if status >= 300: - sys.stderr.write("Error starting syncfiles - {}:\n".format(status)) - sys.stderr.write(rsp.decode('utf8')) - sys.stderr.write('\n') - sys.stderr.flush() - return status - if status == 202: - lastrsp = '' - while status != 204: - time.sleep(1+(2*random.random())) - status, rsp = ac.grab_url_with_status('/confluent-api/self/remotesyncfiles') - if not isinstance(rsp, str): - rsp = rsp.decode('utf8') - if status == 200: - lastrsp = rsp - pendpasswd = os.path.join(tmpdir, 'etc/passwd') - if os.path.exists(pendpasswd): - cm = CredMerger() - cm.read_passwd(pendpasswd, targfile=False) - cm.read_passwd('/etc/passwd', targfile=True) - cm.write_out('/etc/passwd') - pendgroup = os.path.join(tmpdir, 'etc/group') - if os.path.exists(pendgroup): - cm = CredMerger() - cm.read_group(pendgroup, targfile=False) - cm.read_group('/etc/group', targfile=True) - cm.write_out('/etc/group') - pendhosts = os.path.join(tmpdir, 'etc/hosts') - if os.path.exists(pendhosts): - cm = HostMerger() - cm.read_source(pendhosts) - cm.read_target('/etc/hosts') - cm.write_out('/etc/hosts') - for dirn in os.walk(appendoncedir): - for filen in dirn[2]: - appendonce(appendoncedir, os.path.join(dirn[0], filen)) - if lastrsp: - lastrsp = json.loads(lastrsp) - opts = lastrsp.get('options', {}) - for fname in opts: - uid = -1 - gid = -1 - for opt in opts[fname]: - if opt == 'owner': - try: - uid = pwd.getpwnam(opts[fname][opt]['name']).pw_uid - except KeyError: - try: - uid = opts[fname][opt]['id'] - except KeyError: - raise Exception(f"Unable to map owner of {fname}") - elif opt == 'group': - try: - gid = grp.getgrnam(opts[fname][opt]['name']).gr_gid - except KeyError: - gid = opts[fname][opt]['id'] - elif opt == 'permissions': - os.chmod(fname, int(opts[fname][opt], 8)) - if uid != -1 or gid != -1: - os.chown(fname, uid, gid) - return status - finally: - shutil.rmtree(tmpdir) - shutil.rmtree(appendoncedir) - - -if __name__ == '__main__': - status = 202 - while status not in (204, 200): - try: - status = synchronize() - except Exception as e: - sys.stderr.write(str(e)) - sys.stderr.write('\n') - sys.stderr.flush() - status = 300 - if status not in (204, 200): - time.sleep((random.random()*3)+2) diff --git a/confluent_server/confluent/syncfiles.py b/confluent_server/confluent/syncfiles.py index 259799b1..50ac8e52 100644 --- a/confluent_server/confluent/syncfiles.py +++ b/confluent_server/confluent/syncfiles.py @@ -356,7 +356,9 @@ def get_syncresult(nodename): try: result = syncrunners[nodename].result() except Exception as e: + # Report the failure rather than a success with no payload. print(repr(e)) - result = None + del syncrunners[nodename] + return 500, 'Error', {'error': str(e)} del syncrunners[nodename] return 200, 'OK', result