xCAT/confluent
2
0
Fork 0
mirror of https://github.com/xcat2/confluent.git synced 2026-07-18 00:16:50 +00:00
Code Issues Projects Releases Wiki Activity

Fix asynchronous discovery call contracts

Browse Source
This commit is contained in:
Markus Hilger
2026-07-12 23:54:32 +02:00
parent 5fc036a2b7
commit 89d0fa81b9
12 changed files with 51 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
confluent_server/confluent/discovery/core.py
+10 -10
View File
@@ -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
confluent_server/confluent/discovery/handlers/bmc.py
+3 -3
View File
@@ -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(
confluent_server/confluent/discovery/handlers/redfishbmc.py
+3 -2
View File
@@ -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 = []
confluent_server/confluent/discovery/handlers/smm.py
+2 -2
View File
@@ -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):
confluent_server/confluent/discovery/handlers/smm3.py
+1 -2
View File
@@ -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]))
confluent_server/confluent/discovery/handlers/tsm.py
+2 -2
View File
@@ -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
confluent_server/confluent/discovery/protocols/mdns.py
+1 -1
View File
@@ -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:
confluent_server/confluent/discovery/protocols/ssdp.py
+2 -4
View File
@@ -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))
confluent_server/confluent/networking/lldp.py
+2 -3
View File
@@ -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)
confluent_server/confluent/networking/macmap.py
+13 -3
View File
@@ -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):
confluent_server/confluent/plugins/hardwaremanagement/affluent.py
+9 -9
View File
@@ -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):
confluent_server/confluent/selfservice.py
+3 -3
View File
@@ -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')