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

finishing restart_net.py

Browse Source
This commit is contained in:
brent s. 2020-08-02 03:44:02 -04:00
parent c0df7dabe1
commit 50d2f7138e
Signed by: bts
GPG Key ID: 8C004C2F93481F6B
2 changed files with 385 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

223
better_virsh.py Executable file
View File

@ -0,0 +1,223 @@
#!/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_KiB'),
('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, state = 'active', **kwargs)
self.start(target, state = 'inactive', **kwargs)
return()

def start(self, target, **kwargs):
if not self.conn:
self.startConn()
targets = self._getTargets(target, state = 'inactive', **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, state = 'active', **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()

162
restart_net.py Executable file
View File

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

import argparse
import time
import uuid
##
import libvirt
from lxml import etree


# The next dummy function and proceeding statement are because of the fucking worst design decisions.
# https://stackoverflow.com/a/45543887/733214
def libvirt_callback(userdata, err):
pass


libvirt.registerErrorHandler(f = libvirt_callback, ctx = None)


# Defaults.
_def_uri = 'qemu:///system'
_def_rsrt_guests = True

# "Ignore" these libvirt.libvirtError codes in shutdown attempts.
#
_lv_err_pass = (86, 8)


class DomainNetwork(object):
def __init__(self, dom):
# dom is a libvirt.virDomain (or whatever it's called) instance.
self.dom = dom
self.dom_xml = etree.fromstring(dom.XMLDesc())
self.ifaces = []

def get_nets(self, netnames):
for iface in self.dom_xml.xpath('devices/interface'):
if iface.attrib.get('type') == 'network':
src_xml = iface.find('source')
if src_xml and src_xml.attrib.get('network') in netnames:
# Why, oh why, does the libvirt API want the XML??
self.ifaces.append(etree.tostring(iface).decode('utf-8'))
return(None)


class VMManager(object):
def __init__(self, netname, restart_guests = True, uri = _def_uri, *args, **kwargs):
self.netname = netname
self.restart_guests = restart_guests
self.uri = uri
self.networks = []
self._listonly = True
self.conn = None
self._connect()
self._get_networks()
if self.netname:
self._listonly = False
self.networks = [i for i in self.networks if i.name() == self.netname]
self.errs = []

def _connect(self):
if self.conn:
return(None)
self.conn = libvirt.open(self.uri)
return(None)

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

def _get_networks(self):
self._connect()
self.networks = self.conn.listAllNetworks(flags = libvirt.VIR_CONNECT_LIST_NETWORKS_ACTIVE)
self.networks.sort(key = lambda i: i.name())
self._disconnect()
return(None)

def restart(self):
if self._listonly:
return(False)
self._connect()
doms = {}
netnames = [i.name() for i in self.networks]
# https://libvirt.org/html/libvirt-libvirt-domain.html#virConnectListAllDomains
# _flags = (libvirt.VIR_CONNECT_LIST_DOMAINS_ACTIVE | libvirt.VIR_CONNECT_LIST_DOMAINS_RUNNING)
_findflags = libvirt.VIR_CONNECT_LIST_DOMAINS_RUNNING
# https://libvirt.org/html/libvirt-libvirt-domain.html#virDomainDetachDeviceFlags
# https://libvirt.org/html/libvirt-libvirt-domain.html#virDomainAttachDeviceFlags
# TODO: actually use _CURRENT?
_ifaceflags = (libvirt.VIR_DOMAIN_AFFECT_LIVE | libvirt.VIR_DOMAIN_AFFECT_CURRENT)
for dom in self.conn.listAllDomains(flags = _findflags):
domnet = DomainNetwork(dom)
domnet.get_nets(netnames)
if dom.ifaces:
doms[uuid.UUID(bytes = dom.UUID())] = domnet
for n in self.networks:
n.destroy()
n.create()
for dom_uuid, domnet in doms.items():
for iface_xml in domnet.ifaces:
# Nice that we don't actually need to shut the machine down.
# Here be dragons, though, if the OS doesn't understand NIC hotplugging.
domnet.dom.detachDeviceFlags(iface_xml, flags = _ifaceflags)
domnet.dom.attachDeviceFlags(iface_xml, flags = _ifaceflags)
self._disconnect()
if not doms:
doms = None
return(doms)

def print_nets(self):
max_name = len(max([i.name() for i in self.networks], key = len)) + 2
hdr = '| Name{0} | Active | Persistent |'.format((' ' * (max_name - 6)))
sep = '-' * len(hdr)
print(sep)
print(hdr)
print(sep)
vmtpl = '| {{name:<{0}}} | {{isActive}} | {{isPersistent}} |'.format((max_name - 2))
for n in self.networks:
vals = {}
for i in ('name', 'isActive', 'isPersistent'):
v = getattr(n, i)
v = v()
if isinstance(v, int):
v = ('Y' if v else 'N')
vals[i] = v
print(vmtpl.format(**vals))
print(sep)
return(None)


def parseArgs():
args = argparse.ArgumentParser(description = 'Restart a network and all associated guests ("domains")')
args.add_argument('-u', '--uri',
dest = 'uri',
default = _def_uri,
help = ('The URI to use for which libvirt to connect to. '
'Default: {0}').format(_def_uri))
args.add_argument('-n', '--no-guests',
action = 'store_false',
dest = 'restart_guests',
help = ('If specified, suppress rebooting guests and only restart the network'))
args.add_argument('netname',
metavar = 'NETWORK_NAME',
nargs = '?',
help = ('Restart this network name. If not specified, list all networks and quit'))
return(args)


def main():
args = parseArgs().parse_args()
VMM = VMManager(**vars(args))
if not args.netname:
VMM.print_nets()
VMM.restart()
return(None)


if __name__ == '__main__':
main()