From 6650a01a259255585c06417aa86dbe06ea1111be Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sat, 11 Jul 2026 18:43:53 +0200 Subject: [PATCH 01/17] Fix Redfish OEM handler instantiation Fallback paths called OEM handler constructors directly, but these handlers are initialized through async create factories. This caused generic, TSMA, and SMM3 selection to fail with 'OEMHandler() takes no arguments'. Use and await the factories consistently. Also await the asynchronous bmcinfo lookup and forward the TSMA pool argument correctly. --- .../aiohmi/redfish/oem/lenovo/main.py | 10 ++++---- .../aiohmi/redfish/oem/lenovo/tsma.py | 4 ++-- confluent_server/aiohmi/redfish/oem/lookup.py | 24 +++++++++---------- 3 files changed, 19 insertions(+), 19 deletions(-) diff --git a/confluent_server/aiohmi/redfish/oem/lenovo/main.py b/confluent_server/aiohmi/redfish/oem/lenovo/main.py index 35ba6bbf..de04ae1b 100644 --- a/confluent_server/aiohmi/redfish/oem/lenovo/main.py +++ b/confluent_server/aiohmi/redfish/oem/lenovo/main.py @@ -39,7 +39,7 @@ async def get_handler(sysinfo, sysurl, webclient, cache, cmd, rootinfo={}): if not leninf: bmcinfo = await cmd.bmcinfo() if 'Ami' in bmcinfo.get('Oem', {}): - return tsma.TsmHandler(sysinfo, sysurl, webclient, cache) + return await tsma.TsmHandler.create(sysinfo, sysurl, webclient, cache, gpool=cmd._gpool) elif 'xclarity controller' in mgrinfo.get('Model', '').lower(): if mgrinfo['Model'].endswith('3'): return await xcc3.OEMHandler.create(sysinfo, sysurl, webclient, cache, @@ -60,9 +60,9 @@ async def get_handler(sysinfo, sysurl, webclient, cache, cmd, rootinfo={}): devdesc = await webclient.grab_json_response_with_status('/DeviceDescription.json') if devdesc[1] == 200: if devdesc[0]['type'].lower() in ('lenovo-smm3', 'smm3'): - return smm3.OEMHandler(sysinfo, sysurl, webclient, cache, - gpool=cmd._gpool) + return await smm3.OEMHandler.create(sysinfo, sysurl, webclient, cache, + gpool=cmd._gpool) except Exception: pass - return generic.OEMHandler(sysinfo, sysurl, webclient, cache, - gpool=cmd._gpool) + return await generic.OEMHandler.create(sysinfo, sysurl, webclient, cache, + gpool=cmd._gpool) diff --git a/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py b/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py index afa4851f..d36d9355 100644 --- a/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py +++ b/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py @@ -101,8 +101,8 @@ class TsmHandler(generic.OEMHandler): @classmethod async def create(cls, sysinfo, sysurl, webclient, cache=None, fish=None, gpool=None): - self = await super(TsmHandler, cls).create(sysinfo, sysurl, webclient, cache, fish, - gpool) + self = await super(TsmHandler, cls).create(sysinfo, sysurl, webclient, cache, + gpool=gpool) if cache is None: cache = {} self._wc = None diff --git a/confluent_server/aiohmi/redfish/oem/lookup.py b/confluent_server/aiohmi/redfish/oem/lookup.py index 210f2dc2..4f528f8d 100644 --- a/confluent_server/aiohmi/redfish/oem/lookup.py +++ b/confluent_server/aiohmi/redfish/oem/lookup.py @@ -27,23 +27,23 @@ OEMMAP = { } -def get_oem_handler(sysinfo, sysurl, webclient, cache, cmd, rootinfo={}): +async def get_oem_handler(sysinfo, sysurl, webclient, cache, cmd, rootinfo={}): if rootinfo.get('Vendor', None) in OEMMAP: - return OEMMAP[rootinfo['Vendor']].get_handler(sysinfo, sysurl, - webclient, cache, cmd, rootinfo) + return await OEMMAP[rootinfo['Vendor']].get_handler(sysinfo, sysurl, + webclient, cache, cmd, rootinfo) for oem in sysinfo.get('Oem', {}): if oem in OEMMAP: - return OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache, - cmd, rootinfo) + return await OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache, + cmd, rootinfo) for oem in sysinfo.get('Links', {}).get('OEM', []): if oem in OEMMAP: - return OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache, - cmd, rootinfo) + return await OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache, + cmd, rootinfo) if rootinfo: # rootinfo indicates early invocation, bmcinfo not ready yet - return generic.OEMHandler(sysinfo, sysurl, webclient, cache, cmd._gpool, rootinfo) - bmcinfo = cmd.bmcinfo + return await generic.OEMHandler.create(sysinfo, sysurl, webclient, cache, cmd._gpool, rootinfo) + bmcinfo = await cmd.bmcinfo() for oem in bmcinfo.get('Oem', {}): if oem in OEMMAP: - return OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache, - cmd, rootinfo) - return generic.OEMHandler(sysinfo, sysurl, webclient, cache, cmd._gpool, rootinfo) + return await OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache, + cmd, rootinfo) + return await generic.OEMHandler.create(sysinfo, sysurl, webclient, cache, cmd._gpool, rootinfo) From 2117d120d8a10a065e227e7617cdfbd73857f56f Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sat, 11 Jul 2026 20:07:20 +0200 Subject: [PATCH 02/17] Fix remaining aiohmi async call paths Await OEM sensor, NTP, retry, and firmware-update operations that otherwise returned or discarded coroutine objects. Return the initialized energy manager for FAPM systems and update stale utility entry points to use asynchronous Command factories. --- .../aiohmi/ipmi/oem/lenovo/config.py | 4 +-- .../aiohmi/ipmi/oem/lenovo/energy.py | 14 ++++++--- .../aiohmi/ipmi/oem/lenovo/handler.py | 29 ++++++++++--------- .../aiohmi/ipmi/oem/lenovo/imm.py | 4 +-- .../aiohmi/ipmi/oem/lenovo/nextscale.py | 4 +-- confluent_server/aiohmi/redfish/command.py | 10 +++++-- 6 files changed, 38 insertions(+), 27 deletions(-) diff --git a/confluent_server/aiohmi/ipmi/oem/lenovo/config.py b/confluent_server/aiohmi/ipmi/oem/lenovo/config.py index ebf5f7e9..12304f19 100644 --- a/confluent_server/aiohmi/ipmi/oem/lenovo/config.py +++ b/confluent_server/aiohmi/ipmi/oem/lenovo/config.py @@ -74,7 +74,7 @@ async def run_command_with_retry(connection, data): except pygexc.IpmiException as e: if e.ipmicode != 0xa or not tries: raise - connection.ipmi_session.pause(1) + await connection.ipmi_session.pause(1) def _convert_syntax(raw): @@ -198,7 +198,7 @@ class LenovoFirmwareConfig(object): break except KeyError: pass - self.connection.ipmi_session.pause(5) + await self.connection.ipmi_session.pause(5) filehandle = response['data'][3:7] filehandle = struct.unpack(" 99.0: initpct = 99.0 @@ -1039,7 +1039,7 @@ class SMMClient(object): complete = False tries = 0 while not complete: - ipmisession.Session.pause(3) + await ipmisession.Session.pause(3) wc.request('POST', '/data', 'get=fwProgress,fwUpdate') try: rsp = wc.getresponse() diff --git a/confluent_server/aiohmi/redfish/command.py b/confluent_server/aiohmi/redfish/command.py index adc7288d..e602393f 100644 --- a/confluent_server/aiohmi/redfish/command.py +++ b/confluent_server/aiohmi/redfish/command.py @@ -1586,6 +1586,10 @@ class Command(object): return await oem.apply_license(filename, self, progress, data) if __name__ == '__main__': - print(repr( - Command(sys.argv[1], os.environ['BMCUSER'], os.environ['BMCPASS'], - verifycallback=lambda x: True).get_power())) + async def main(): + cmd = await Command.create( + sys.argv[1], os.environ['BMCUSER'], os.environ['BMCPASS'], + verifycallback=lambda x: True) + print(repr(await cmd.get_power())) + + asyncio.run(main()) From 2c41841efc44409a313ec1fc03b549006846471d Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sat, 11 Jul 2026 22:28:40 +0200 Subject: [PATCH 03/17] Fix additional missing awaits in aiohmi Await the channel access raw command, the TSMA remote media settings requests, and the XCC3 volume creation responses. These calls returned or unpacked coroutine objects, breaking set_channel_access, TSMA virtual media attach, and RAID volume creation at runtime. --- confluent_server/aiohmi/ipmi/command.py | 2 +- confluent_server/aiohmi/redfish/oem/lenovo/tsma.py | 10 +++++----- confluent_server/aiohmi/redfish/oem/lenovo/xcc3.py | 4 ++-- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/confluent_server/aiohmi/ipmi/command.py b/confluent_server/aiohmi/ipmi/command.py index 5efb7b4a..d4f0cfaf 100644 --- a/confluent_server/aiohmi/ipmi/command.py +++ b/confluent_server/aiohmi/ipmi/command.py @@ -1581,7 +1581,7 @@ class Command(object): } b |= privilege_levels[privilege_level] & 0b00000111 data.append(b) - response = self.raw_command(netfn=0x06, command=0x40, data=data) + response = await self.raw_command(netfn=0x06, command=0x40, data=data) if 'error' in response: raise Exception(response['error']) return True diff --git a/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py b/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py index d36d9355..49d0fe8c 100644 --- a/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py +++ b/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py @@ -731,7 +731,7 @@ class TsmHandler(generic.OEMHandler): hddslots += 1 else: raise exc.UnsupportedFunctionality('Unknown slot type requested') - gensettings = wc.grab_json_response('/api/settings/media/general') + gensettings = await wc.grab_json_response('/api/settings/media/general') samesettings = gensettings['same_settings'] == 1 if samesettings: hds = gensettings['cd_remote_server_address'] @@ -772,10 +772,10 @@ class TsmHandler(generic.OEMHandler): gensettings['remote_media_support'] = 1 gensettings['cd_remote_password'] = '' gensettings['hd_remote_password'] = '' - wc.grab_json_response_with_status('/api/settings/media/general', + await wc.grab_json_response_with_status('/api/settings/media/general', gensettings, method='PUT') # need to calibrate instances correctly - currinfo, status = wc.grab_json_response_with_status( + currinfo, status = await wc.grab_json_response_with_status( '/api/settings/media/instance') currinfo['num_cd'] = cdslots currinfo['num_hd'] = hddslots @@ -783,9 +783,9 @@ class TsmHandler(generic.OEMHandler): currinfo['kvm_num_cd'] = cdslots if currinfo['kvm_num_hd'] > hddslots: currinfo['kvm_num_hd'] = hddslots - wc.grab_json_response_with_status( + await wc.grab_json_response_with_status( '/api/settings/media/instance', currinfo, method='PUT') - images = wc.grab_json_response('/api/settings/media/remote/images') + images = await wc.grab_json_response('/api/settings/media/remote/images') tries = 20 while tries and not images: tries -= 1 diff --git a/confluent_server/aiohmi/redfish/oem/lenovo/xcc3.py b/confluent_server/aiohmi/redfish/oem/lenovo/xcc3.py index 3c465a34..a109931c 100644 --- a/confluent_server/aiohmi/redfish/oem/lenovo/xcc3.py +++ b/confluent_server/aiohmi/redfish/oem/lenovo/xcc3.py @@ -696,14 +696,14 @@ class OEMHandler(generic.OEMHandler): if write_policy: request_data["WriteCachePolicy"] = write_policy - msg, code=self.webclient.grab_json_response_with_status( + msg, code = await self.webclient.grab_json_response_with_status( f'/redfish/v1/Systems/1/Storage/{cid}/Volumes', method='POST', data=request_data) if code == 500 and not stripsize: # Mystery error can be a mandatory strip size, default to 64k to match WebUI behavior request_data["StripSizeBytes"] = 65536 - msg, code=self.webclient.grab_json_response_with_status( + msg, code = await self.webclient.grab_json_response_with_status( f'/redfish/v1/Systems/1/Storage/{cid}/Volumes', method='POST', data=request_data) From 91654ea0d19c6f624c9a64cb56037dd7dfe2e694 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sun, 12 Jul 2026 23:53:39 +0200 Subject: [PATCH 04/17] Fix aiohmi async call contracts --- confluent_server/aiohmi/ipmi/command.py | 9 ++-- .../aiohmi/ipmi/oem/lenovo/handler.py | 42 +++++++++---------- .../aiohmi/ipmi/oem/lenovo/nextscale.py | 8 ++-- .../aiohmi/ipmi/private/localsession.py | 1 + confluent_server/aiohmi/ipmi/sdr.py | 16 +++---- .../aiohmi/redfish/oem/generic.py | 13 +++--- .../aiohmi/redfish/oem/lenovo/smm3.py | 3 +- .../aiohmi/redfish/oem/lenovo/xcc.py | 16 +++---- 8 files changed, 55 insertions(+), 53 deletions(-) diff --git a/confluent_server/aiohmi/ipmi/command.py b/confluent_server/aiohmi/ipmi/command.py index d4f0cfaf..a7f4723b 100644 --- a/confluent_server/aiohmi/ipmi/command.py +++ b/confluent_server/aiohmi/ipmi/command.py @@ -738,7 +738,8 @@ class Command(object): This provides a detailed view of the LEDs of the managed system. """ await self.oem_init() - return await self._oem.get_leds() + async for led in self._oem.get_leds(): + yield led async def get_ntp_enabled(self): await self.oem_init() @@ -806,7 +807,7 @@ class Command(object): rsp = await self.raw_command(command=0x2d, netfn=4, rslun=currsensor.sensor_lun, data=(currsensor.sensor_number,)) - return self._sdr.sensors[sensor].decode_sensor_reading( + return await self._sdr.sensors[sensor].decode_sensor_reading( self, rsp['data']) await self.oem_init() return await self._oem.get_sensor_reading(sensorname) @@ -1091,7 +1092,7 @@ class Command(object): if rsp['code'] == 203: # Sensor does not exist, optional dev continue raise exc.IpmiException(rsp['error'], code=rsp['code']) - yield self._sdr.sensors[sensor].\ + yield await self._sdr.sensors[sensor].\ decode_sensor_reading(self, rsp['data']) await self.oem_init() async for reading in self._oem.get_sensor_data(): @@ -2311,4 +2312,4 @@ class Command(object): """ await self.oem_init() - return await self._oem.set_oem_extended_privilleges(uid) \ No newline at end of file + return await self._oem.set_oem_extended_privilleges(uid) diff --git a/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py b/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py index eaff7789..65186e69 100755 --- a/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py +++ b/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py @@ -336,9 +336,9 @@ class OEMHandler(generic.OEMHandler): event['component'] += ' {0}'.format(evdata[1] & 0b11111) async def reseat_bay(self, bay): - if self.is_fpc: + if await self.is_fpc(): return await self.smmhandler.reseat_bay(bay) - elif self.has_xcc and bay == -1: + elif await self.has_xcc() and bay == -1: return await self.immhandler.reseat() return await super(OEMHandler, self).reseat_bay(bay) @@ -347,7 +347,7 @@ class OEMHandler(generic.OEMHandler): ntpres = await self.ipmicmd.raw_command(netfn=0x32, command=0xa7) return ntpres['data'][0] == '\x01' elif await self.is_fpc(): - return await self.smmhandler.get_ntp_enabled(self._fpc_variant) + return self.smmhandler.get_ntp_enabled(self._fpc_variant) elif self.has_tsma: return await self.tsmahandler.get_ntp_enabled() return None @@ -360,7 +360,7 @@ class OEMHandler(generic.OEMHandler): srvs.append(ntpres['data'][129:257].rstrip('\x00')) return srvs if await self.is_fpc(): - return await self.smmhandler.get_ntp_servers() + return self.smmhandler.get_ntp_servers() if self.has_tsma: return await self.tsmahandler.get_ntp_servers() return () @@ -375,7 +375,7 @@ class OEMHandler(generic.OEMHandler): netfn=0x32, command=0xa8, data=(3, 0), timeout=15) return True if await self.is_fpc(): - await self.smmhandler.set_ntp_enabled(enabled) + self.smmhandler.set_ntp_enabled(enabled) return True if self.has_tsma: await self.tsmahandler.set_ntp_enabled(enabled) @@ -393,7 +393,7 @@ class OEMHandler(generic.OEMHandler): if not 0 <= index <= 2: raise pygexc.InvalidParameterValue( 'SMM supports indexes 0 through 2') - await self.smmhandler.set_ntp_server(server, index) + self.smmhandler.set_ntp_server(server, index) return True elif self.has_tsma: if not (0 <= index <= 1): @@ -516,7 +516,7 @@ class OEMHandler(generic.OEMHandler): elif await self.is_fpc(): async for compname in self.smmhandler.get_inventory_descriptions( self.ipmicmd, await self.is_fpc()): - yield (compname, self.smmhandler.get_inventory_of_component( + yield (compname, await self.smmhandler.get_inventory_of_component( self.ipmicmd, compname)) async def get_sensor_data(self): @@ -561,14 +561,14 @@ class OEMHandler(generic.OEMHandler): async def get_inventory_of_component(self, component): if await self.has_tsm() or await self.has_ami() or await self.has_asrock(): await self._collect_tsm_inventory() - return await self.oem_inventory_info.get(component, None) + return self.oem_inventory_info.get(component, None) if await self.has_imm(): return await self.immhandler.get_component_inventory(component) if await self.is_fpc(): - return await self.smmhandler.get_inventory_of_component(component) + return await self.smmhandler.get_inventory_of_component(self.ipmicmd, component) - def get_cmd_type(self, categorie_item, catspec): - if self.has_asrock: + async def get_cmd_type(self, categorie_item, catspec): + if await self.has_asrock(): cmd_type = catspec["command"]["asrock"] elif categorie_item in categorie_items: cmd_type = catspec["command"]["lenovo"] @@ -579,9 +579,7 @@ class OEMHandler(generic.OEMHandler): async def _collect_tsm_inventory(self): self.oem_inventory_info = {} - asrock = False - if self.has_asrock: - asrock = True + asrock = await self.has_asrock() for catid, catspec in inventory.categories.items(): # skip the inventory fields if the system is RS160 if asrock and catid not in categorie_items: @@ -590,7 +588,7 @@ class OEMHandler(generic.OEMHandler): and catspec["workaround_bmc_bug"]( "ami" if await self.has_ami() else "lenovo")): rsp = None - cmd = self.get_cmd_type(catid, catspec) + cmd = await self.get_cmd_type(catid, catspec) tmp_command = dict(cmd) tmp_command["data"] = list(tmp_command["data"]) count = 0 @@ -617,7 +615,7 @@ class OEMHandler(generic.OEMHandler): rsp["data"] = buffer(bytearray(rsp["data"])) else: try: - cmd = self.get_cmd_type(catid, catspec) + cmd = await self.get_cmd_type(catid, catspec) rsp = await self.ipmicmd.raw_command(**cmd) except pygexc.IpmiException: continue @@ -852,7 +850,7 @@ class OEMHandler(generic.OEMHandler): async def get_oem_firmware(self, bmcver, components, category): if await self.has_tsm() or await self.has_ami() or await self.has_asrock(): command = firmware.get_categories()["firmware"] - fw_cmd = self.get_cmd_type("firmware", command) + fw_cmd = await self.get_cmd_type("firmware", command) rsp = await self.ipmicmd.raw_command(**fw_cmd) # the newest Lenovo ThinkServer versions are returning Bios version @@ -860,7 +858,7 @@ class OEMHandler(generic.OEMHandler): bios_versions = None if await self.has_tsm() or await self.has_asrock(): bios_command = firmware.get_categories()["bios_version"] - bios_cmd = self.get_cmd_type("bios_version", bios_command) + bios_cmd = await self.get_cmd_type("bios_version", bios_command) bios_rsp = await self.ipmicmd.raw_command(**bios_cmd) if await self.has_asrock(): bios_versions = bios_rsp['data'] @@ -943,7 +941,7 @@ class OEMHandler(generic.OEMHandler): name += rsp['data'][:] return name.rstrip('\x00') elif await self.is_fpc(): - return await self.smmhandler.get_domain() + return self.smmhandler.get_domain() async def set_oem_domain_name(self, name): if await self.has_tsm(): @@ -962,20 +960,20 @@ class OEMHandler(generic.OEMHandler): await self._restart_dns() return elif await self.is_fpc(): - await self.smmhandler.set_domain(name) + self.smmhandler.set_domain(name) async def set_hostname(self, hostname): if await self.has_xcc(): return await self.immhandler.set_hostname(hostname) elif await self.is_fpc(): - return await self.smmhandler.set_hostname(hostname) + return self.smmhandler.set_hostname(hostname) return await super(OEMHandler, self).set_hostname(hostname) async def get_hostname(self): if await self.has_xcc(): return await self.immhandler.get_hostname() elif await self.is_fpc(): - return await self.smmhandler.get_hostname() + return self.smmhandler.get_hostname() return await super(OEMHandler, self).get_hostname() """ Gets a remote console launcher for a Lenovo ThinkServer. diff --git a/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py b/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py index 56adf001..1c15d507 100644 --- a/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py +++ b/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py @@ -1061,16 +1061,16 @@ class SMMClient(object): complete = percent >= 100.0 return 'complete' - def get_inventory_descriptions(self, ipmicmd, variant): + async def get_inventory_descriptions(self, ipmicmd, variant): if variant >> 5 == 0: return - psucount = get_psu_count(ipmicmd, variant) + psucount = await get_psu_count(ipmicmd, variant) for idx in range(psucount): yield 'PSU {}'.format(idx + 1) - def get_inventory_of_component(self, ipmicmd, component): + async def get_inventory_of_component(self, ipmicmd, component): psuidx = int(component.replace('PSU ', '')) - return self.get_psu_info(ipmicmd, psuidx) + return await self.get_psu_info(ipmicmd, psuidx) async def get_psu_info(self, ipmicmd, psunum): psuinfo = await ipmicmd.raw_command(0x34, 0x6, data=(psunum,)) diff --git a/confluent_server/aiohmi/ipmi/private/localsession.py b/confluent_server/aiohmi/ipmi/private/localsession.py index 2489d272..82b0a7b5 100644 --- a/confluent_server/aiohmi/ipmi/private/localsession.py +++ b/confluent_server/aiohmi/ipmi/private/localsession.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import asyncio import ctypes import fcntl from select import select diff --git a/confluent_server/aiohmi/ipmi/sdr.py b/confluent_server/aiohmi/ipmi/sdr.py index d2855525..fcf05d3c 100644 --- a/confluent_server/aiohmi/ipmi/sdr.py +++ b/confluent_server/aiohmi/ipmi/sdr.py @@ -475,7 +475,7 @@ class SDREntry(object): health = const.Health.Ok return desc, health - def decode_sensor_reading(self, ipmicmd, reading): + async def decode_sensor_reading(self, ipmicmd, reading): numeric = None output = { 'name': self.sensor_name, @@ -495,8 +495,8 @@ class SDREntry(object): if numeric is not None: lowerbound = numeric - (0.5 + (self.tolerance / 2.0)) upperbound = numeric + (0.5 + (self.tolerance / 2.0)) - lowerbound = self.decode_value(ipmicmd, lowerbound) - upperbound = self.decode_value(ipmicmd, upperbound) + lowerbound = await self.decode_value(ipmicmd, lowerbound) + upperbound = await self.decode_value(ipmicmd, upperbound) output['value'] = (lowerbound + upperbound) / 2.0 output['imprecision'] = output['value'] - lowerbound discrete = False @@ -552,13 +552,13 @@ class SDREntry(object): output['state_ids'].append(self.assert_trap_value(6)) return SensorReading(output, self.unit_suffix) - def _set_tmp_formula(self, ipmicmd, value): - rsp = ipmicmd.raw_command(netfn=4, command=0x23, - data=(self.sensor_number, value)) + async def _set_tmp_formula(self, ipmicmd, value): + rsp = await ipmicmd.raw_command(netfn=4, command=0x23, + data=(self.sensor_number, value)) # skip next reading field, not used in on-demand situation self.decode_formula(rsp['data'][1:]) - def decode_value(self, ipmicmd, value): + async def decode_value(self, ipmicmd, value): # Take the input value and return meaningful value linearization = self.linearization if linearization > 11: # direct calling code to get factors @@ -568,7 +568,7 @@ class SDREntry(object): # fashion. However for now opt for retrieving rows as needed # rather than tracking all that information for a relatively # rare behavior - self._set_tmp_formula(ipmicmd, value) + await self._set_tmp_formula(ipmicmd, value) linearization = 0 # time to compute the pre-linearization value. decoded = float((value * self.m + self.b) diff --git a/confluent_server/aiohmi/redfish/oem/generic.py b/confluent_server/aiohmi/redfish/oem/generic.py index 87c052a2..df053980 100644 --- a/confluent_server/aiohmi/redfish/oem/generic.py +++ b/confluent_server/aiohmi/redfish/oem/generic.py @@ -660,8 +660,8 @@ class OEMHandler(object): procurl = sysinfo.get('Processors', {}).get('@odata.id', None) if procurl: - for cpu in await fishclient._do_web_request(procurl).get( - 'Members', []): + procinfo = await fishclient._do_web_request(procurl) + for cpu in procinfo.get('Members', []): cinfo = await fishclient._do_web_request(cpu['@odata.id']) if cinfo.get('Status', {}).get( 'State', None) == 'Absent': @@ -1004,8 +1004,9 @@ class OEMHandler(object): def _extract_fwinfo(self, inf): return {} - def get_firmware_inventory(self, components, fishclient, category=None): - return [] + async def get_firmware_inventory(self, components, fishclient, category=None): + return + yield def set_credentials(self, username, password): try: @@ -1240,7 +1241,7 @@ class OEMHandler(object): if not foundmacs: # No PCIe device inventory, but *maybe* ethernet inventory... idxsbyname = {} - for nicinfo in self._get_eth_urls(): + for nicinfo in await self._get_eth_urls(): nicinfo = await self._do_web_request(nicinfo) nicname = nicinfo.get('Name', None) nicinfo = nicinfo.get('MACAddress', nicinfo.get('PermanentAddress', None)) @@ -1675,7 +1676,7 @@ class OEMHandler(object): await fishclient._do_web_request(licenses, licinfo) - def get_user_expiration(self, uid): + async def get_user_expiration(self, uid): return None async def reseat_bay(self, bay): diff --git a/confluent_server/aiohmi/redfish/oem/lenovo/smm3.py b/confluent_server/aiohmi/redfish/oem/lenovo/smm3.py index 959c967d..1da4165e 100644 --- a/confluent_server/aiohmi/redfish/oem/lenovo/smm3.py +++ b/confluent_server/aiohmi/redfish/oem/lenovo/smm3.py @@ -286,7 +286,8 @@ class OEMHandler(generic.OEMHandler): rsp = await self._do_web_request(url, method='POST') async def get_event_log(self, clear=False, fishclient=None): - return await super().get_event_log(clear, fishclient, extraurls=[{'@odata.id':'/redfish/v1/Chassis/chassis1/LogServices/EventLog'}]) + async for event in super().get_event_log(clear, fishclient, extraurls=[{'@odata.id':'/redfish/v1/Chassis/chassis1/LogServices/EventLog'}]): + yield event async def get_description(self, fishclient): return {'height': 13, 'slot': 0, 'slots': [8, 2]} diff --git a/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py b/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py index 575b3726..09fd1a11 100644 --- a/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py +++ b/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py @@ -1757,18 +1757,18 @@ class OEMHandler(generic.OEMHandler): else: return days - def get_inventory_descriptions(self, withids=False): - hwmap = self.hardware_inventory_map() + async def get_inventory_descriptions(self, withids=False): + hwmap = await self.hardware_inventory_map() yield "System" for key in natural_sort(hwmap): yield key - for cpuinv in self._get_cpu_inventory(): + async for cpuinv in self._get_cpu_inventory(): yield cpuinv[0] - for meminv in self._get_mem_inventory(): + async for meminv in self._get_mem_inventory(): yield meminv[0] - def get_inventory_of_component(self, compname): + async def get_inventory_of_component(self, compname): if compname.lower() == 'system': sysinfo = { 'UUID': self._varsysinfo.get('UUID', ''), @@ -1779,14 +1779,14 @@ class OEMHandler(generic.OEMHandler): 'SKU', self._varsysinfo.get('PartNumber', '')), } return sysinfo - hwmap = self.hardware_inventory_map() + hwmap = await self.hardware_inventory_map() try: return hwmap[compname] except KeyError: - for cpuinv in self._get_cpu_inventory(): + async for cpuinv in self._get_cpu_inventory(): if cpuinv[0] == compname: return cpuinv[1] - for meminv in self._get_mem_inventory(): + async for meminv in self._get_mem_inventory(): if meminv[0] == compname: return meminv[1] From 5fc036a2b737341491fd2965e328c5f2a58e9ded Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sun, 12 Jul 2026 23:54:16 +0200 Subject: [PATCH 05/17] 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'}) - From 89d0fa81b98ea0da377bb285a09a8f28ae0a6da9 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sun, 12 Jul 2026 23:54:32 +0200 Subject: [PATCH 06/17] 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') From 9c4f9e19354bc8965319fa7d84bdc6c9ca479ea9 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sun, 12 Jul 2026 23:55:32 +0200 Subject: [PATCH 07/17] Fix hardware management async dispatch --- .../plugins/hardwaremanagement/deltapdu.py | 4 +-- .../plugins/hardwaremanagement/ipmi.py | 25 +++++++++---------- .../plugins/hardwaremanagement/pdu.py | 10 +++++--- .../plugins/hardwaremanagement/redfish.py | 14 +++++------ 4 files changed, 27 insertions(+), 26 deletions(-) diff --git a/confluent_server/confluent/plugins/hardwaremanagement/deltapdu.py b/confluent_server/confluent/plugins/hardwaremanagement/deltapdu.py index 58d3b785..6a4b7bc0 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/deltapdu.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/deltapdu.py @@ -163,7 +163,7 @@ class PDUClient(object): self.wc.grab_response('/logout_wait.htm') async def get_outlet(self, outlet): - rsp = await self.wc.grab_response('/setting_admin4.xml') + rsp = self.wc.grab_response('/setting_admin4.xml') xd = fromstring(rsp[0]) for ch in xd: if 'relay' not in ch.tag: @@ -177,7 +177,7 @@ class PDUClient(object): outlet = int(outlet) ident = self.map_outlets[outlet] sitem = '/SetParm?item={}?content={}'.format(ident, state) - await self.wc.grab_response(sitem) + self.wc.grab_response(sitem) async def retrieve(nodes, element, configmanager, inputdata): if 'outlets' not in element: diff --git a/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py b/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py index 04894407..7333d170 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py @@ -554,16 +554,16 @@ class IpmiHandler: if (self.error == 'timeout' or 'Insufficient resources' in self.error): self.error = self.error.replace(' reported in RAKP4', '') - self.output.put(msg.ConfluentTargetTimeout( + await self.output.put(msg.ConfluentTargetTimeout( self.node, self.error)) return elif 'Invalid Session ID' in self.error: - self.output.put(msg.ConfluentTargetTimeout( + await self.output.put(msg.ConfluentTargetTimeout( self.node, 'Temporary Login Error')) return elif ('Unauthorized' in self.error or 'Incorrect password' in self.error): - self.output.put( + await self.output.put( msg.ConfluentTargetInvalidCredentials(self.node)) return else: @@ -708,7 +708,7 @@ class IpmiHandler: if tmpvarbind.endswith('3183.1.1'): varbinddata = inputdata[tmpvarbind] varbinddata = hex2bin(varbinddata) - event = self.ipmicmd.decode_pet(specifictrap, varbinddata) + event = await self.ipmicmd.decode_pet(specifictrap, varbinddata) self.pyghmi_event_to_confluent(event) await self.output.put(msg.EventCollection((event,), name=self.node)) @@ -1007,7 +1007,7 @@ class IpmiHandler: if activeupdates: await self.output.put(msg.KeyValueData({'status': 'active'}, self.node)) else: - status = self.ipmicmd.get_update_status() + status = await self.ipmicmd.get_update_status() await self.output.put(msg.KeyValueData({'status': status}, self.node)) async def handle_inventory(self): @@ -1025,12 +1025,12 @@ class IpmiHandler: async def list_leds(self): await self.output.put(msg.ChildCollection('all')) - for category, info in self.ipmicmd.get_leds(): + async for category, info in self.ipmicmd.get_leds(): await self.output.put(msg.ChildCollection(simplify_name(category))) async def read_leds(self, component): led_categories = [] - for category, info in self.ipmicmd.get_leds(): + async for category, info in self.ipmicmd.get_leds(): if component == 'all' or component == simplify_name(category): led_categories.append({category: info}) await self.output.put(msg.LEDStatus(led_categories, self.node)) @@ -1054,7 +1054,7 @@ class IpmiHandler: await self.make_inventory_map() compname = self.invmap.get(component, None) if compname is None: - self.output.put(msg.ConfluentTargetNotFound()) + await self.output.put(msg.ConfluentTargetNotFound()) return invdata = await self.ipmicmd.get_inventory_of_component(compname) if invdata is None: @@ -1118,7 +1118,7 @@ class IpmiHandler: if len(storelem) < 2 or storelem[0] != 'volumes': raise exc.InvalidArgumentException('Must target a specific volume') volname = storelem[-1] - curr = self.ipmicmd.get_storage_configuration() + curr = await self.ipmicmd.get_storage_configuration() volumes = [] volsfound = False toremove = storage.ConfigSpec(arrays=[storage.Array(volumes=volumes)]) @@ -1131,7 +1131,7 @@ class IpmiHandler: await self.output.put(msg.ConfluentTargetNotFound( self.node, "No volume named '{0}' found".format(volname))) return - self.ipmicmd.remove_storage_configuration(toremove) + await self.ipmicmd.remove_storage_configuration(toremove) await self.output.put(msg.DeletedResource(volname)) async def _create_storage(self, storelem): @@ -1139,7 +1139,7 @@ class IpmiHandler: raise exc.InvalidArgumentException('Can only create volumes') vols = [] thedisks = None - currcfg = self.ipmicmd.get_storage_configuration() + currcfg = await self.ipmicmd.get_storage_configuration() currnames = [] for arr in currcfg.arrays: arrname = '{0}-{1}'.format(*arr.id) @@ -1345,7 +1345,7 @@ class IpmiHandler: async def list_sensors(self): try: - sensors = await self.ipmicmd.get_sensor_descriptions() + sensors = [sensor async for sensor in self.ipmicmd.get_sensor_descriptions()] except pygexc.IpmiException: await self.output.put(msg.ConfluentTargetTimeout(self.node)) return @@ -1760,4 +1760,3 @@ def delete(nodes, element, configmanager, inputdata): element, type='ffdc') return perform_requests( 'delete', nodes, element, configmanager, inputdata, 'delete') - diff --git a/confluent_server/confluent/plugins/hardwaremanagement/pdu.py b/confluent_server/confluent/plugins/hardwaremanagement/pdu.py index 7a5a5376..cefb9c1b 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/pdu.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/pdu.py @@ -71,9 +71,10 @@ async def readpdu(pdu, outletmap, configmanager, rspq): for outlet in outletmap: node, pgroup = outletmap[outlet] try: - for rsp in core.handle_path( + responses = await core.handle_path( '/nodes/{0}/power/outlets/{1}'.format(pdu, outlet), - 'retrieve', configmanager): + 'retrieve', configmanager) + async for rsp in core.iterate_responses(responses): await rspq.put(msg.KeyValueData({pgroup: rsp.kvpairs['state']['value']}, node)) except exc.TargetEndpointBadCredentials: await rspq.put(msg.ConfluentTargetInvalidCredentials(pdu)) @@ -137,8 +138,9 @@ async def updatepdu(pdu, outletmap, configmanager, inputdata, rspq): try: for outlet in outletmap: node, pgroup = outletmap[outlet] - for rsp in core.handle_path('/nodes/{0}/power/outlets/{1}'.format(pdu, outlet), - 'update', configmanager, inputdata={'state': inputdata.powerstate(node)}): + responses = await core.handle_path('/nodes/{0}/power/outlets/{1}'.format(pdu, outlet), + 'update', configmanager, inputdata={'state': inputdata.powerstate(node)}) + async for rsp in core.iterate_responses(responses): await rspq.put(msg.KeyValueData({pgroup: rsp.kvpairs['state']['value']}, node)) finally: await rspq.put(TaskDone()) diff --git a/confluent_server/confluent/plugins/hardwaremanagement/redfish.py b/confluent_server/confluent/plugins/hardwaremanagement/redfish.py index d95ff71e..0ef6781f 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/redfish.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/redfish.py @@ -399,7 +399,7 @@ class IpmiHandler: tenant = cfg.tenant if (node, tenant) not in persistent_ipmicmds: try: - await persistent_ipmicmds[(node, tenant)].close_confluent() + persistent_ipmicmds[(node, tenant)].close_confluent() except KeyError: # was no previous session pass try: @@ -830,7 +830,7 @@ class IpmiHandler: async def make_sensor_map(self, sensors=None): if sensors is None: - sensors = await self.ipmicmd.get_sensor_descriptions() + sensors = self.ipmicmd.get_sensor_descriptions() async for sensor in sensors: resourcename = sensor['name'] self.sensormap[simplify_name(resourcename)] = resourcename @@ -1075,7 +1075,7 @@ class IpmiHandler: volsfound = True volumes.append(vol) if not volsfound: - self.output.put(msg.ConfluentTargetNotFound( + await self.output.put(msg.ConfluentTargetNotFound( self.node, "No volume named '{0}' found".format(volname))) return await self.ipmicmd.remove_storage_configuration(toremove) @@ -1128,14 +1128,14 @@ class IpmiHandler: vol.status, arrname)) return else: - self._show_storage(storelem[:1] + [vol['name']]) + await self._show_storage(storelem[:1] + [vol['name']]) async def _update_storage(self, storelem): if storelem[0] == 'disks': if len(storelem) == 1: raise exc.InvalidArgumentException('Must target a disk') - self.set_disk(storelem[-1], - self.inputdata.inputbynode[self.node]) + await self.set_disk(storelem[-1], + self.inputdata.inputbynode[self.node]) await self._show_storage(storelem) async def _show_storage(self, storelem): @@ -1292,7 +1292,7 @@ class IpmiHandler: async def list_sensors(self): try: - sensors = await self.ipmicmd.get_sensor_descriptions() + sensors = [sensor async for sensor in self.ipmicmd.get_sensor_descriptions()] except pygexc.IpmiException: await self.output.put(msg.ConfluentTargetTimeout(self.node)) return From b5e0e9f9e4cda42f7d998b333d73320d20e48dea Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sun, 12 Jul 2026 23:56:03 +0200 Subject: [PATCH 08/17] Fix asynchronous console and shell contracts --- confluent_server/confluent/asynchttp.py | 4 +-- confluent_server/confluent/consoleserver.py | 5 ++- confluent_server/confluent/httpapi.py | 37 ++++++++++++--------- confluent_server/confluent/shellmodule.py | 17 ++++++---- confluent_server/confluent/shellserver.py | 6 ++-- 5 files changed, 38 insertions(+), 31 deletions(-) diff --git a/confluent_server/confluent/asynchttp.py b/confluent_server/confluent/asynchttp.py index 25a89938..ff94029c 100644 --- a/confluent_server/confluent/asynchttp.py +++ b/confluent_server/confluent/asynchttp.py @@ -57,8 +57,8 @@ class AsyncTermRelation(object): self.asynchdl = asynchdl self.termid = termid - def got_data(self, data): - self.asynchdl.add(self.termid, data) + async def got_data(self, data): + await self.asynchdl.add(self.termid, data) class AsyncSession(object): diff --git a/confluent_server/confluent/consoleserver.py b/confluent_server/confluent/consoleserver.py index d9707a36..691afa74 100644 --- a/confluent_server/confluent/consoleserver.py +++ b/confluent_server/confluent/consoleserver.py @@ -451,8 +451,7 @@ class ConsoleHandler(object): await self._send_rcpts({'deleting': True}) await self._disconnect() if self._console: - - self._console.close() + await self._console.close() self._console = None if self.connectionthread: self.connectionthread.cancel() @@ -871,7 +870,7 @@ class ConsoleSession(object): await self.conshdl.attachsession(self) self.write = self.conshdl.write - def got_data(self, data): + async def got_data(self, data): """Receive data from console and buffer If the caller does not provide a callback and instead will be polling diff --git a/confluent_server/confluent/httpapi.py b/confluent_server/confluent/httpapi.py index 27ac3c98..c1c6ded3 100644 --- a/confluent_server/confluent/httpapi.py +++ b/confluent_server/confluent/httpapi.py @@ -585,21 +585,24 @@ async def wsock_handler(req): if asess: await asess.destroy() return rsp - if '/console/session' in ws.path or '/shell/sessions/' in ws.path: - def datacallback(data): - ws.send(websockify_data(data)) - geom = ws.wait() - geom = geom[1:] + path = req.rel_url.path + if '/console/session' in path or '/shell/sessions/' in path: + async def datacallback(data): + await rsp.send_str(websockify_data(data)) + geom = await rsp.receive() + if geom.type != WSMsgType.TEXT: + return rsp + geom = geom.data[1:] geom = json.loads(geom) width = geom['width'] height = geom['height'] skipreplay = geom.get('skipreplay', False) #hard bake JSON into this path, do not support other incarnations - if '/console/session' in ws.path: - prefix, _, _ = ws.path.partition('/console/session') + if '/console/session' in path: + prefix, _, _ = path.partition('/console/session') shellsession = False - elif '/shell/sessions/' in ws.path: - prefix, _, _ = ws.path.partition('/shell/sessions') + elif '/shell/sessions/' in path: + prefix, _, _ = path.partition('/shell/sessions') shellsession = True _, _, nodename = prefix.rpartition('/') @@ -618,11 +621,12 @@ async def wsock_handler(req): ) except exc.NotFoundException: return - clientmsg = ws.wait() + clientmsg = await rsp.receive() try: - while clientmsg is not None: + while clientmsg.type == WSMsgType.TEXT: + clientmsg = clientmsg.data if clientmsg[0] == ' ': - consession.write(clientmsg[1:]) + await consession.write(clientmsg[1:]) elif clientmsg[0] == '!': cmd = json.loads(clientmsg[1:]) action = cmd.get('action', None) @@ -632,10 +636,11 @@ async def wsock_handler(req): consession.resize( width=cmd['width'], height=cmd['height']) elif clientmsg[0] == '?': - ws.send(u'?') - clientmsg = ws.wait() + await rsp.send_str(u'?') + clientmsg = await rsp.receive() finally: - consession.destroy() + await consession.destroy() + return rsp async def resourcehandler(request): @@ -927,7 +932,7 @@ async def resourcehandler_backend(req, make_response): await rsp.write(json.dumps({'session': querydict['session']})) return rsp # client has requests to send or receive, not both... elif 'closesession' in querydict: - consolesessions[querydict['session']]['session'].destroy() + await consolesessions[querydict['session']]['session'].destroy() del consolesessions[querydict['session']] rsp = await make_response('application/json', 200) await rsp.write(b'{"sessionclosed": true}') diff --git a/confluent_server/confluent/shellmodule.py b/confluent_server/confluent/shellmodule.py index 89bfb7e7..34bc7e37 100644 --- a/confluent_server/confluent/shellmodule.py +++ b/confluent_server/confluent/shellmodule.py @@ -30,6 +30,7 @@ import fcntl import os import pty import random +import select import subprocess @@ -54,12 +55,12 @@ class ExecConsole(conapi.Console): try: somedata = os.read(self._master, 128) while somedata: - self._datacallback(somedata) + await self._datacallback(somedata) await asyncio.sleep(0) somedata = os.read(self._master, 128) except OSError as e: if e.errno == 5: - self._datacallback(conapi.ConsoleEvent.Disconnect) + await self._datacallback(conapi.ConsoleEvent.Disconnect) self.subproc = None return if e.errno != 11: @@ -68,7 +69,7 @@ class ExecConsole(conapi.Console): try: somedata = self.subproc.stderr.read() while somedata: - self._datacallback(somedata) + await self._datacallback(somedata) await asyncio.sleep(0) somedata = self.subproc.stderr.read() except IOError as e: @@ -76,10 +77,10 @@ class ExecConsole(conapi.Console): raise childstate = self.subproc.poll() if childstate is not None: - self._datacallback(conapi.ConsoleEvent.Disconnect) + await self._datacallback(conapi.ConsoleEvent.Disconnect) self.subproc = None - def connect(self, callback): + async def connect(self, callback): self._datacallback = callback master, slave = pty.openpty() self._master = master @@ -90,14 +91,16 @@ class ExecConsole(conapi.Console): stderr=subprocess.PIPE, close_fds=True) except OSError: print("Unable to execute " + self.executable + " (permissions?)") - self.close() + os.close(master) + os.close(slave) + self._master = None return os.close(slave) fcntl.fcntl(master, fcntl.F_SETFL, os.O_NONBLOCK) fcntl.fcntl(self.subproc.stderr.fileno(), fcntl.F_SETFL, os.O_NONBLOCK) self.readerthread = tasks.spawn(self.relaydata()) - def write(self, data): + async def write(self, data): os.write(self._master, data) async def close(self): diff --git a/confluent_server/confluent/shellserver.py b/confluent_server/confluent/shellserver.py index cdcdcfe2..3bb6a086 100644 --- a/confluent_server/confluent/shellserver.py +++ b/confluent_server/confluent/shellserver.py @@ -54,7 +54,7 @@ class _ShellHandler(consoleserver.ConsoleHandler): _reaper = tasks.spawn(reapsessions()) - def check_collective(self, attrvalue): + async def check_collective(self, attrvalue): return def log(self, *args, **kwargs): @@ -69,7 +69,7 @@ class _ShellHandler(consoleserver.ConsoleHandler): # #retdata, connstate = await super(_ShellHandler, self).get_recent() # return '', {} # connstate - def _got_disconnected(self): + async def _got_disconnected(self): self.connectstate = 'closed' tasks.spawn(self._bgdisconnect()) @@ -77,7 +77,7 @@ class _ShellHandler(consoleserver.ConsoleHandler): await self._send_rcpts({'connectstate': self.connectstate}) for session in list(self.livesessions): await session.destroy() - self.feedbuffer('\x1bc') + await self.feedbuffer('\x1bc') From 38746b19d5a90be82199e1612b0865c64df97b01 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Sun, 12 Jul 2026 23:56:33 +0200 Subject: [PATCH 09/17] Import signal for SSH agent cleanup --- confluent_server/confluent/sshutil.py | 1 + 1 file changed, 1 insertion(+) diff --git a/confluent_server/confluent/sshutil.py b/confluent_server/confluent/sshutil.py index 53ec9d26..7f0a81b9 100644 --- a/confluent_server/confluent/sshutil.py +++ b/confluent_server/confluent/sshutil.py @@ -7,6 +7,7 @@ import confluent.collective.manager as collective import confluent.util as util import glob import os +import signal import shutil import subprocess import tempfile From 0165fc993569da8ca2382e0f0a4a372ba9c1c522 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 13 Jul 2026 00:59:17 +0200 Subject: [PATCH 10/17] Fix IPMI coroutine result handling --- confluent_server/aiohmi/ipmi/command.py | 12 +++++----- confluent_server/aiohmi/ipmi/console.py | 4 ++-- .../aiohmi/ipmi/private/session.py | 23 +++++++++---------- 3 files changed, 19 insertions(+), 20 deletions(-) diff --git a/confluent_server/aiohmi/ipmi/command.py b/confluent_server/aiohmi/ipmi/command.py index a7f4723b..7c08f4ef 100644 --- a/confluent_server/aiohmi/ipmi/command.py +++ b/confluent_server/aiohmi/ipmi/command.py @@ -329,8 +329,8 @@ class Command(object): await self.oem_init() if hasattr(self._oem, 'set_power'): - return self._oem.set_power(powerstate, - bridge_request=bridge_request) + return await self._oem.set_power(powerstate, + bridge_request=bridge_request) if hasattr(self._oem, 'process_power_state'): powerstate = self._oem.process_power_state( @@ -675,7 +675,7 @@ class Command(object): await self.init_sdr() for fruid in self._sdr.fru: if self._sdr.fru[fruid].fru_name == component: - return self._oem.process_fru(fru.FRU( + return await self._oem.process_fru(fru.FRU( ipmicmd=self, fruid=fruid, sdr=self._sdr.fru[fruid]).info, component) return await self._oem.get_inventory_of_component(component) @@ -854,9 +854,9 @@ class Command(object): else: raise Exception("Unrecognized data format " + repr(fetchdata)) - def get_extended_bmc_configuration(self): - self.oem_init() - return self._oem.get_extended_bmc_configuration() + async def get_extended_bmc_configuration(self): + await self.oem_init() + return await self._oem.get_extended_bmc_configuration() async def get_bmc_configuration(self): await self.oem_init() diff --git a/confluent_server/aiohmi/ipmi/console.py b/confluent_server/aiohmi/ipmi/console.py index 294a84a9..b7465ea4 100644 --- a/confluent_server/aiohmi/ipmi/console.py +++ b/confluent_server/aiohmi/ipmi/console.py @@ -133,7 +133,7 @@ class Console(object): response['code']) return if 'error' in response: - self._print_error(response['error']) + await self._print_error(response['error']) return self.activated = True # data[0:3] is reserved except for the test mode, which we don't use @@ -381,7 +381,7 @@ class Console(object): await self.send_payload(ackpayload, retry=False) except exc.IpmiException: # if the session is broken, then close the SOL session - self.close() + await self.close() if self.myseq != 0 and ackseq == self.myseq: # the bmc has something # to say about last xmit self.awaitingack = False diff --git a/confluent_server/aiohmi/ipmi/private/session.py b/confluent_server/aiohmi/ipmi/private/session.py index 0725fdf3..c6891184 100644 --- a/confluent_server/aiohmi/ipmi/private/session.py +++ b/confluent_server/aiohmi/ipmi/private/session.py @@ -1023,11 +1023,11 @@ class Session(object): # tried ipmi 2.0 against a 1.5 which should work, but some bmcs # thought 'reserved' meant 'must be zero' self.ipmi15only = 1 - return self._get_channel_auth_cap() + return await self._get_channel_auth_cap() mysuffix = " while trying to get channel authentication capabalities" errstr = get_ipmi_error(response, suffix=mysuffix) if errstr: - self.onlogon({'error': errstr}) + await self.onlogon({'error': errstr}) return data = response['data'] self.currentchannel = data[0] @@ -1035,7 +1035,7 @@ class Session(object): self.ipmiversion = 2.0 if self.ipmiversion == 1.5: if not (data[1] & 0b100): - self.onlogon( + await self.onlogon( {'error': "MD5 required but not enabled/available on target BMC"}) return @@ -1047,12 +1047,11 @@ class Session(object): errstr = get_ipmi_error(response, suffix=" while getting session challenge") if errstr: - self.onlogon({'error': errstr}) - return + return self.onlogon({'error': errstr}) data = response['data'] self.sessionid = struct.unpack(" Date: Mon, 13 Jul 2026 00:59:38 +0200 Subject: [PATCH 11/17] Fix OEM asynchronous operation dispatch --- .../aiohmi/ipmi/oem/lenovo/handler.py | 13 ++++--- .../aiohmi/ipmi/oem/lenovo/imm.py | 34 +++++++++---------- .../aiohmi/ipmi/oem/lenovo/nextscale.py | 18 +++++----- confluent_server/aiohmi/redfish/command.py | 14 ++++---- .../aiohmi/redfish/oem/dell/idrac.py | 10 +++--- .../aiohmi/redfish/oem/lenovo/tsma.py | 2 +- .../aiohmi/redfish/oem/lenovo/xcc.py | 24 ++++++------- .../aiohmi/redfish/oem/lenovo/xcc3.py | 12 +++---- 8 files changed, 62 insertions(+), 65 deletions(-) diff --git a/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py b/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py index 65186e69..40fc0380 100755 --- a/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py +++ b/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py @@ -493,8 +493,7 @@ class OEMHandler(generic.OEMHandler): if await self.has_tsm() or await self.has_ami() or await self.has_asrock(): # Thinkserver with TSM if not self.oem_inventory_info: - async for desc in self._collect_tsm_inventory(): - yield desc + await self._collect_tsm_inventory() for compname in self.oem_inventory_info: yield compname elif await self.has_imm(): @@ -767,7 +766,7 @@ class OEMHandler(generic.OEMHandler): except (AttributeError, KeyError, IndexError): pass if await self.has_xcc() and name and name.startswith('PSU '): - self.immhandler.augment_psu_info(fru, name) + await self.immhandler.augment_psu_info(fru, name) if (await self.has_xcc() and 'memory_type' in fru and fru['memory_type'] == 'Unknown'): await self.immhandler.fetch_dimm(name, fru) @@ -1028,14 +1027,14 @@ class OEMHandler(generic.OEMHandler): async def add_extra_net_configuration(self, netdata, channel=None): if await self.has_tsm(): - ipv6_addr = await self.ipmicmd.raw_command( + ipv6_addr = (await self.ipmicmd.raw_command( netfn=0x0c, command=0x02, - data=(0x01, 0xc5, 0x00, 0x00))["data"][1:] + data=(0x01, 0xc5, 0x00, 0x00)))['data'][1:] if not ipv6_addr: return - rspdata = await self.ipmicmd.raw_command( + rspdata = (await self.ipmicmd.raw_command( netfn=0xc, command=0x02, - data=(0x1, 0xc6, 0, 0))['data'] + data=(0x1, 0xc6, 0, 0)))['data'] ipv6_prefix_ba = bytearray(rspdata) ipv6_prefix = ipv6_prefix_ba[1] diff --git a/confluent_server/aiohmi/ipmi/oem/lenovo/imm.py b/confluent_server/aiohmi/ipmi/oem/lenovo/imm.py index 9451cce8..5fd5848a 100644 --- a/confluent_server/aiohmi/ipmi/oem/lenovo/imm.py +++ b/confluent_server/aiohmi/ipmi/oem/lenovo/imm.py @@ -839,7 +839,7 @@ class IMMClient(object): hwmap[aname] = bdata self.datacache['lenovo_cached_hwmap'] = (hwmap, util._monotonic_time()) - self.weblogout() + await self.weblogout() return hwmap async def get_firmware_inventory(self, bmcver, components, category): @@ -1434,7 +1434,7 @@ class XCCClient(IMMClient): if controller != disk.id[0]: raise pygexc.UnsupportedFunctionality( 'Cannot span arrays across controllers') - raidmap = self._raid_number_map(controller) + raidmap = await self._raid_number_map(controller) if not raidmap: raise pygexc.InvalidParameterValue( 'There are no available drives for a new array') @@ -1487,21 +1487,21 @@ class XCCClient(IMMClient): # TODO(): adding new volume to existing array would be here pass - def _make_jbod(self, disk, realcfg): + async def _make_jbod(self, disk, realcfg): currstatus = self._get_status(disk, realcfg) if currstatus.lower() == 'jbod': return - self._make_available(disk, realcfg) - self._set_drive_state(disk, 16) + await self._make_available(disk, realcfg) + await self._set_drive_state(disk, 16) - def _make_global_hotspare(self, disk, realcfg): + async def _make_global_hotspare(self, disk, realcfg): currstatus = self._get_status(disk, realcfg) if currstatus.lower() == 'global hot spare': return - self._make_available(disk, realcfg) - self._set_drive_state(disk, 1) + await self._make_available(disk, realcfg) + await self._set_drive_state(disk, 1) - def _make_available(self, disk, realcfg): + async def _make_available(self, disk, realcfg): # 8 if jbod, 4 if hotspare.., leave alone if already... currstatus = self._get_status(disk, realcfg) newstate = None @@ -1511,7 +1511,7 @@ class XCCClient(IMMClient): newstate = 4 elif currstatus.lower() == 'jbod': newstate = 8 - self._set_drive_state(disk, newstate) + await self._set_drive_state(disk, newstate) def _get_status(self, disk, realcfg): for cfgdisk in realcfg.disks: @@ -1536,7 +1536,7 @@ class XCCClient(IMMClient): wc = await self.wc() rsp = await wc.grab_json_response( '/api/function', {'raidlink_ClearRaidConf': '1'}) - self.weblogout() + await self.weblogout() if rsp['return'] != 0: raise Exception('Unexpected return to clear config: ' + repr(rsp)) @@ -1560,10 +1560,10 @@ class XCCClient(IMMClient): if rsp.get('return', -1) != 0: raise Exception( 'Unexpected return to volume deletion: ' + repr(rsp)) - self._wait_storage_async() + await self._wait_storage_async() for disk in cfgspec.disks: await self._make_available(disk, realcfg) - self.weblogout() + await self.weblogout() async def apply_storage_configuration(self, cfgspec): realcfg = await self.get_storage_configuration(False) @@ -2168,7 +2168,7 @@ class XCCClient(IMMClient): retry = 3 while not complete and retry > 0: try: - pgress, status = self.grab_redfish_response_with_status( + pgress, status = await self.grab_redfish_response_with_status( monitorurl) except socket.timeout: pgress = None @@ -2205,10 +2205,10 @@ class XCCClient(IMMClient): return 'complete' return 'pending' finally: - self.grab_redfish_response_with_status( + await self.grab_redfish_response_with_status( '/redfish/v1/UpdateService', {'HttpPushUriTargetsBusy': False}, method='PATCH') - self.grab_redfish_response_with_status( + await self.grab_redfish_response_with_status( '/redfish/v1/UpdateService', {'HttpPushUriTargets': []}, method='PATCH') @@ -2410,7 +2410,7 @@ class XCCClient(IMMClient): errmsg = repr(rsp) if rsp else wc.lastjsonerror raise Exception( 'Unexpected result from PCI select: ' + errmsg) - self.set_property('/v2/ibmc/uefi/force-inventory', 1) + await self.set_property('/v2/ibmc/uefi/force-inventory', 1) else: await self._refresh_token() rsp = await wc.grab_json_response( diff --git a/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py b/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py index 1c15d507..53fe238e 100644 --- a/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py +++ b/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py @@ -671,14 +671,14 @@ class SMMClient(object): for bayn in range(1, numbays + 1): if fnmatch.fnmatch('bay{0}_user_cap'.format(bayn), key.lower()): - self.set_bay_cap(bayn, changeset[key]['value']) + await self.set_bay_cap(bayn, changeset[key]['value']) if fnmatch.fnmatch( 'bay{0}_user_cap_active'.format(bayn), key.lower()): - self.set_bay_cap_active(bayn, changeset[key]['value']) + await self.set_bay_cap_active(bayn, changeset[key]['value']) if fnmatch.fnmatch('chassis_user_cap', key.lower()): - self.set_bay_cap(numbays + 1, changeset[key]['value']) + await self.set_bay_cap(numbays + 1, changeset[key]['value']) if fnmatch.fnmatch('chassis_user_cap_active', key.lower()): - self.set_bay_cap_active(numbays + 1, changeset[key]['value']) + await self.set_bay_cap_active(numbays + 1, changeset[key]['value']) if fnmatch.fnmatch('fanspeed', key.lower()): for mode in self.fanmodes: byteval = mode @@ -816,10 +816,10 @@ class SMMClient(object): async def process_fru(self, fru): smmv1 = self.smm_variant & 0xf0 == 0 # TODO(jjohnson2): can also get EIOM, SMM, and riser data if warranted - snum = bytes(await self.ipmicmd.raw_command( - netfn=0x32, command=0xb0, data=(5, 1))['data'][:]) - mnum = bytes(await self.ipmicmd.raw_command( - netfn=0x32, command=0xb0, data=(5, 0))['data'][:]) + snum = bytes((await self.ipmicmd.raw_command( + netfn=0x32, command=0xb0, data=(5, 1)))['data'][:]) + mnum = bytes((await self.ipmicmd.raw_command( + netfn=0x32, command=0xb0, data=(5, 0)))['data'][:]) if not smmv1: snum = snum[2:] mnum = mnum[2:] @@ -1115,5 +1115,5 @@ class SMMClient(object): async def wc(self): if (not self._wc or self._wc.broken or self._wc.vintage < util._monotonic_time() + 30): - self._wc = await self.get_webclient() + self._wc = self.get_webclient() return self._wc diff --git a/confluent_server/aiohmi/redfish/command.py b/confluent_server/aiohmi/redfish/command.py index e602393f..dd0cdfa4 100644 --- a/confluent_server/aiohmi/redfish/command.py +++ b/confluent_server/aiohmi/redfish/command.py @@ -1035,7 +1035,7 @@ class Command(object): 'Address': static_gateway, }] if patch: - nicurl = self._get_bmc_nic_url(name) + nicurl = await self._get_bmc_nic_url(name) await self._do_web_request(nicurl, patch, 'PATCH') async def set_net_configuration(self, ipv4_address=None, ipv4_configuration=None, @@ -1458,22 +1458,22 @@ class Command(object): del self.wc.stdheaders['Authorization'] return for vmurl in vmurls: - vminfo = self._do_web_request(vmurl, cache=False) + vminfo = await self._do_web_request(vmurl, cache=False) if vminfo.get('ConnectedVia', None) != 'NotConnected': continue inserturl = vminfo.get( 'Actions', {}).get( '#VirtualMedia.InsertMedia', {}).get('target', None) if inserturl: - self._do_web_request(inserturl, {'Image': url}) + await self._do_web_request(inserturl, {'Image': url}) else: try: - self._do_web_request(vmurl, - {'Image': url, 'Inserted': True}, - 'PATCH') + await self._do_web_request(vmurl, + {'Image': url, 'Inserted': True}, + 'PATCH') except exc.RedfishError as re: if re.msgid.endswith(u'PropertyUnknown'): - self._do_web_request(vmurl, {'Image': url}, 'PATCH') + await self._do_web_request(vmurl, {'Image': url}, 'PATCH') else: raise break diff --git a/confluent_server/aiohmi/redfish/oem/dell/idrac.py b/confluent_server/aiohmi/redfish/oem/dell/idrac.py index e52eacdf..9863dae7 100644 --- a/confluent_server/aiohmi/redfish/oem/dell/idrac.py +++ b/confluent_server/aiohmi/redfish/oem/dell/idrac.py @@ -17,8 +17,8 @@ import aiohmi.redfish.oem.generic as generic class OEMHandler(generic.OEMHandler): - def set_bootdev(self, bootdev, persist=False, uefiboot=None, - fishclient=None): + async def set_bootdev(self, bootdev, persist=False, uefiboot=None, + fishclient=None): # gleaned from web console, under configuration, system settings, # hardware, first boot device. iDrac presumes that the standard # explicitly refers only to physical devices. I think the intent @@ -26,8 +26,8 @@ class OEMHandler(generic.OEMHandler): # the 'physical' standard to the vFDD/VCD-DVD seen in the idrac # web gui if bootdev not in ('floppy', 'cd'): - return super(OEMHandler, self).set_bootdev(bootdev, persist, - uefiboot, fishclient) + return await super(OEMHandler, self).set_bootdev(bootdev, persist, + uefiboot, fishclient) payload = {'Attributes': {}} if persist: payload['Attributes']['ServerBoot.1.BootOnce'] = 'Disabled' @@ -37,7 +37,7 @@ class OEMHandler(generic.OEMHandler): payload['Attributes']['ServerBoot.1.FirstBootDevice'] = 'vFDD' elif bootdev == 'cd': payload['Attributes']['ServerBoot.1.FirstBootDevice'] = 'VCD-DVD' - fishclient._do_web_request( + await fishclient._do_web_request( '/redfish/v1/Managers/iDRAC.Embedded.1/Attributes', payload, method='PATCH') return {'bootdev': bootdev} diff --git a/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py b/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py index 49d0fe8c..0136bc13 100644 --- a/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py +++ b/confluent_server/aiohmi/redfish/oem/lenovo/tsma.py @@ -186,7 +186,7 @@ class TsmHandler(generic.OEMHandler): if 'dns_domain'.startswith(key.lower()): dnschgs['domain_name'] = currval if 'password_complexity'.startswith(key.lower()): - self._set_pass_complexity(currval, wc) + await self._set_pass_complexity(currval, wc) if 'password_login_failures'.startswith(key.lower()): await self._set_pass_lockout(currval, wc) if dnschgs: diff --git a/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py b/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py index 09fd1a11..8dad831f 100644 --- a/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py +++ b/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py @@ -184,7 +184,7 @@ class OEMHandler(generic.OEMHandler): try: self.fwo = await self.fwc.get_fw_options(fetchimm=fetchimm) except config.Unsupported: - return super(OEMHandler, self).get_system_configuration( + return await super(OEMHandler, self).get_system_configuration( hideadvanced, fishclient) except Exception: raise Exception('%s failed to retrieve UEFI configuration' @@ -831,7 +831,7 @@ class OEMHandler(generic.OEMHandler): wc.set_header('X-XSRF-TOKEN', cookie.value) wc.vintage = util._monotonic_time() - def _make_available(self, disk, realcfg): + async def _make_available(self, disk, realcfg): # 8 if jbod, 4 if hotspare.., leave alone if already... currstatus = self._get_status(disk, realcfg) newstate = None @@ -841,21 +841,21 @@ class OEMHandler(generic.OEMHandler): newstate = 4 elif currstatus.lower() == 'jbod': newstate = 8 - self._set_drive_state(disk, newstate) + await self._set_drive_state(disk, newstate) - def _make_jbod(self, disk, realcfg): + async def _make_jbod(self, disk, realcfg): currstatus = self._get_status(disk, realcfg) if currstatus.lower() == 'jbod': return - self._make_available(disk, realcfg) - self._set_drive_state(disk, 16) + await self._make_available(disk, realcfg) + await self._set_drive_state(disk, 16) - def _make_global_hotspare(self, disk, realcfg): + async def _make_global_hotspare(self, disk, realcfg): currstatus = self._get_status(disk, realcfg) if currstatus.lower() == 'global hot spare': return - self._make_available(disk, realcfg) - self._set_drive_state(disk, 1) + await self._make_available(disk, realcfg) + await self._set_drive_state(disk, 1) def _get_status(self, disk, realcfg): for cfgdisk in realcfg.disks: @@ -978,7 +978,7 @@ class OEMHandler(generic.OEMHandler): pass async def _create_array(self, pool): - params = self._parse_array_spec(pool) + params = await self._parse_array_spec(pool) cid = params['controller'].split(',')[0] cslotno = params['controller'].split(',')[1] url = '/api/function/raid_conf?params=raidlink_GetDefaultVolProp' @@ -1443,7 +1443,7 @@ class OEMHandler(generic.OEMHandler): async def update_firmware_backend(self, filename, data=None, progress=None, bank=None): - self._refresh_token() + await self._refresh_token() wc = await self.wc() rsv = await wc.grab_json_response('/api/providers/fwupdate', json.dumps( {'UPD_WebReserve': 1})) @@ -1470,7 +1470,7 @@ class OEMHandler(generic.OEMHandler): raise Exception('File is larger than supported') raise Exception('Unexpected result:' + repr(rsp)) uploadstate = rsp['state'] - self._refresh_token() + await self._refresh_token() while uploadstate != 'done': rsp = await wc.grab_json_response( '/upload/progress?X-Progress-ID={0}'.format(xid)) diff --git a/confluent_server/aiohmi/redfish/oem/lenovo/xcc3.py b/confluent_server/aiohmi/redfish/oem/lenovo/xcc3.py index a109931c..342b5e92 100644 --- a/confluent_server/aiohmi/redfish/oem/lenovo/xcc3.py +++ b/confluent_server/aiohmi/redfish/oem/lenovo/xcc3.py @@ -432,7 +432,7 @@ class OEMHandler(generic.OEMHandler): await self._make_available(disk, realcfg) self._urlcache.clear() - def _parse_array_spec(self, arrayspec): + async def _parse_array_spec(self, arrayspec): controller = None if arrayspec.disks: for disk in list(arrayspec.disks) + list(arrayspec.hotspares): @@ -441,7 +441,7 @@ class OEMHandler(generic.OEMHandler): if controller != disk.id[0]: raise pygexc.UnsupportedFunctionality( 'Cannot span arrays across controllers') - raidmap = self._raid_number_map(controller) + raidmap = await self._raid_number_map(controller) if not raidmap: raise pygexc.InvalidParameterValue( 'No RAID Type supported on this controller') @@ -511,7 +511,7 @@ class OEMHandler(generic.OEMHandler): return themap async def _create_array(self, pool): - params = self._parse_array_spec(pool) + params = await self._parse_array_spec(pool) cid = params['controller'].split(',')[0] c_capabilities, code = await self.webclient.grab_json_response_with_status( f'/redfish/v1/Systems/1/Storage/{cid}/Volumes/Capabilities') @@ -906,7 +906,7 @@ class OEMHandler(generic.OEMHandler): cache=False) rawsettings = rawsettings.get('Attributes', {}) pendingsettings = {} - ret = self._set_redfish_settings( + ret = await self._set_redfish_settings( changeset, fishclient, currsettings, rawsettings, pendingsettings, self.lenovobmcattrdeps, reginfo, '/redfish/v1/Managers/1/Oem/Lenovo/BMCSettings') @@ -1039,7 +1039,7 @@ class OEMHandler(generic.OEMHandler): if usbsettings: await self.apply_usb_configuration(usbsettings) if bmchangeset: - self._set_xcc3_settings(bmchangeset, self) + await self._set_xcc3_settings(bmchangeset, self) async def apply_usb_configuration(self, usbsettings): bmcattribs = {} @@ -1220,5 +1220,3 @@ class OEMHandler(generic.OEMHandler): 'Name': 'HPM-FPGA Pending', 'build': pendinghpm} raise pygexc.BypassGenericBehavior() - - From 3c6e7d202f03c86a5ed1fd60efc9d890c255bcbb Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 13 Jul 2026 01:00:03 +0200 Subject: [PATCH 12/17] Await BMC discovery configuration operations --- .../confluent/discovery/handlers/bmc.py | 30 +++++++++---------- .../confluent/discovery/handlers/imm.py | 7 ++--- .../confluent/discovery/handlers/xcc.py | 5 ++-- 3 files changed, 20 insertions(+), 22 deletions(-) diff --git a/confluent_server/confluent/discovery/handlers/bmc.py b/confluent_server/confluent/discovery/handlers/bmc.py index db083822..fcb59ecc 100644 --- a/confluent_server/confluent/discovery/handlers/bmc.py +++ b/confluent_server/confluent/discovery/handlers/bmc.py @@ -108,7 +108,7 @@ class NodeHandler(generic.NodeHandler): # Use existing account that has been created newuserslot = uid if newpass != passwd: # don't mess with existing if no change - ic.set_user_password(newuserslot, password=newpass) + await ic.set_user_password(newuserslot, password=newpass) ic = await self._get_ipmicmd(user, passwd) if vc: ic.register_key_handler(vc) @@ -118,8 +118,8 @@ class NodeHandler(generic.NodeHandler): if newuserslot < 2: newuserslot = 2 if newpass != passwd: # don't mess with existing if no change - ic.set_user_password(newuserslot, password=newpass) - ic.set_user_name(newuserslot, newuser) + await ic.set_user_password(newuserslot, password=newpass) + await ic.set_user_name(newuserslot, newuser) if havecustomcreds: ic = await self._get_ipmicmd(user, passwd) if vc: @@ -132,17 +132,17 @@ class NodeHandler(generic.NodeHandler): for uid in currusers: if uid != newuserslot: if uid <= lockedusers: # we cannot delete, settle for disable - ic.disable_user(uid, 'disable') + await ic.disable_user(uid, 'disable') else: # lead with the most critical thing, removing user access - ic.set_user_access(uid, channel=None, callback=False, - link_auth=False, ipmi_msg=False, - privilege_level='no_access') + await ic.set_user_access(uid, channel=None, callback=False, + link_auth=False, ipmi_msg=False, + privilege_level='no_access') # next, try to disable the password - ic.set_user_password(uid, mode='disable', password=None) + await ic.set_user_password(uid, mode='disable', password=None) # ok, now we can be less paranoid try: - ic.user_delete(uid) + await ic.user_delete(uid) except pygexc.IpmiException as ie: if ie.ipmicode != 0xd5: # some response to the 0xff # name... @@ -165,14 +165,14 @@ class NodeHandler(generic.NodeHandler): netconfig = await netutil.get_nic_config(cfg, nodename, ip=newip) plen = netconfig['prefix'] newip = '{0}/{1}'.format(newip, plen) - currcfg = ic.get_net_configuration() + currcfg = await ic.get_net_configuration() if currcfg['ipv4_address'] != newip: # do not change the ipv4_config if the current config looks # like it is already accurate - ic.set_net_configuration(ipv4_address=newip, - ipv4_configuration='static', - ipv4_gateway=netconfig[ - 'ipv4_gateway']) + await ic.set_net_configuration(ipv4_address=newip, + ipv4_configuration='static', + ipv4_gateway=netconfig[ + 'ipv4_gateway']) elif self.ipaddr.startswith('fe80::'): await cfg.set_node_attributes( {nodename: {'hardwaremanagement.manager': self.ipaddr}}) @@ -180,5 +180,5 @@ class NodeHandler(generic.NodeHandler): raise exc.TargetEndpointUnreachable( 'hardwaremanagement.manager must be set to desired address') if reset: - ic.reset_bmc() + await ic.reset_bmc() return ic diff --git a/confluent_server/confluent/discovery/handlers/imm.py b/confluent_server/confluent/discovery/handlers/imm.py index e7099860..a6946e81 100644 --- a/confluent_server/confluent/discovery/handlers/imm.py +++ b/confluent_server/confluent/discovery/handlers/imm.py @@ -89,13 +89,13 @@ class NodeHandler(bmchandler.NodeHandler): guiddata = await ipmicmd.xraw_command(netfn=6, command=8) self.info['uuid'] = pygutil.decode_wireformat_uuid( guiddata['data']).lower() - ipmicmd.oem_init() - bayid = ipmicmd._oem.immhandler.get_property( + await ipmicmd.oem_init() + bayid = await ipmicmd._oem.immhandler.get_property( '/v2/cmm/sp/7') if not bayid: return self.info['enclosure.bay'] = int(bayid) - smmid = ipmicmd._oem.immhandler.get_property( + smmid = await ipmicmd._oem.immhandler.get_property( '/v2/ibmc/smm/chassis/uuid') if not smmid: return @@ -108,4 +108,3 @@ class NodeHandler(bmchandler.NodeHandler): except pygexc.IpmiException as ie: print(repr(ie)) raise - diff --git a/confluent_server/confluent/discovery/handlers/xcc.py b/confluent_server/confluent/discovery/handlers/xcc.py index d0353a60..391de256 100644 --- a/confluent_server/confluent/discovery/handlers/xcc.py +++ b/confluent_server/confluent/discovery/handlers/xcc.py @@ -171,8 +171,8 @@ class NodeHandler(immhandler.NodeHandler): {'IPMI': {'ProtocolEnabled': True}}, method='PATCH') ipmicmd = None try: - ipmicmd = self._get_ipmicmd(self._currcreds[0], self._currcreds[1]) - ipmicmd.xraw_command(netfn=0x3a, command=0xf1, data=(1,)) + ipmicmd = await self._get_ipmicmd(self._currcreds[0], self._currcreds[1]) + await ipmicmd.xraw_command(netfn=0x3a, command=0xf1, data=(1,)) except pygexc.IpmiException as e: if (e.ipmicode != 193 and 'Unauthorized name' not in str(e) and 'Incorrect password' not in str(e) and @@ -711,4 +711,3 @@ async def remote_nodecfg(nodename, cfm): else: nh = xcc3handler.NodeHandler(info, cfm) await nh.config(nodename) - From 430260becf481180e080b1b1721da6ba599c65b6 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 13 Jul 2026 01:00:17 +0200 Subject: [PATCH 13/17] Await collective address propagation --- confluent_server/confluent/collective/manager.py | 4 ++-- confluent_server/confluent/config/configmanager.py | 10 +++++----- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/confluent_server/confluent/collective/manager.py b/confluent_server/confluent/collective/manager.py index 4cdedeef..5b71e917 100644 --- a/confluent_server/confluent/collective/manager.py +++ b/confluent_server/confluent/collective/manager.py @@ -656,8 +656,8 @@ async def handle_connection(connection, cert, request, local=False): retrythread = None async with leader_init: cnn = tlvdata.get_socket(connection) - cfm.update_collective_address(request['name'], - cnn.getpeername()[0]) + await cfm.update_collective_address(request['name'], + cnn.getpeername()[0]) await tlvdata.send(connection, cfm._dump_keys(None, False)) await tlvdata.send(connection, cfm._cfgstore['collective']) await tlvdata.send(connection, {'confluent_uuid': cfm.get_global('confluent_uuid')}) # cfm.get_globals()) diff --git a/confluent_server/confluent/config/configmanager.py b/confluent_server/confluent/config/configmanager.py index d5fef21a..6e743a5a 100644 --- a/confluent_server/confluent/config/configmanager.py +++ b/confluent_server/confluent/config/configmanager.py @@ -743,7 +743,7 @@ async def relay_slaved_requests(name, listener): _newquorum = has_quorum() _hasquorum = _newquorum if _hasquorum and _pending_collective_updates: - apply_pending_collective_updates() + await apply_pending_collective_updates() msg = await lh.get_next_msg() while msg: if name not in cfgstreams: @@ -1050,23 +1050,23 @@ def _true_del_collective_member(name, sync=True): _pending_collective_updates = {} -def update_collective_address(name ,address): +async def update_collective_address(name, address): fprint = _cfgstore['collective'][name]['fingerprint'] oldaddress = _cfgstore['collective'][name]['address'] if oldaddress == address: return try: check_quorum() - add_collective_member(name, address, fprint) + await add_collective_member(name, address, fprint) except exc.DegradedCollective: _pending_collective_updates[name] = address -def apply_pending_collective_updates(): +async def apply_pending_collective_updates(): for name in list(_pending_collective_updates): fprint = _cfgstore['collective'][name]['fingerprint'] address = _pending_collective_updates[name] - add_collective_member(name, address, fprint) + await add_collective_member(name, address, fprint) del _pending_collective_updates[name] From ca6a54ab05b8264367837a98377ec385006ade76 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 13 Jul 2026 01:02:23 +0200 Subject: [PATCH 14/17] Fix asynchronous console control dispatch --- confluent_server/confluent/consoleserver.py | 26 ++++++++++----------- confluent_server/confluent/httpapi.py | 6 ++--- confluent_server/confluent/shellserver.py | 2 +- confluent_server/confluent/sockapi.py | 2 +- 4 files changed, 18 insertions(+), 18 deletions(-) diff --git a/confluent_server/confluent/consoleserver.py b/confluent_server/confluent/consoleserver.py index 691afa74..15a2703d 100644 --- a/confluent_server/confluent/consoleserver.py +++ b/confluent_server/confluent/consoleserver.py @@ -225,13 +225,13 @@ class ConsoleHandler(object): 'value', None) if list(configmodule.list_collective()) and not myc: self._is_local = False - self._detach() + await self._detach() await self._disconnect() if myc and myc != collective.get_myname(): # Do not do console connect for nodes managed by another # confluent collective member self._is_local = False - self._detach() + await self._detach() await self._disconnect() else: self._is_local = True @@ -301,9 +301,9 @@ class ConsoleHandler(object): 'or the console simply not having any output since last connection]') self.clearpending = True - def _detach(self): + async def _detach(self): for ses in list(self.livesessions): - ses.detach() + await ses.detach() async def _disconnect(self): if self.connectionthread: @@ -669,7 +669,7 @@ class ProxyConsole(object): def _attribschanged(self, nodeattribs, configmanager, **kwargs): if self.clisession: - self.clisession.detach() + tasks.spawn(self.clisession.detach()) self.clisession = None async def relay_data(self): @@ -742,15 +742,15 @@ class ProxyConsole(object): pass self.clisession = None - def send_break(self): - tlvdata.send(self.remote, {'operation': 'break'}) + async def send_break(self): + await tlvdata.send(self.remote, {'operation': 'break'}) - def reopen(self): - tlvdata.send(self.remote, {'operation': 'reopen'}) + async def reopen(self): + await tlvdata.send(self.remote, {'operation': 'reopen'}) def resize(self, width, height): - tlvdata.send(self.remote, {'operation': 'resize', 'width': width, - 'height': height}) + tasks.spawn(tlvdata.send(self.remote, {'operation': 'resize', 'width': width, + 'height': height})) # this represents some api view of a console handler. This handles things like @@ -824,10 +824,10 @@ class ConsoleSession(object): self.conshdl = await connect_node(self.node, self.configmanager, self.username, self.direct, self.width, self.height) - def send_break(self): + async def send_break(self): """Send break to remote system """ - self.conshdl.send_break() + await self.conshdl.send_break() def resize(self, width, height): self.conshdl.resize(width, height) diff --git a/confluent_server/confluent/httpapi.py b/confluent_server/confluent/httpapi.py index c1c6ded3..a9cecd6f 100644 --- a/confluent_server/confluent/httpapi.py +++ b/confluent_server/confluent/httpapi.py @@ -571,7 +571,7 @@ async def wsock_handler(req): width=msg['width'], height=msg['height']) if action == 'break': clientsessid = '{0}'.format(msg['sessid']) - myconsoles[clientsessid].send_break() + await myconsoles[clientsessid].send_break() elif action == 'stop': sessid = '{0}'.format(msg.get('sessid', None)) if sessid in myconsoles: @@ -631,7 +631,7 @@ async def wsock_handler(req): cmd = json.loads(clientmsg[1:]) action = cmd.get('action', None) if action == 'break': - consession.send_break() + await consession.send_break() elif action == 'resize': consession.resize( width=cmd['width'], height=cmd['height']) @@ -939,7 +939,7 @@ async def resourcehandler_backend(req, make_response): return rsp elif 'action' in querydict: if querydict['action'] == 'break': - consolesessions[querydict['session']]['session'].send_break() + await consolesessions[querydict['session']]['session'].send_break() elif querydict['action'] == 'resize': consolesessions[querydict['session']]['session'].resize( width=querydict['width'], height=querydict['height']) diff --git a/confluent_server/confluent/shellserver.py b/confluent_server/confluent/shellserver.py index 3bb6a086..76a3f6cd 100644 --- a/confluent_server/confluent/shellserver.py +++ b/confluent_server/confluent/shellserver.py @@ -37,7 +37,7 @@ async def reapsessions(): for sesshdl in list(currcli): currsess = currcli[sesshdl] if currsess.numusers == 0 and currsess.expiry < time.time(): - currsess.close() + await currsess.close() del activesessions[clientid][sesshdl] class _ShellHandler(consoleserver.ConsoleHandler): diff --git a/confluent_server/confluent/sockapi.py b/confluent_server/confluent/sockapi.py index 3e7ace46..f880d751 100644 --- a/confluent_server/confluent/sockapi.py +++ b/confluent_server/confluent/sockapi.py @@ -319,7 +319,7 @@ async def term_interact(authdata, authname, ccons, cfm, connection, consession, await consession.destroy() break elif data['operation'] == 'break': - consession.send_break() + await consession.send_break() continue elif data['operation'] == 'reopen': await consession.reopen() From 670a11666d74060018f62120af1ae96dd4ebd9ae Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 13 Jul 2026 01:10:31 +0200 Subject: [PATCH 15/17] Fix remaining hardware async responses --- confluent_server/confluent/plugins/hardwaremanagement/ipmi.py | 2 +- .../confluent/plugins/hardwaremanagement/srlinux.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py b/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py index 7333d170..d82e5a9e 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py @@ -1668,7 +1668,7 @@ class IpmiHandler: async def handle_ikvm(self): methods = await self.ipmicmd.get_ikvm_methods() if 'openbmc' in methods: - url = vinzmanager.get_url(self.node, self.inputdata) + url = await vinzmanager.get_url(self.node, self.inputdata) await self.output.put(msg.ChildCollection(url)) return launchdata = await self.ipmicmd.get_ikvm_launchdata() diff --git a/confluent_server/confluent/plugins/hardwaremanagement/srlinux.py b/confluent_server/confluent/plugins/hardwaremanagement/srlinux.py index c3f895ab..4c60fde3 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/srlinux.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/srlinux.py @@ -55,7 +55,7 @@ async def retrieve_node_backend(node, element, user, pwd, configmanager, inputda if element[-1] == 'all' or simplify_name(sensor['name']) == element[-1]: await results.put(msg.SensorReadings([sensor], node)) else: - results.put(msg.ConfluentNodeError(node, 'Not supported')) + await results.put(msg.ConfluentNodeError(node, 'Not supported')) async def retrieve(nodes, element, configmanager, inputdata): From dc906b9cd008451525bccee1ea4ccb469a936323 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 13 Jul 2026 16:18:15 +0200 Subject: [PATCH 16/17] Await IPMI session challenge callbacks The IPMI 1.5 session-challenge callback returned coroutine objects from error reporting and session activation instead of expressing an asynchronous callback contract directly. That made completion depend on the dispatch path noticing and awaiting the returned object. Make the callback asynchronous and explicitly await both onlogon() and _activate_session(), ensuring failure notification and activation finish before callback dispatch continues. --- confluent_server/aiohmi/ipmi/private/session.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/confluent_server/aiohmi/ipmi/private/session.py b/confluent_server/aiohmi/ipmi/private/session.py index c6891184..b39fe33e 100644 --- a/confluent_server/aiohmi/ipmi/private/session.py +++ b/confluent_server/aiohmi/ipmi/private/session.py @@ -1043,15 +1043,16 @@ class Session(object): elif self.ipmiversion == 2.0: await self._open_rmcpplus_request() - def _got_session_challenge(self, response): + async def _got_session_challenge(self, response): errstr = get_ipmi_error(response, suffix=" while getting session challenge") if errstr: - return self.onlogon({'error': errstr}) + await self.onlogon({'error': errstr}) + return data = response['data'] self.sessionid = struct.unpack(" Date: Mon, 13 Jul 2026 17:15:45 +0200 Subject: [PATCH 17/17] Revert changes flagged by review Restore pre-PR behavior for three changes flagged by @jjohnson42. Bigger changes are needed for these. Will be done in a separate PR. --- confluent_server/aiohmi/ipmi/oem/lenovo/handler.py | 2 +- confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py b/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py index 40fc0380..d8833ed4 100755 --- a/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py +++ b/confluent_server/aiohmi/ipmi/oem/lenovo/handler.py @@ -347,7 +347,7 @@ class OEMHandler(generic.OEMHandler): ntpres = await self.ipmicmd.raw_command(netfn=0x32, command=0xa7) return ntpres['data'][0] == '\x01' elif await self.is_fpc(): - return self.smmhandler.get_ntp_enabled(self._fpc_variant) + return await self.smmhandler.get_ntp_enabled(self._fpc_variant) elif self.has_tsma: return await self.tsmahandler.get_ntp_enabled() return None diff --git a/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py b/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py index 53fe238e..9fa64fc7 100644 --- a/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py +++ b/confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py @@ -1115,5 +1115,5 @@ class SMMClient(object): async def wc(self): if (not self._wc or self._wc.broken or self._wc.vintage < util._monotonic_time() + 30): - self._wc = self.get_webclient() + self._wc = await self.get_webclient() return self._wc