From 5fc036a2b737341491fd2965e328c5f2a58e9ded Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sun, 12 Jul 2026 23:54:16 +0200 Subject: [PATCH] Await asynchronous configuration mutations --- confluent_server/confluent/collective/manager.py | 4 ++-- confluent_server/confluent/config/configmanager.py | 2 +- confluent_server/confluent/credserver.py | 4 ++-- .../confluent/plugins/configuration/attributes.py | 4 ++-- confluent_server/confluent/plugins/deployment/identimage.py | 3 +-- confluent_server/confluent/plugins/shell/ssh.py | 6 +++--- confluent_server/confluent/webauthn.py | 3 +-- 7 files changed, 12 insertions(+), 14 deletions(-) diff --git a/confluent_server/confluent/collective/manager.py b/confluent_server/confluent/collective/manager.py index d9a44828..4cdedeef 100644 --- a/confluent_server/confluent/collective/manager.py +++ b/confluent_server/confluent/collective/manager.py @@ -187,7 +187,7 @@ async def connect_to_leader(cert=None, name=None, leader=None, remote=None, isre await cfm.ConfigManager(tenant=None)._load_from_json(dbjson, sync=False) cfm.commit_clear() - except Exception: + except Exception as e: print(repr(e)) await cfm.stop_following() cfm.rollback_clear() @@ -895,7 +895,7 @@ async def check_managers(): targets = sorted(expandednoderanges[targets], key=availmanagers.get) if not targets: continue - c.set_node_attributes({node: {'collective.manager': {'value': targets[0]}}}) + await c.set_node_attributes({node: {'collective.manager': {'value': targets[0]}}}) availmanagers[targets[0]] += 1 await _assimilate_missing() failovercheck = None diff --git a/confluent_server/confluent/config/configmanager.py b/confluent_server/confluent/config/configmanager.py index dd5430a3..d5fef21a 100644 --- a/confluent_server/confluent/config/configmanager.py +++ b/confluent_server/confluent/config/configmanager.py @@ -3232,7 +3232,7 @@ async def dump_db_to_directory(location, password, redact=None, skipkeys=False, try: for tenant in os.listdir( os.path.join(ConfigManager._cfgdir, '/tenants/')): - tenant_data = ConfigManager(tenant=tenant)._dump_to_json(redact=redact) + tenant_data = await ConfigManager(tenant=tenant)._dump_to_json(redact=redact) with open(os.path.join(location, 'tenants', tenant, f'main.{format}'), 'wb' if format == 'json' else 'w') as cfgfile: if format == 'json': cfgfile.write(tenant_data) diff --git a/confluent_server/confluent/credserver.py b/confluent_server/confluent/credserver.py index f42b09a1..2697f3db 100644 --- a/confluent_server/confluent/credserver.py +++ b/confluent_server/confluent/credserver.py @@ -103,7 +103,7 @@ class CredServer(object): now = datetime.datetime.utcnow() expiry = datetime.datetime.strptime(apiarmed, "%Y-%m-%dT%H:%M:%SZ") if now > expiry: - self.cfm.set_node_attributes({nodename: {'deployment.apiarmed': ''}}) + await self.cfm.set_node_attributes({nodename: {'deployment.apiarmed': ''}}) client.close() return await cloop.sock_sendall(client, b'\x02\x20') @@ -133,7 +133,7 @@ class CredServer(object): await cloop.sock_recv(client, 2) # drain end of message await cloop.sock_sendall(client, b'\x05\x00') # report success if hmackey and apiarmed != 'continuous': - self.cfm.clear_node_attributes([nodename], ['secret.selfapiarmtoken']) + await self.cfm.clear_node_attributes([nodename], ['secret.selfapiarmtoken']) if apiarmed != 'continuous': disarm = {nodename: {'deployment.sealedapikey': '', 'deployment.apiarmed': ''}} finally: diff --git a/confluent_server/confluent/plugins/configuration/attributes.py b/confluent_server/confluent/plugins/configuration/attributes.py index c241619b..4e6ce50c 100644 --- a/confluent_server/confluent/plugins/configuration/attributes.py +++ b/confluent_server/confluent/plugins/configuration/attributes.py @@ -254,10 +254,10 @@ def yield_rename_resources(namemap, isnode): else: yield msg.RenamedResource(node, namemap[node]) -def update_locks(nodes, configmanager, inputdata): +async def update_locks(nodes, configmanager, inputdata): for node in nodes: updatestate = inputdata.inputbynode[node] - configmanager.set_node_attributes({node: {'deployment.lock': updatestate}}) + await configmanager.set_node_attributes({node: {'deployment.lock': updatestate}}) yield msg.DeploymentLock(node, updatestate) async def update_nodes(nodes, element, configmanager, inputdata): diff --git a/confluent_server/confluent/plugins/deployment/identimage.py b/confluent_server/confluent/plugins/deployment/identimage.py index 9801da1f..c8e7474e 100644 --- a/confluent_server/confluent/plugins/deployment/identimage.py +++ b/confluent_server/confluent/plugins/deployment/identimage.py @@ -46,7 +46,7 @@ async def create_ident_image(node, configmanager): tmpd = tempfile.mkdtemp() ident = { 'nodename': node } apikey = create_apikey() - configmanager.set_node_attributes({node: {'secret.selfapiarmtoken': apikey}}) + await configmanager.set_node_attributes({node: {'secret.selfapiarmtoken': apikey}}) ident['apitoken'] = apikey # This particular mechanism does not (yet) do anything smart with collective # It would be a reasonable enhancement to list all collective server addresses @@ -79,4 +79,3 @@ async def update(nodes, element, configmanager, inputdata): yield msg.CreatedResource( 'nodes/{0}/deployment/ident_image'.format(node)) - diff --git a/confluent_server/confluent/plugins/shell/ssh.py b/confluent_server/confluent/plugins/shell/ssh.py index 73186531..0c2e6a18 100644 --- a/confluent_server/confluent/plugins/shell/ssh.py +++ b/confluent_server/confluent/plugins/shell/ssh.py @@ -146,7 +146,7 @@ class SshShell(conapi.Console): if b'\r' in self.keyaction: action = self.keyaction.split(b'\r')[0] if action.lower() == b'accept': - self.nodeconfig.set_node_attributes( + await self.nodeconfig.set_node_attributes( {self.node: {self.keyattrname: self.candidatefprint}}) await self.datacallback('\r\n') @@ -157,10 +157,10 @@ class SshShell(conapi.Console): self.keyaction = b'' await self.datacallback('\r\nEnter "disconnect" or "accept": ') elif len(data) > 0: - self.datacallback(data) + await self.datacallback(data) elif self.inputmode == 0: while len(data) and data[0:1] == b'\x7f' and len(self.username): - self.datacallback('\b \b') # erase previously echoed value + await self.datacallback('\b \b') # erase previously echoed value self.username = self.username[:-1] data = data[1:] while len(data) and data[0:1] == b'\x7f': diff --git a/confluent_server/confluent/webauthn.py b/confluent_server/confluent/webauthn.py index 3cbf7c13..dbd1abb7 100644 --- a/confluent_server/confluent/webauthn.py +++ b/confluent_server/confluent/webauthn.py @@ -248,7 +248,7 @@ async def handle_api_request(url, req, username, cfm, reqbody, authorized): if url == '/registration_options': userinfo = cfm.get_user(username) if not userinfo: - cfm.create_user(username, role='Stub') + await cfm.create_user(username, role='Stub') userinfo = cfm.get_user(username) authid = userinfo.get('webauthid', None) if not authid: # TODO: index users by authid as well as name @@ -293,4 +293,3 @@ async def handle_api_request(url, req, username, cfm, reqbody, authorized): if rsp.get('verified', False): return json.dumps({'status': 'Success'}) -