xCAT/confluent
2
0
Fork 0
mirror of https://github.com/xcat2/confluent.git synced 2026-07-30 17:49:48 +00:00
Code Issues Projects Releases Wiki Activity

Commence work on async

Browse Source
This commit is contained in:
Jarrod Johnson
2024-02-23 11:56:07 -05:00
parent 91e0aa938c
commit d6bff637db
6 changed files with 83 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
confluent_server/bin/confluent
+3
View File
@@ -17,6 +17,9 @@
import sys
import os
import eventlet.hubs
eventlet.hubs.use_hub("eventlet.hubs.asyncio")
path = os.path.dirname(os.path.realpath(__file__))
path = os.path.realpath(os.path.join(path, '..', 'lib', 'python'))
if path.startswith('/opt'):
confluent_server/confluent/core.py
+21 -16
View File
@@ -946,7 +946,7 @@ def _forward_rsp(connection, res):
connection.sendall(r)
def handle_node_request(configmanager, inputdata, operation,
async def handle_node_request(configmanager, inputdata, operation,
pathcomponents, autostrip=True):
if log.logfull:
raise exc.TargetResourceUnavailable('Filesystem full, free up space and restart confluent service')
@@ -1027,7 +1027,7 @@ def handle_node_request(configmanager, inputdata, operation,
else:
raise Exception("TODO here")
del pathcomponents[0:2]
passvalues = queue.Queue()
passvalues = asyncio.Queue()
plugroute = routespec.routeinfo
_plugin = None
@@ -1095,23 +1095,23 @@ def handle_node_request(configmanager, inputdata, operation,
numworkers = 0
for hfunc in nodesbyhandler:
numworkers += 1
workers.spawn(addtoqueue, passvalues, hfunc, {'nodes': nodesbyhandler[hfunc],
asyncio.create_task(addtoqueue(passvalues, hfunc, {'nodes': nodesbyhandler[hfunc],
'element': pathcomponents,
'configmanager': configmanager,
'inputdata': _get_input_data(_plugin, pathcomponents,
operation, inputdata,nodes,
isnoderange, configmanager)})
isnoderange, configmanager)}))
for manager in nodesbymanager:
numworkers += 1
workers.spawn(addtoqueue, passvalues, dispatch_request, {
asyncio.create_task(addtoqueue(passvalues, dispatch_request, {
'nodes': nodesbymanager[manager], 'manager': manager,
'element': pathcomponents, 'configmanager': configmanager,
'inputdata': inputdata, 'operation': operation, 'isnoderange': isnoderange})
'inputdata': inputdata, 'operation': operation, 'isnoderange': isnoderange}))
if isnoderange or not autostrip:
return iterate_queue(numworkers, passvalues)
return await iterate_queue(numworkers, passvalues)
else:
if numworkers > 0:
return iterate_queue(numworkers, passvalues, nodes[0])
return await iterate_queue(numworkers, passvalues, nodes[0])
else:
raise exc.NotImplementedException()
@@ -1132,10 +1132,10 @@ def _get_input_data(plugin_ext, pathcomponents, operation, inputdata,
nodes, isnoderange,configmanager)
def iterate_queue(numworkers, passvalues, strip=False):
async def iterate_queue(numworkers, passvalues, strip=False):
completions = 0
while completions < numworkers:
nv = passvalues.get()
nv = await passvalues.get()
if nv == 'theend':
completions += 1
else:
@@ -1146,18 +1146,23 @@ def iterate_queue(numworkers, passvalues, strip=False):
yield nv
def addtoqueue(theq, fun, kwargs):
async def addtoqueue(theq, fun, kwargs):
try:
result = fun(**kwargs)
if isinstance(result, console.Console):
theq.put(result)
await theq.put(result)
else:
for pv in result:
theq.put(pv)
if isinstance(result, types.AsyncGeneratorType):
async for pv in result:
await theq.put(pv)
else:
for pv in result:
await theq.put(pv)
except Exception as e:
theq.put(e)
await theq.put(e)
finally:
theq.put('theend')
await theq.put('theend')
def dispatch_request(nodes, manager, element, configmanager, inputdata,
confluent_server/confluent/messages.py
+5
View File
@@ -590,6 +590,11 @@ class InputFirmwareUpdate(ConfluentMessage):
@property
def filename(self):
# TODO: get the currennt_user and cross reference if that user is allowed to
# read... however, not sure wwhat to do if user is pure confluent user
# though the staging may get an explicit pass, which should cover the web case...
# media and firmware are ways to currently push things out, but if we allow profile export
# what then?
if self._complexname:
raise Exception('User requested substitutions, but code is '
'written against old api, code must be fixed or '
confluent_server/confluent/plugins/hardwaremanagement/enclosure.py
+12 -15
View File
@@ -15,11 +15,9 @@ import confluent.core as core
import confluent.messages as msg
import pyghmi.exceptions as pygexc
import confluent.exceptions as exc
import eventlet.queue as queue
import eventlet.greenpool as greenpool
def reseat_bays(encmgr, bays, configmanager, rspq):
async def reseat_bays(encmgr, bays, configmanager, rspq):
try:
for encbay in bays:
node = bays[encbay]
@@ -28,13 +26,13 @@ def reseat_bays(encmgr, bays, configmanager, rspq):
'/nodes/{0}/_enclosure/reseat_bay'.format(encmgr),
'update', configmanager,
inputdata={'reseat': int(encbay)}):
rspq.put(rsp)
await rspq.put(rsp)
except pygexc.UnsupportedFunctionality as uf:
rspq.put(msg.ConfluentNodeError(node, str(uf)))
await rspq.put(msg.ConfluentNodeError(node, str(uf)))
except exc.TargetEndpointUnreachable as uf:
rspq.put(msg.ConfluentNodeError(node, str(uf)))
await rspq.put(msg.ConfluentNodeError(node, str(uf)))
finally:
rspq.put(None)
await rspq.put(None)
def update(nodes, element, configmanager, inputdata):
emebs = configmanager.get_node_attributes(
@@ -54,17 +52,16 @@ def update(nodes, element, configmanager, inputdata):
if em not in baysbyencmgr:
baysbyencmgr[em] = {}
baysbyencmgr[em][eb] = node
rspq = queue.Queue()
gp = greenpool.GreenPool(64)
reseattasks = []
rspq = asyncio.Queue()
for encmgr in baysbyencmgr:
gp.spawn_n(reseat_bays, encmgr, baysbyencmgr[encmgr], configmanager, rspq)
while gp.running():
nrsp = rspq.get()
currtask = asyncio.create_task(reseat_bays(encmgr, baysbyencmgr[encmgr], configmanager, rspq))
reseattasks.append(currtask)
while not all([task.done() for task in reseattasks]):;
nrsp = await rspq.get()
if nrsp is not None:
yield nrsp
while not rspq.empty():
nrsp = rspq.get()
nrsp = await rspq.get()
if nrsp is not None:
yield nrsp
confluent_server/confluent/plugins/hardwaremanagement/ipmi.py
+19 -15
View File
@@ -30,11 +30,11 @@ import eventlet.support.greendns
from fnmatch import fnmatch
import os
import pwd
import pyghmi.constants as pygconstants
import pyghmi.exceptions as pygexc
import pyghmi.storage as storage
import aiohmi.constants as pygconstants
import aiohmi.exceptions as pygexc
import aiohmi.storage as storage
console = eventlet.import_patched('pyghmi.ipmi.console')
ipmicommand = eventlet.import_patched('pyghmi.ipmi.command')
import aiohmi.ipmi.command as ipmicommand
import socket
import ssl
import traceback
@@ -174,7 +174,9 @@ def sanitize_invdata(indata):
class IpmiCommandWrapper(ipmicommand.Command):
def __init__(self, node, cfm, **kwargs):
@classmethod
async def create(cls, node, cfm, **kwargs):
self = cls()
self.cfm = cfm
self.node = node
self.sensormap = {}
@@ -185,7 +187,7 @@ class IpmiCommandWrapper(ipmicommand.Command):
(node,), ('secret.hardwaremanagementuser', 'collective.manager',
'secret.hardwaremanagementpassword', 'secret.ipmikg',
'hardwaremanagement.manager'), self._attribschanged)
super(self.__class__, self).__init__(**kwargs)
amait super().create(**kwargs)
self.setup_confluent_keyhandler()
try:
os.makedirs('/var/cache/confluent/ipmi/')
@@ -380,7 +382,7 @@ class IpmiConsole(conapi.Console):
self.solconnection.send_break()
def perform_requests(operator, nodes, element, cfg, inputdata, realop):
async def perform_requests(operator, nodes, element, cfg, inputdata, realop):
cryptit = cfg.decrypt
cfg.decrypt = True
configdata = cfg.get_node_attributes(nodes, _configattributes)
@@ -430,7 +432,7 @@ def perform_requests(operator, nodes, element, cfg, inputdata, realop):
def perform_request(operator, node, element,
configdata, inputdata, cfg, results, realop):
try:
return IpmiHandler(operator, node, element, configdata, inputdata,
return IpmiHandler.create(operator, node, element, configdata, inputdata,
cfg, results, realop).handle_request()
except pygexc.IpmiException as ipmiexc:
excmsg = str(ipmiexc)
@@ -460,9 +462,11 @@ def perform_request(operator, node, element,
persistent_ipmicmds = {}
class IpmiHandler(object):
def __init__(self, operation, node, element, cfd, inputdata, cfg, output,
class IpmiHandler:
@classmethod
async def create(cls, operation, node, element, cfd, inputdata, cfg, output,
realop):
self = cls()
self.cfm = cfg
self.invmap = {}
self.output = output
@@ -492,7 +496,7 @@ class IpmiHandler(object):
except KeyError: # was no previous session
pass
try:
persistent_ipmicmds[(node, tenant)] = IpmiCommandWrapper(
persistent_ipmicmds[(node, tenant)] = await IpmiCommandWrapper.create(
node, cfg, bmc=connparams['bmc'],
userid=connparams['username'],
password=connparams['passphrase'], kg=connparams['kg'],
@@ -1636,14 +1640,14 @@ def initthread():
_ipmithread = eventlet.spawn(_ipmi_evtloop)
def create(nodes, element, configmanager, inputdata, realop='create'):
async def create(nodes, element, configmanager, inputdata, realop='create'):
initthread()
if element == ['_console', 'session']:
if len(nodes) > 1:
raise Exception("_console/session does not support multiple nodes")
return IpmiConsole(nodes[0], configmanager)
else:
return perform_requests(
return await perform_requests(
'update', nodes, element, configmanager, inputdata, realop)
@@ -1652,7 +1656,7 @@ def update(nodes, element, configmanager, inputdata):
return create(nodes, element, configmanager, inputdata, 'update')
def retrieve(nodes, element, configmanager, inputdata):
async def retrieve(nodes, element, configmanager, inputdata):
initthread()
if '/'.join(element).startswith('inventory/firmware/updates/active'):
return firmwaremanager.list_updates(nodes, configmanager.tenant,
@@ -1664,7 +1668,7 @@ def retrieve(nodes, element, configmanager, inputdata):
return firmwaremanager.list_updates(nodes, configmanager.tenant,
element, 'ffdc')
else:
return perform_requests('read', nodes, element, configmanager,
return await perform_requests('read', nodes, element, configmanager,
inputdata, 'read')
def delete(nodes, element, configmanager, inputdata):
confluent_server/confluent/plugins/shell/ssh.py
+23 -52
View File
@@ -22,35 +22,29 @@
import confluent.exceptions as cexc
import confluent.interface.console as conapi
import confluent.log as log
try:
import cryptography
except ImportError:
# Using older, non-crypography based paramiko
cryptography = None
import eventlet
import hashlib
import sys
sys.modules['gssapi'] = None
paramiko = eventlet.import_patched('paramiko')
warnhostkey = False
if cryptography and cryptography.__version__.split('.') < ['1', '5']:
# older cryptography with paramiko breaks most key support except
# ed25519
warnhostkey = True
paramiko.transport.Transport._preferred_keys = filter(
lambda x: 'ed25519' in x,
paramiko.transport.Transport._preferred_keys)
#paramiko = eventlet.import_patched('paramiko')
import asyncio
import asyncssh
class HostKeyHandler(paramiko.client.MissingHostKeyPolicy):
class HostKeyHandler:
def __init__(self, configmanager, node):
self.cfm = configmanager
self.node = node
def missing_host_key(self, client, hostname, key):
# have to catch the valueerror and use ssh-keyscan to trigger this, asyncssh host key handling
# is a bit more limited compared to paramiko
#but... leverage /etc/ssh/ssh_known_hosts, we can try that way, and if it fails, fallback to our
#confluent db based handler
fingerprint = 'sha512$' + hashlib.sha512(key.asbytes()).hexdigest()
cfg = self.cfm.get_node_attributes(
self.node, ('pubkeys.ssh', 'pubkeys.addpolicy'))
@@ -94,11 +88,12 @@ class SshShell(conapi.Console):
self.height = height
if not self.connected:
return
# asyncssh channel has change_terminal_size, hurray
self.shell.resize_pty(width=width, height=height)
def recvdata(self):
async def recvdata(self):
while self.connected:
pendingdata = self.shell.recv(8192)
pendingdata = await self.shell.stdout.read(8192)
if not pendingdata:
self.ssh.close()
if self.datacallback:
@@ -123,31 +118,18 @@ class SshShell(conapi.Console):
self.inputmode = -3
eventlet.spawn_n(self.do_logon)
def do_logon(self):
self.ssh = paramiko.SSHClient()
self.ssh.set_missing_host_key_policy(
HostKeyHandler(self.nodeconfig, self.node))
async def do_logon(self):
sco = asyncssh.SSHClientConnectionOptions()
#The below would be to support the confluent db, and only fallback if the SSH CA do not work
#sco.client_fatory = SSHKnownHostsLookup
try:
self.datacallback('\r\nConnecting to {}...'.format(self.node))
self.ssh.connect(self.node, username=self.username,
password=self.password, allow_agent=False,
look_for_keys=False)
except paramiko.AuthenticationException as e:
self.ssh.close()
self.inputmode = 0
self.username = b''
self.password = b''
self.datacallback('\r\nError connecting to {0}:\r\n {1}\r\n'.format(self.node, str(e)))
self.datacallback('\r\nlogin as: ')
return
except paramiko.ssh_exception.NoValidConnectionsError as e:
self.ssh.close()
self.datacallback('\r\nError connecting to {0}:\r\n {1}\r\n'.format(self.node, str(e)))
self.inputmode = 0
self.username = b''
self.password = b''
self.datacallback('\r\nlogin as: ')
return
try:
self.ssh = await asyncssh.connect(self,node, username=self.username, password=self.password, known_hosts='/etc/ssh/ssh_known_hosts')
except ValueError:
#TODO: non-cert ssh targets
raise
except cexc.PubkeyInvalid as pi:
self.ssh.close()
self.keyaction = b''
@@ -158,17 +140,6 @@ class SshShell(conapi.Console):
self.inputmode = -1
self.datacallback('\r\nEnter "disconnect" or "accept": ')
return
except paramiko.SSHException as pi:
self.ssh.close()
self.inputmode = -2
warn = str(pi)
if warnhostkey:
warn += ' (Older cryptography package on this host only ' \
'works with ed25519, check ssh startup on target ' \
'and permissions on /etc/ssh/*key)\r\n' \
'Press Enter to close...'
self.datacallback('\r\n' + warn)
return
except Exception as e:
self.ssh.close()
self.ssh.close()