From 89d0fa81b98ea0da377bb285a09a8f28ae0a6da9 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sun, 12 Jul 2026 23:54:32 +0200 Subject: [PATCH] Fix asynchronous discovery call contracts --- confluent_server/confluent/discovery/core.py | 20 +++++++++---------- .../confluent/discovery/handlers/bmc.py | 6 +++--- .../discovery/handlers/redfishbmc.py | 5 +++-- .../confluent/discovery/handlers/smm.py | 4 ++-- .../confluent/discovery/handlers/smm3.py | 3 +-- .../confluent/discovery/handlers/tsm.py | 4 ++-- .../confluent/discovery/protocols/mdns.py | 2 +- .../confluent/discovery/protocols/ssdp.py | 6 ++---- confluent_server/confluent/networking/lldp.py | 5 ++--- .../confluent/networking/macmap.py | 16 ++++++++++++--- .../plugins/hardwaremanagement/affluent.py | 18 ++++++++--------- confluent_server/confluent/selfservice.py | 6 +++--- 12 files changed, 51 insertions(+), 44 deletions(-) diff --git a/confluent_server/confluent/discovery/core.py b/confluent_server/confluent/discovery/core.py index 0652bf30..ef56dd07 100644 --- a/confluent_server/confluent/discovery/core.py +++ b/confluent_server/confluent/discovery/core.py @@ -580,14 +580,14 @@ async def handle_api_request(configmanager, inputdata, operation, pathcomponents return (msg.KeyValueData({'rescan': 'started'}),) elif operation in ('update', 'create') and pathcomponents[:2] == ['discovery', 'subscriptions']: target = pathcomponents[2] - affluent.subscribe_discovery(target, configmanager, collective.get_myname()) + await affluent.subscribe_discovery(target, configmanager, collective.get_myname()) currsubs = get_subscriptions() currsubs[target] = {} save_subscriptions(currsubs) return (msg.KeyValueData({'status': 'subscribed'}),) elif operation == 'delete' and pathcomponents[:2] == ['discovery', 'subscriptions']: target = pathcomponents[2] - affluent.unsubscribe_discovery(target, configmanager, collective.get_myname()) + await affluent.unsubscribe_discovery(target, configmanager, collective.get_myname()) currsubs = get_subscriptions() if target in currsubs: del currsubs[target] @@ -597,7 +597,7 @@ async def handle_api_request(configmanager, inputdata, operation, pathcomponents if pathcomponents == ['discovery', 'register']: if 'addresses' not in inputdata: raise exc.InvalidArgumentException('Missing address in input') - return await register_remote_addrs(inputdata['addresses'], configmanager) + return register_remote_addrs(inputdata['addresses'], configmanager) if 'node' not in inputdata: raise exc.InvalidArgumentException('Missing node name in input') mac = _get_mac_from_query(pathcomponents) @@ -920,7 +920,7 @@ async def detected(info): else: policies = set([]) if policies & {'open', 'permissive'}: - cfg.set_node_attributes({nodename: {'id.uuid': info['uuid']}}) + await cfg.set_node_attributes({nodename: {'id.uuid': info['uuid']}}) return # already known, no need for more #TODO(jjohnson2): We might have to get UUID for certain searches... #for now defer probe until inside eval_node. We might not have @@ -975,7 +975,7 @@ async def get_chained_smm_name(nodename, cfg, handler, nl=None, checkswitch=True mycert = await handler.get_https_cert() if checkswitch: fprints = macmap.get_node_fingerprints(nodename, cfg) - for fprint in fprints: + async for fprint in fprints: if util.cert_matches(fprint[0], mycert): # ok we have a direct match, it is this node return nodename, fprint[1] @@ -1380,7 +1380,7 @@ async def eval_node(cfg, handler, info, nodename, manual=False): # The candidate nodename is the head of a chain, we must # validate the smm certificate by the switch fprints = macmap.get_node_fingerprints(nodename, cfg) - for fprint in fprints: + async for fprint in fprints: if util.cert_matches(fprint[0], await handler.get_https_cert()): if not await discover_node(cfg, handler, info, nodename, manual): @@ -1434,7 +1434,7 @@ async def discover_node(cfg, handler, info, nodename, manual): # in some product or another. curruuid = info.get('uuid', False) if 'pxe' in policies and info['handler'] == pxeh: - return do_pxe_discovery(cfg, handler, info, manual, nodename, policies) + return await do_pxe_discovery(cfg, handler, info, manual, nodename, policies) elif ('permissive' in policies and handler.https_supported and lastfp and not util.cert_matches(lastfp, await handler.get_https_cert()) and not manual): info['discofailure'] = 'fingerprint' @@ -1455,7 +1455,7 @@ async def discover_node(cfg, handler, info, nodename, manual): return False info['nodename'] = nodename if info['handler'] == pxeh: - return do_pxe_discovery(cfg, handler, info, manual, nodename, policies) + return await do_pxe_discovery(cfg, handler, info, manual, nodename, policies) elif manual or not util.cert_matches(lastfp, await handler.get_https_cert()): # only 'discover' if it is not the same as last time try: @@ -1554,7 +1554,7 @@ async def wait_for_connection(bmcaddr): continue await asyncio.sleep(1) -def do_pxe_discovery(cfg, handler, info, manual, nodename, policies): +async def do_pxe_discovery(cfg, handler, info, manual, nodename, policies): # use uuid based scheme in lieu of tls cert, ideally only # for stateless 'discovery' targets like pxe, where data does not # change @@ -1585,7 +1585,7 @@ def do_pxe_discovery(cfg, handler, info, manual, nodename, policies): for checkattr in attribs: checkval = currattrs.get(nodename, {}).get(checkattr, {}).get('value', None) if checkval != attribs[checkattr]: - cfg.set_node_attributes({nodename: attribs}) + await cfg.set_node_attributes({nodename: attribs}) break if info['uuid'] in known_pxe_uuids: return True diff --git a/confluent_server/confluent/discovery/handlers/bmc.py b/confluent_server/confluent/discovery/handlers/bmc.py index eedf9ab5..db083822 100644 --- a/confluent_server/confluent/discovery/handlers/bmc.py +++ b/confluent_server/confluent/discovery/handlers/bmc.py @@ -48,8 +48,8 @@ class NodeHandler(generic.NodeHandler): return # TODO(jjohnson2): probe serial number and uuid - def config(self, nodename, reset=False): - self._bmcconfig(nodename, reset) + async def config(self, nodename, reset=False): + await self._bmcconfig(nodename, reset) async def _bmcconfig(self, nodename, reset=False, customconfig=None, vc=None): # TODO(jjohnson2): set ip parameters, user/pass, alert cfg maybe @@ -174,7 +174,7 @@ class NodeHandler(generic.NodeHandler): ipv4_gateway=netconfig[ 'ipv4_gateway']) elif self.ipaddr.startswith('fe80::'): - cfg.set_node_attributes( + await cfg.set_node_attributes( {nodename: {'hardwaremanagement.manager': self.ipaddr}}) else: raise exc.TargetEndpointUnreachable( diff --git a/confluent_server/confluent/discovery/handlers/redfishbmc.py b/confluent_server/confluent/discovery/handlers/redfishbmc.py index 2fd8941d..4576798c 100644 --- a/confluent_server/confluent/discovery/handlers/redfishbmc.py +++ b/confluent_server/confluent/discovery/handlers/redfishbmc.py @@ -266,8 +266,9 @@ class NodeHandler(generic.NodeHandler): newip = newipinfo[-1][0] if ':' in newip: raise exc.NotImplementedException('IPv6 remote config TODO') - hifurls = await get_host_interface_urls(wc, self.mgrinfo(wc)) - mgtnicinfo = self.mgrinfo(wc)['EthernetInterfaces']['@odata.id'] + mgrinfo = await self.mgrinfo(wc) + hifurls = await get_host_interface_urls(wc, mgrinfo) + mgtnicinfo = mgrinfo['EthernetInterfaces']['@odata.id'] mgtnicinfo = await wc.grab_json_response(mgtnicinfo) mgtnics = [x['@odata.id'] for x in mgtnicinfo.get('Members', [])] actualnics = [] diff --git a/confluent_server/confluent/discovery/handlers/smm.py b/confluent_server/confluent/discovery/handlers/smm.py index 97b777d6..86b8ab65 100644 --- a/confluent_server/confluent/discovery/handlers/smm.py +++ b/confluent_server/confluent/discovery/handlers/smm.py @@ -99,7 +99,7 @@ class NodeHandler(bmchandler.NodeHandler): if smmip: smmip = smmip.split('/', 1)[0] if smmip and ':' not in smmip: - smmip = await asyncio.get_running_loop().getaddrinfo(smmip, 0)[0] + smmip = (await asyncio.get_running_loop().getaddrinfo(smmip, 0))[0] smmip = smmip[-1][0] if smmip and ':' in smmip: raise exc.NotImplementedException('IPv6 not supported') @@ -125,7 +125,7 @@ class NodeHandler(bmchandler.NodeHandler): if smmip and ':' in smmip and not smmip.startswith('fe80::'): raise exc.NotImplementedException('IPv6 configuration TODO') if self.ipaddr.startswith('fe80::'): - cfg.set_node_attributes( + await cfg.set_node_attributes( {nodename: {'hardwaremanagement.manager': self.ipaddr}}) def _webconfigcreds(self, username, password): diff --git a/confluent_server/confluent/discovery/handlers/smm3.py b/confluent_server/confluent/discovery/handlers/smm3.py index 45682d29..9cf43443 100644 --- a/confluent_server/confluent/discovery/handlers/smm3.py +++ b/confluent_server/confluent/discovery/handlers/smm3.py @@ -46,7 +46,7 @@ async def remote_nodecfg(nodename, cfm): ipaddr = cfg.get(nodename, {}).get('hardwaremanagement.manager', {}).get( 'value', None) ipaddr = ipaddr.split('/', 1)[0] - ipaddr = await asyncio.get_running_loop().getaddrinfo(ipaddr, 0)[0][-1] + ipaddr = (await asyncio.get_running_loop().getaddrinfo(ipaddr, 0))[0][-1] if not ipaddr: raise Exception('Cannot remote configure a system without known ' 'address') @@ -63,4 +63,3 @@ if __name__ == '__main__': print(repr(info)) testr = NodeHandler(info, c) asyncio.run(testr.config(sys.argv[2])) - diff --git a/confluent_server/confluent/discovery/handlers/tsm.py b/confluent_server/confluent/discovery/handlers/tsm.py index 4d0fe9a9..71ef812f 100644 --- a/confluent_server/confluent/discovery/handlers/tsm.py +++ b/confluent_server/confluent/discovery/handlers/tsm.py @@ -225,7 +225,7 @@ class NodeHandler(generic.NodeHandler): rsp, status = await wc.grab_json_response_with_status('/api/session', method='DELETE') -def remote_nodecfg(nodename, cfm): +async def remote_nodecfg(nodename, cfm): cfg = cfm.get_node_attributes( nodename, 'hardwaremanagement.manager') ipaddr = cfg.get(nodename, {}).get('hardwaremanagement.manager', {}).get( @@ -237,7 +237,7 @@ def remote_nodecfg(nodename, cfm): 'address') info = {'addresses': [ipaddr]} nh = NodeHandler(info, cfm) - nh.config(nodename) + await nh.config(nodename) if __name__ == '__main__': import confluent.config.configmanager as cfm diff --git a/confluent_server/confluent/discovery/protocols/mdns.py b/confluent_server/confluent/discovery/protocols/mdns.py index e359a9b0..290ed462 100644 --- a/confluent_server/confluent/discovery/protocols/mdns.py +++ b/confluent_server/confluent/discovery/protocols/mdns.py @@ -336,7 +336,7 @@ async def check_fish(urldata, port=443, verifycallback=None): except (IndexError, KeyError): return None url = '/redfish/v1/' - peerinfo = wc.grab_json_response('/redfish/v1/') + peerinfo = await wc.grab_json_response('/redfish/v1/') if url == '/redfish/v1/': if 'UUID' in peerinfo: if 'services' not in data: diff --git a/confluent_server/confluent/discovery/protocols/ssdp.py b/confluent_server/confluent/discovery/protocols/ssdp.py index c71000bc..9b3c8eb5 100644 --- a/confluent_server/confluent/discovery/protocols/ssdp.py +++ b/confluent_server/confluent/discovery/protocols/ssdp.py @@ -298,7 +298,7 @@ async def snoop(handler, byehandler=None, protocol=None, uuidlookup=None): if await netutil.ip_on_same_subnet(theip, 'fe80::', 64): if '%' in peer[0]: ifidx = peer[0].split('%', 1)[1] - iface = await cloop.getaddrinfo(peer[0], 0, socket.AF_INET6, socket.SOCK_DGRAM)[0][-1][-1] + iface = (await cloop.getaddrinfo(peer[0], 0, socket.AF_INET6, socket.SOCK_DGRAM))[0][-1][-1] else: ifidx = '{}'.format(peer[-1]) iface = peer[-1] @@ -502,7 +502,7 @@ async def check_fish(urldata, port=443, verifycallback=None): if url == '/DeviceDescription.json': if not peerinfo: if data.get('services', None) == ['urn::dmtf-org:service:redfish-rest:']: - peerinfo = wc.grab_json_response('/redfish/v1/') + peerinfo = await wc.grab_json_response('/redfish/v1/') if peerinfo: data['services'] = ['lenovo-smm3'] data['uuid'] = peerinfo['UUID'].lower() @@ -609,5 +609,3 @@ if __name__ == '__main__': def printit(rsp): pass # print(repr(rsp)) asyncio.run(active_scan(printit)) - - diff --git a/confluent_server/confluent/networking/lldp.py b/confluent_server/confluent/networking/lldp.py index 7eb2cac5..fa5ab588 100644 --- a/confluent_server/confluent/networking/lldp.py +++ b/confluent_server/confluent/networking/lldp.py @@ -132,8 +132,8 @@ def b64tohex(b64str): bd = bytearray(bd) return ''.join(['{0:02x}'.format(x) for x in bd]) -def get_fingerprint(switch, port, configmanager, portmatch): - update_switch_data(switch, configmanager) +async def get_fingerprint(switch, port, configmanager, portmatch): + await update_switch_data(switch, configmanager) for neigh in _neighbypeerid: info = _neighbypeerid[neigh] if neigh == '!!vintage' or info.get('switch', None) != switch: @@ -491,4 +491,3 @@ async def _handle_neighbor_query(pathcomponents, configmanager): if isinstance(x, Exception): raise x return list_info(parms, listrequested) - diff --git a/confluent_server/confluent/networking/macmap.py b/confluent_server/confluent/networking/macmap.py index d6a47d6f..39f1daa3 100644 --- a/confluent_server/confluent/networking/macmap.py +++ b/confluent_server/confluent/networking/macmap.py @@ -255,6 +255,7 @@ async def _start_offloader(): async def _recv_offload(): + global _offloader try: upacker = msgpack.Unpacker(encoding='utf8') except TypeError: @@ -262,6 +263,15 @@ async def _recv_offload(): #instream = _offloader.stdout.fileno() while True: datum = await _offloader.stdout.read(512) + if not datum: + _offloader = None + pending = list(_offloadevts.values()) + _offloadevts.clear() + for future in pending: + if not future.done(): + future.set_exception( + RuntimeError('MAC map offload process exited')) + return upacker.feed(datum) for result in upacker: if result[0] not in _offloadevts: @@ -628,15 +638,15 @@ async def handle_api_request(configmanager, inputdata, operation, pathcomponents operation, '/'.join(pathcomponents))) -def get_node_fingerprints(nodename, configmanager): +async def get_node_fingerprints(nodename, configmanager): cfg = configmanager.get_node_attributes(nodename, ['net*.switch', 'net*.switchport']) for attrkey in cfg[nodename]: if attrkey.endswith('switch'): switch = cfg[nodename][attrkey]['value'] port = cfg[nodename][attrkey + 'port']['value'] - yield get_fingerprint(switch, port, configmanager, - _namesmatch) + yield await get_fingerprint(switch, port, configmanager, + _namesmatch) async def handle_read_api_request(pathcomponents, configmanager): diff --git a/confluent_server/confluent/plugins/hardwaremanagement/affluent.py b/confluent_server/confluent/plugins/hardwaremanagement/affluent.py index d286ea73..d7c4383f 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/affluent.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/affluent.py @@ -72,10 +72,10 @@ async def renotify_me(node, configmanager, myname): creds = configmanager.get_node_attributes( node, ['secret.hardwaremanagementuser', 'secret.hardwaremanagementpassword'], decrypt=True) wc = WebClient(node, configmanager, creds) - res, status = wc.wc.grab_json_response_with_status('/affluent/systems/renotify', {'subscriber': myname}) + await wc.wc.grab_json_response_with_status('/affluent/systems/renotify', {'subscriber': myname}) -def subscribe_discovery(node, configmanager, myname): +async def subscribe_discovery(node, configmanager, myname): creds = configmanager.get_node_attributes( node, ['secret.hardwaremanagementuser', 'secret.hardwaremanagementpassword'], decrypt=True) tsock = socket.create_connection((node, 443)) @@ -87,19 +87,19 @@ def subscribe_discovery(node, configmanager, myname): wc = WebClient(node, configmanager, creds) with open('/etc/confluent/tls/cacert.pem') as cain: cacert = cain.read() - wc.wc.grab_json_response('/affluent/cert_authorities/{0}'.format(myname), cacert) - res, status = wc.wc.grab_json_response_with_status('/affluent/discovery_subscribers/{0}'.format(myname), {'url': myurl, 'authname': node}) + await wc.wc.grab_json_response('/affluent/cert_authorities/{0}'.format(myname), cacert) + res, status = await wc.wc.grab_json_response_with_status('/affluent/discovery_subscribers/{0}'.format(myname), {'url': myurl, 'authname': node}) if status == 200: agentkey = res['cryptkey'] - configmanager.set_node_attributes({node: {'crypted.selfapikey': {'hashvalue': agentkey}}}) - res, status = wc.wc.grab_json_response_with_status('/affluent/systems/renotify', {'subscriber': myname}) + await configmanager.set_node_attributes({node: {'crypted.selfapikey': {'hashvalue': agentkey}}}) + await wc.wc.grab_json_response_with_status('/affluent/systems/renotify', {'subscriber': myname}) -def unsubscribe_discovery(node, configmanager, myname): +async def unsubscribe_discovery(node, configmanager, myname): creds = configmanager.get_node_attributes( node, ['secret.hardwaremanagementuser', 'secret.hardwaremanagementpassword'], decrypt=True) wc = WebClient(node, configmanager, creds) - res, status = wc.wc.grab_json_response_with_status('/affluent/cert_authorities/{0}'.format(myname), method='DELETE') - res, status = wc.wc.grab_json_response_with_status('/affluent/discovery_subscribers/{0}'.format(myname), method='DELETE') + await wc.wc.grab_json_response_with_status('/affluent/cert_authorities/{0}'.format(myname), method='DELETE') + await wc.wc.grab_json_response_with_status('/affluent/discovery_subscribers/{0}'.format(myname), method='DELETE') def update(nodes, element, configmanager, inputdata): diff --git a/confluent_server/confluent/selfservice.py b/confluent_server/confluent/selfservice.py index b99c126e..c9499a7f 100644 --- a/confluent_server/confluent/selfservice.py +++ b/confluent_server/confluent/selfservice.py @@ -252,7 +252,7 @@ async def handle_request(req, make_response, mimetype): if not bmcaddr: return await make_response(mimetype, 500, 'Internal Server Error', body='Missing value in hardwaremanagement.manager') bmcaddr = bmcaddr.split('/', 1)[0] - bmcaddr = await asyncio.get_running_loop().getaddrinfo(bmcaddr, 0)[0] + bmcaddr = (await asyncio.get_running_loop().getaddrinfo(bmcaddr, 0))[0] bmcaddr = bmcaddr[-1][0] if '.' in bmcaddr: # ipv4 is allowed netconfig = await netutil.get_nic_config(cfg, nodename, ip=bmcaddr) @@ -456,9 +456,9 @@ async def handle_request(req, make_response, mimetype): reqbody = None cfgmod = reqbody.get('configmod', 'unspecified') if cfgmod == 'xcc': - xcc.remote_nodecfg(nodename, cfg) + await xcc.remote_nodecfg(nodename, cfg) elif cfgmod == 'tsm': - tsm.remote_nodecfg(nodename, cfg) + await tsm.remote_nodecfg(nodename, cfg) else: return await make_response(mimetype, 500, 'unsupported configmod', body='Unsupported configmod "{}"'.format(cfgmod)) return await make_response(mimetype, 200, 'Ok', body='complete')