From c0bc33e494c35ec586ecec77e00f7a36dda4b2bd Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sat, 1 Aug 2026 23:02:47 +0200 Subject: [PATCH] Report syncfiles failures instead of discarding them get_syncresult() caught the sync task's exception, logged a repr server side and returned 200 OK with a null body. The node then called .get('options') on that null resulted in: c1: 'NoneType' object has no attribute 'get' and syncfileclient still exited 0 as if syncing had succeeded. Return the error to the requestor as a 500 with an error payload. On the node, unwrap the body that grab_url_with_status raises for a non-success status, print it once and exit non-zero. Only a failure the server deliberately reported for this sync is terminal. Anything else, such as a dropped connection, is re-raised so the existing retry loop handles it as before. The same case now reports c1: Error performing syncfiles: Syncing failed due to unreadable files: /etc/dangling.conf c1: 'syncfileclient' exited with code 1 --- .../common/profile/scripts/syncfileclient | 34 ++++++++++++++++++- .../profiles/default/scripts/syncfileclient | 32 ++++++++++++++++- .../profiles/default/scripts/syncfileclient | 32 ++++++++++++++++- confluent_server/confluent/syncfiles.py | 4 ++- 4 files changed, 98 insertions(+), 4 deletions(-) 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/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_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