optools/libvirt/better_virsh.py

#!/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()