advancelibvirt vm clock

vm_clock_jump.py

#!/usr/bin/env python3
import argparse
import functools
from xml.etree import ElementTree

import arrow
import libvirt

ONE_YEAR = 365 * 24 * 60 * 60
CERT_REFRESH = ONE_YEAR / 4
TWENTY_MINS = 20 * 60


def time_offset(time_expression: str) -> int:
    now = arrow.utcnow()
    future_time = now.dehumanize(time_expression)
    delta = future_time - now
    return int(delta.total_seconds())


def seconds_from_timestamp(timestamp: str, time_expression='') -> int:
    # dt = datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S %Z")
    if timestamp:
        stamp = arrow.get(timestamp)
    else:
        stamp = arrow.utcnow()
    future_time = stamp
    if time_expression:
        future_time = stamp.dehumanize(time_expression)
    time_difference = future_time - arrow.utcnow()
    return int(time_difference.total_seconds())


def update_dom_clock(conn, dom, new_adjustment: str):
    # libvirt.libvirtError: argument unsupported: QEMU guest agent is not configured
    # print(dom.getTime())
    # Get the XML description of the domain
    domain_xml = dom.XMLDesc(0)
    # Parse the XML using ElementTree
    root = ElementTree.fromstring(domain_xml)
    # Find the 'clock' element
    clock = root.find('./clock')
    if clock is None:
        raise RuntimeError('Failed to find the clock element')
    # Set the new adjustment attribute of the 'clock' element
    # use whole seconds
    clock.set('adjustment', "%d" % new_adjustment)
    # Generate the new domain XML
    new_domain_xml = ElementTree.tostring(root, encoding='unicode')
    # Redefine the domain with the new XML
    print(new_domain_xml)
    if conn.defineXML(new_domain_xml) is None:
        raise RuntimeError('Failed to update the domain XML')
    print('Successfully updated the domain XML')


def main():
    parser = argparse.ArgumentParser(description='VM Clock Jump')
    parser.add_argument('domain', type=str, help='The domain for the VM')
    parser.add_argument('time_offset', type=str, help='The time offset to jump')
    parser.add_argument(
        '--from', dest='timestamp', type=str, default=None, help='The base timestamp'
    )
    parser.add_argument(
        "--dryrun", "-d", action='store_true', default=False, help='Just '
                                                                   'print '
                                                                   'the XML'
    )

    args = parser.parse_args()

    # print the arguments to verify they're working correctly
    print('Domain:', args.domain)
    print('Time Offset:', args.time_offset)

    # Open connection to libvirt
    conn = libvirt.open('qemu:///system')
    if conn is None:
        raise RuntimeError('Failed to open connection to qemu:///system')

    # Lookup the domain
    doms = conn.listAllDomains()
    print(doms)
    matched = [d for d in doms if args.domain in d.name()]
    time_func = time_offset
    if args.timestamp:
        time_func = functools.partial(seconds_from_timestamp, args.timestamp)

    for dom in matched:
        time_adjustment = time_func(args.time_offset)

        if args.dryrun:
            print(dom)
            print("<clock offset='variable' adjustment='%d'/>" % time_adjustment)
        else:
            update_dom_clock(conn, dom, time_adjustment)


if __name__ == "__main__":
    main()

"""
# arg 1: domain (string),  arg 2: time offset (string)
vm_clock_jump.py master "in 1 hours"
vm click jump.py --from="2023-12-05T08:54:01Z" master "in 1 months"
vm_clock_jump.py  --from "2023-07-31T18:28:39Z" master "in 131380 minutes"

"""