r00t2/optools
1
0
Fork 0
Code Issues Pull Requests Projects Releases Activity

better_virsh.py. ASMD, redhat.

Browse Source
This commit is contained in:
brent s. 2019-11-29 04:47:25 -05:00
parent 701949b8f7
commit 66ece65699
3 changed files with 224 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

222
libvirt/better_virsh.py Normal file
View File

@ -0,0 +1,222 @@
#!/usr/bin/env python3

import argparse
# import os
# import getpass
import re
##
import libvirt
# from lxml import etree

# NOTE: docs URLS are super long. Extrapolate using following:
# docsurl = 'https://libvirt.org/docs/libvirt-appdev-guide-python/en-US/html'

# TODO: flesh this out. only supports guests atm
# TODO: use openAuth?
# {docsurl}/libvirt_application_development_guide_using_python-Connections.html#idp13928160

# I would like to take the moment to point out that I did in three hours with exactly NO prior knowledge of the libvirt
# API what Red Hat couldn't do in four YEARS. https://bugzilla.redhat.com/show_bug.cgi?id=1244093


def libvirt_callback(userdata, err):
# fucking worst design decision.
# https://stackoverflow.com/a/45543887/733214
pass


# fucking worst design decision.
# https://stackoverflow.com/a/45543887/733214
libvirt.registerErrorHandler(f = libvirt_callback, ctx = None)


class LV(object):
def __init__(self, uri, *args, **kwargs):
self.uri = uri
self.conn = None
self._args = args
self._kwargs = kwargs

def _getTargets(self, target, regex = False, ttype = 'guest',
state = None, nocase = False, *args, **kwargs):
targets = []
# TODO: ..._RUNNING as well? can add multiple flags
state_flags = {'guest': (libvirt.VIR_CONNECT_LIST_DOMAINS_ACTIVE,
libvirt.VIR_CONNECT_LIST_DOMAINS_INACTIVE),
'net': (libvirt.VIR_CONNECT_LIST_NETWORKS_ACTIVE,
libvirt.VIR_CONNECT_LIST_NETWORKS_INACTIVE),
'storage': (libvirt.VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE,
libvirt.VIR_CONNECT_LIST_STORAGE_POOLS_INACTIVE)}
re_flags = re.UNICODE # The default
if nocase:
re_flags += re.IGNORECASE
if not self.conn:
self.startConn()
search_funcs = {'guest': self.conn.listAllDomains,
'net': self.conn.listAllNetworks,
'storage': self.conn.listAllStoragePools}
if not regex:
ptrn = r'^{0}$'.format(target)
else:
ptrn = target
ptrn = re.compile(ptrn, re_flags)
if state == 'active':
flag = state_flags[ttype][0]
elif state == 'inactive':
flag = state_flags[ttype][1]
else:
flag = 0
for t in search_funcs[ttype](flag):
if ptrn.search(t.name()):
targets.append(t)
targets.sort(key = lambda i: i.name())
return(targets)

def list(self, target, verbose = False, *args, **kwargs):
# {docsurl}/libvirt_application_development_guide_using_python-Guest_Domains-Information-Info.html
if not self.conn:
self.startConn()
targets = self._getTargets(target, **kwargs)
results = []
# Each attr is a tuple; the name of the attribute and the key name the result should use (if defined)
attr_map = {'str': (('name', None),
('OSType', 'os'),
('UUIDString', 'uuid'),
('hostname', None)),
'bool': (('autostart', None),
('hasCurrentSnapshot', 'current_snapshot'),
('hasManagedSaveImage', 'managed_save_image'),
('isActive', 'active'),
('isPersistent', 'persistent'),
('isUpdated', 'updated')),
'int': (('ID', 'id'),
('maxMemory', 'max_memory_bytes'),
('maxVcpus', 'max_vCPUs'))}
for t in targets:
if not verbose:
results.append(t.name())
else:
r = {}
for attrname, newkey in attr_map['str']:
keyname = (newkey if newkey else attrname)
try:
r[keyname] = str(getattr(t, attrname)())
except libvirt.libvirtError:
r[keyname] = '(N/A)'
for attrname, newkey in attr_map['bool']:
keyname = (newkey if newkey else attrname)
try:
r[keyname] = bool(getattr(t, attrname)())
except (libvirt.libvirtError, ValueError):
r[keyname] = None
for attrname, newkey in attr_map['int']:
keyname = (newkey if newkey else attrname)
try:
r[keyname] = int(getattr(t, attrname)())
if r[keyname] == -1:
r[keyname] = None
except (libvirt.libvirtError, ValueError):
r[keyname] = None
results.append(r)
return(results)

def restart(self, target, *args, **kwargs):
self.stop(target, **kwargs)
self.start(target, **kwargs)
return()

def start(self, target, **kwargs):
if not self.conn:
self.startConn()
targets = self._getTargets(target, **kwargs)
for t in targets:
t.create()
return()

def stop(self, target, force = False, *args, **kwargs):
if not self.conn:
self.startConn()
targets = self._getTargets(target, **kwargs)
for t in targets:
if not force:
t.shutdown()
else:
t.destroy()
return ()

def startConn(self):
self.conn = libvirt.open(self.uri)
return()

def stopConn(self):
if self.conn:
self.conn.close()
self.conn = None
return()


def parseArgs():
args = argparse.ArgumentParser(description = 'Some better handling of libvirt guests')
common_args = argparse.ArgumentParser(add_help = False)
common_args.add_argument('-u', '--uri',
dest = 'uri',
default = 'qemu:///system',
help = 'The URI for the libvirt to connect to. Default: qemu:///system')
common_args.add_argument('-r', '--regex',
action = 'store_true',
help = 'If specified, use a regex pattern for TARGET instead of exact match')
common_args.add_argument('-i', '--case-insensitive',
action = 'store_true',
dest = 'nocase',
help = 'If specified, match the target name/regex pattern case-insensitive')
common_args.add_argument('-T', '--target-type',
# choices = ['guest', 'net', 'storage'],
choices = ['guest'],
default = 'guest',
dest = 'ttype',
help = 'The type of TARGET')
common_args.add_argument('-t', '--target',
dest = 'target',
metavar = 'TARGET',
default = '.*',
help = ('The guest, network, etc. to manage. '
'If not specified, operate on all (respecting other filtering)'))
subparsers = args.add_subparsers(help = 'Operation to perform',
dest = 'oper',
metavar = 'OPERATION',
required = True)
start_args = subparsers.add_parser('start', help = 'Start the target(s)', parents = [common_args])
restart_args = subparsers.add_parser('restart', help = 'Restart the target(s)', parents = [common_args])
stop_args = subparsers.add_parser('stop', help = 'Stop ("destroy") the target(s)', parents = [common_args])
stop_args.add_argument('-f', '--force',
dest = 'force',
action = 'store_true',
help = 'Hard poweroff instead of send a shutdown/ACPI powerdown signal')
list_args = subparsers.add_parser('list', help = 'List the target(s)', parents = [common_args])
list_args.add_argument('-v', '--verbose',
dest = 'verbose',
action = 'store_true',
help = 'Display more output')
list_args.add_argument('-s', '--state',
dest = 'state',
choices = ['active', 'inactive'],
default = None,
help = 'Filter results by state. Default is all states')
return(args)


def main():
args = parseArgs().parse_args()
varargs = vars(args)
lv = LV(**varargs)
f = getattr(lv, args.oper)(**varargs)
if args.oper == 'list':
if args.verbose:
import json
print(json.dumps(f, indent = 4, sort_keys = True))
else:
print('\n'.join(f))
return()

if __name__ == '__main__':
main()

2
net/connchk.py Normal file
View File

@ -0,0 +1,2 @@
#!/usr/bin/env python3

0
ref/snippets/mailer.py Normal file
View File