223 lines
9.0 KiB
Python
Executable File
223 lines
9.0 KiB
Python
Executable File
#!/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()
|