mirror of
https://github.com/xcat2/confluent.git
synced 2026-08-25 08:06:46 +00:00
Markus Hilger
8a3fce85c0
Every one of these raises NameError if its code path is reached: - nodeapply: run_automation accumulated into an exitcode that only existed in run(), so any automation error crashed instead of being reported. It now keeps and returns its own, tracked separately from the exit code of the ssh commands: the early exit after the spawn loop tests that one, and folding automation failures into it would exit with children already running and their pipes abandoned. Both are reported at the real exits. - nodeconsole: redraw() reads firstnodename, which was local to do_screenshot(); promote it to a module global like the other drawing state. - nodedeploy: the redeploy path appended to a lockednodes list that did not exist yet. The block that follows re-reads the same lock state and acts on it, so drop the dead duplicate. - samples/nodeattrib_from_switch.py, misc/filterpasswd: missing import sys. - xcc3: fixuuid was never imported. xcc imports xcc3, so take a local copy the way the smm handler does instead of creating an import cycle. - httpapi: the async session call still passed the WSGI-era env and an extra argument to handle_async(), which has taken only querydict since the aiohttp port. Calling it correctly exposed that handle_async() registers an AsyncSession before raising on the discontinued long poll path, so every request to it would leak a session that is never reaped. It now only creates one when there is a websocket handler to yield it to. - messages: the InputFirmwareUpdate.filename property checked self.filebynode[node] with no node in scope. __init__ already validates every expanded path and nodefile() rechecks per node, so drop the checks. - pam: drop the python2 branches referencing unicode and raw_input. The server has been python3 only since the asyncio port. - cooltera: the sensor-name listing referenced a nonexistent sensors dict. The available sensors depend on the model, which is only known after reading the device, so list them from the same status data the readings use. - deltapdu, eatonpdu, geist: the not-implemented response in update() used node outside the loop, unlike retrieve() in the same files and unlike raritan/enlogic. - confluentdbgcli: stray self. on a module-level socket connect.
219 lines
8.6 KiB
Python
Executable File
219 lines
8.6 KiB
Python
Executable File
#!/usr/bin/python3
|
|
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright 2016-2017 Lenovo
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
from collections import deque
|
|
import optparse
|
|
import os
|
|
import select
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
|
|
try:
|
|
signal.signal(signal.SIGPIPE, signal.SIG_DFL)
|
|
except AttributeError:
|
|
pass
|
|
path = os.path.dirname(os.path.realpath(__file__))
|
|
path = os.path.realpath(os.path.join(path, '..', 'lib', 'python'))
|
|
if path.startswith('/opt'):
|
|
sys.path.append(path)
|
|
|
|
import confluent.client as client
|
|
import confluent.sortutil as sortutil
|
|
devnull = None
|
|
|
|
def run_automation(noderange, category, c):
|
|
exitcode = 0
|
|
automationbynode = {}
|
|
for res in c.update('/noderange/{0}/deployment/remote_config/run'.format(noderange), {
|
|
'category': category,
|
|
}):
|
|
if 'error' in res:
|
|
sys.stderr.write(res['error'] + '\n')
|
|
exitcode |= res.get('errorcode', 1)
|
|
if 'created' in res:
|
|
nodename = res['created'].split('/')[2]
|
|
automationbynode[nodename] = res['created']
|
|
while automationbynode:
|
|
for node in list(automationbynode):
|
|
for res in c.read(automationbynode[node]):
|
|
if 'error' in res:
|
|
sys.stderr.write(res['error'] + '\n')
|
|
exitcode |= res.get('errorcode', 1)
|
|
for result in res.get('results', []):
|
|
sys.stdout.write('{0}: Task [{1}] {2}\n'.format(
|
|
node, result['task_name'], result['state']))
|
|
for warning in result.get('warnings', []):
|
|
sys.stderr.write('{0}: [WARNING] {1}\n'.format(node, warning))
|
|
if 'errorinfo' in result:
|
|
for errorline in result['errorinfo'].splitlines():
|
|
sys.stderr.write('{0}: [ERROR] {1}\n'.format(node, errorline))
|
|
if res.get('complete', False):
|
|
del automationbynode[node]
|
|
sys.stdout.write('{0}: Automation complete\n'.format(node))
|
|
return exitcode
|
|
|
|
|
|
def run():
|
|
global devnull
|
|
devnull = open(os.devnull, 'rb')
|
|
argparser = optparse.OptionParser(
|
|
usage="Usage: %prog [options] noderange commandexpression",
|
|
epilog="Expressions are the same as in attributes, e.g. "
|
|
"'ipmitool -H {hardwaremanagement.manager}' will be expanded.")
|
|
argparser.add_option('-f', '-c', '--count', type='int', default=168,
|
|
help='Number of commands to run at a time')
|
|
argparser.add_option('-k', '--security', action='store_true',
|
|
help='Update SSH setup')
|
|
argparser.add_option('-F', '--sync', action='store_true',
|
|
help='Run the syncfiles associated with the currently completed OS profile on the noderange')
|
|
argparser.add_option('-P', '--scripts',
|
|
help='Re-run specified scripts, with full path under scripts, e.g. post.d/first,firstboot.d/second')
|
|
argparser.add_option('-A', '--automation',
|
|
help='Run the automation scripts associated with the current OS profile on the noderange, specifying category (onboot.d/firstboot.d/post.d)')
|
|
argparser.add_option('-m', '--maxnodes', type='int',
|
|
help='Specify a maximum number of '
|
|
'nodes to run remote ssh command to, '
|
|
'prompting if over the threshold')
|
|
# among other things, FD_SETSIZE limits. Besides, spawning too many
|
|
# processes can be unkind for the unaware on memory pressure and such...
|
|
#argparser.disable_interspersed_args()
|
|
(options, args) = argparser.parse_args()
|
|
if len(args) < 1:
|
|
argparser.print_help()
|
|
sys.exit(1)
|
|
client.check_globbing(args[0])
|
|
concurrentprocs = options.count
|
|
c = client.Command()
|
|
|
|
currprocs = 0
|
|
all = set([])
|
|
poller = select.epoll()
|
|
pipedesc = {}
|
|
pendingexecs = deque()
|
|
exitcode = 0
|
|
# Kept apart from exitcode: a failed automation run must not trip the
|
|
# early exit below, which would abandon ssh children already spawned.
|
|
autoexitcode = 0
|
|
|
|
c.stop_if_noderange_over(args[0], options.maxnodes)
|
|
if options.automation:
|
|
autoexitcode = run_automation(args[0], options.automation, c)
|
|
|
|
nodemap = {}
|
|
cmdparms = []
|
|
nodes = []
|
|
|
|
|
|
cmdstorun = []
|
|
if options.security:
|
|
cmdstorun.append(['run_remote', 'setupssh'])
|
|
if options.sync:
|
|
cmdstorun.append(['run_remote_python', 'syncfileclient'])
|
|
if options.scripts:
|
|
for script in options.scripts.split(','):
|
|
cmdstorun.append(['run_remote', script])
|
|
if not cmdstorun:
|
|
if options.automation:
|
|
sys.exit(autoexitcode)
|
|
argparser.print_help()
|
|
sys.exit(1)
|
|
for res in c.read('/noderange/{0}/nodes/'.format(args[0])):
|
|
if 'error' in res:
|
|
sys.stderr.write(res['error'] + '\n')
|
|
exitcode |= res.get('errorcode', 1)
|
|
break
|
|
node = res['item']['href'][:-1]
|
|
nodes.append(node)
|
|
idxbynode = {}
|
|
cmdvbase = ['bash', '/etc/confluent/functions']
|
|
for sshnode in nodes:
|
|
idxbynode[sshnode] = 1
|
|
cmdv = ['ssh', sshnode] + cmdvbase + cmdstorun[0]
|
|
if currprocs < concurrentprocs:
|
|
currprocs += 1
|
|
run_cmdv(sshnode, cmdv, all, poller, pipedesc)
|
|
else:
|
|
pendingexecs.append((sshnode, cmdv))
|
|
if not all or exitcode:
|
|
sys.exit(exitcode | autoexitcode)
|
|
rdy = poller.poll(10)
|
|
while all:
|
|
pernodeout = {}
|
|
for r in rdy:
|
|
r = r[0]
|
|
desc = pipedesc[r]
|
|
r = desc['file']
|
|
node = desc['node']
|
|
data = True
|
|
singlepoller = select.epoll()
|
|
singlepoller.register(r, select.EPOLLIN)
|
|
while data and singlepoller.poll(0):
|
|
data = r.readline()
|
|
if data:
|
|
if desc['type'] == 'stdout':
|
|
if node not in pernodeout:
|
|
pernodeout[node] = []
|
|
pernodeout[node].append(data)
|
|
else:
|
|
data = client.stringify(data)
|
|
sys.stderr.write('{0}: {1}'.format(node, data))
|
|
sys.stderr.flush()
|
|
else:
|
|
pop = desc['popen']
|
|
ret = pop.poll()
|
|
if ret is not None:
|
|
exitcode = exitcode | ret
|
|
all.discard(r)
|
|
poller.unregister(r)
|
|
r.close()
|
|
if desc['type'] == 'stdout':
|
|
if idxbynode[node] < len(cmdstorun):
|
|
cmdv = ['ssh', sshnode] + cmdvbase + cmdstorun[idxbynode[node]]
|
|
idxbynode[node] += 1
|
|
run_cmdv(node, cmdv, all, poller, pipedesc)
|
|
elif pendingexecs:
|
|
node, cmdv = pendingexecs.popleft()
|
|
run_cmdv(node, cmdv, all, poller, pipedesc)
|
|
singlepoller.close()
|
|
for node in sortutil.natural_sort(pernodeout):
|
|
for line in pernodeout[node]:
|
|
line = client.stringify(line)
|
|
line = line.lstrip('\x08')
|
|
sys.stdout.write('{0}: {1}'.format(node, line))
|
|
sys.stdout.flush()
|
|
if all:
|
|
rdy = poller.poll(10)
|
|
sys.exit(exitcode | autoexitcode)
|
|
|
|
|
|
def run_cmdv(node, cmdv, all, poller, pipedesc):
|
|
nopen = subprocess.Popen(
|
|
cmdv, stdin=devnull, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
pipedesc[nopen.stdout.fileno()] = {'node': node, 'popen': nopen,
|
|
'type': 'stdout', 'file': nopen.stdout}
|
|
pipedesc[nopen.stderr.fileno()] = {'node': node, 'popen': nopen,
|
|
'type': 'stderr', 'file': nopen.stderr}
|
|
all.add(nopen.stdout)
|
|
poller.register(nopen.stdout, select.EPOLLIN)
|
|
all.add(nopen.stderr)
|
|
poller.register(nopen.stderr, select.EPOLLIN)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run()
|