Various helper scripts for libvirt.
https://r00t2.io/
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
177 lines
6.1 KiB
177 lines
6.1 KiB
#!/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 = None |
|
self.ifaces = [] |
|
self._get_xml() |
|
|
|
def _get_xml(self): |
|
self.dom_xml = etree.fromstring(self.dom.XMLDesc()) |
|
return(None) |
|
|
|
def get_nets(self, netnames, chk = None): |
|
if not self.ifaces and chk: |
|
return(None) |
|
_netnames = [] |
|
for iface in self.dom_xml.xpath('devices/interface'): |
|
if iface.attrib.get('type') == 'network': |
|
src_xml = iface.find('source') |
|
netname = src_xml.attrib.get('network') |
|
if src_xml is not None and netname in netnames and not chk: |
|
# Why, oh why, does the libvirt API want the XML?? |
|
self.ifaces.append(etree.tostring(iface).decode('utf-8')) |
|
elif src_xml is not None and netname in netnames: |
|
_netnames.append(netname) |
|
if chk: |
|
pass # TODO: confirm that network is removed so we can remove the time.sleep() below. |
|
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 domnet.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) |
|
# TODO: modify DomainNetwork so we can check if an interface is removed or not. How? |
|
time.sleep(3) |
|
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()
|
|
|