virt-manager/virt-xml

#!/usr/bin/env python3
#
# Copyright 2013-2014 Red Hat, Inc.
# Cole Robinson <crobinso@redhat.com>
#
# This work is licensed under the GNU GPLv2 or later.
# See the COPYING file in the top-level directory.

import difflib
import logging
import os
import re
import sys

import libvirt

import virtinst
from virtinst import cli
from virtinst import util
from virtinst.cli import fail, print_stdout, print_stderr


###################
# Utility helpers #
###################

def prompt_yes_or_no(msg):
    while 1:
        printmsg = msg + " (y/n): "
        sys.stdout.write(printmsg)
        sys.stdout.flush()

        if "VIRTINST_TEST_SUITE" in os.environ:
            inp = "yes"
        else:
            inp = sys.stdin.readline().lower().strip()

        if inp in ["y", "yes"]:
            return True
        elif inp in ["n", "no"]:
            return False
        else:
            print_stdout(_("Please enter 'yes' or 'no'."))


def get_diff(origxml, newxml):
    ret = "".join(difflib.unified_diff(origxml.splitlines(1),
                                       newxml.splitlines(1),
                                       fromfile="Original XML",
                                       tofile="Altered XML"))

    if ret:
        logging.debug("XML diff:\n%s", ret)
    else:
        logging.debug("No XML diff, didn't generate any change.")
    return ret


def get_domain_and_guest(conn, domstr):
    try:
        int(domstr)
        isint = True
    except ValueError:
        isint = False

    uuidre = "[a-fA-F0-9]{8}[-]([a-fA-F0-9]{4}[-]){3}[a-fA-F0-9]{12}$"
    isuuid = bool(re.match(uuidre, domstr))

    try:
        if isint:
            domain = conn.lookupByID(int(domstr))
        elif isuuid:
            domain = conn.lookupByUUIDString(domstr)
        else:
            domain = conn.lookupByName(domstr)
    except libvirt.libvirtError as e:
        fail(_("Could not find domain '%s': %s") % (domstr, e))

    state = domain.info()[0]
    active_xmlobj = None
    inactive_xmlobj = virtinst.Guest(conn, parsexml=domain.XMLDesc(0))
    if state != libvirt.VIR_DOMAIN_SHUTOFF:
        active_xmlobj = inactive_xmlobj
        inactive_xmlobj = virtinst.Guest(conn,
                parsexml=domain.XMLDesc(libvirt.VIR_DOMAIN_XML_INACTIVE))

    return (domain, inactive_xmlobj, active_xmlobj)


################
# Change logic #
################

def _find_objects_to_edit(guest, action_name, editval, parserclass):
    objlist = util.listify(parserclass.lookup_prop(guest))
    idx = None

    if editval is None:
        idx = 1
    elif (editval.isdigit() or
          editval.startswith("-") and editval[1:].isdigit()):
        idx = int(editval)

    if idx is not None:
        # Edit device by index
        if idx == 0:
            fail(_("Invalid --edit option '%s'") % editval)

        if not objlist:
            fail(_("No --%s objects found in the XML") %
                parserclass.cli_arg_name)
        if len(objlist) < abs(idx):
            fail(_("--edit %s requested but there's only %s "
                   "--%s object in the XML") %
                (idx, len(objlist), parserclass.cli_arg_name))

        if idx > 0:
            idx -= 1
        inst = objlist[idx]

    elif editval == "all":
        # Edit 'all' devices
        inst = objlist[:]

    else:
        # Lookup device by the passed prop string
        parserobj = parserclass(guest, editval)
        inst = parserobj.lookup_child_from_option_string()
        if not inst:
            fail(_("No matching objects found for --%s %s") %
                 (action_name, editval))

    return inst


def check_action_collision(options):
    actions = ["edit", "add-device", "remove-device", "build-xml"]

    collisions = []
    for cliname in actions:
        optname = cliname.replace("-", "_")
        if getattr(options, optname) not in [False, -1]:
            collisions.append(cliname)

    if len(collisions) == 0:
        fail(_("One of %s must be specified.") %
             ", ".join(["--" + c for c in actions]))
    if len(collisions) > 1:
        fail(_("Conflicting options %s") %
             ", ".join(["--" + c for c in collisions]))


def check_xmlopt_collision(options):
    collisions = []
    for parserclass in cli.VIRT_PARSERS:
        if getattr(options, parserclass.cli_arg_name):
            collisions.append(parserclass)

    if len(collisions) == 0:
        fail(_("No change specified."))
    if len(collisions) != 1:
        fail(_("Only one change operation may be specified "
               "(conflicting options %s)") %
               ["--" + c.cli_arg_name for c in collisions])

    return collisions[0]


def action_edit(guest, options, parserclass):
    if parserclass.propname:
        inst = _find_objects_to_edit(guest, "edit", options.edit, parserclass)
    else:
        inst = guest
        if options.edit and options.edit != '1' and options.edit != 'all':
            fail(_("'--edit %s' doesn't make sense with --%s, "
                   "just use empty '--edit'") %
            (options.edit, parserclass.cli_arg_name))

    return cli.parse_option_strings(options, guest, inst, update=True)


def action_add_device(guest, options, parserclass):
    if not parserclass.prop_is_list(guest):
        fail(_("Cannot use --add-device with --%s") % parserclass.cli_arg_name)
    return cli.parse_option_strings(options, guest, None)


def action_remove_device(guest, options, parserclass):
    if not parserclass.prop_is_list(guest):
        fail(_("Cannot use --remove-device with --%s") %
             parserclass.cli_arg_name)

    devs = _find_objects_to_edit(guest, "remove-device",
        getattr(options, parserclass.cli_arg_name)[-1], parserclass)

    devs = util.listify(devs)
    for dev in devs:
        guest.remove_device(dev)
    return devs


def action_build_xml(conn, options, parserclass):
    if not parserclass.propname:
        fail(_("--build-xml not supported for --%s") %
             parserclass.cli_arg_name)

    guest = virtinst.Guest(conn)
    inst = parserclass.lookup_prop(guest)
    if parserclass.prop_is_list(guest):
        inst = inst.new()
    else:
        inst = inst.__class__(conn)
    return cli.parse_option_strings(options, guest, inst)


def setup_device(dev):
    if getattr(dev, "DEVICE_TYPE", None) != "disk":
        return
    if getattr(dev, "virt_xml_setup", None) is True:
        return

    logging.debug("Doing setup for disk=%s", dev)

    dev.setup(cli.get_meter())
    dev.virt_xml_setup = True


def define_changes(conn, inactive_xmlobj, devs, action, confirm):
    if confirm:
        if not prompt_yes_or_no(
                _("Define '%s' with the changed XML?") % inactive_xmlobj.name):
            return False

    if action == "hotplug":
        for dev in devs:
            setup_device(dev)

    conn.defineXML(inactive_xmlobj.get_xml_config())
    print_stdout(_("Domain '%s' defined successfully.") % inactive_xmlobj.name)
    return True


def update_changes(domain, devs, action, confirm):
    for dev in devs:
        xml = dev.get_xml_config()

        if confirm:
            if action == "hotplug":
                prep = "to"
            elif action == "hotunplug":
                prep = "from"
            else:
                prep = "for"

            msg = ("%s\n\n%s this device %s guest '%s'?" %
                   (xml, action.capitalize(), prep, domain.name()))
            if not prompt_yes_or_no(msg):
                continue

        if action == "hotplug":
            setup_device(dev)

        try:
            if action == "hotplug":
                domain.attachDeviceFlags(xml, libvirt.VIR_DOMAIN_AFFECT_LIVE)
            elif action == "hotunplug":
                domain.detachDeviceFlags(xml, libvirt.VIR_DOMAIN_AFFECT_LIVE)
            elif action == "update":
                domain.updateDeviceFlags(xml, libvirt.VIR_DOMAIN_AFFECT_LIVE)
        except libvirt.libvirtError as e:
            fail(_("Error attempting device %s: %s") % (action, e))

        print_stdout(_("Device %s successful.") % action)
        if confirm:
            print_stdout("")


def prepare_changes(xmlobj, options, parserclass):
    origxml = xmlobj.get_xml_config()

    if options.edit != -1:
        devs = action_edit(xmlobj, options, parserclass)
        action = "update"

    elif options.add_device:
        devs = action_add_device(xmlobj, options, parserclass)
        action = "hotplug"

    elif options.remove_device:
        devs = action_remove_device(xmlobj, options, parserclass)
        action = "hotunplug"

    newxml = xmlobj.get_xml_config()
    diff = get_diff(origxml, newxml)

    if options.print_diff:
        if diff:
            print_stdout(diff)
    elif options.print_xml:
        print_stdout(newxml)

    return devs, action


#######################
# CLI option handling #
#######################

def parse_args():
    parser = cli.setupParser(
        "%(prog)s [options]",
        _("Edit libvirt XML using command line options."),
        introspection_epilog=True)

    cli.add_connect_option(parser, "virt-xml")

    parser.add_argument("domain", nargs='?',
        help=_("Domain name, id, or uuid"))

    actg = parser.add_argument_group(_("XML actions"))
    actg.add_argument("--edit", nargs='?', default=-1,
        help=_("Edit VM XML. Examples:\n"
        "--edit --disk ...     (edit first disk device)\n"
        "--edit 2 --disk ...   (edit second disk device)\n"
        "--edit all --disk ... (edit all disk devices)\n"
        "--edit target=hda --disk ... (edit disk 'hda')\n"))
    actg.add_argument("--remove-device", action="store_true",
        help=_("Remove specified device. Examples:\n"
        "--remove-device --disk 1 (remove first disk)\n"
        "--remove-device --disk all (remove all disks)\n"
        "--remove-device --disk /some/path"))
    actg.add_argument("--add-device", action="store_true",
        help=_("Add specified device. Example:\n"
        "--add-device --disk ..."))
    actg.add_argument("--build-xml", action="store_true",
        help=_("Just output the built device XML, no domain required."))

    outg = parser.add_argument_group(_("Output options"))
    outg.add_argument("--update", action="store_true",
        help=_("Apply changes to the running VM.\n"
               "With --add-device, this is a hotplug operation.\n"
               "With --remove-device, this is a hotunplug operation.\n"
               "With --edit, this is an update device operation."))
    outg.add_argument("--define", action="store_true",
        help=_("Force defining the domain. Only required if a --print "
               "option was specified."))
    outg.add_argument("--print-diff", action="store_true",
        help=_("Only print the requested change, in diff format"))
    outg.add_argument("--print-xml", action="store_true",
        help=_("Only print the requested change, in full XML format"))
    outg.add_argument("--confirm", action="store_true",
        help=_("Require confirmation before saving any results."))

    g = parser.add_argument_group(_("XML options"))
    cli.add_disk_option(g, editexample=True)
    cli.add_net_option(g)
    cli.add_gfx_option(g)
    cli.add_metadata_option(g)
    cli.add_memory_option(g)
    cli.vcpu_cli_options(g, editexample=True)
    cli.add_guest_xml_options(g)
    cli.add_boot_options(g)
    cli.add_device_options(g)

    misc = parser.add_argument_group(_("Miscellaneous Options"))
    cli.add_misc_options(misc, prompt=False, printxml=False, dryrun=False)

    return parser.parse_args()


###################
# main() handling #
###################

def main(conn=None):
    cli.earlyLogging()
    options = parse_args()

    if (options.confirm or options.print_xml or
        options.print_diff or options.build_xml):
        options.quiet = False
    cli.setupLogging("virt-xml", options.debug, options.quiet)

    if cli.check_option_introspection(options):
        return 0

    options.stdinxml = None
    if not options.domain and not options.build_xml:
        if not sys.stdin.closed and not sys.stdin.isatty():
            if options.confirm:
                fail(_("Can't use --confirm with stdin input."))
            if options.update:
                fail(_("Can't use --update with stdin input."))
            options.stdinxml = sys.stdin.read()
        else:
            fail(_("A domain must be specified"))

    if not options.print_xml and not options.print_diff:
        if options.stdinxml:
            if not options.define:
                options.print_xml = True
        else:
            options.define = True
    if options.confirm and not options.print_xml:
        options.print_diff = True

    if conn is None:
        conn = cli.getConnection(options.connect)

    domain = None
    active_xmlobj = None
    inactive_xmlobj = None
    if options.domain:
        domain, inactive_xmlobj, active_xmlobj = get_domain_and_guest(
            conn, options.domain)
    elif not options.build_xml:
        inactive_xmlobj = virtinst.Guest(conn, options.stdinxml)

    check_action_collision(options)
    parserclass = check_xmlopt_collision(options)

    if options.update and not parserclass.propname:
        fail(_("Don't know how to --update for --%s") %
             (parserclass.cli_arg_name))

    if options.build_xml:
        devs = action_build_xml(conn, options, parserclass)
        for dev in devs:
            # pylint: disable=no-member
            print_stdout(dev.get_xml_config())
        return 0

    if options.update:
        if active_xmlobj:
            devs, action = prepare_changes(active_xmlobj, options, parserclass)
            update_changes(domain, devs, action, options.confirm)
        else:
            logging.warning(
                _("The VM is not running, --update is inapplicable."))
    if options.define:
        devs, action = prepare_changes(inactive_xmlobj, options, parserclass)
        applied = define_changes(conn, inactive_xmlobj,
                                 devs, action, options.confirm)
        if not options.update and active_xmlobj and applied:
            print_stdout(
                _("Changes will take effect after the domain is fully powered off."))
    if not options.update and not options.define:
        prepare_changes(inactive_xmlobj, options, parserclass)

    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except SystemExit as sys_e:
        sys.exit(sys_e.code)
    except KeyboardInterrupt:
        logging.debug("", exc_info=True)
        print_stderr(_("Aborted at user request"))
    except Exception as main_e:
        fail(main_e)